summaryrefslogtreecommitdiff
path: root/buildscripts/smoke/executor.py
blob: bae5f10b77fdc1263859492e6a8ae9c76be85cc5 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
"""
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()