simulation_test_ng: initial CL

recipe_engine/test.py

+# Copyright 2017 The LUCI Authors. All rights reserved.
+# Use of this source code is governed under the Apache License, Version 2.0
+# that can be found in the LICENSE file.
+
+from __future__ import print_function
+
+import argparse
+import cStringIO
+import contextlib
+import copy
+import coverage
+import datetime
+import difflib
+import functools
+import json
+import multiprocessing
+import os
+import pprint
+import signal
+import sys
+import tempfile
+import traceback
+
+from . import checker
+from . import config_types
+from . import loader
+from . import run
+from . import step_runner
+from . import stream
+
+
+# These variables must be set in the dynamic scope of the functions in this
+# file.  We do this instead of passing because they're not picklable, and
+# that's required by multiprocessing.
+_UNIVERSE_VIEW = None
+_ENGINE_FLAGS = None
+
+
+# An event to signal exit, for example on Ctrl-C.
+_KILL_SWITCH = multiprocessing.Event()
+
+
+class PostProcessError(ValueError):
+  """Exception raised when any of the post-process hooks fails."""
+  pass
+
+
+@contextlib.contextmanager
+def coverage_context(include=None):
+  """Context manager that records coverage data."""
+  c = coverage.coverage(config_file=False, include=include)
+
+  # Sometimes our strict include lists will result in a run
+  # not adding any coverage info. That's okay, avoid output spam.
+  c._warn_no_data = False
+
+  c.start()
+  try:
+    yield c
+  finally:
+    c.stop()
+
+
+class TestFailure(object):
+  """Generic class for different kinds of test failures."""
+
+  def format(self):
+    """Returns a human-readable description of the failure."""
+    raise NotImplementedError()
+
+
+class DiffFailure(TestFailure):
+  """Failure when simulated recipe commands don't match recorded expectations.
+  """
+
+  def __init__(self, diff):
+    self.diff = diff
+
+  def format(self):
+    return self.diff
+
+
+class CheckFailure(TestFailure):
+  """Failure when any of the post-process checks fails."""
+
+  def __init__(self, check):
+    self.check = check
+
+  def format(self):
+    return self.check.format(indent=4)
+
+
+# TODO(phajdan.jr): Consider namedtuple instead.
+class TestResult(object):
+  def __init__(self, test, failures, coverage_data):
+    self.test = test
+    self.failures = failures
+    self.coverage_data = coverage_data
+
+
+class Test(object):
+  def __init__(self, recipe_name, test_data, expect_dir, covers):
+    self.recipe_name = recipe_name
+    self.test_data = test_data
+    self.expect_dir = expect_dir
+    self.covers = covers
+
+  @property
+  def test_name(self):
+    return self.test_data.name
+
+  @property
+  def full_name(self):
+    return '%s.%s' % (self.recipe_name, self.test_name)
+
+  @property
+  def expectation_path(self):
+    """Returns path where serialized expectation data is stored."""
+    # TODO(phajdan.jr): move expectation tree outside of the recipe tree.
+    return os.path.join(self.expect_dir, self.test_name + '.json')
+
+  def run(self):
+    """Runs the test. Returns TestResults object."""
+    # TODO(phajdan.jr): handle missing expectation files.
+    with open(self.expectation_path) as f:
+      # TODO(phajdan.jr): why do we need to re-encode golden data files?
+      expected = re_encode(json.load(f))
+
+    actual, failed_checks, coverage_data = self._run_recipe()
+    actual = re_encode(actual)
+
+    failures = []
+
+    # TODO(phajdan.jr): handle exception (errors) in the recipe execution.
+    if failed_checks:
+      sys.stdout.write('C')
+      failures.extend([CheckFailure(c) for c in failed_checks])
+    elif actual != expected:
+      diff = '\n'.join(list(difflib.unified_diff(
+          pprint.pformat(expected).splitlines(),
+          pprint.pformat(actual).splitlines(),
+          fromfile='expected', tofile='actual',
+          n=4, lineterm='')))
+
+      failures.append(DiffFailure(diff))
+      sys.stdout.write('F')
+    else:
+      sys.stdout.write('.')
+    sys.stdout.flush()
+
+    return TestResult(self, failures, coverage_data)
+
+  def _run_recipe(self):
+    """Runs the recipe under test in simulation mode.
+
+    Returns a tuple:
+      - expectation data
+      - failed post-process checks (if any)
+      - coverage data
+    """
+    config_types.ResetTostringFns()
+
+    annotator = SimulationAnnotatorStreamEngine()
+    with stream.StreamEngineInvariants.wrap(annotator) as stream_engine:
+      runner = step_runner.SimulationStepRunner(
+          stream_engine, self.test_data, annotator)
+
+      props = self.test_data.properties.copy()
+      props['recipe'] = self.recipe_name
+      engine = run.RecipeEngine(
+          runner, props, _UNIVERSE_VIEW, engine_flags=_ENGINE_FLAGS)
+      with coverage_context(include=self.covers) as cov:
+        # Run recipe loading under coverage context. This ensures we collect
+        # coverage of all definitions and globals.
+        recipe_script = _UNIVERSE_VIEW.load_recipe(
+            self.recipe_name, engine=engine)
+
+        api = loader.create_recipe_api(
+          _UNIVERSE_VIEW.universe.package_deps.root_package,
+          recipe_script.LOADED_DEPS,
+          recipe_script.path, engine, self.test_data)
+        result = engine.run(recipe_script, api, self.test_data.properties)
+
+      raw_expectations = runner.steps_ran.copy()
+      # Don't include tracebacks in expectations because they are too sensitive
+      # to change.
+      result.result.pop('traceback', None)
+      raw_expectations[result.result['name']] = result.result
+
+      failed_checks = []
+
+      for hook, args, kwargs, filename, lineno \
+          in self.test_data.post_process_hooks:

