frontend/afe/doctests/003_misc_rpc_features.txt

# this doctest contains tests for miscellaneous features of the RPC interface
# that would clutter the main rpc_test

# setup
>>> from autotest_lib.frontend.afe import rpc_interface

>>> rpc_interface.add_profiler(name='oprofile')
1
>>> rpc_interface.add_profiler(name='iostat')
2

# profiler support in control file generation
>>> cf_info = rpc_interface.generate_control_file(
...     tests=['sleeptest'],
...     profilers=['oprofile', 'iostat'])
>>> print cf_info['control_file']
def step_init():
    job.next_step('step0')
    job.next_step('step1')
    job.next_step('step2')
    job.next_step('step3')
    job.next_step('step4')
<BLANKLINE>
def step0():
    job.profilers.add('oprofile')
<BLANKLINE>
def step1():
    job.profilers.add('iostat')
<BLANKLINE>
def step2():
    job.run_test('testname')
<BLANKLINE>
def step3():
    job.profilers.delete('oprofile')
<BLANKLINE>
def step4():
    job.profilers.delete('iostat')

# profile_only=False support in control file generation
>>> cf_info = rpc_interface.generate_control_file(
...     tests=['sleeptest'],
...     profilers=['oprofile'],
...     profile_only=False)
>>> print cf_info['control_file']
def step_init():
    job.next_step('step0')
    job.next_step('step1')
    job.next_step('step2')
    job.next_step('step3')
<BLANKLINE>
def step0():
    job.default_profile_only = False
<BLANKLINE>
def step1():
    job.profilers.add('oprofile')
<BLANKLINE>
def step2():
    job.run_test('testname')
<BLANKLINE>
def step3():
    job.profilers.delete('oprofile')

# profile_only=True support in control file generation
>>> cf_info = rpc_interface.generate_control_file(
...     tests=['sleeptest'],
...     profilers=['iostat'],
...     profile_only=True)
>>> print cf_info['control_file']
def step_init():
    job.next_step('step0')
    job.next_step('step1')
    job.next_step('step2')
    job.next_step('step3')
<BLANKLINE>
def step0():
    job.default_profile_only = True
<BLANKLINE>
def step1():
    job.profilers.add('iostat')
<BLANKLINE>
def step2():
    job.run_test('testname')
<BLANKLINE>
def step3():
    job.profilers.delete('iostat')

# test preserving of the kernel config file setting
>>> cf_info = rpc_interface.generate_control_file(
...     tests=['sleeptest'],
...     kernel=[{'version': '2.6.18', 'config_file': 'foo/bar'}])
>>> print cf_info['control_file'] #doctest: +NORMALIZE_WHITESPACE
kernel_list = [{'version': '2.6.18', 'config_file': 'foo/bar'}]
def step_init():
    for kernel_info in kernel_list:
        job.next_step(boot_kernel, kernel_info)
        job.next_step(step_test, kernel_info['version'])
    if len(kernel_list) > 1:
        job.use_sequence_number = True  # include run numbers in directory names
def boot_kernel(kernel_info):
    # remove kernels (and associated data) not referenced by the bootloader
    for host in job.hosts:
        host.cleanup_kernels()
    testkernel = job.kernel(kernel_info['version'])
    if kernel_info['config_file']:
        testkernel.config(kernel_info['config_file'])
    testkernel.build()
    testkernel.install()
    cmdline = ' '.join((kernel_info.get('cmdline', ''), ''))
    testkernel.boot(args=cmdline)
def step_test(kernel_version):
    global kernel
    kernel = kernel_version  # Set the global in case anyone is using it.
    if len(kernel_list) > 1:
        # this is local to a machine, safe to assume there's only one host
        host, = job.hosts
        job.automatic_test_tag = host.get_kernel_ver()
    job.next_step('step0')
def step0():
    job.run_test('testname')

# upload_kernel_config requires server side test
>>> rpc_interface.generate_control_file(tests=['sleeptest'],
...        kernel=[{'version': '2.6.18', 'config_file': 'foo/bar'}],
...        upload_kernel_config=True)
Traceback (most recent call last):
ValidationError: {'upload_kernel_config': 'Cannot use upload_kernel_config with client side tests'}

