blob: 258f34f77cf512fb3c55ed6599140acd7309c743 [file] [log] [blame]
#!/usr/bin/env python
# Copyright 2016 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import logging
import os
import pipes
import unittest
import factory_common # pylint: disable=unused-import
from cros.factory.test import dut as dut_module
from cros.factory.test.args import Arg
from cros.factory.test.args import Args
from cros.factory.test.utils import deploy_utils
from cros.factory.test.utils import pytest_utils
from cros.factory.utils import type_utils
class FunctionMapper(object):
"""A decorator class for registering alias names for functions.
mapper = FunctionMapper()
def func(a, b, c):
mapper.CallFunction('foo', 4, 5, 6)
def __init__(self):
self._FUNCTION_MAP = {}
def Register(self, alias_name):
if alias_name in self._FUNCTION_MAP:
logging.warn('`%s` already registered, overriding old value', alias_name)
def wrapper(func):
self._FUNCTION_MAP[alias_name] = func
return func
return wrapper
def CallFunction(self, alias_name, *args, **kargs):
"""Call a function registered with alias name `alias_name`.
The function registered with name `alias_name` will be called with keyword
arguments `kargs` and bind to `obj`.
alias_name: a str, the registered name of the function.
obj: a object which the function should bind to.
kargs: additional arguments for the function.
return self._FUNCTION_MAP[alias_name](*args, **kargs)
class ScriptBuilder(object):
"""Builder class for shell offline test.
builder = ScriptBuilder(dut.Create())
builder.TestStressApp(0.5, 20, True)
for unused_i in xrange(10):
builder.TestStressApp(0.5, 20, True)
print builder.Build()
ShellTestCase = FunctionMapper()
def __init__(self, dut):
"""Constructor of ScriptBuilder.
:type dut: cros.factory.test.dut.board.DUTBoard
self.tasks = []
self.dut = dut
def cpu_count(self):
return or 1
def memory_total_kb(self):
return or (100 * 1024)
def Clear(self):
self.tasks = []
def _AddTask(self, template_file, **kargs):
next_task_id = len(self.tasks) + 1 # task index is 1 based.
def _FormatTemplate(self, template, *args, **kargs):
"""Formats template file by replacing place holders by arguments.
Place holders in template file look like: {%name%}.
template: file path under py/test/pytests/offline_test/shell/
args, kargs: arguments for str.format.
path = os.path.join(os.path.dirname(os.path.realpath(__file__)), template)
with open(path) as f:
template =
# escape all braces
template = template.replace("{", "{{")
template = template.replace("}", "}}")
# now {%...%} becomes {{%...%}}
template = template.replace("{{%", "{")
template = template.replace("%}}", "}")
return template.format(*args, **kargs)
def Build(self):
"""Generates the final script.
The script will execute each declared tasks in the order they are declared.
tasks = '\n'.join(self.tasks)
return self._FormatTemplate('',,
def AddShellTestCase(self, test_name, **kargs):
self.ShellTestCase.CallFunction(test_name, self, **kargs)
return self
def AddPythonTestCase(self, pytest_name, **kargs):
"""Generates a test case using `factory.par run_pytest`.
Push factory.par to DUT, and the shell script will use factory.par to run
`cros.factory.test.pytests.<pytest_name>` with arguments `kargs`.
The `board_class` of `dut_options` will be the same with `self.dut`, and
`link_class` will be `LocalLink`.
The pytest is not guaranteed to be able to run on DUT, since it might
requires Goofy, or it requires Chrome OS envrionment.
archive = deploy_utils.FactoryPythonArchive(self.dut)
dut_options = {
'board_class': self.dut.__class__.__name__,
'link_class': 'LocalLink'}
cmd = archive.DryRun(['run_pytest', '--args', repr(kargs), '--dut-options',
repr(dut_options), '--no-use-goofy', pytest_name])
cmd = ' '.join(map(pipes.quote, cmd))
self._AddTask('', cmd=cmd)
return self
def WaitFor(self, wait_seconds):
"""Waits for `wait_seconds` seconds."""
self._AddTask('', seconds=wait_seconds)
return self
def TestStressApp(self, memory_ratio, seconds, disk_thread):
"""Generate stressapptest script by formatting `./`."""
if disk_thread:
disk_thread = ('-f "${tmpdir}/sat.disk_thread.a" '
'-f "${tmpdir}/sat.disk_thread.b"')
disk_thread = ''
mem_usage = max(int(self.memory_total_kb * memory_ratio / 1024), 32)
self._AddTask('', cpu_count=self.cpu_count,
mem_usage=mem_usage, seconds=seconds, disk_thread=disk_thread)
return self
def TestShutdown(self):
"""Generate shutdown script by formatting `./`."""
return self
def TestThermalLoad(self):
return self
def TestBadBlocks(self, **kargs):
"""Generates the bad blocks test.
Please refer to `cros.factory.test.pytests.bad_blocks.BadBlocksTest.ARGS`
for the argument list.
module = pytest_utils.LoadPytestModule('bad_blocks')
test = module.BadBlocksTest()
test.dut = self.dut
test.args = Args(*test.ARGS).Parse(kargs)
params = test.DetermineParameters()
is_file=('true' if test.args.mode == 'file' else 'false'),
return self
class OfflineTestError(Exception):
class DeployShellOfflineTest(unittest.TestCase):
ARGS = [
Arg('test_spec', list,
'Please refer to _`py/test/pytests/offline_test/shell/README`.'),
Arg('next_action', NEXT_ACTION,
('What to do after tests are deployed (One of %s)' % NEXT_ACTION)),
Arg('start_up_service', bool, 'Do you want to run the tests on start up?',
default=True, optional=True)]
def setUp(self):
self.dut = dut_module.Create()
self.builder = ScriptBuilder(self.dut)
def runTest(self):
factory_root =
# make sure factory_root is writable
if not
raise OfflineTestError('failed to make dut:%s writable' % factory_root)
for spec in self.args.test_spec:
dargs = spec.get('dargs', {})
if 'shtest_name' in spec:
self.builder.AddShellTestCase(spec['shtest_name'], **dargs)
elif 'pytest_name' in spec:
self.builder.AddPythonTestCase(spec['pytest_name'], **dargs)
raise ValueError('You must specify one of `shtest_name` and '
script_path = self.dut.path.join(factory_root, self.SCRIPT_FILE_NAME)
self.dut.WriteFile(script_path, self.builder.Build())
# TODO(stimim): make the script a start-up application
if self.args.next_action == self.NEXT_ACTION.POWEROFF:
self.dut.Call(['shutdown', 'now'])
elif self.args.next_action == self.NEXT_ACTION.REBOOT:
self.dut.Call(['shutdown', 'reboot'])
self.dut.Call(['sh', script_path])