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 re
+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()
+
+
+# This maps from (recipe_name,test_name) -> yielded test_data. It's outside of
+# run_recipe so that it can persist between RunRecipe calls in the same process.
+_GEN_TEST_CACHE = {}
+
+
+# Allow regex patterns to be 'deep copied' by using them as-is.
+copy._deepcopy_dispatch[re._pattern_type] = copy._deepcopy_atomic
+
+
+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):

[iannucci, 2017/03/10 01:20:09]

wdyt about putting some of this test code in a subdirectory of recipe_engine?
e.g. recipe_engine/test_impl/exceptions.py, etc.

[Paweł Hajdan Jr., 2017/03/10 17:20:30]

With slightly over 500 lines, I don't see big benefits of splitting.

+  """Generic class for different kinds of test failures."""

[iannucci, 2017/03/10 01:20:08]

Generic or 'Base' class? I assume this is never meant to be raised directly?

[Paweł Hajdan Jr., 2017/03/10 17:20:29]

Yeah, changed to base. Obviously it's not meant to be raised directly - format()
raises NotImplementedError().

+
+  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)
+
+
+class TestResult(object):
+  """Result of running a test."""
+
+  def __init__(self, test_description, failures, coverage_data):
+    self.test_description = test_description
+    self.failures = failures
+    self.coverage_data = coverage_data
+
+
+class TestDescription(object):
+  """Identifies a specific test.
+
+  Deliberately small and picklable for use with multiprocessing."""
+
+  def __init__(self, recipe_name, test_name, expect_dir, covers):
+    self.recipe_name = recipe_name
+    self.test_name = test_name
+    self.expect_dir = expect_dir
+    self.covers = covers
+
+  @property
+  def full_name(self):
+    return '%s.%s' % (self.recipe_name, self.test_name)
+
+
+def expectation_path(expect_dir, test_name):
+  """Returns path where serialized expectation data is stored."""
+  return os.path.join(expect_dir, test_name + '.json')
+
+
+def run_test(test_description):
+  """Runs a test. Returns TestResults object."""
+  expected = None
+  path = expectation_path(
+      test_description.expect_dir, test_description.test_name)
+  if os.path.exists(path):
+    with open(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 = run_recipe(
+      test_description.recipe_name, test_description.test_name,
+      test_description.covers)
+  actual = re_encode(actual)
+
+  failures = []
+
+  # TODO(phajdan.jr): handle exception (errors) in the recipe execution.
+  if failed_checks:
+    sys.stdout.write('C')

[iannucci, 2017/03/10 01:20:09]

are these writes going to happen from multiple threads/processes? This might get
confusing if so.

[Paweł Hajdan Jr., 2017/03/10 17:20:29]

Yes.
Why?

+    failures.extend([CheckFailure(c) for c in failed_checks])
+  elif actual != expected:
+    diff = '\n'.join(list(difflib.unified_diff(

[iannucci, 2017/03/10 01:20:09]

is the 'list' necessary? afaik, ''.join will work with a generator just fine
without buffering the output into an extra copy first 

[Paweł Hajdan Jr., 2017/03/10 17:20:30]

Done.

+        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(test_description, failures, coverage_data)
+
+
+def run_recipe(recipe_name, test_name, covers):
+  """Runs the recipe under test in simulation mode.
+
+  Returns a tuple:
+    - expectation data
+    - failed post-process checks (if any)
+    - coverage data
+  """
+  config_types.ResetTostringFns()
+
+  # Reload test data in the worker to avoid pickling complex data structures.
+  cache_key = (recipe_name, test_name)
+  if cache_key not in _GEN_TEST_CACHE:
+    recipe_script = _UNIVERSE_VIEW.load_recipe(recipe_name)
+    test_api = loader.create_test_api(recipe_script.LOADED_DEPS, _UNIVERSE_VIEW)
+    for test_data in recipe_script.gen_tests(test_api):
+      _GEN_TEST_CACHE[(recipe_name, test_data.name)] = copy.deepcopy(test_data)

[iannucci, 2017/03/10 01:20:09]

use cache_key? we had a bug before where we were writing into the cache with key
X, but doing a lookup using key Y; as a result it was continually reloading and
copying everything :(

Actually now that I look at this (and the existing implementation), I would
recommend making this a 2-tier cache of CACHE[recipe_name][test_name]. Then the
initial lookup will be for the recipe, and then it will either load the full
recipe into the cache, or use the existing one, then do a direct lookup for the
test name. Otherwise in the case of a bug, we might run a given recipe's
GenTests function more than once, and in the case of flake, this might lead to
things somehow working some of the time.

An example of this might be if the user's recipe generates test names
non-deterministically. That SHOULD be a failure, but in this implementation it
MIGHT succeed.

[Paweł Hajdan Jr., 2017/03/10 17:20:30]

We can't use cache_key on this line, because test_name is only equal to
test_data.name for _one_ of the loop iterations, not _all_ of them.

Obviously cache_key is used on the line right below.
I don't see an advantage of this. If a recipe is present in the cache, so are
all its tests.
We're already running GenTests more than once: first in gen_tests, then in each
of the worker processes.
I think in general the recipe engine should be robust against user recipe code
doing weird things. Realistically, there are limits of what we can do, including
whether it's worth the complexity.

[iannucci, 2017/03/10 18:14:33]

Oh, heh, yeah I'm dumb.
Only if GenTests is deterministic. Say that it generates 99% the same test names
when you run it. In this model, the main thread will generate test name A, B, C
and X, and this worker will generate A, B, C and Z. The worker generates
everything when it sees the first test A, and then re-generates everything again
for test X (because (recipe, X) isn't in the cache).
Yes, and what I'm saying is that we should run them AT MOST 1+N times (one in
the main thread, one per worker). Right now there are scenarios where they could
run more than that (and not fail where they should).
In general, sure, but a 2D map isn't complex.

+  test_data = _GEN_TEST_CACHE[cache_key]
+
+  annotator = SimulationAnnotatorStreamEngine()
+  with stream.StreamEngineInvariants.wrap(annotator) as stream_engine:
+    runner = step_runner.SimulationStepRunner(
+        stream_engine, test_data, annotator)
+
+    props = test_data.properties.copy()
+    props['recipe'] = recipe_name
+    engine = run.RecipeEngine(
+        runner, props, _UNIVERSE_VIEW, engine_flags=_ENGINE_FLAGS)
+    with coverage_context(include=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(recipe_name, engine=engine)
+
+      api = loader.create_recipe_api(
+        _UNIVERSE_VIEW.universe.package_deps.root_package,
+        recipe_script.LOADED_DEPS,
+        recipe_script.path, engine, test_data)
+      result = engine.run(recipe_script, api, test_data.properties)
+    coverage_data = cov.get_data()
+
+    raw_expectations = runner.steps_ran.copy()
+    # Don't include tracebacks in expectations because they are too sensitive
+    # to change.
+    result.result.pop('traceback', None)

[iannucci, 2017/03/10 01:20:09]

should we replace the traceback with something indicating there WAS a traceback?
I wouldn't do that in this CL (it will probably change expectations). maybe
TODO?

[Paweł Hajdan Jr., 2017/03/10 17:20:29]

Added a TODO.

+    raw_expectations[result.result['name']] = result.result
+
+    failed_checks = []
+
+    for hook, args, kwargs, filename, lineno in test_data.post_process_hooks:
+      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)
+
+      with coverage_context(include=covers) as cov:
+        # Run the hook itself under coverage. There may be custom post-process
+        # functions in recipe test code.

[iannucci, 2017/03/10 01:20:09]

+1

+        rslt = hook(checker_obj, input_odict, *args, **kwargs)
+      coverage_data.update(cov.get_data())
+
+      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, coverage_data)
+
+
+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 = []
+
+  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')

[iannucci, 2017/03/10 01:20:09]

nit: this could be moved outside the loop

[Paweł Hajdan Jr., 2017/03/10 17:20:30]

Done.

FWIW, there's a slight tradeoff between "efficiency", and the variable being
closer to where it's used. Whatever.

+    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)
+      # 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.
+      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]
+        # TODO(phajdan.jr): move expectation tree outside of the recipe tree.
+        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(TestDescription(
+            recipe_name, test_data.name, 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', '*'))

[iannucci, 2017/03/10 01:20:09]

is the final * necessary? because some modules have subdirs under resources too
(this is the recursive loading thing). Does this accept **?

[Paweł Hajdan Jr., 2017/03/10 17:20:30]

Yes. Otherwise simulation tests inside recipe engine itself fail.

Also note this is what the current code does, see cover_omit in
recipe_engine/simulation_test.py .
It does, but not sure if it means what you want.

Again, this matches what current code does, I tested against all infra repos,
and https://codereview.chromium.org/2738053005/ will remove recursive loading.

+
+  # 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 run_test(test)
+
+
+def run_run(jobs):
+  """Implementation of the 'run' command."""
+  start_time = datetime.datetime.now()
+
+  report_coverage_version()
+
+  tests, coverage_data, uncovered_modules = get_tests()

[iannucci, 2017/03/10 01:20:09]

should this run with the kill switch too? or does ctrl-c work fine to cancel
get_tests?

[Paweł Hajdan Jr., 2017/03/10 17:20:30]

kill_switch is only needed for multiprocessing.

+  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_description.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())
+      print('FATAL: Insufficient coverage (%.f%%)' % int(percentage))
+  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.

[iannucci, 2017/03/10 01:20:09]

print "Got Ctrl-C: Stopping. Ctrl-C again to shut down uncleanly" ?

[Paweł Hajdan Jr., 2017/03/10 17:20:29]

This would be non-trivial. If done as-is, it'd print this _after_ bash prompt,
possibly more than once.

As far as I know, the current code doesn't do this, and I'm not sure what's the
specific advantage of printing this message. From user POV, if things hang,
there doesn't seem to be a big difference between "clean" and "unclean"
shutdown.

+  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):

[iannucci, 2017/03/10 01:20:08]

todo: look into custom decoder class:
https://docs.python.org/2/library/json.html#json.JSONDecoder to just do all this
on the first pass.

We should incorporate it into the 'json' recipe module too.

[Paweł Hajdan Jr., 2017/03/10 17:20:29]

Added a TODO. FWIW, this matches what current code does.

+  """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)