Merge "Adds new Brew file runner."

This commit is contained in:
Jenkins 2016-05-10 21:03:45 +00:00 committed by Gerrit Code Review
commit 2b0ec92892
9 changed files with 658 additions and 1 deletions

View File

View File

@ -0,0 +1,94 @@
import argparse
import sys
from cafe.drivers.unittest.arguments import ConfigAction, VerboseAction
from cafe.drivers.base import print_exception, get_error
class ArgumentParser(argparse.ArgumentParser):
"""
Parses all arguments.
"""
def __init__(self):
desc = "Open Common Automation Framework Engine BrewFile Runner"
usage_string = """
cafe-brew <config> <runfile(s)>...
[--failfast]
[--dry-run]
[--parallel=(class|test)]
[--result=(json|xml)] [--result-directory=RESULT_DIRECTORY]
[--verbose=VERBOSE]
[--exit-on-error]
[--workers=NUM]
[--help]
"""
super(ArgumentParser, self).__init__(
usage=usage_string, description=desc)
self.prog = "Argument Parser"
self.add_argument(
"config",
action=ConfigAction,
metavar="<config>",
help="test config. Looks in the .opencafe/configs directory."
"Example: bsl/uk.json")
self.add_argument(
"runfiles",
nargs="*",
default=[],
metavar="<runfiles>...",
help="A list of paths to opencafe runfiles.")
self.add_argument(
"--dry-run",
action="store_true",
help="dry run. Don't run tests just print them. Will run data"
" generators.")
self.add_argument(
"--failfast",
action="store_true",
help="fail fast")
self.add_argument(
"--verbose", "-v",
action=VerboseAction,
choices=[1, 2, 3],
default=2,
type=int,
help="Set unittest output verbosity")
self.add_argument(
"--parallel",
choices=["class", "test"],
help="Runs test in parallel by class grouping or test")
self.add_argument(
"--result",
choices=["json", "xml", "subunit"],
help="Generates a specified formatted result file")
self.add_argument(
"--result-directory",
default="./",
metavar="RESULT_DIRECTORY",
help="Directory for result file to be stored")
self.add_argument(
"--workers",
nargs="?",
default=10,
type=int,
help="Set number of workers for --parallel option")
def error(self, message):
self.print_usage(sys.stderr)
print_exception("Argument Parser", message)
exit(get_error())
def parse_args(self, *args, **kwargs):
args = super(ArgumentParser, self).parse_args(*args, **kwargs)
args.all_tags = False
return args

View File

