mirror of
https://github.com/mongodb/mongo.git
synced 2024-12-01 09:32:32 +01:00
fac5571f67
Includes: - Smoke.py options for filtering by jstest tags - New resmoke.py test utility in v0 alpha - Lots of example resmoke configurations - SCons integration for unittests - Sample tagged jstests
138 lines
4.3 KiB
Python
138 lines
4.3 KiB
Python
"""
|
|
Module which allows execution of a suite of tests with customizable fixtures and testers.
|
|
|
|
Fixtures are set up per-suite, and register APIs per-test. Generally this is custom setup code.
|
|
|
|
Testers encapsulate test code of different types in a standard, UnitTest object.
|
|
"""
|
|
|
|
import inspect
|
|
import logging
|
|
import traceback
|
|
import unittest
|
|
|
|
import fixtures
|
|
import testers
|
|
|
|
|
|
def exec_suite(suite, logger, **kwargs):
|
|
"""Main entry point, executes a suite of tests with the given logger and executor arguments."""
|
|
|
|
suite_executor = TestSuiteExecutor(logger, **kwargs)
|
|
|
|
try:
|
|
successful_setup = suite_executor.setup_suite(suite)
|
|
|
|
if successful_setup:
|
|
suite_executor.exec_suite()
|
|
|
|
finally:
|
|
suite_executor.teardown_suite(suite)
|
|
|
|
|
|
def instantiate(class_name, *args, **kwargs):
|
|
"""Helper to dynamically instantiate a class from a name."""
|
|
split_name = class_name.split(".")
|
|
module_name = split_name[0]
|
|
class_name = ".".join(split_name[1:])
|
|
|
|
module = __import__(module_name)
|
|
class_ = getattr(module, class_name)
|
|
return class_(*args, **kwargs)
|
|
|
|
|
|
class TestSuiteExecutor(object):
|
|
|
|
"""The state of execution of a suite of tests.
|
|
|
|
The job of the TestSuiteExecutor is to convert the incoming fixtures and tester configuration
|
|
into Fixture and TestCase objects, then execute them using the standard unittest framework.
|
|
|
|
"""
|
|
|
|
def __init__(self, logger, testers={}, fixtures={}, fail_fast=False, **kwargs):
|
|
|
|
self.logger = logger
|
|
self.testers = testers
|
|
self.fixtures = fixtures
|
|
self.fail_fast = fail_fast
|
|
|
|
if len(kwargs) > 0:
|
|
raise optparse.OptionValueError("Unrecognized options for executor: %s" % kwargs)
|
|
|
|
for fixture_name in self.fixtures:
|
|
self.fixtures[fixture_name] = \
|
|
self.build_fixture(fixture_name, **self.fixtures[fixture_name])
|
|
|
|
def build_fixture(self, fixture_name, fixture_class=None, fixture_logger=None,
|
|
**fixture_kwargs):
|
|
|
|
if not fixture_class:
|
|
fixture_class = fixtures.DEFAULT_FIXTURE_CLASSES[fixture_name]
|
|
|
|
if not fixture_logger:
|
|
fixture_logger = self.logger.getChild("fixtures.%s" % fixture_name)
|
|
else:
|
|
fixture_logger = logging.getLogger(fixture_logger)
|
|
|
|
return instantiate(fixture_class, fixture_logger, **fixture_kwargs)
|
|
|
|
def build_tester(self, test):
|
|
|
|
tester_type = test.test_type
|
|
|
|
def extract_tester_args(tester_class=None, tester_logger=None, **tester_kwargs):
|
|
return tester_class, tester_logger, tester_kwargs
|
|
|
|
tester_class, tester_logger, tester_kwargs = \
|
|
extract_tester_args(
|
|
**(self.testers[tester_type] if tester_type in self.testers else {}))
|
|
|
|
if not tester_class:
|
|
tester_class = testers.DEFAULT_TESTER_CLASSES[tester_type]
|
|
|
|
if not tester_logger:
|
|
tester_logger = self.logger.getChild("testers.%s.%s" % (tester_type, test.uri))
|
|
else:
|
|
tester_logger = logging.getLogger(tester_logger)
|
|
|
|
test_apis = []
|
|
for fixture_name, fixture in self.fixtures.items():
|
|
test_api = fixture.build_api(tester_type, tester_logger)
|
|
if test_api:
|
|
test_apis.append(test_api)
|
|
|
|
return instantiate(tester_class, test, test_apis, tester_logger, **tester_kwargs)
|
|
|
|
def setup_suite(self, suite):
|
|
|
|
self.setup_fixtures = {}
|
|
for fixture_name, fixture in self.fixtures.items():
|
|
try:
|
|
fixture.setup()
|
|
self.setup_fixtures[fixture_name] = fixture
|
|
except:
|
|
print "Suite setup failed: %s" % fixture_name
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
self.unittest_suite = unittest.TestSuite()
|
|
for test in suite:
|
|
self.unittest_suite.addTest(self.build_tester(test))
|
|
|
|
return True
|
|
|
|
def exec_suite(self):
|
|
# TODO: More stuff here?
|
|
unittest.TextTestRunner(
|
|
verbosity=2, failfast=self.fail_fast).run(self.unittest_suite)
|
|
|
|
def teardown_suite(self, suite):
|
|
|
|
for fixture_name, fixture in self.setup_fixtures.items():
|
|
try:
|
|
fixture.teardown()
|
|
except:
|
|
print "Suite teardown failed: %s" % fixture_name
|
|
traceback.print_exc()
|