simulation_test_ng: initial CL

unittests/test_test.py

+#!/usr/bin/env python
+# 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.
+
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+
+
+ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, ROOT_DIR)
+import recipe_engine.env
+
+
+from recipe_engine import package
+from recipe_engine import package_pb2
+
+
+class RecipeWriter(object):
+  """Helper to write a recipe for tests."""
+
+  def __init__(self, root_dir, name):
+    self.root_dir = root_dir
+    self.name = name
+
+    # These are expected to be set appropriately by the caller.
+    self.DEPS = []
+    self.RunStepsLines = ['pass']
+    self.GenTestsLines = ['yield api.test("basic")']
+    self.expectations = {}
+
+  @property
+  def recipes_dir(self):
+    return os.path.join(self.root_dir, 'recipes')
+
+  @property
+  def expect_dir(self):
+    return os.path.join(self.recipes_dir, '%s.expected' % self.name)
+
+  def add_expectation(self, test_name, commands=None, recipe_result=None,
+                      status_code=0):
+    """Adds expectation for a simulation test.
+
+    Arguments:
+      test_name(str): name of the test
+      commands(list): list of expected commands

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

I think this is a list of expectation dictionaries? is that correct?

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

Done.

+      recipe_result(object): expected result of the recipe
+      status_code(int): expected exit code
+    """
+    self.expectations[test_name] = (commands or []) + [{
+        'name': '$result',
+        'recipe_result': recipe_result,
+        'status_code': status_code
+    }]
+
+  def write(self):
+    """Writes the recipe to disk."""
+    for d in (self.recipes_dir, self.expect_dir):
+      if not os.path.exists(d):
+        os.makedirs(d)
+    with open(os.path.join(self.recipes_dir, '%s.py' % self.name), 'w') as f:
+      f.write('\n'.join([
+        'from recipe_engine import post_process',
+        '',
+        'DEPS = %r' % self.DEPS,
+        '',
+        'def RunSteps(api):',
+      ] + ['  %s' % l for l in self.RunStepsLines] + [
+        '',
+        'def GenTests(api):',
+      ] + ['  %s' % l for l in self.GenTestsLines]))
+    for test_name, test_contents in self.expectations.iteritems():
+      with open(os.path.join(self.expect_dir, '%s.json' % test_name), 'w') as f:
+        json.dump(test_contents, f)
+
+
+class RecipeModuleWriter(object):
+  """Helper to write a recipe module for tests."""
+
+  def __init__(self, root_dir, name):
+    self.root_dir = root_dir
+    self.name = name
+
+    # These are expected to be set appropriately by the caller.
+    self.DEPS = []
+    self.disable_strict_coverage = False
+    self.methods = {}
+
+    self.example = type('example_placeholder', (), {})()
+    self.example.enabled = False
+    self.example.DEPS = []
+    self.example.RunStepsLines = ['pass']
+    self.example.GenTestsLines = ['yield api.test("basic")']
+    self.example.expectations = {}
+
+  @property
+  def module_dir(self):
+    return os.path.join(self.root_dir, 'recipe_modules', self.name)
+
+  @property
+  def expect_dir(self):
+    return os.path.join(self.module_dir, 'example.expected')
+
+  def add_expectation(self, test_name, commands=None, recipe_result=None,
+                      status_code=0):
+    """Adds expectation for a simulation test.
+
+    Arguments:
+      test_name(str): name of the test
+      commands(list): list of expected commands
+      recipe_result(object): expected result of the recipe
+      status_code(int): expected exit code
+    """
+    self.example.expectations[test_name] = (commands or []) + [{
+        'name': '$result',
+        'recipe_result': recipe_result,
+        'status_code': status_code
+    }]
+
+  def write(self):
+    """Writes the recipe module to disk."""
+
+    for d in (self.module_dir, self.expect_dir):
+      if not os.path.exists(d):
+        os.makedirs(d)
+
+    with open(os.path.join(self.module_dir, '__init__.py'), 'w') as f:
+      f.write('DEPS = %r\n' % self.DEPS)
+      if self.disable_strict_coverage:
+        f.write('\nDISABLE_STRICT_COVERAGE = True')
+
+    api_lines = [
+        'from recipe_engine import recipe_api',
+        '',
+        'class MyApi(recipe_api.RecipeApi):',
+    ]
+    if self.methods:
+      for m_name, m_lines in self.methods.iteritems():
+        api_lines.extend([
+            '',
+            '  def %s(self):' % m_name,
+          ] + ['    %s' % l for l in m_lines] + [
+            '',
+          ])
+    else:
+      api_lines.append('  pass')
+    with open(os.path.join(self.module_dir, 'api.py'), 'w') as f:
+      f.write('\n'.join(api_lines))
+    if self.example.enabled:
+      with open(os.path.join(self.module_dir, 'example.py'), 'w') as f:
+        f.write('\n'.join([
+          'from recipe_engine import post_process',
+          '',
+          'DEPS = %r' % self.example.DEPS,
+          '',
+          'def RunSteps(api):',
+        ] + ['  %s' % l for l in self.example.RunStepsLines] + [
+          '',
+          'def GenTests(api):',

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

this looks duplicated from RecipeWriter... could a RecipeModuleWriter have-a
RecipeWriter instead?

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

I was considering that, but IMO it could make things unnecessarily slightly more
complicated.

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

How is it more complex to reuse a well-defined class instead of copy and pasting
its implementation in multiple places?

+        ] + ['  %s' % l for l in self.example.GenTestsLines]))
+      for test_name, test_contents in self.example.expectations.iteritems():
+        with open(os.path.join(
+            self.expect_dir, '%s.json' % test_name), 'w') as f:
+          json.dump(test_contents, f)
+
+
+class TestTest(unittest.TestCase):
+  def setUp(self):
+    root_dir = tempfile.mkdtemp()
+    config_dir = os.path.join(root_dir, 'infra', 'config')
+    os.makedirs(config_dir)
+
+    self._root_dir = root_dir
+    self._recipes_cfg = os.path.join(config_dir, 'recipes.cfg')
+    self._recipe_tool = os.path.join(ROOT_DIR, 'recipes.py')
+
+    test_pkg = package_pb2.Package(
+        api_version=1,
+        project_id='test_pkg',
+        recipes_path='',
+        deps=[
+            package_pb2.DepSpec(
+                project_id='recipe_engine',
+                url='file://'+ROOT_DIR),
+        ],
+    )
+    package.ProtoFile(self._recipes_cfg).write(test_pkg)
+
+  def tearDown(self):
+    shutil.rmtree(self._root_dir)
+
+  def _run_recipes(self, *args):
+    return subprocess.check_output((
+        sys.executable,
+        self._recipe_tool,
+        '--use-bootstrap',
+        '--package', self._recipes_cfg,
+    ) + args, stderr=subprocess.STDOUT)
+
+  def test_list(self):
+    rw = RecipeWriter(self._root_dir, 'foo')
+    rw.RunStepsLines = ['pass']
+    rw.write()
+    self.assertEqual(
+        ['foo.basic'],
+        self._run_recipes('test', 'list').splitlines())
+
+  def test_test(self):
+    rw = RecipeWriter(self._root_dir, 'foo')
+    rw.RunStepsLines = ['pass']
+    rw.add_expectation('basic')
+    rw.write()
+    self._run_recipes('test', 'run')
+
+  def test_test_expectation_failure_empty(self):
+    rw = RecipeWriter(self._root_dir, 'foo')
+    rw.RunStepsLines = ['pass']
+    rw.write()
+    with self.assertRaises(subprocess.CalledProcessError) as cm:
+      self._run_recipes('test', 'run')
+    self.assertNotIn('FATAL: Insufficient coverage', cm.exception.output)

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

