blob: 6e82cf66ff1ed3d2f4731d54ec50f3a2e67f4ec4 [file] [log] [blame]
# Copyright 2019 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.
"""Provides a 'fake' version of the RecipeDeps object.
This is pretty heavyweight and should be used for integration-testing recipe
functionality.
The objects here manipulate full 'recipe repos' on disk (complete with git
repositories and the ability to make commits in them).
Access this via test_env.RecipeEngineUnitTest.FakeRecipeDeps().
"""
import contextlib
import errno
from io import StringIO
import json
import os
import shutil
import subprocess
import sys
import textwrap
import attr
from google.protobuf import json_format as jsonpb
from PB.recipe_engine.recipes_cfg import RepoSpec
from recipe_engine import __path__ as RECIPE_ENGINE_PATH
from recipe_engine.internal.fetch import GitBackend, CommitMetadata
from recipe_engine.internal.simple_cfg import RECIPES_CFG_LOCATION_REL
from recipe_engine.internal.test.test_util import filesystem_safe
ROOT_DIR = os.path.dirname(RECIPE_ENGINE_PATH[0])
DEVNULL = open(os.devnull, 'w')
REAL_STDERR = sys.stderr # capture stderr before tests potentially mess with it
def _get_suite(buf, default, indent=' '):
"""This is a helper function to extract a python code suite from a StringIO
object.
This will dedent the buffer's content, split it into lines, and then re-indent
it by the amount indicated.
Args:
* buf (StringIO) - The buffer to take the content from.
* default (str) - The content to use if `buf` is empty.
* indent (str) - The string to prepend to all de-indented lines.
Returns the transformed string.
"""
value = textwrap.dedent(buf.getvalue() or default).strip('\n')
return '\n'.join(indent + l for l in value.splitlines())
@attr.s
class FakeRecipeRepo:
"""Manipulates a recipe repo on disk (as a git repository)."""
# The FakeRecipeDeps which owns this repo.
fake_recipe_deps = attr.ib() # type: FakeRecipeDeps
# The name of this repo.
name = attr.ib() # type: str
# The absolute path on disk to the root of this repo.
path = attr.ib()
# The GitBackend for this FakeRecipeRepo.
backend = attr.ib(default=attr.Factory(
lambda self: GitBackend(self.path, None),
takes_self=True))
@contextlib.contextmanager
def edit_recipes_cfg_pb2(self):
"""Context manager for read/modify/write'ing the recipes.cfg file in this
repo.
Usage:
with repo.edit_recipes_cfg_pb2() as pb:
pb.deps['some_repo'].revision = 'abcdefg'
Yields a recipes_cfg_pb2.RepoSpec object decoded from the current state of
the recipes.cfg file. Any modifications done to this object will be recorded
back to disk.
"""
spec = self.recipes_cfg_pb2
yield spec
cfg_path = os.path.join(self.path, RECIPES_CFG_LOCATION_REL)
with open(cfg_path, 'w') as fil:
fil.write(jsonpb.MessageToJson(spec, preserving_proto_field_name=True))
@property
def recipes_cfg_pb2(self):
"""Returns the current recipes_cfg_pb2.RepoSpec decoded from the recipes.cfg
file in this repo."""
cfg_path = os.path.join(self.path, RECIPES_CFG_LOCATION_REL)
with open(cfg_path, 'r') as fil:
return jsonpb.Parse(fil.read(), RepoSpec())
@contextlib.contextmanager
def write_file(self, path):
"""Context manager for writing a file inside the repo.
Any missing directories will be automatically created.
Usage:
with repo.write_file('relative/path/filename.ext') as fil:
fil.write('''
Content is de-indented.
So you don't have ugly tests.
''')
Args:
* path (str) - Path relative to the root of the repo of the file to write.
Yields a StringIO object which you may write to. For convenience, the data
written to this StringIO object will be `textwrap.dedent`d to allow nice
inline strings in test files.
"""
full_path = os.path.join(self.path, path)
try:
os.makedirs(os.path.dirname(full_path))
except OSError as ex:
if ex.errno != errno.EEXIST:
raise
buf = StringIO()
yield buf
with open(full_path, 'w') as fil:
fil.write(textwrap.dedent(buf.getvalue()))
def read_file(self, path):
"""Reads a file inside the repo.
Args:
* path (str) - Path relative to the root of the repo of the file to read.
Returns the file contents as a str. If there is an error reading the file
this will return None.
"""
full_path = os.path.join(self.path, path)
try:
with open(full_path, 'r') as fil:
return fil.read()
except: # pylint: disable=bare-except
return None
def exists(self, path):
"""Checks to see if a path exists in the repo.
Args:
* path (str) - Path relative to the root of the repo to check.
Returns True iff the item exists.
"""
return os.path.exists(os.path.join(self.path, path))
def is_dir(self, path):
"""Checks to see if a path is a directory in the repo.
Args:
* path (str) - Path relative to the root of the repo to check.
Returns True iff the item exists and is a directory.
"""
return os.path.isdir(os.path.join(self.path, path))
def is_file(self, path):
"""Checks to see if a path is a file in the repo.
Args:
* path (str) - Path relative to the root of the repo to check.
Returns True iff the item exists and is a regular file.
"""
return os.path.isfile(os.path.join(self.path, path))
@attr.s(slots=True)
class WriteableRecipe:
"""Yielded from the `write_recipe` method. Used to generate a recipe on
disk.
This contains two string buffers which you may write to, RunSteps and
GenTests. These represent the body of the RunSteps and GenTests functions in
the recipe, respectively. Both buffers will be dedent'd as described in
`write_file`. If you do not write to them, the function bodies will be
generated with default values:
RunSteps: pass
GenTests: yield api.test('basic')
The RunSteps_args attribute is a list of argument names to be used in the
RunSteps function definition. It defaults to ['api'] and should generally
only be extended.
The imports attribute is a list of 'import' lines you want at the top of the
recipe. These lines should be an entire valid python import line (i.e
starting with `import` or `from`).
The DEPS attribute corresponds exactly to the recipe DEPS field, and you may
put a python list or dict here, depending on how you want your recipe's DEPS
to look. The DEPS comes, by default, as a list containing the
'recipe_engine/step' module. This list may be appended to, or replaced
entirely (with assignment), if you choose.
The PROPERTIES attribute is a Python expression to assign to the recipe
PROPERTIES field.
The `expectation` dictionary can be populated with expectation JSON data to
write to disk. The key is the test name (e.g. "basic", "test_on_luci", etc.)
and the value is list of dicts (i.e. the expectation format). It defaults to
a single expectation for the test "basic" which has an empty (`None`)
final result (i.e. the RunSteps function returned None without running
steps).
"""
base_path = attr.ib()
imports = attr.ib(factory=list)
RunSteps = attr.ib(factory=StringIO)
RunSteps_args = attr.ib(factory=lambda: ['api'])
GenTests = attr.ib(factory=StringIO)
DEPS = attr.ib(factory=lambda: ['recipe_engine/step'])
PROPERTIES = attr.ib(default='{}')
ENV_PROPERTIES = attr.ib(default='None')
expectation = attr.ib(factory=lambda: {
'basic': [{'name': '$result'}],
})
@property
def path(self):
"""Returns the repo-relative path to the recipe file."""
return self.base_path + '.py'
@property
def expect_path(self):
"""Returns the repo-relative path to the recipe's expectation dir."""
return self.base_path + '.expected'
@contextlib.contextmanager
def write_recipe(self, name_or_module, name=None):
"""Context manager for writing a recipe to the disk in this testing repo.
Overwrites any existing recipe. This can be called like:
write_recipe('recipe_name')
write_recipe('module_name', 'recipe_name')
If you need to write a malformed recipe, or one with additional top-level
functions, please use `write_file` to directly write the entire recipe file.
Usage:
with repo.write_recipe('my_recipe') as recipe:
recipe.DEPS.append('my_module')
recipe.RunSteps.write('''
api.step('stepname', ['echo', 'something'])
api.my_module.cool_function()
''')
Args:
* name_or_module (str) - If the only argument, this is the name of the
recipe to write. If provided in tandem with `name`, then this is the
module to write the recipe under.
* name (str|None) - If provided, this is the name of the recipe, and
`name_or_module` is the name of the recipe module.
Yields a WriteableRecipe object. You may manipulate the DEPS, GenTests and
RunSteps members to craft the recipe you want for your test. See
WriteableRecipe for details.
"""
if name:
base_path = os.path.join('recipe_modules', name_or_module, name)
else:
base_path = os.path.join('recipes', name_or_module)
recipe = self.WriteableRecipe(base_path)
yield recipe
dump = self.fake_recipe_deps._ambient_toplevel_code_dump()
with self.write_file(base_path + '.py') as buf:
buf.write(textwrap.dedent('''
{imports}
{ambient_toplevel_code}
DEPS = {DEPS!r}
PROPERTIES = {PROPERTIES}
ENV_PROPERTIES = {ENV_PROPERTIES}
def RunSteps({RunSteps_args}):
{RunSteps}
def GenTests(api):
{GenTests}
''').format(
imports='\n'.join(recipe.imports),
ambient_toplevel_code=dump,
DEPS=recipe.DEPS,
PROPERTIES=recipe.PROPERTIES,
ENV_PROPERTIES=recipe.ENV_PROPERTIES,
RunSteps_args=', '.join(recipe.RunSteps_args),
RunSteps=_get_suite(recipe.RunSteps, 'pass'),
GenTests=_get_suite(
recipe.GenTests, "yield api.test('basic')")))
for test_name, expectation in recipe.expectation.items():
test_name = filesystem_safe(test_name)
expect_path = os.path.join(base_path + '.expected', test_name + '.json')
with self.write_file(expect_path) as buf:
json.dump(expectation, buf, indent=2)
self.recipes_py('doc')
@attr.s(slots=True)
class WriteableModule:
"""Yielded from the `write_module` method. Used to generate a recipe module
on disk.
This contains three string buffers which you may write to, `api`, `test_api`
and `config`. All buffers will be dedent'd as described in `write_file`.
* `api` represents the body of the {modname.title}Api class inside of
`api.py`. If not written to, then the body of this class will be `pass`.
* `test_api` represents the body of the {modname.title}TestApi class
inside of `test_api.py`. If not written to, then the body of this class
will be `pass`.
* `config` represents the body of `config.py`. If not written to then
`config.py` will not be written to disk.
The imports attribute is a list of 'import' lines you want at the top of the
module files. These lines should be an entire valid python import line (i.e
starting with `import` or `from`).
The DEPS attribute corresponds exactly to the recipe module's DEPS field,
and you may put a python list or dict here, depending on how you want your
recipe's DEPS to look. The DEPS comes, by default, as a list containing the
'recipe_engine/step' module. This list may be appended to, or replaced
entirely (with assignment), if you choose.
The PROPERTIES attribute is a Python expression to assign to the module's
PROPERTIES field.
The DISABLE_STRICT_COVERAGE maps directly to the same-named option in
`__init__.py`.
"""
path = attr.ib() # base path of the module folder
api = attr.ib(factory=StringIO)
test_api = attr.ib(factory=StringIO)
config = attr.ib(factory=StringIO)
imports = attr.ib(factory=list)
DEPS = attr.ib(factory=lambda: ['recipe_engine/step'])
PROPERTIES = attr.ib(default='{}')
GLOBAL_PROPERTIES = attr.ib(default='None')
ENV_PROPERTIES = attr.ib(default='None')
WARNINGS = attr.ib(factory=list)
DISABLE_STRICT_COVERAGE = attr.ib(default=False)
@contextlib.contextmanager
def write_module(self, mod_name):
"""Context manager for writing a recipe module to the disk in this testing
repo.
Overwrites any existing recipe module (i.e. the existing module, if any,
will be removed).
If you need to write a malformed recipe module, or one with additional
customizations, please use this method, and then use `write_file` to
overwrite whichever files need to be adjusted.
Args:
* mod_name (str) - The name of the module to write.
Yields a WriteableModule object. You may manipulate its various fields to
craft the recipe module you want for your test. See WriteableModule for
details.
"""
base = os.path.join('recipe_modules', mod_name)
mod = self.WriteableModule(base)
yield mod
if self.exists(base):
shutil.rmtree(os.path.join(self.path, base))
with self.write_file(os.path.join(base, '__init__.py')) as buf:
buf.write('''
{imports}
DEPS = {DEPS!r}
WARNINGS = {WARNINGS!r}
DISABLE_STRICT_COVERAGE = {DISABLE_STRICT_COVERAGE!r}
PROPERTIES = {PROPERTIES}
GLOBAL_PROPERTIES = {GLOBAL_PROPERTIES}
ENV_PROPERTIES = {ENV_PROPERTIES}
'''.format(
imports='\n'.join(mod.imports),
DEPS=mod.DEPS,
WARNINGS = mod.WARNINGS,
DISABLE_STRICT_COVERAGE=mod.DISABLE_STRICT_COVERAGE,
PROPERTIES=mod.PROPERTIES,
GLOBAL_PROPERTIES=mod.GLOBAL_PROPERTIES,
ENV_PROPERTIES=mod.ENV_PROPERTIES,
))
dump = self.fake_recipe_deps._ambient_toplevel_code_dump()
with self.write_file(os.path.join(base, 'api.py')) as buf:
buf.write(textwrap.dedent('''
from recipe_engine.recipe_api import RecipeApi
{imports}
{ambient_toplevel_code}
class {mod_name}Api(RecipeApi):
{api}
''').format(
imports='\n'.join(mod.imports),
ambient_toplevel_code=dump,
mod_name=mod_name.title().replace(' ', ''),
api=_get_suite(mod.api, 'pass')))
with self.write_file(os.path.join(base, 'test_api.py')) as buf:
buf.write(textwrap.dedent('''
from recipe_engine.recipe_test_api import RecipeTestApi
{imports}
{ambient_toplevel_code}
class {mod_name}Api(RecipeTestApi):
{test_api}
''').format(
imports='\n'.join(mod.imports),
ambient_toplevel_code=dump,
mod_name=mod_name.title().replace(' ', ''),
test_api=_get_suite(mod.test_api, 'pass')))
config_body = _get_suite(mod.config, '', indent='')
if config_body:
with self.write_file(os.path.join(base, 'config.py')) as buf:
buf.write(textwrap.dedent('''
from recipe_engine.config import ConfigGroup, ConfigList, Dict, List
from recipe_engine.config import Set, Single, config_item_context
{imports}
{config_body}
''').format(
imports='\n'.join(mod.imports),
config_body=config_body,
))
def add_dep(self, *depnames):
"""Adds new repo-level dependencies to this repo.
The recipes_cfg_pb2 file will be updated to contain all the new entries;
they'll point to the current HEAD version of the repos.
Args:
* depnames (List[str]) - The names of the other repos to depend on. These
must already have been created with `RecipeDeps.add_repo`.
"""
with self.edit_recipes_cfg_pb2() as pkg_pb:
for depname in depnames:
dep_repo = self.fake_recipe_deps.repos[depname]
dep_entry = pkg_pb.deps[depname]
dep_entry.url = 'file://' + dep_repo.path
dep_entry.branch = 'refs/heads/main'
dep_entry.revision = dep_repo.backend.commit_metadata('HEAD').revision
def recipes_py(self, *args, **kwargs):
"""Runs `recipes.py` in this repo with the given args, just like a user
might run it.
Args:
* args (List[str]) - the arguments to pass to recipes.py.
Kwargs:
* env (Dict[str, str]) - Extra environment variables to set while invoking
recipes.py.
* py3 (bool) - Instruct recipe to run with python3 interpreter.
Returns (output, retcode) where 'output' is the combined stdout/stderr from
the command and retcode it's return code.
"""
env = os.environ.copy()
env.update(kwargs.pop('env', {}))
proc = subprocess.Popen(
('python3', 'recipes.py') + args,
cwd=self.path,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
env=env,
text=True)
output, _ = proc.communicate()
return output, proc.returncode
class TestCommitMetadata(CommitMetadata):
"""Commit metadata for a git commit.
Identical to `fetch.CommitMetadata`, except that it has a helper function
to make writing autoroller tests less tedious.
"""
def as_roll_info(self):
"""Returns a dict of author_email, message_lines and revision which
is JSON serializable.
In particular, message_lines is a list instead of a tuple.
"""
return {
'author_email': self.author_email,
'message_lines': list(self.message_lines),
'revision': self.revision,
}
def commit(self, msg,
author_name='Phinley Pfeiffer',
author_email='ph.pf@example.com'):
"""Adds all files in the repo, then commits it with the given message.
Returns a TestCommitMetadata object describing the commit we just created.
"""
# pylint: disable=protected-access
self.backend._git('add', '.')
self.backend._git(
'-c', 'user.name='+author_name,
'-c', 'user.email='+author_email,
'commit', '--allow-empty', '-m', msg)
return self.TestCommitMetadata(
**self.backend.commit_metadata('HEAD')._asdict())
@attr.s
class FakeRecipeDeps:
"""FakeRecipeDeps is a heavyweight testing object which controls a collection
of git repos on disk.
Every FakeRecipeDeps has a 'main' repo. This is analogous to the user-facing
recipe repo that you might run recipes from (i.e. it has deps on 'upstream'
repos, including, but not limited to, the recipe engine repo).
When you create a new FakeRecipeDeps, it auto-creates this main repo with
a dependency on the recipe_engine (the repo you're reading this from).
You can add additional repos with the add_repo method, and create dependencies
between repos with the 'FakeRecipeRepo.add_dep' method.
"""
# _root will contain:
# * sources/ - All non-main repos are created here, with their name
# as a subfolder.
# * main/ - The main repo is created here
# * main/.recipe_deps - The .recipe_deps folder of the main repo.
_root = attr.ib()
# A map of repo_name -> FakeRecipeRepo
repos = attr.ib(factory=dict)
# A list of textwrap.deindent-able python snippets which will be injected into
# every recipe, module api and test_api written with this FakeRecipeDeps.
#
# This is useful to add helper methods and imports which are implicitly
# available everywhere in the test.
ambient_toplevel_code = attr.ib(factory=list)
def _ambient_toplevel_code_dump(self):
return '\n'.join(map(textwrap.dedent, self.ambient_toplevel_code))
ENGINE_REVISION = None
@classmethod
def _get_engine_revision(cls):
if not cls.ENGINE_REVISION:
if subprocess.call(['git', 'diff-index', '--quiet', 'HEAD', '--']):
print('*' * 6, file=REAL_STDERR)
print(textwrap.dedent('''
WARNING: Tests may rely on current recipe engine repo, but you have
un-committed changes. If you see unexpected behavior in the tests please
try committing your changes to the engine repo first and then running
the tests again.
''').lstrip(), end=' ', file=REAL_STDERR)
print('*' * 6, file=REAL_STDERR)
print(file=REAL_STDERR)
REAL_STDERR.flush()
cls.ENGINE_REVISION = subprocess.check_output(
['git', 'rev-parse', 'HEAD'], text=True).strip()
return cls.ENGINE_REVISION
def _create_repo(self, name, path):
"""Creates a recipe repo with the given name at the given path.
This generates an `infra/config/recipes.cfg` with a file:// dependency on
the current recipe engine, repo_name set to $name.
Returns the commit of the created repo.
"""
assert isinstance(name, str)
assert name not in self.repos, (
'duplicate repo_name: %r' % (name,))
os.makedirs(path)
subprocess.check_call(
['git', 'init', '-b', 'main'], cwd=path, stdout=DEVNULL)
cfg_path = os.path.join(path, RECIPES_CFG_LOCATION_REL)
os.makedirs(os.path.dirname(cfg_path))
with open(cfg_path, 'w') as fil:
json.dump(
{
'api_version': 2,
'repo_name': name,
'deps': {
'recipe_engine': {
'url': 'file://' + ROOT_DIR,
'branch': 'HEAD',
'revision': self._get_engine_revision(),
},
},
}, fil)
readme_path = os.path.join(path, 'README.recipes.md')
readme_str = ('<!--- AUTOGENERATED BY `./recipes.py test train` -->\n'
'# Repo documentation for [main]()\n## Table of Contents')
with open(readme_path, 'w') as f:
f.write(readme_str)
shutil.copy(os.path.join(ROOT_DIR, 'recipes.py'), path)
subprocess.check_call(['git', 'add', '.'], cwd=path, stdout=DEVNULL)
subprocess.check_call(['git', 'commit', '-m', 'init '+name], cwd=path,
stdout=DEVNULL)
self.repos[name] = FakeRecipeRepo(self, name, path)
return subprocess.check_output(
['git', 'rev-parse', 'HEAD'], cwd=path).strip()
def __attrs_post_init__(self):
"""Makes a new RecipeDeps temp folder on disk with a single (main) repo
called 'main' which has no dependencies."""
self._create_repo('main', os.path.join(self._root, 'main'))
@property
def recipe_deps_path(self):
"""Returns the absolute path to the `.recipe_deps` folder of the main repo.
"""
return os.path.join(self.main_repo.path, '.recipe_deps')
def add_repo(self, name, detached=False):
"""Adds a new repo to the RecipeDeps.
This is created in `{FakeRecipeDeps.recipe_deps_path}/{name}`.
This adds a dependency of the main repo onto this new repo, AND COMMITS THE
CHANGE TO recipes.cfg.
Returns created FakeRecipeRepo.
"""
assert isinstance(name, str)
self._create_repo(name, os.path.join(self._root, 'sources', name))
if not detached:
self.main_repo.add_dep(name)
self.main_repo.commit('add dep on ' + name)
return self.repos[name]
@property
def main_repo(self):
"""Returns the main FakeRecipeRepo for this FakeRecipeDeps."""
return self.repos['main']