[iannucci, 2017/03/07 19:13:38]

I think you misunderstood me :). This implementation WILL NOT WORK for repos
other than recipe_engine, for the reasons I said in my previous comment. Users
pass closures via the post process hooks, and it is not possible to pickle
closures across processes.

The cost of loading the recipes' tests and caching them is small in repos like
build, compared to the number of tests that need to run, and fast enough for
small repos when cached that it shouldn't matter.

[Paweł Hajdan Jr., 2017/03/07 22:53:02]

Unless I'm severely missing something, this did in fact work with recipe_engine,
depot_tools, build, and infra repos.

Anyway, I changed the code to only pickle TestDescription so we avoid the
controversy.
I'd prefer not to worry about premature optimization right now.

+        input_odict = copy.deepcopy(raw_expectations)
+        # We ignore the input_odict so that it never gets printed in full.
+        # Usually the check invocation itself will index the input_odict or
+        # will use it only for a key membership comparison, which provides
+        # enough debugging context.
+        checker_obj = checker.Checker(
+            filename, lineno, hook, args, kwargs, input_odict)
+        rslt = hook(checker_obj, input_odict, *args, **kwargs)
+        failed_checks += checker_obj.failed_checks
+        if rslt is not None:
+          msg = checker.VerifySubset(rslt, raw_expectations)
+          if msg:
+            raise PostProcessError('post_process: steps'+msg)
+          # restore 'name'
+          for k, v in rslt.iteritems():
+            if 'name' not in v:
+              v['name'] = k
+          raw_expectations = rslt
+
+      # empty means drop expectation
+      result_data = raw_expectations.values() if raw_expectations else None
+      return (result_data, failed_checks, cov.get_data())

[iannucci, 2017/03/07 19:13:38]

pickling the entire result data across processes is one of the things that makes
the current implementation slow. I would consider having this function
write/compare the result directly.

[Paweł Hajdan Jr., 2017/03/07 22:53:02]

Again, I'd prefer not to worry about premature optimization right now.