maybe TODO: use jsonproto output of 'test' instead of parsing stdout? Internally
test could produce its entire result as jsonproto, and then there would be a
small, independently tested printer which does proto->human stdout.

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

Yes, I plan to add JSON output support in subsequent CLs.

Already done for list command (was necessary to pass presubmit tests on trybots
- otherwise there was some warning noise).

+    self.assertNotIn('CHECK(FAIL)', cm.exception.output)
+    self.assertIn(
+        'foo.basic failed',
+        cm.exception.output)
+
+  def test_test_expectation_failure_different(self):
+    rw = RecipeWriter(self._root_dir, 'foo')
+    rw.DEPS = ['recipe_engine/step']
+    rw.RunStepsLines = ['api.step("test", ["echo", "bar"])']
+    rw.add_expectation('basic')
+    rw.write()
+    with self.assertRaises(subprocess.CalledProcessError) as cm:
+      self._run_recipes('test', 'run')
+    self.assertNotIn('FATAL: Insufficient coverage', cm.exception.output)
+    self.assertNotIn('CHECK(FAIL)', cm.exception.output)
+    self.assertIn(
+        'foo.basic failed',
+        cm.exception.output)
+    self.assertIn(
+        '+[{\'cmd\': [\'echo\', \'bar\'], \'name\': \'test\'},\n',
+        cm.exception.output)
+
+  def test_test_expectation_pass(self):
+    rw = RecipeWriter(self._root_dir, 'foo')
+    rw.DEPS = ['recipe_engine/step']
+    rw.RunStepsLines = ['api.step("test", ["echo", "bar"])']
+    rw.add_expectation('basic', [{'cmd': ['echo', 'bar'], 'name': 'test'}])
+    rw.write()
+    self._run_recipes('test', 'run')
+
+  def test_test_recipe_not_covered(self):
+    rw = RecipeWriter(self._root_dir, 'foo')
+    rw.RunStepsLines = ['if False:', '  pass']
+    rw.add_expectation('basic')
+    rw.write()
+    with self.assertRaises(subprocess.CalledProcessError) as cm:
+      self._run_recipes('test', 'run')
+    self.assertIn('FATAL: Insufficient coverage', cm.exception.output)
+    self.assertNotIn('CHECK(FAIL)', cm.exception.output)
+    self.assertNotIn('foo.basic failed', cm.exception.output)
+
+  def test_test_check_failure(self):
+    rw = RecipeWriter(self._root_dir, 'foo')
+    rw.RunStepsLines = ['pass']
+    rw.GenTestsLines = [
+        'yield api.test("basic") + \\',
+        '  api.post_process(post_process.MustRun, "bar")'
+    ]
+    rw.add_expectation('basic')
+    rw.write()
+    with self.assertRaises(subprocess.CalledProcessError) as cm:
+      self._run_recipes('test', 'run')
+    self.assertNotIn('FATAL: Insufficient coverage', cm.exception.output)
+    self.assertIn('CHECK(FAIL)', cm.exception.output)
+    self.assertIn('foo.basic failed', cm.exception.output)
+
+  def test_test_check_success(self):
+    rw = RecipeWriter(self._root_dir, 'foo')
+    rw.RunStepsLines = ['pass']
+    rw.GenTestsLines = [
+        'yield api.test("basic") + \\',
+        '  api.post_process(post_process.DoesNotRun, "bar")'
+    ]
+    rw.add_expectation('basic')
+    rw.write()
+    self._run_recipes('test', 'run')
+
+  def test_test_recipe_syntax_error(self):
+    rw = RecipeWriter(self._root_dir, 'foo')
+    rw.RunStepsLines = ['baz']
+    rw.add_expectation('basic')
+    rw.write()
+    with self.assertRaises(subprocess.CalledProcessError) as cm:
+      self._run_recipes('test', 'run')
+    self.assertIn('NameError: global name \'baz\' is not defined',
+                  cm.exception.output)
+
+  def test_test_recipe_module_uncovered(self):
+    mw = RecipeModuleWriter(self._root_dir, 'foo')
+    mw.write()
+    with self.assertRaises(subprocess.CalledProcessError) as cm:
+      self._run_recipes('test', 'run')
+    self.assertIn('The following modules lack test coverage: foo',
+                  cm.exception.output)
+
+  def test_test_recipe_module_syntax_error(self):
+    mw = RecipeModuleWriter(self._root_dir, 'foo_module')
+    mw.methods['foo'] = ['baz']
+    mw.example.enabled = True
+    mw.example.DEPS = ['foo_module']
+    mw.example.RunStepsLines = ['api.foo_module.foo()']
+    mw.write()
+    with self.assertRaises(subprocess.CalledProcessError) as cm:
+      self._run_recipes('test', 'run')
+    self.assertIn('NameError: global name \'baz\' is not defined',
+                  cm.exception.output)
+    self.assertIn('FATAL: Insufficient coverage', cm.exception.output)
+
+  def test_test_recipe_module_syntax_error_in_example(self):
+    mw = RecipeModuleWriter(self._root_dir, 'foo_module')
+    mw.methods['foo'] = ['pass']
+    mw.example.enabled = True
+    mw.example.DEPS = ['foo_module']
+    mw.example.RunStepsLines = ['baz']
+    mw.write()
+    with self.assertRaises(subprocess.CalledProcessError) as cm:
+      self._run_recipes('test', 'run')
+    self.assertIn('NameError: global name \'baz\' is not defined',
+                  cm.exception.output)
+    self.assertIn('FATAL: Insufficient coverage', cm.exception.output)
+
+  def test_test_recipe_module_example_not_covered(self):
+    mw = RecipeModuleWriter(self._root_dir, 'foo_module')
+    mw.methods['foo'] = ['pass']
+    mw.example.enabled = True
+    mw.example.DEPS = ['foo_module']
+    mw.example.RunStepsLines = ['if False:', '  pass']
+    mw.write()
+    with self.assertRaises(subprocess.CalledProcessError) as cm:
+      self._run_recipes('test', 'run')
+    self.assertIn('FATAL: Insufficient coverage', cm.exception.output)
+
+  def test_test_recipe_module_uncovered_not_strict(self):
+    mw = RecipeModuleWriter(self._root_dir, 'foo')
+    mw.disable_strict_coverage = True
+    mw.write()
+    self._run_recipes('test', 'run')
+
+  def test_test_recipe_module_covered_by_recipe_not_strict(self):
+    mw = RecipeModuleWriter(self._root_dir, 'foo_module')
+    mw.methods['bar'] = ['pass']
+    mw.disable_strict_coverage = True
+    mw.write()
+    rw = RecipeWriter(self._root_dir, 'foo_recipe')
+    rw.DEPS = ['foo_module']
+    rw.RunStepsLines = ['api.foo_module.bar()']
+    rw.add_expectation('basic')
+    rw.write()
+    self._run_recipes('test', 'run')
+
+  def test_test_recipe_module_covered_by_recipe(self):
+    mw = RecipeModuleWriter(self._root_dir, 'foo_module')
+    mw.methods['bar'] = ['pass']
+    mw.write()
+    rw = RecipeWriter(self._root_dir, 'foo_recipe')
+    rw.DEPS = ['foo_module']
+    rw.RunStepsLines = ['api.foo_module.bar()']
+    rw.add_expectation('basic')
+    rw.write()
+    with self.assertRaises(subprocess.CalledProcessError) as cm:
+      self._run_recipes('test', 'run')
+    self.assertIn('The following modules lack test coverage: foo_module',
+                  cm.exception.output)
+
+  def test_test_recipe_module_partially_covered_by_recipe_not_strict(self):
+    mw = RecipeModuleWriter(self._root_dir, 'foo_module')
+    mw.methods['bar'] = ['pass']
+    mw.methods['baz'] = ['pass']
+    mw.disable_strict_coverage = True
+    mw.example.enabled = True
+    mw.example.DEPS = ['foo_module']
+    mw.example.RunStepsLines = ['api.foo_module.baz()']
+    mw.add_expectation('basic')
+    mw.write()
+    rw = RecipeWriter(self._root_dir, 'foo_recipe')
+    rw.DEPS = ['foo_module']
+    rw.RunStepsLines = ['api.foo_module.bar()']
+    rw.add_expectation('basic')
+    rw.write()
+    self._run_recipes('test', 'run')
+
+  def test_test_recipe_module_partially_covered_by_recipe(self):
+    mw = RecipeModuleWriter(self._root_dir, 'foo_module')
+    mw.methods['bar'] = ['pass']
+    mw.methods['baz'] = ['pass']
+    mw.example.enabled = True
+    mw.example.DEPS = ['foo_module']
+    mw.example.RunStepsLines = ['api.foo_module.baz()']
+    mw.add_expectation('basic')
+    mw.write()
+    rw = RecipeWriter(self._root_dir, 'foo_recipe')
+    rw.DEPS = ['foo_module']
+    rw.RunStepsLines = ['api.foo_module.bar()']
+    rw.add_expectation('basic')
+    rw.write()
+    with self.assertRaises(subprocess.CalledProcessError) as cm:
+      self._run_recipes('test', 'run')
+    self.assertIn('FATAL: Insufficient coverage', cm.exception.output)

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

I don't see any tests for the post_process hooks? or are those covered
elsewhere?

edit: I guess that's covered by the stdlib_test.py test above?

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

Did you see test_test_check_failure and test_test_check_success?
That too.

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

Yes, but they don't cover the case of a user-provided function, but it's covered
here
https://github.com/luci/recipes-py/blob/master/recipes/engine_tests/whitelist...
so it's fine.

+
+
+if __name__ == '__main__':
+  sys.exit(unittest.main())