@ -0,0 +1,348 @@
import collections
import importlib
import os
import string
import sys
import unittest
from collections import OrderedDict
from six import string_types
from six.moves import configparser
from types import ModuleType
import types
from cafe.drivers.unittest.decorators import DataDrivenClass
from cafe.common.reporting import cclogging
RESERVED_SECTION_NAMES = [
"defaults",
"cli-defaults",
]
FIXTURE_ATTR = 'fixture_class'
DATASETLIST_ATTR = 'dsl'
TESTCLASSES_ATTR = 'mixin_test_classes'
BREW_SECTION_ATTR_LIST = [FIXTURE_ATTR, DATASETLIST_ATTR, TESTCLASSES_ATTR]
REQUIRED_BREW_SECTION_ATTR_LIST = [FIXTURE_ATTR]
class RunFileNotFoundError(Exception):
pass
class RunFileSectionCollisionError(Exception):
pass
class RunFileIncompleteBrewError(Exception):
pass
class MalformedClassImportPathError(Exception):
pass
class ModuleNotImportableError(Exception):
pass
class ClassNotImportableError(Exception):
pass
class BrewMissingTestClassesError(Exception):
pass
class _ImportablePathWrapper(object):
"""
Provides convenience methods for importing module and class object
denoted by a class_import_path string.
"""
def __init__(self, class_import_path):
"""Accepts a dotted class path."""
split_path = class_import_path.rsplit(".", 1)
if len(split_path) != 2:
raise MalformedClassImportPathError(
"Path '{cip}' was malformed.\nA dotted path ending in a "
"class name was expected.".format(cip=class_import_path))
self.module_path = split_path[0]
self.class_name = split_path[1]
self._original_class_import_path = class_import_path
self._module = None
self._class = None
def __repr__(self):
s = "{ip}.{cn}".format(ip=self.module_path, cn=self.class_name)
return s
def import_module(self):
"""Import the module at import_path and return the module object"""
if self._module is None:
try:
self._module = importlib.import_module(self.module_path)
except ImportError as ie:
msg = "Could not import module from path '{p}: {e}".format(
p=self.module_path,
e=str(ie))
raise ModuleNotImportableError(msg)
return self._module
def import_class(self):
"""Import the module at import_path and extract the class_name class"""
if self._class is None:
try:
self._class = getattr(self.import_module(), self.class_name)
except AttributeError as err:
msg = (
"Could not import class '{c}' from path '{p}': {e}".format(
c=self.class_name,
p=self._original_class_import_path,
e=str(err)))
raise ClassNotImportableError(msg)
return self._class
class _Brew(object):
"""
Returns a module object containing all generated classes
"""
def __init__(
self, name, fixture_class=None, dsl=None, mixin_test_classes=None):
self.name = name
self.fixture_class = None
self.dsl = None
self.mixin_test_classes = list()
if fixture_class is not None:
self.fixture_class = _ImportablePathWrapper(fixture_class)
if dsl is not None:
self.dsl = _ImportablePathWrapper(dsl)
if mixin_test_classes is not None:
if (not isinstance(mixin_test_classes, collections.Iterable) or
isinstance(mixin_test_classes, string_types)):
raise BrewMissingTestClassesError(
"Brew was instantiated with an uniterable "
"mixin_test_classes object of type {t}".format(
t=type(mixin_test_classes)))
self.mixin_test_classes = [
_ImportablePathWrapper(tc) for tc in mixin_test_classes]
self.automodule_name = "{name}_automodule".format(name=self.name)
def __repr__(self):
return (
"\n{name}:\n\t"
"automodule: {automodule_name}\n\t"
"{fixture_attr}: {fixture}\n\t"
"{dsl_attr}: {dsl}\n\t"
"{testclasses_attr}:\n{tcs}\n".format(
fixture_attr=FIXTURE_ATTR,
dsl_attr=DATASETLIST_ATTR,
testclasses_attr=TESTCLASSES_ATTR,
name=self.name,
automodule_name=self.automodule_name,
fixture=self.fixture_class,
dsl=self.dsl,
tcs='\n'.join(["\t\t{s}{t}".format(
s=" " * 5, t=t) for t in self.mixin_test_classes])))
def _generate_module(self, module_name):
"""Create the container module for autgenerated classes."""
return types.ModuleType(
module_name,
"This module was auto-generated as a container for BrewFile-driven"
" classes generated at runtime.")
def _register_module(self, module_object):
"""Add the module to sys.modules so that it's importable elsewhere."""
sys.modules[module_object.__name__] = module_object
def _generate_test_class(self, module_name=None):
"""
Create the aggregate test class by generating a new class that inherits
from the fixture and all the test classes.
The generated class's __module__ attribute will be set to module_name
"""
# Import and append the test class objects to the bases list
bases = list()
# Make sure the fixture is first in the method resolution order
if self.fixture_class is not None:
bases.append(self.fixture_class.import_class())
for tc in self.mixin_test_classes:
bases.append(tc.import_class())
# Create the new test class from the bases list and register it as a
# member of the automodule so that it can be properly imported later
# (type requires that bases be a tuple)
class_dict = {}
if module_name:
class_dict['__module__'] = module_name
return type(
self.name, tuple(bases), class_dict)
def __call__(self):
"""Generates a module to contain the generated test fixture and
all data generated test classes.
Returns the module object"""
# Generate the automodule
automodule = self._generate_module(self.automodule_name)
# add it to sys.modules
self._register_module(automodule)
# Generate the aggregate test class
test_class = self._generate_test_class(
module_name=self.automodule_name)
if self.dsl is not None:
# Instantiate the DataDrivenClass decorator with an instance of the
# dsl_class
dsl_class = self.dsl.import_class()
# Generate final data driven class aggregate by decorating the
# aggregate test_class with the the dsl_class-driven
# DataDrivenClass decorator.
test_class = DataDrivenClass(dsl_class())(test_class)
# Add the ddtest_class to the automodule
setattr(automodule, test_class.__name__, test_class)
return automodule
class BrewFile(object):
def __init__(self, files):
"""Accepts mutiple (config-like) run files and generates a
consolidated representation of them, enforcing rules during parsing.
A BrewFile is a SafeConfigParser file, except:
The section 'cli-defaults' is special and can only be used for
defining defaults for optional command-line arguments.
(NOTE: This feature is not yet implemented)
All keys in any given section must be unique.
All section names across all files passed into BrewFile must be
unique, with the exception of 'defaults' and 'cli-defaults', which
are special and not vetted.
The section 'cli-defaults' should only appear once across all
files passed into BrewFile.
"""
self._log = cclogging.getLogger(
cclogging.get_object_namespace(self.__class__))
self.files = files
self._data = self._validate_runfiles(files)
def __repr__(self):
files = self._files_string()
brews = self._brews_string()
return "Files:\n{files}Brews:{brews}".format(files=files, brews=brews)
@property
def cli_defaults(self):
return dict(self._data.items('cli-defaults'))
def _brews_string(self):
sub_brews = "\n\t".join(
["\n\t".join(str(b).splitlines()) for b in self.iterbrews()])
return "{brews}\n".format(brews=sub_brews)
def brews_to_strings(self):
return self._brews_string().splitlines()
def _files_string(self):
return "{files}\n".format(
files='\n'.join(["{space}{file}".format(
space="\t", file=f)for f in self.files]))
def brew_list(self):
return [
s for s in self._data.sections()
if s.lower() not in RESERVED_SECTION_NAMES]
def iterbrews(self):
""" Iterates through runfile sections and yields each individual
section as a Brew object. You have to call .brew() on the individual
Brews to get them to generate a module that contains the aggregate
test class, so these should be safe to store in a list regardless of
dataset size.
"""
for s in self.brew_list():
attr_dict = dict(name=s)
for attr in BREW_SECTION_ATTR_LIST:
try:
attr_dict[attr] = self._data.get(s, attr)
except Exception:
attr_dict[attr] = None
attr_dict[TESTCLASSES_ATTR] = (attr_dict.get(
TESTCLASSES_ATTR) or "").strip().splitlines()
b = _Brew(**attr_dict)
yield b
def brew_modules(self):
"""Returns a list of generated modules each with mixin_test_classes
inside, based on the BrewFile's contents. For a large enough dataset
this could be memory intensive, so it's recommended to use iterbrews
and yield the module/mixin_test_classes to your runner one at a time.
"""
modules = list()
for brew in self.iterbrews():
module = brew()
modules.append(module)
return modules
@staticmethod
def _validate_runfiles(files):
""" Enforces the BrewFile rules on all provided files.
If all files pass validation, a SafeConfigParser is returned.
"""
# Validate the config files individually
for f in files:
# Make sure the file is actually there since config parser
# fails silently when loading non-existant files
if not os.path.isfile(f):
msg = "Could not locate file '{f}'".format(f=f)
raise RunFileNotFoundError(msg)
# TODO: Add checks for duplicate sections among multiple files
cfg = configparser.SafeConfigParser()
cfg.read(f)
# Check for incomplete sections, excluding reserved
# sections.
for section in [
s for s in cfg.sections()
if s not in RESERVED_SECTION_NAMES]:
for attr in REQUIRED_BREW_SECTION_ATTR_LIST:
try:
cfg.get(section, attr)
except configparser.NoOptionError:
msg = (
"\nSection '{sec}' in runfile '{filename}' is "
"missing the '{attr}' option".format(
filename=f, sec=s, attr=attr))
raise RunFileIncompleteBrewError(msg)
# config files are valid, return aggregate config parser object
cfg = configparser.SafeConfigParser()
cfg.read(files)
return cfg

