Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use publish_result method in localstack

Best Python code snippet using localstack_python

validate.py

Source:validate.py

...13    print("pycodestyle is not available!")14def print_stderr(message):15    sys.stderr.write(message)16    sys.stderr.write("\n")17def publish_result(result_message_list, args):18    result_message = '\n'.join(result_message_list)19    try:20        f = open(args.result_file, 'a')21        f.write("\n\n")22        f.write(result_message)23        f.write("\n\n")24        f.close()25    except IOError as e:26        print_stderr("Cannot write to result file: %s" % args.result_file)27    print_stderr(result_message)28def detect_abi():29    # Retrieve the current canonical abi from30    # automated/lib/sh-test-lib:detect_abi31    return subprocess.check_output(32        ". automated/lib/sh-test-lib && detect_abi && echo $abi",33        shell=True).decode('utf-8').strip()34def pycodestyle_check(filepath, args):35    _fmt = "%(row)d:%(col)d: %(code)s %(text)s"36    options = {37        'ignore': args.pycodestyle_ignore,38        "show_source": True}39    pycodestyle_checker = pycodestyle.StyleGuide(options)40    fchecker = pycodestyle_checker.checker_class(41        filepath,42        options=pycodestyle_checker.options)43    fchecker.check_all()44    if fchecker.report.file_errors > 0:45        result_message_list = []46        result_message_list.append("* PYCODESTYLE: [FAILED]: " + filepath)47        fchecker.report.print_statistics()48        for line_number, offset, code, text, doc in fchecker.report._deferred_print:49            result_message_list.append(50                _fmt % {51                    'path': filepath,52                    'row': fchecker.report.line_offset + line_number,53                    'col': offset + 1,54                    'code': code, 'text': text,55                })56        publish_result(result_message_list, args)57        return 158    else:59        message = "* PYCODESTYLE: [PASSED]: " + filepath60        print_stderr(message)61    return 062def validate_yaml_contents(filepath, args):63    def validate_lava_yaml(y, args):64        result_message_list = []65        if 'metadata' not in y.keys():66            result_message_list.append("* METADATA [FAILED]: " + filepath)67            result_message_list.append("\tmetadata section missing")68            publish_result(result_message_list, args)69            exit(1)70        metadata_dict = y['metadata']71        mandatory_keys = set([72            'name',73            'format',74            'description',75            'maintainer',76            'os',77            'devices'])78        if not mandatory_keys.issubset(set(metadata_dict.keys())):79            result_message_list.append("* METADATA [FAILED]: " + filepath)80            result_message_list.append("\tmandatory keys missing: %s" %81                                       mandatory_keys.difference(set(metadata_dict.keys())))82            result_message_list.append("\tactual keys present: %s" %83                                       metadata_dict.keys())84            publish_result(result_message_list, args)85            return 186        for key in mandatory_keys:87            if len(metadata_dict[key]) == 0:88                result_message_list.append("* METADATA [FAILED]: " + filepath)89                result_message_list.append("\t%s has no content" % key)90                publish_result(result_message_list, args)91                return 192        result_message_list.append("* METADATA [PASSED]: " + filepath)93        publish_result(result_message_list, args)94        return 095    def validate_skipgen_yaml(filepath, args):96        abi = detect_abi()97        # Run skipgen on skipgen yaml file to check for output and errors98        skips = subprocess.check_output(99            "automated/bin/{}/skipgen {}".format(abi, filepath),100            shell=True).decode('utf-8').strip()101        if len(skips.split('\n')) < 1:102            publish_result(["* SKIPGEN [FAILED]: " + filepath + " - No skips found"], args)103            return 1104        publish_result(["* SKIPGEN [PASSED]: " + filepath], args)105        return 0106    filecontent = None107    try:108        with open(filepath, "r") as f:109            filecontent = f.read()110    except FileNotFoundError:111        publish_result(["* YAMLVALIDCONTENTS [PASSED]: " + filepath + " - deleted"], args)112        return 0113    y = yaml.load(filecontent)114    if 'metadata' in y.keys():115        # lava yaml file116        return validate_lava_yaml(y, args)117    elif 'skiplist' in y.keys():118        # skipgen yaml file119        return validate_skipgen_yaml(filepath, args)120    else:121        publish_result(["* YAMLVALIDCONTENTS [FAILED]: " + filepath + " - Unknown yaml type detected"], args)122        return 1123def validate_yaml(filename, args):124    filecontent = None125    try:126        with open(filename, "r") as f:127            filecontent = f.read()128    except FileNotFoundError:129        publish_result(["* YAMLVALID [PASSED]: " + filename + " - deleted"], args)130        return 0131    try:132        yaml.load(filecontent)133        message = "* YAMLVALID: [PASSED]: " + filename134        print_stderr(message)135    except yaml.YAMLError:136        message = "* YAMLVALID: [FAILED]: " + filename137        result_message_list = []138        result_message_list.append(message)139        result_message_list.append("\n\n")140        exc_type, exc_value, exc_traceback = sys.exc_info()141        for line in traceback.format_exception_only(exc_type, exc_value):142            result_message_list.append(' ' + line)143        publish_result(result_message_list, args)144        return 1145    return 0146def validate_shell(filename, ignore_options):147    ignore_string = ""148    if args.shellcheck_ignore is not None:149        # Exclude types of warnings in the following format:150        # -e CODE1,CODE2..151        ignore_string = "-e %s" % ",".join(args.shellcheck_ignore)152    if len(ignore_string) < 4:  # contains only "-e "153        ignore_string = ""154    cmd = 'shellcheck %s' % ignore_string155    return validate_external(cmd, filename, "SHELLCHECK", args)156def validate_php(filename, args):157    cmd = 'php -l'158    return validate_external(cmd, filename, "PHPLINT", args)159def validate_external(cmd, filename, prefix, args):160    final_cmd = "%s %s 2>&1" % (cmd, filename)161    status, output = subprocess.getstatusoutput(final_cmd)162    if status == 0:163        message = '* %s: [PASSED]: %s' % (prefix, filename)164        print_stderr(message)165    else:166        result_message_list = []167        result_message_list.append('* %s: [FAILED]: %s' % (prefix, filename))168        result_message_list.append('* %s: [OUTPUT]:' % prefix)169        for line in output.splitlines():170            result_message_list.append(' ' + line)171        publish_result(result_message_list, args)172        return 1173    return 0174def validate_file(args, path):175    exitcode = 0176    if path.endswith(".yaml"):177        exitcode = validate_yaml(path, args)178        if exitcode == 0:179            # if yaml isn't valid there is no point in checking metadata180            exitcode = validate_yaml_contents(path, args)181    elif run_pycodestyle and path.endswith(".py"):182        exitcode = pycodestyle_check(path, args)183    elif path.endswith(".php"):184        exitcode = validate_php(path, args)185    elif path.endswith(".sh") or \...