# server-side control file generation
>>> rpc_interface.modify_test('sleeptest', test_type='Server')
>>> cf_info = rpc_interface.generate_control_file(tests=['sleeptest'],
...        kernel=[{'version': '2.6.18'}, {'version': '2.6.22'}])
>>> print cf_info['control_file'] #doctest: +NORMALIZE_WHITESPACE
kernel_list = [{'version': '2.6.18', 'config_file': None}, {'version': '2.6.22', 'config_file': None}]
kernel_install_control = """
kernel_list = %(client_kernel_list)s
def step_init():
    for kernel_info in kernel_list:
        job.next_step(boot_kernel, kernel_info)
        job.next_step(step_test, kernel_info['version'])
    if len(kernel_list) > 1:
        job.use_sequence_number = True  # include run numbers in directory names
def boot_kernel(kernel_info):
    # remove kernels (and associated data) not referenced by the bootloader
    for host in job.hosts:
        host.cleanup_kernels()
    testkernel = job.kernel(kernel_info['version'])
    if kernel_info['config_file']:
        testkernel.config(kernel_info['config_file'])
    testkernel.build()
    testkernel.install()
    cmdline = ' '.join((kernel_info.get('cmdline', ''), ''))
    testkernel.boot(args=cmdline)
def step_test(kernel_version):
    global kernel
    kernel = kernel_version  # Set the global in case anyone is using it.
    if len(kernel_list) > 1:
        # this is local to a machine, safe to assume there's only one host
        host, = job.hosts
        job.automatic_test_tag = host.get_kernel_ver()
    pass
"""
from autotest_lib.client.common_lib import error
at = autotest.Autotest()
def install_kernel(machine, kernel_info):
    host = hosts.create_host(machine)
    at.install(host=host)
    at.run(kernel_install_control %
           {'client_kernel_list': repr([kernel_info])}, host=host)
num_machines_required = len(machines)
if len(machines) > 4:
    # Allow a large multi-host tests to proceed despite a couple of hosts
    # failing to properly install the desired kernel (exclude those hosts).
    # TODO(gps): Figure out how to get and use SYNC_COUNT here.  It is defined
    # within some control files and will end up inside of stepN functions below.
    num_machines_required = len(machines) - 2
def step_init():
    # a host object we use solely for the purpose of finding out the booted
    # kernel version, we use machines[0] since we already check that the same
    # kernel has been booted on all machines
    if len(kernel_list) > 1:
        kernel_host = hosts.create_host(machines[0])
    for kernel_info in kernel_list:
        func = lambda machine: install_kernel(machine, kernel_info)
        good_machines = job.parallel_on_machines(func, machines)
        if len(good_machines) < num_machines_required:
            raise error.TestError(
                    "kernel installed on only %d of %d machines."
                    % (len(good_machines), num_machines_required))
        # Replace the machines list that step_test() will use with the
        # ones that successfully installed the kernel.
        machines[:] = good_machines
        # have server_job.run_test() automatically add the kernel version as
        # a suffix to the test name otherwise we cannot run the same test on
        # different kernel versions
        if len(kernel_list) > 1:
            job.automatic_test_tag = kernel_host.get_kernel_ver()
        step_test()
def step_test():
    step0()
def step0():
    job.run_test('testname')
step_init()
>>> cf_info['is_server']
True