+
+
+def get_tests():
+  """Returns a list of tests for current recipe package."""
+  tests = []
+  coverage_data = coverage.CoverageData()
+
+  all_modules = set(_UNIVERSE_VIEW.loop_over_recipe_modules())
+  covered_modules = set()
+
+  base_covers = []
+
+  # Make sure disabling strict coverage also disables our additional check
+  # for module coverage. Note that coverage will still raise an error if
+  # the module is executed by any of the tests, but having less than 100%
+  # coverage.
+  for module in all_modules:
+    # Run module loading under coverage context. This ensures we collect
+    # coverage of all definitions and globals.
+    coverage_include = os.path.join(_UNIVERSE_VIEW.module_dir, '*', '*.py')
+    with coverage_context(include=coverage_include) as cov:
+      mod = _UNIVERSE_VIEW.load_recipe_module(module)
+    coverage_data.update(cov.get_data())
+
+    # Recipe modules can only be covered by tests inside the same module.
+    # To make transition possible for existing code (which will require
+    # writing tests), a temporary escape hatch is added.
+    # TODO(phajdan.jr): remove DISABLE_STRICT_COVERAGE (crbug/693058).
+    if mod.DISABLE_STRICT_COVERAGE:
+      covered_modules.add(module)
+      base_covers.append(os.path.join(
+          _UNIVERSE_VIEW.module_dir, module, '*.py'))
+
+  for recipe_path, recipe_name in _UNIVERSE_VIEW.loop_over_recipes():
+    try:
+      covers = [recipe_path] + base_covers
+
+      # Example/test recipes in a module always cover that module.
+      if ':' in recipe_name:
+        module, _ = recipe_name.split(':', 1)
+        covered_modules.add(module)
+        covers.append(os.path.join(_UNIVERSE_VIEW.module_dir, module, '*.py'))
+
+      with coverage_context(include=covers) as cov:
+        # Run recipe loading under coverage context. This ensures we collect
+        # coverage of all definitions and globals.
+        recipe = _UNIVERSE_VIEW.load_recipe(recipe_name)
+        test_api = loader.create_test_api(recipe.LOADED_DEPS, _UNIVERSE_VIEW)
+
+        root, name = os.path.split(recipe_path)
+        name = os.path.splitext(name)[0]
+        expect_dir = os.path.join(root, '%s.expected' % name)
+
+        # Immediately convert to list to force running the generator under
+        # coverage context. Otherwise coverage would only report executing
+        # the function definition, not GenTests body.
+        recipe_tests = list(recipe.gen_tests(test_api))
+      coverage_data.update(cov.get_data())
+
+      for test_data in recipe_tests:
+        tests.append(Test(recipe_name, test_data, expect_dir, covers))
+    except:
+      info = sys.exc_info()
+      new_exec = Exception('While generating results for %r: %s: %s' % (
+        recipe_name, info[0].__name__, str(info[1])))
+      raise new_exec.__class__, new_exec, info[2]
+
+  uncovered_modules = all_modules.difference(covered_modules)
+  return (tests, coverage_data, uncovered_modules)
+
+
+def run_list():
+  """Implementation of the 'list' command."""
+  tests, _coverage_data, _uncovered_modules = get_tests()
+  print('\n'.join(sorted(t.full_name for t in tests)))
+  return 0
+
+
+def cover_omit():
+  """Returns list of patterns to omit from coverage analysis."""
+  omit = [ ]
+
+  mod_dir_base = _UNIVERSE_VIEW.module_dir
+  if os.path.isdir(mod_dir_base):
+      omit.append(os.path.join(mod_dir_base, '*', 'resources', '*'))
+
+  # Exclude recipe engine files from simulation test coverage. Simulation tests
+  # should cover "user space" recipe code (recipes and modules), not the engine.
+  # The engine is covered by unit tests, not simulation tests.
+  omit.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), '*'))
+
+  return omit
+
+
+def report_coverage_version():
+  """Prints info about coverage module (for debugging)."""
+  print('Using coverage %s from %r' % (coverage.__version__, coverage.__file__))
+
+
+def worker(f):
+  """Wrapper for a multiprocessing worker function.
+
+  This addresses known issues with multiprocessing workers:
+
+    - they can hang on uncaught exceptions
+    - we need explicit kill switch to clearly terminate parent"""
+  @functools.wraps(f)
+  def wrapper(*args, **kwargs):
+    try:
+      if _KILL_SWITCH.is_set():
+        return (False, 'kill switch')
+      return (True, f(*args, **kwargs))
+    except Exception:
+      return (False, traceback.format_exc())
+  return wrapper
+
+
+@worker
+def run_worker(test):
+  """Worker for 'run' command (note decorator above)."""
+  return test.run()
+
+
+def run_run(jobs):
+  """Implementation of the 'run' command."""
+  start_time = datetime.datetime.now()
+
+  report_coverage_version()
+
+  tests, coverage_data, uncovered_modules = get_tests()
+  if uncovered_modules:
+    raise Exception('The following modules lack test coverage: %s' % (
+        ','.join(sorted(uncovered_modules))))
+
+  with kill_switch():
+    pool = multiprocessing.Pool(jobs)
+    results = pool.map(run_worker, tests)
+
+  print()
+
+  rc = 0
+  for success, details in results:
+    if success:
+      assert isinstance(details, TestResult)
+      if details.failures:
+        rc = 1
+        print('%s failed:' % details.test.full_name)
+        for failure in details.failures:
+          print(failure.format())
+      coverage_data.update(details.coverage_data)
+    else:
+      rc = 1
+      print('Internal failure:')
+      print(details)
+
+  try:
+    # TODO(phajdan.jr): Add API to coverage to load data from memory.
+    with tempfile.NamedTemporaryFile(delete=False) as coverage_file:
+      coverage_data.write_file(coverage_file.name)
+
+    cov = coverage.coverage(
+        data_file=coverage_file.name, config_file=False, omit=cover_omit())
+    cov.load()
+    outf = cStringIO.StringIO()
+    percentage = cov.report(file=outf, show_missing=True, skip_covered=True)
+    if int(percentage) != 100:
+      rc = 1
+      print(outf.getvalue())
+  finally:
+    os.unlink(coverage_file.name)
+
+  finish_time = datetime.datetime.now()
+  print('-' * 70)
+  print('Ran %d tests in %0.3fs' % (
+      len(tests), (finish_time - start_time).total_seconds()))
+  print()
+  print('OK' if rc == 0 else 'FAILED')
+
+  return rc
+
+
+class SimulationAnnotatorStreamEngine(stream.AnnotatorStreamEngine):
+  """Stream engine which just records generated commands."""
+
+  def __init__(self):
+    self._step_buffer_map = {}
+    super(SimulationAnnotatorStreamEngine, self).__init__(
+        self.step_buffer(None))
+
+  def step_buffer(self, step_name):
+    return self._step_buffer_map.setdefault(step_name, cStringIO.StringIO())
+
+  def new_step_stream(self, step_config):
+    return self._create_step_stream(step_config,
+                                    self.step_buffer(step_config.name))
+
+
+def handle_killswitch(*_):
+  """Function invoked by ctrl-c. Signals worker processes to exit."""
+  _KILL_SWITCH.set()
+
+  # Reset the signal to DFL so that double ctrl-C kills us for sure.
+  signal.signal(signal.SIGINT, signal.SIG_DFL)
+  signal.signal(signal.SIGTERM, signal.SIG_DFL)
+
+
+@contextlib.contextmanager
+def kill_switch():
+  """Context manager to handle ctrl-c properly with multiprocessing."""
+  orig_sigint = signal.signal(signal.SIGINT, handle_killswitch)
+  try:
+    orig_sigterm = signal.signal(signal.SIGTERM, handle_killswitch)
+    try:
+      yield
+    finally:
+      signal.signal(signal.SIGTERM, orig_sigterm)
+  finally:
+    signal.signal(signal.SIGINT, orig_sigint)
+
+  if _KILL_SWITCH.is_set():
+    sys.exit(1)
+
+
+def re_encode(obj):
+  """Ensure consistent encoding for common python data structures."""
+  if isinstance(obj, dict):
+    return {re_encode(k): re_encode(v) for k, v in obj.iteritems()}
+  elif isinstance(obj, list):
+    return [re_encode(i) for i in obj]
+  elif isinstance(obj, (unicode, str)):
+    if isinstance(obj, str):
+      obj = obj.decode('utf-8', 'replace')
+    return obj.encode('utf-8', 'replace')
+  else:
+    return obj
+
+
+def parse_args(args):
+  """Returns parsed command line arguments."""
+  parser = argparse.ArgumentParser()
+
+  subp = parser.add_subparsers()
+
+  list_p = subp.add_parser('list', description='Print all test names')
+  list_p.set_defaults(func=lambda opts: run_list())
+
+  # TODO(phajdan.jr): support running a subset of tests.
+  run_p = subp.add_parser('run', description='Run the tests')
+  run_p.set_defaults(func=lambda opts: run_run(opts.jobs))
+  run_p.add_argument(
+      '--jobs', metavar='N', type=int,
+      default=multiprocessing.cpu_count(),
+      help='run N jobs in parallel (default %(default)s)')
+
+  return parser.parse_args(args)
+
+
+def main(universe_view, raw_args, engine_flags):
+  """Runs simulation tests on a given repo of recipes.
+
+  Args:
+    universe_view: an UniverseView object to operate on
+    raw_args: command line arguments to simulation_test_ng
+    engine_flags: recipe engine command-line flags
+  Returns:
+    Exit code
+  """
+  global _UNIVERSE_VIEW
+  _UNIVERSE_VIEW = universe_view
+  global _ENGINE_FLAGS
+  _ENGINE_FLAGS = engine_flags
+
+  args = parse_args(raw_args)
+  return args.func(args)