lava-project-ci.py

Source:lava-project-ci.py

...51    except subprocess.CalledProcessError as e:52        message = '* SETUP STEP: [FAILED]: %s' % cmd53        result_message_list.append(message)54        print e.output55        publish_result()56        exit(1)57def notify_committer():58    message_list= []59    message_list.append('* Hello %s' % os.environ['GERRIT_CHANGE_OWNER_NAME'])60    message_list.append('* Your patch set %s has triggered automated testing.' % os.environ['GERRIT_PATCHSET_REVISION'])61    message_list.append('* Please do not merge this commit until after I have reviewed the results with you.')62    message_list.append('* %s' % os.environ['BUILD_URL'])63    message = '\n'.join(message_list)64    if debug:65        print message66    add_gerrit_comment(message, 0)67def publish_result(result=None):68    test_results = os.environ['BUILD_URL'] + 'console'69    result_message_list.append('* TEST RESULTS: %s' % test_results)70    result_message = '\n'.join(result_message_list)71    if result is None:72        add_gerrit_comment(result_message, 0)73    elif result:74        add_gerrit_comment(result_message, +1)75    else:76        add_gerrit_comment(result_message, -1)77def checkout_and_rebase():78    os.chdir(os.environ['WORKSPACE'])79    cmd = 'git clone %s/%s' % (git_server, os.environ['GERRIT_PROJECT'])80    if debug:81        print cmd82    try:83        output = subprocess.check_output(cmd, shell=True)84        if debug:85            print output86        message = '* SETUP STEP: [PASSED]: %s' % cmd87        result_message_list.append(message)88    except subprocess.CalledProcessError as e:89        message = '* SETUP STEP: [FAILED]: %s' % cmd90        result_message_list.append(message)91        publish_result()92        exit(1)93    os.chdir(os.environ['GERRIT_PROJECT'].split('/')[1])94    cmd = 'git fetch ssh://%s/%s %s && git checkout FETCH_HEAD && git rebase master' % (gerrit_server, os.environ['GERRIT_PROJECT'], os.environ['GERRIT_REFSPEC'])95    if debug:96        print cmd97    try:98        output = subprocess.check_output(cmd, shell=True)99        message = '* SETUP STEP: [PASSED]: %s' % cmd100        result_message_list.append(message)101    except subprocess.CalledProcessError as e:102        message = '* SETUP STEP: [FAILED]: %s' % cmd103        result_message_list.append(message)104        print e.output105        publish_result()106        exit(1)107def pep8_check(ignore_options):108    os.chdir(os.environ['WORKSPACE'])109    os.chdir(os.environ['GERRIT_PROJECT'].split('/')[1])110    cmd = 'pep8 --ignore %s .' % ignore_options111    try:112        output = subprocess.check_output(cmd, shell=True)113        if debug:114            print output115        message = '* PEP8 CHECK: [PASSED]: %s' % cmd116        result_message_list.append(message)117    except subprocess.CalledProcessError as e:118        message_list = []119        message_list.append('* PEP8 CHECK: [FAILED]: %s' % cmd)120        message_list.append('* PEP8 CHECK: [OUTPUT]:')121        for line in e.output.splitlines():122            message_list.append('* ' + line)123        message = '\n'.join(message_list)124        result_message_list.append(message)125        cmd_verbose = 'pep8 --ignore %s --show-source --show-pep8 .' % ignore_options126        subprocess.call(cmd_verbose, shell=True)127        publish_result(False)128        exit(1)129def run_unit_tests():130    lava_project = os.environ['GERRIT_PROJECT'].split('/')[1]131    os.chdir(os.environ['WORKSPACE'])132    os.chdir(lava_project)133    if lava_project == 'lava-server':134        cmd = './ci-run'135    elif lava_project == 'lava-dispatcher':136        cmd = './ci-run'137    elif lava_project == 'lava-tool':138        cmd = './ci-build'139    else:140        cmd = './ci-run'141    try:142        output = subprocess.check_output(cmd, shell=True)143        if debug:144            print output145        message = '* UNIT TESTS: [PASSED]: %s' % cmd146        result_message_list.append(message)147    except subprocess.CalledProcessError as e:148        message = '* UNIT TEST: [FAILED]: %s' % cmd149        result_message_list.append(message)150        drop_test_db()151        print e.output152        publish_result(False)153        exit(1)154def drop_test_db():155    lava_project = os.environ['GERRIT_PROJECT'].split('/')[1]156    if lava_project == 'lava-server':157        cmd = "dropdb 'test_lava-lavabot'"158        try:159            output = subprocess.check_output(cmd, shell=True)160            if debug:161                print output162            message = '* TEAR DOWN STEP: [PASSED]: %s' % cmd163            result_message_list.append(message)164        except subprocess.CalledProcessError as e:165            message = '* TEAR DOWN STEP: [FAILED]: %s' % cmd166            result_message_list.append(message)167            print e.output168            publish_result(False)169            exit(1)170def init():171    result_message_list.append('* LAVABOT: [RESULTS]: Automated test results for patch set: %s' % os.environ['GERRIT_PATCHSET_REVISION'])172def main(ignore_options):173    # Uncomment the following to run locally174    #debug = True175    #dummy_env()176    if check_enviroment():177        print 'All required environment variables have been set...continuing'178    else:179        print 'All required environment variables have not been set...exiting'180        exit(1)181    notify_committer()182    init()183    checkout_and_rebase()184    pep8_check(ignore_options)185    run_unit_tests()186    publish_result(True)187    exit(0)188if __name__ == '__main__':...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.