xref: /aosp_15_r20/external/autotest/client/bin/setup_job.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7)

# Lint as: python2, python3
# Copyright 2007 Google Inc. Released under the GPL v2
#
# Eric Li <[email protected]>

import logging, os, pickle, re, sys
import common

from autotest_lib.client.bin import job as client_job
from autotest_lib.client.common_lib import base_job
from autotest_lib.client.common_lib import error
from autotest_lib.client.common_lib import logging_manager
from autotest_lib.client.common_lib import packages


class setup_job(client_job.job):
    """
    setup_job is a job which runs client test setup() method at server side.

    This job is used to pre-setup client tests when development toolchain is not
    available at client.
    """

    def __init__(self, options):
        """
        Since setup_job is a client job but run on a server, it takes no control
        file as input. So client_job.__init__ is by-passed.

        @param options: an object passed in from command line OptionParser.
                        See all options defined on client/bin/autotest.
        """
        base_job.base_job.__init__(self, options=options)
        self._cleanup_debugdir_files()
        self._cleanup_results_dir()
        self.machine_dict_list = [{'hostname' : options.hostname}]
        # Client side tests should always run the same whether or not they are
        # running in the lab.
        self.in_lab = False
        self.pkgmgr = packages.PackageManager(
            self.autodir, run_function_dargs={'timeout':3600})


def init_test(options, testdir):
    """
    Instantiate a client test object from a given test directory.

    @param options Command line options passed in to instantiate a setup_job
                   which associates with this test.
    @param testdir The test directory.
    @returns A test object or None if failed to instantiate.
    """

    locals_dict = locals().copy()
    globals_dict = globals().copy()

    locals_dict['testdir'] = testdir

    job = setup_job(options=options)
    locals_dict['job'] = job

    test_name = os.path.split(testdir)[-1]
    outputdir = os.path.join(job.resultdir, test_name)
    try:
        os.makedirs(outputdir)
    except OSError:
        pass
    locals_dict['outputdir'] = outputdir

    sys.path.insert(0, testdir)
    client_test = None
    try:
        try:
            import_stmt = 'import %s' % test_name
            init_stmt = ('auto_test = %s.%s(job, testdir, outputdir)' %
                         (test_name, test_name))
            exec(import_stmt + '\n' + init_stmt, locals_dict, globals_dict)
            client_test = globals_dict['auto_test']
        except ImportError as e:
            # skips error if test is control file without python test
            if re.search(test_name, str(e)):
                pass
            # give the user a warning if there is an import error.
            else:
                logging.exception('%s import error: %s.  Skipping %s' %
                              (test_name, e, test_name))
        except Exception as e:
            # Log other errors (e.g., syntax errors) and collect the test.
            logging.exception("%s: %s", test_name, e)
    finally:
        sys.path.pop(0) # pop up testbindir
    return client_test


def load_all_client_tests(options):
    """
    Load and instantiate all client tests.

    This function is inspired from runtest() on client/common_lib/test.py.

    @param options: an object passed in from command line OptionParser.
                    See all options defined on client/bin/autotest.

    @return a tuple containing the list of all instantiated tests and
            a list of tests that failed to instantiate.
    """

    local_namespace = locals().copy()
    global_namespace = globals().copy()

    all_tests = []
    broken_tests = []
    for test_base_dir in ['tests', 'site_tests']:
        testdir = os.path.join(os.environ['AUTODIR'], test_base_dir)
        for test_name in sorted(os.listdir(testdir)):
            client_test = init_test(options, os.path.join(testdir, test_name))
            if client_test:
                all_tests.append(client_test)
            else:
                broken_tests.append(test_name)
    return all_tests, broken_tests


def setup_test(client_test):
    """
    Direct invoke test.setup() method.

    @returns A boolean to represent success or not.
    """

    # TODO: check if its already build. .version? hash?
    test_name = client_test.__class__.__name__
    cwd = os.getcwd()
    good_setup = False
    try:
        try:
            outputdir = os.path.join(client_test.job.resultdir, test_name)
            try:
                os.makedirs(outputdir)
                os.chdir(outputdir)
            except OSError:
                pass
            logging.info('setup %s.' % test_name)
            client_test.setup()

            # Touch .version file under src to prevent further setup on client
            # host. See client/common_lib/utils.py update_version()
            if os.path.exists(client_test.srcdir):
                versionfile = os.path.join(client_test.srcdir, '.version')
                pickle.dump(client_test.version, open(versionfile, 'wb'))
            good_setup = True
        except Exception as err:
            logging.error(err)
            raise error.AutoservError('Failed to build client test %s on '
                                      'server.' % test_name)
    finally:
        # back to original working dir
        os.chdir(cwd)
    return good_setup


def setup_tests(options):
    """
    Load and instantiate all client tests.

    This function is inspired from runtest() on client/common_lib/test.py.

    @param options: an object passed in from command line OptionParser.
                    See all options defined on client/bin/autotest.
    """

    assert options.client_test_setup, 'Specify prebuild client tests on the ' \
                                      'command line.'

    requested_tests = options.client_test_setup.split(',')
    candidates, broken_tests = load_all_client_tests(options)

    failed_tests = []
    if 'all' in requested_tests:
        need_to_setup = candidates
        failed_tests += broken_tests
    else:
        need_to_setup = []
        for candidate in candidates:
            if candidate.__class__.__name__ in requested_tests:
                need_to_setup.append(candidate)
        for broken_test in broken_tests:
            if broken_test in requested_tests:
                failed_tests.append(broken_test)

    if need_to_setup:
        cwd = os.getcwd()
        os.chdir(need_to_setup[0].job.clientdir)
        os.system('tools/make_clean')
        os.chdir(cwd)
    elif not failed_tests:
        logging.error('### No test setup candidates ###')
        raise error.AutoservError('No test setup candidates.')

    for client_test in need_to_setup:
        good_setup = setup_test(client_test)
        if not good_setup:
            failed_tests.append(client_test.__class__.__name__)

    logging.info('############################# SUMMARY '
                 '#############################')

    # Print out tests that failed
    if failed_tests:
        logging.info('Finished setup -- The following tests failed')
        for failed_test in failed_tests:
            logging.info(failed_test)
    else:
        logging.info('Finished setup -- All tests built successfully')
    logging.info('######################### END SUMMARY '
                 '##############################')
    if failed_tests:
        raise error.AutoservError('Finished setup with errors.')