View File

@ -0,0 +1,89 @@
import os
import time
from cafe.configurator.managers import TestEnvManager
from cafe.common.reporting import cclogging
from cafe.drivers.base import print_exception, get_error
from cafe.drivers.unittest.brew.arguments import ArgumentParser
from cafe.drivers.unittest.brew.parser import BrewFile
from cafe.drivers.unittest.runner_parallel import UnittestRunner
from cafe.drivers.unittest.suite_builder import SuiteBuilder
class BrewRunner(UnittestRunner):
"""OpenCafe BrewFile Runner"""
def __init__(self):
self.print_mug()
self.cl_args = ArgumentParser().parse_args()
self.test_env = TestEnvManager(
"", self.cl_args.config, test_repo_package_name="")
self.test_env.test_data_directory = self.test_env.test_data_directory
self.test_env.finalize()
cclogging.init_root_log_handler()
# This is where things diverge from the regular parallel runner
# Extract the runfile contents
self._log = cclogging.getLogger(
cclogging.get_object_namespace(self.__class__))
self.datagen_start = time.time()
self.run_file = BrewFile(self.cl_args.runfiles)
# TODO: Once the parallel_runner is changed to a yielding model,
# change this to yielding brews instead of generating a list
self.suites = SuiteBuilder(
testrepos=self.run_file.brew_modules(),
dry_run=self.cl_args.dry_run,
exit_on_error=True).get_suites()
self.print_configuration(self.test_env, runfile=self.run_file)
def print_configuration(self, test_env, repos=None, runfile=None):
"""Prints the config/logs/repo/data_directory"""
print("=" * 150)
print("Percolated Configuration")
print("-" * 150)
if runfile:
print("BREW FILES........:")
print("\t\t " + "\n\t\t ".join(runfile.files))
if self.cl_args.verbose == 3:
print("BREWS............:")
print "\t" + "\n\t".join(runfile.brews_to_strings())
if repos:
print("BREWING FROM: ....: {0}".format(repos[0]))
for repo in repos[1:]:
print("{0}{1}".format(" " * 20, repo))
self._log.debug(str(runfile))
print("ENGINE CONFIG FILE: {0}".format(test_env.engine_config_path))
print("TEST CONFIG FILE..: {0}".format(test_env.test_config_file_path))
print("DATA DIRECTORY....: {0}".format(test_env.test_data_directory))
print("LOG PATH..........: {0}".format(test_env.test_log_dir))
print("=" * 150)
@staticmethod
def print_mug():
"""Prints the cafe mug"""
print("""
/~~~~~~~~~~~~~~~~~~~~~~~/|
/ /######/ / |
/ /______/ / |
========================= /||
|_______________________|/ ||
| \****/ \__,,__/ ||
|===\**/ __,,__ ||
|______________\====/%____||
| ___ /~~~~\ % / |
_| |===|=== / \%_/ |
| | |###| |########| | /
|____\###/______\######/__|/
~~~~~~~~~~~~~~~~~~~~~~~~~~
=== CAFE Brewfile Runner ===""")
def entry_point():
"""setup.py script entry point"""
try:
runner = BrewRunner()
exit(runner.run())
except KeyboardInterrupt:
print_exception(
"BrewRunner", "run", "Keyboard Interrupt, exiting...")
os.killpg(0, 9)

