Robot Framework 源码解析（2）- 执行测试的入
接着上一章，我们先来看 src/robot/run.py 的 run_cli方法。
def run_cli(arguments=None, exit=True):
if arguments is None:
arguments = sys.argv[1:]
return RobotFramework().execute_cli(arguments, exit=exit)
方法很简单，只是调用了RobotFramework类中的execute_cli方法。RobotFramework是run.py的一个内部类，也是Application的子类。通过from robot.utils import Application, unic, text可查看Application是做什么的。
src/robot/utils/application.py
摘录部分代码：
class Application(object):
......
def main(self, arguments, **options):
raise NotImplementedError
....
def execute_cli(self, cli_arguments, exit=True):
with self._logger:
self._logger.info('%s %s' % (self._ap.name, self._ap.version))
options, arguments = self._parse_arguments(cli_arguments)
rc = self._execute(arguments, options)
if exit:
self._exit(rc)
return rc
......
def _parse_arguments(self, cli_args):
try:
options, arguments = self.parse_arguments(cli_args)
except Information as msg:
self._report_info(msg.message)
except DataError as err:
self._report_error(err.message, help=True, exit=True)
else:
self._logger.info('Arguments: %s' % ','.join(arguments))
return options, arguments
def parse_arguments(self, cli_args):
"""Public interface for parsing command line arguments.
:param cli_args: Command line arguments as a list
:returns: options (dict), arguments (list)
:raises: :class:`~robot.errors.Information` when --help or --version used
:raises: :class:`~robot.errors.DataError` when parsing fails
"""
return self._ap.parse_args(cli_args)
def execute(self, *arguments, **options):
with self._logger:
self._logger.info('%s %s' % (self._ap.name, self._ap.version))
return self._execute(list(arguments), options)
def _execute(self, arguments, options):
try:
rc = self.main(arguments, **options)
.....
Application的execute_cli方法其实也只是做了参数的解析工作，具体的任务交给了本实例的main方法。仍然回到 src/robot/run.py 看RobotFramework的main方法：
def main(self, datasources, **options):
settings = RobotSettings(options)
LOGGER.register_console_logger(**settings.console_output_config)
LOGGER.info('Settings:\n%s' % unic(settings))
builder = TestSuiteBuilder(settings['SuiteNames'],
extension=settings.extension,
rpa=settings.rpa)
suite = builder.build(*datasources)
settings.rpa = builder.rpa
suite.configure(**settings.suite_config)
if settings.pre_run_modifiers:
suite.visit(ModelModifier(settings.pre_run_modifiers,
settings.run_empty_suite, LOGGER))
with pyloggingconf.robot_handler_enabled(settings.log_level):
old_max_error_lines = text.MAX_ERROR_LINES
text.MAX_ERROR_LINES = settings.max_error_lines
try:
result = suite.run(settings)
finally:
text.MAX_ERROR_LINES = old_max_error_lines
LOGGER.info("Tests execution ended. Statistics:\n%s"
% result.suite.stat_message)
if settings.log or settings.report or settings.xunit:
writer = ResultWriter(settings.output if settings.log
else result)
writer.write_results(settings.get_rebot_settings())
return result.return_code
在这个方法里，进行了设置项的赋值，并且真正开始了执行测试。看TestSuiteBuilder是做什么的。
class TestSuiteBuilder(object):
def __init__(self, include_suites=None, warn_on_skipped='DEPRECATED',
extension=None, rpa=None):
self.include_suites = include_suites
self.extensions = self._get_extensions(extension)
builder = StepBuilder()
self._build_steps = builder.build_steps
self._build_step = builder.build_step
self.rpa = rpa
self._rpa_not_given = rpa is None
......
def build(self, *paths):
"""
:param paths: Paths to test data files or directories.
:return: :class:`~robot.running.model.TestSuite` instance.
"""
if not paths:
raise DataError('One or more source paths required.')
if len(paths) == 1:
return self._parse_and_build(paths[0])
root = TestSuite()
for path in paths:
root.suites.append(self._parse_and_build(path))
root.rpa = self.rpa
return root
......
这个TestSuiteBuilder的目的是通过解析datasource来构建一个TestSuite。那TestSuite又是什么的呢？
从from .model import Keyword, TestCase, TestSuite可知，TestSuite是在
class TestSuite(model.TestSuite):
"""Represents a single executable test suite.
See the base class for documentation of attributes not documented here.
"""
__slots__ = ['resource']
test_class = TestCase #: Internal usage only.
keyword_class = Keyword #: Internal usage only.
def __init__(self, name='', doc='', metadata=None, source=None, rpa=False):
model.TestSuite.__init__(self, name, doc, metadata, source, rpa)
#: :class:`ResourceFile` instance containing imports, variables and
#: keywords the suite owns. When data is parsed from the file system,
#: this data comes from the same test case file that creates the suite.
self.resource = ResourceFile(source=source)
def configure(self, randomize_suites=False, randomize_tests=False,
randomize_seed=None, **options)
model.TestSuite.configure(self, **options)
self.randomize(randomize_suites, randomize_tests, randomize_seed)
def randomize(self, suites=True, tests=True, seed=None):
"""Randomizes the order of suites and/or tests, recursively.
:param suites: Boolean controlling should suites be randomized.
:param tests: Boolean controlling should tests be randomized.
:param seed: Random seed. Can be given if previous random order needs
to be re-created. Seed value is always shown in logs and reports.
"""
self.visit(Randomizer(suites, tests, seed))
def run(self, settings=None, **options):
.......
from .namespace import IMPORTER
from .signalhandler import STOP_SIGNAL_MONITOR
from .runner import Runner
with LOGGER:
if not settings:
settings = RobotSettings(options)
LOGGER.register_console_logger(**settings.console_output_config)
with pyloggingconf.robot_handler_enabled(settings.log_level):
with STOP_SIGNAL_MONITOR:
IMPORTER.reset()
output = Output(settings)
runner = Runner(output, settings)
self.visit(runner)
output.close(runner.result)
return runner.result
这里TestSuite是model.TestSuite的子类
def visit(self, visitor):
""":mod:`Visitor interface <robot.model.visitor>` entry-point."""
visitor.visit_suite(self)
这里只是调用了Runner的visit_suite方法，来看一下
class Runner(SuiteVisitor):
Runner只是SuiteVisitor的一个子类，看看SuiteVisitor
class SuiteVisitor(object):
"""Abstract class to ease traversing through the test suite structure.
See the :mod:`module level <robot.model.visitor>` documentation for more
information and an example.
"""
def visit_suite(self, suite):
"""Implements traversing through the suite and its direct children.
Can be overridden to allow modifying the passed in ``suite`` without
calling :func:`start_suite` or :func:`end_suite` nor visiting child
suites, tests or keywords (setup and teardown) at all.
"""
if self.start_suite(suite) is not False:
suite.keywords.visit(self)
suite.suites.visit(self)
suite.tests.visit(self)
self.end_suite(suite)
.......
start_suite / end_suite 就是在Runner具体实现的。
具体看start_suite是做什么的：
def start_suite(self, suite):
self._output.library_listeners.new_suite_scope()
result = TestSuite(source=suite.source,
name=suite.name,
doc=suite.doc,
metadata=suite.metadata,
starttime=get_timestamp(),
rpa=self._settings.rpa)
if not self.result:
result.set_criticality(self._settings.critical_tags,
self._settings.non_critical_tags)
self.result = Result(root_suite=result, rpa=self._settings.rpa)
self.result.configure(status_rc=self._settings.status_rc,
stat_config=self._settings.statistics_config)
else:
self._suite.suites.append(result)
self._suite = result
self._suite_status = SuiteStatus(self._suite_status,
self._settings.exit_on_failure,
self._settings.exit_on_error,
self._settings.skip_teardown_on_exit)
ns = Namespace(self._variables, result, suite.resource)
ns.start_suite()
ns.variables.set_from_variable_table(suite.resource.variables)
EXECUTION_CONTEXTS.start_suite(result, ns, self._output,
self._settings.dry_run)
self._context.set_suite_variables(result)
if not self._suite_status.failures:
ns.handle_imports()
ns.variables.resolve_delayed()
result.doc = self._resolve_setting(result.doc)
result.metadata = [(self._resolve_setting(n), self._resolve_setting(v))
for n, v in result.metadata.items()]
self._context.set_suite_variables(result)
self._output.start_suite(ModelCombiner(suite, result,
tests=suite.tests,
suites=suite.suites,
test_count=suite.test_count))
self._output.register_error_listener(self._suite_status.error_occurred)
self._run_setup(suite.keywords.setup, self._suite_status)
self._executed_tests = NormalizedDict(ignore='_')
def end_suite(self, suite):
self._suite.message = self._suite_status.message
self._context.report_suite_status(self._suite.status,
self._suite.full_message)
with self._context.suite_teardown():
failure = self._run_teardown(suite.keywords.teardown, self._suite_status)
if failure:
self._suite.suite_teardown_failed(unic(failure))
if self._suite.statistics.critical.failed:
self._suite_status.critical_failure_occurred()
self._suite.endtime = get_timestamp()
self._suite.message = self._suite_status.message
self._context.end_suite(ModelCombiner(suite, self._suite))
self._suite = self._suite.parent
self._suite_status = self._suite_status.parent
self._output.library_listeners.discard_suite_scope()
执行到这里，会根据设置和datasource 已经开始了收集测试结果。
回到最初的 src/robot/run.py
根据
if __name__ == '__main__':
run_cli(sys.argv[1:])
可以看出，run.py不仅可以通过java -jar robotframework.jar run mytests.robot，被最终调用，还可以直接使用命令，例如：
robot path/to/tests.robot
robot --include tag1 --include tag2 --splitlog tests.robot
robot --name Example --log NONE t1.robot t2.robot > stdout.txt
来执行robotframework的测试用例。
如果喜欢作者的文章，请关注"写代码的猿"订阅号以便第一时间获得最新内容。本文版权归作者所有，欢迎转载.