# server-side control file generation with kernel config upload code
>>> cf_info = rpc_interface.generate_control_file(tests=['sleeptest'],
...        kernel=[{'version': '2.6.18', 'config_file': 'foo/bar'}],
...        upload_kernel_config=True)
>>> print cf_info['control_file'] #doctest: +NORMALIZE_WHITESPACE
kernel_list = [{'version': '2.6.18', 'config_file': 'foo/bar'}]
kernel_install_control = """
kernel_list = %(client_kernel_list)s
def step_init():
    for kernel_info in kernel_list:
        job.next_step(boot_kernel, kernel_info)
        job.next_step(step_test, kernel_info['version'])
    if len(kernel_list) > 1:
        job.use_sequence_number = True  # include run numbers in directory names
def boot_kernel(kernel_info):
    # remove kernels (and associated data) not referenced by the bootloader
    for host in job.hosts:
        host.cleanup_kernels()
    testkernel = job.kernel(kernel_info['version'])
    if kernel_info['config_file']:
        testkernel.config(kernel_info['config_file'])
    testkernel.build()
    testkernel.install()
    cmdline = ' '.join((kernel_info.get('cmdline', ''), ''))
    testkernel.boot(args=cmdline)
def step_test(kernel_version):
    global kernel
    kernel = kernel_version  # Set the global in case anyone is using it.
    if len(kernel_list) > 1:
        # this is local to a machine, safe to assume there's only one host
        host, = job.hosts
        job.automatic_test_tag = host.get_kernel_ver()
    pass
"""
from autotest_lib.client.common_lib import error
at = autotest.Autotest()
def upload_kernel_config(host, kernel_info):
    """
    If the kernel_info['config_file'] is a URL it will be downloaded
    locally and then uploaded to the client and a copy of the original
    dictionary with the new path to the config file will be returned.
    If the config file is not a URL the function returns the original
    dictionary.
    """
    import os
    from autotest_lib.client.common_lib import autotemp, utils
    config_orig = kernel_info.get('config_file')
    # if the file is not an URL then we assume it's a local client path
    if not config_orig or not utils.is_url(config_orig):
        return kernel_info
    # download it locally (on the server) and send it to the client
    config_tmp = autotemp.tempfile('kernel_config_upload', dir=job.tmpdir)
    try:
        utils.urlretrieve(config_orig, config_tmp.name)
        config_new = os.path.join(host.get_autodir(), 'tmp',
                                  os.path.basename(config_orig))
        host.send_file(config_tmp.name, config_new)
    finally:
        config_tmp.clean()
    return dict(kernel_info, config_file=config_new)
def install_kernel(machine, kernel_info):
    host = hosts.create_host(machine)
    at.install(host=host)
    kernel_info = upload_kernel_config(host, kernel_info)
    at.run(kernel_install_control %
           {'client_kernel_list': repr([kernel_info])}, host=host)
num_machines_required = len(machines)
if len(machines) > 4:
    # Allow a large multi-host tests to proceed despite a couple of hosts
    # failing to properly install the desired kernel (exclude those hosts).
    # TODO(gps): Figure out how to get and use SYNC_COUNT here.  It is defined
    # within some control files and will end up inside of stepN functions below.
    num_machines_required = len(machines) - 2
def step_init():
    # a host object we use solely for the purpose of finding out the booted
    # kernel version, we use machines[0] since we already check that the same
    # kernel has been booted on all machines
    if len(kernel_list) > 1:
        kernel_host = hosts.create_host(machines[0])
    for kernel_info in kernel_list:
        func = lambda machine: install_kernel(machine, kernel_info)
        good_machines = job.parallel_on_machines(func, machines)
        if len(good_machines) < num_machines_required:
            raise error.TestError(
                    "kernel installed on only %d of %d machines."
                    % (len(good_machines), num_machines_required))
        # Replace the machines list that step_test() will use with the
        # ones that successfully installed the kernel.
        machines[:] = good_machines
        # have server_job.run_test() automatically add the kernel version as
        # a suffix to the test name otherwise we cannot run the same test on
        # different kernel versions
        if len(kernel_list) > 1:
            job.automatic_test_tag = kernel_host.get_kernel_ver()
        step_test()
def step_test():
    step0()
def step0():
    job.run_test('testname')
step_init()

# test that multiline quoted strings are not indented
>>> import common
>>> from autotest_lib.frontend.afe import test, control_file
>>> import os
>>> control_path = os.path.join(os.path.dirname(test.__file__),
...                             'doctests', 'test.control.3')
>>> control_path = os.path.abspath(control_path)
>>> class FakeTest(object):
...   path = control_path
...
>>> print control_file.generate_control([FakeTest()], is_server=True) #doctest: +NORMALIZE_WHITESPACE
def step_init():
    step0()
def step0():
    client_code = """
some content\"""quoted content\"""
'''other quoted content\"""'''
\\"""
    client_code2 = '''
some content\'''quoted content\'''
"""other quoted content\'''"""
\\'''
    job.run_test('testname')
step_init()