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

How to use run_autoserv method in autotest

Best Python code snippet using autotest_python

autoserv.py

Source:autoserv.py

...38from autotest.server import server_logging_config39from autotest.server import server_job, autoserv_parser40from autotest.server import autotest_remote41from autotest.client.shared import pidfile, logging_manager42def run_autoserv(pid_file_manager, results, parser):43    # send stdin to /dev/null44    dev_null = os.open(os.devnull, os.O_RDONLY)45    os.dup2(dev_null, sys.stdin.fileno())46    os.close(dev_null)47    # Create separate process group48    os.setpgrp()49    # Implement SIGTERM handler50    def handle_sigterm(signum, frame):51        if pid_file_manager:52            pid_file_manager.close_file(1, signal.SIGTERM)53        os.killpg(os.getpgrp(), signal.SIGKILL)54    # Set signal handler55    signal.signal(signal.SIGTERM, handle_sigterm)56    # Ignore SIGTTOU's generated by output from forked children.57    signal.signal(signal.SIGTTOU, signal.SIG_IGN)58    # Server side tests that call shell scripts often depend on $USER being set59    # but depending on how you launch your autotest scheduler it may not be set.60    os.environ['USER'] = getpass.getuser()61    if parser.options.machines:62        machines = parser.options.machines.replace(',', ' ').strip().split()63    else:64        machines = []65    machines_file = parser.options.machines_file66    label = parser.options.label67    group_name = parser.options.group_name68    user = parser.options.user69    client = parser.options.client70    server = parser.options.server71    install_before = parser.options.install_before72    install_after = parser.options.install_after73    verify = parser.options.verify74    repair = parser.options.repair75    cleanup = parser.options.cleanup76    no_tee = parser.options.no_tee77    parse_job = parser.options.parse_job78    execution_tag = parser.options.execution_tag79    if not execution_tag:80        execution_tag = parse_job81    host_protection = parser.options.host_protection82    ssh_user = parser.options.ssh_user83    ssh_port = parser.options.ssh_port84    ssh_pass = parser.options.ssh_pass85    collect_crashinfo = parser.options.collect_crashinfo86    control_filename = parser.options.control_filename87    # can't be both a client and a server side test88    if client and server:89        parser.parser.error("Can not specify a test as both server and client!")90    if len(parser.args) < 1 and not (verify or repair or cleanup or91                                     collect_crashinfo):92        parser.parser.error("Missing argument: control file")93    # We have a control file unless it's just a verify/repair/cleanup job94    if len(parser.args) > 0:95        control = parser.args[0]96    else:97        control = None98    if machines_file:99        machines = []100        for m in open(machines_file, 'r').readlines():101            # remove comments, spaces102            m = re.sub('#.*', '', m).strip()103            if m:104                machines.append(m)105        print "Read list of machines from file: %s" % machines_file106        print ','.join(machines)107    if machines:108        for machine in machines:109            if not machine or re.search('\s', machine):110                parser.parser.error("Invalid machine: %s" % str(machine))111        machines = list(set(machines))112        machines.sort()113    if group_name and len(machines) < 2:114        parser.parser.error("-G %r may only be supplied with more than one machine."115                            % group_name)116    kwargs = {'group_name': group_name, 'tag': execution_tag}117    if control_filename:118        kwargs['control_filename'] = control_filename119    job = server_job.server_job(control, parser.args[1:], results, label,120                                user, machines, client, parse_job,121                                ssh_user, ssh_port, ssh_pass, **kwargs)122    job.logging.start_logging()123    job.init_parser()124    # perform checks125    job.precheck()126    # run the job127    exit_code = 0128    try:129        try:130            if repair:131                job.repair(host_protection)132            elif verify:133                job.verify()134            else:135                job.run(cleanup, install_before, install_after,136                        only_collect_crashinfo=collect_crashinfo)137        finally:138            while job.hosts:139                host = job.hosts.pop()140                host.close()141    except Exception:142        exit_code = 1143        traceback.print_exc()144    if pid_file_manager:145        pid_file_manager.num_tests_failed = job.num_tests_failed146        pid_file_manager.close_file(exit_code)147    job.cleanup_parser()148    sys.exit(exit_code)149def main():150    # grab the parser151    parser = autoserv_parser.autoserv_parser152    parser.parse_args()153    if len(sys.argv) == 1:154        parser.parser.print_help()155        sys.exit(1)156    if parser.options.no_logging:157        results = None158    else:159        output_dir = settings.get_value('COMMON', 'test_output_dir', default="")160        results = parser.options.results161        if not results:162            results = 'results.' + time.strftime('%Y-%m-%d-%H.%M.%S')163            if output_dir:164                results = os.path.join(output_dir, results)165        results = os.path.abspath(results)166        resultdir_exists = False167        for filename in ('control.srv', 'status.log', '.autoserv_execute'):168            if os.path.exists(os.path.join(results, filename)):169                resultdir_exists = True170        if not parser.options.use_existing_results and resultdir_exists:171            error = "Error: results directory already exists: %s\n" % results172            sys.stderr.write(error)173            sys.exit(1)174        # Now that we certified that there's no leftover results dir from175        # previous jobs, lets create the result dir since the logging system176        # needs to create the log file in there.177        if not os.path.isdir(results):178            os.makedirs(results)179    logging_manager.configure_logging(180        server_logging_config.ServerLoggingConfig(), results_dir=results,181        use_console=not parser.options.no_tee,182        verbose=parser.options.verbose,183        no_console_prefix=parser.options.no_console_prefix)184    if results:185        logging.info("Results placed in %s" % results)186        # wait until now to perform this check, so it get properly logged187        if parser.options.use_existing_results and not resultdir_exists:188            logging.error("No existing results directory found: %s", results)189            sys.exit(1)190    if parser.options.write_pidfile:191        pid_file_manager = pidfile.PidFileManager(parser.options.pidfile_label,192                                                  results)193        pid_file_manager.open_file()194    else:195        pid_file_manager = None196    autotest_remote.BaseAutotest.set_install_in_tmpdir(197        parser.options.install_in_tmpdir)198    exit_code = 0199    try:200        try:201            run_autoserv(pid_file_manager, results, parser)202        except SystemExit as e:203            exit_code = e.code204        except Exception:205            traceback.print_exc()206            # If we don't know what happened, we'll classify it as207            # an 'abort' and return 1.208            exit_code = 1209    finally:210        if pid_file_manager:211            pid_file_manager.close_file(exit_code)...