7
examples/brewfile Normal file
View File

@ -0,0 +1,7 @@
[DoStuff_WithData]
fixture_class=testrepo.fixtures.SomeTestFixtureA
dsl=testrepo.dsl.SomeDatasetList
mixin_test_classes=
testrepo.test_mixins.TestMixinClass1
testrepo.test_mixins.TestMixinClass2
testrepo.test_mixins.TestMixinClass3

View File

@ -72,6 +72,7 @@ setup(
'console_scripts':
['cafe-runner = cafe.drivers.unittest.runner:entry_point',
'cafe-parallel = cafe.drivers.unittest.runner_parallel:entry_point',
'cafe-brew = cafe.drivers.unittest.brew.runner:entry_point',
'behave-runner = cafe.drivers.behave.runner:entry_point',
'vows-runner = cafe.drivers.pyvows.runner:entry_point',
'specter-runner = cafe.drivers.specter.runner:entry_point',

View File

@ -1,5 +1,6 @@
pip
virtualenv
tox
mock
flake8
nose
virtualenv

View File

View File

@ -0,0 +1,117 @@
import importlib
import sys
import unittest
from cafe.drivers.unittest.brew.parser import (
_ImportablePathWrapper, MalformedClassImportPathError,
ModuleNotImportableError, ClassNotImportableError, _Brew,
BrewMissingTestClassesError)
class ImportablePathWrapper_Tests(unittest.TestCase):
def test_raises_MalformedClassImportPathError_on_malformed_path(self):
self.assertRaises(
MalformedClassImportPathError, _ImportablePathWrapper,
'unittest/TestCase')
def test_raises_ClassNotImportableError_on_non_existant_class(self):
i = _ImportablePathWrapper("collections.aaaaDoesNotExistaaa")
self.assertRaises(ClassNotImportableError, i.import_class)
def test_raises_ModuleNotImportableError_on_non_existant_module(self):
i = _ImportablePathWrapper("aaaa.aaaa.aaaa")
self.assertRaises(ModuleNotImportableError, i.import_module)
def test_import_module_happy_path(self):
i = _ImportablePathWrapper("collections.OrderedDict")
m = i.import_module()
self.assertEqual(type(m.OrderedDict()).__name__, 'OrderedDict')
def test_import_class_happy_path(self):
i = _ImportablePathWrapper("collections.OrderedDict")
c = i.import_class()
self.assertEqual(type(c()).__name__, 'OrderedDict')
class Brew_Tests(unittest.TestCase):
test_fixture = "unittest.TestCase"
test_class = "collections.OrderedDict"
dsl = "cafe.drivers.unittest.datasets.DatasetList"
module_name = "FakeBrew"
def test_init_name_only(self):
try:
_Brew(self.module_name)
except:
self.fail("Unable to instantiate Brew with only a name")
def test_call_brew_initialized_with_only_a_name(self):
b = _Brew(self.module_name)
module_ = b()
new_name = "FakeBrew_automodule"
self.assertEquals(module_.__name__, new_name)
def test_init_raises_BrewMissingTestClassesError_non_iterable(self):
self.assertRaises(
BrewMissingTestClassesError, _Brew, "FakeBrew",
"collections.OrderedDict", "collections.OrderedDict", 1)
def test_init_raises_BrewMissingTestClassesError_string(self):
self.assertRaises(
BrewMissingTestClassesError, _Brew, "FakeBrew",
"collections.OrderedDict", "collections.OrderedDict", "string")
def test_init_BrewMissingTestClassesError_bool_non_iterable(self):
self.assertRaises(
BrewMissingTestClassesError, _Brew, "FakeBrew",
"collections.OrderedDict", "collections.OrderedDict", True)
def test_generate_module_correct_type(self):
b = _Brew(
self.module_name, self.test_fixture, self.dsl,
[self.test_class])
m = b._generate_module(self.module_name)
self.assertEqual(
type(m).__name__, 'module',
"_generate_module did not return a module")
def test_generate_module_correct_name(self):
b = _Brew(
self.module_name, self.test_fixture, self.dsl,
[self.test_class])
m = b._generate_module(self.module_name)
self.assertEqual(
m.__name__, self.module_name,
"_generate_module did not return a module with the correct name")
def test_module_registration(self):
b = _Brew(
self.module_name, self.test_fixture, self.dsl,
[self.test_fixture])
m = b._generate_module(self.module_name)
b._register_module(m)
self.assertIn(
self.module_name, sys.modules.keys(),
"_register_module failed to register {} in sys.modules")
def test_registered_module_is_importable(self):
b = _Brew(
self.module_name, self.test_fixture, self.dsl,
[self.test_fixture])
m = b._generate_module(self.module_name)
b._register_module(m)
try:
importlib.import_module(m.__name__)
except ImportError:
self.fail("Unable to import registered module")
def test_generate_test_class_name_and_fixture_only(self):
b = _Brew(self.module_name, fixture_class=self.test_fixture)
gclass = b._generate_test_class()
self.assertEqual(gclass.__name__, b.name)
self.assertEqual(gclass.__name__, self.module_name)
self.assertTrue(issubclass(gclass, unittest.TestCase))