autoserv

Source:autoserv

...8import common9from autotest_lib.server import server_logging_config10from autotest_lib.server import server_job, utils, autoserv_parser, autotest11from autotest_lib.client.common_lib import pidfile, logging_manager12def run_autoserv(pid_file_manager, results, parser):13    # send stdin to /dev/null14    dev_null = os.open(os.devnull, os.O_RDONLY)15    os.dup2(dev_null, sys.stdin.fileno())16    os.close(dev_null)17    # Create separate process group18    os.setpgrp()19    # Implement SIGTERM handler20    def handle_sigterm(signum, frame):21        if pid_file_manager:22            pid_file_manager.close_file(1, signal.SIGTERM)23        os.killpg(os.getpgrp(), signal.SIGKILL)24    # Set signal handler25    signal.signal(signal.SIGTERM, handle_sigterm)26    if parser.options.machines:27        machines = parser.options.machines.replace(',', ' ').strip().split()28    else:29        machines = []30    machines_file = parser.options.machines_file31    label = parser.options.label32    group_name = parser.options.group_name33    user = parser.options.user34    client = parser.options.client35    server = parser.options.server36    install_before = parser.options.install_before37    install_after = parser.options.install_after38    verify = parser.options.verify39    repair = parser.options.repair40    cleanup = parser.options.cleanup41    no_tee = parser.options.no_tee42    parse_job = parser.options.parse_job43    execution_tag = parser.options.execution_tag44    if not execution_tag:45        execution_tag = parse_job46    host_protection = parser.options.host_protection47    ssh_user = parser.options.ssh_user48    ssh_port = parser.options.ssh_port49    ssh_pass = parser.options.ssh_pass50    collect_crashinfo = parser.options.collect_crashinfo51    # can't be both a client and a server side test52    if client and server:53        print "Can not specify a test as both server and client!"54        sys.exit(1)55    if len(parser.args) < 1 and not (verify or repair or cleanup56                                     or collect_crashinfo):57        print parser.parser.print_help()58        sys.exit(1)59    # We have a control file unless it's just a verify/repair/cleanup job60    if len(parser.args) > 0:61        control = parser.args[0]62    else:63        control = None64    if machines_file:65        machines = []66        for m in open(machines_file, 'r').readlines():67            # remove comments, spaces68            m = re.sub('#.*', '', m).strip()69            if m:70                machines.append(m)71        print "Read list of machines from file: %s" % machines_file72        print ','.join(machines)73    if machines:74        for machine in machines:75            if not machine or re.search('\s', machine):76                print "Invalid machine %s" % str(machine)77                sys.exit(1)78        machines = list(set(machines))79        machines.sort()80    if group_name and len(machines) < 2:81        print ("-G %r may only be supplied with more than one machine."82               % group_name)83        sys.exit(1)84    job = server_job.server_job(control, parser.args[1:], results, label,85                                user, machines, client, parse_job,86                                ssh_user, ssh_port, ssh_pass,87                                group_name=group_name, tag=execution_tag)88    job.logging.start_logging()89    # perform checks90    job.precheck()91    # run the job92    exit_code = 093    try:94        try:95            if repair:96                job.repair(host_protection)97            elif verify:98                job.verify()99            else:100                job.run(cleanup, install_before, install_after,101                        only_collect_crashinfo=collect_crashinfo)102        finally:103            while job.hosts:104                host = job.hosts.pop()105                host.close()106    except:107        exit_code = 1108        traceback.print_exc()109    if pid_file_manager:110        pid_file_manager.num_tests_failed = job.num_tests_failed111        pid_file_manager.close_file(exit_code)112    job.cleanup_parser()113    sys.exit(exit_code)114def main():115    # grab the parser116    parser = autoserv_parser.autoserv_parser117    parser.parse_args()118    if len(sys.argv) == 1:119        parser.parser.print_help()120        sys.exit(1)121    if parser.options.no_logging:122        results = None123    else:124        results = parser.options.results125        if not results:126            results = 'results.' + time.strftime('%Y-%m-%d-%H.%M.%S')127        results  = os.path.abspath(results)128        resultdir_exists = os.path.exists(os.path.join(results, 'control.srv'))129        if not parser.options.collect_crashinfo and resultdir_exists:130            error = "Error: results directory already exists: %s\n" % results131            sys.stderr.write(error)132            sys.exit(1)133        # Now that we certified that there's no leftover results dir from134        # previous jobs, lets create the result dir since the logging system135        # needs to create the log file in there.136        if not os.path.isdir(results):137            os.makedirs(results)138    logging_manager.configure_logging(139            server_logging_config.ServerLoggingConfig(), results_dir=results,140            use_console=not parser.options.no_tee,141            verbose=parser.options.verbose,142            no_console_prefix=parser.options.no_console_prefix)143    if results:144        logging.info("Results placed in %s" % results)145    if parser.options.write_pidfile:146        if parser.options.collect_crashinfo:147            pidfile_label = 'collect_crashinfo'148        else:149            pidfile_label = 'autoserv'150        pid_file_manager = pidfile.PidFileManager(pidfile_label, results)151        pid_file_manager.open_file()152    else:153        pid_file_manager = None154    autotest.BaseAutotest.set_install_in_tmpdir(155        parser.options.install_in_tmpdir)156    exit_code = 0157    try:158        try:159            run_autoserv(pid_file_manager, results, parser)160        except SystemExit, e:161            exit_code = e.code162        except:163            traceback.print_exc()164            # If we don't know what happened, we'll classify it as165            # an 'abort' and return 1.166            exit_code = 1167    finally:168        if pid_file_manager:169            pid_file_manager.close_file(exit_code)170    sys.exit(exit_code)171if __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.