Statistics
| Branch: | Tag: | Revision:

root / snf-tools / synnefo_tools / burnin / common.py @ 0c1833c8

History | View | Annotate | Download (7.7 kB)

1
# Copyright 2013 GRNET S.A. All rights reserved.
2
#
3
# Redistribution and use in source and binary forms, with or
4
# without modification, are permitted provided that the following
5
# conditions are met:
6
#
7
#   1. Redistributions of source code must retain the above
8
#      copyright notice, this list of conditions and the following
9
#      disclaimer.
10
#
11
#   2. Redistributions in binary form must reproduce the above
12
#      copyright notice, this list of conditions and the following
13
#      disclaimer in the documentation and/or other materials
14
#      provided with the distribution.
15
#
16
# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
17
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
18
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
19
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
20
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
21
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
22
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
23
# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
24
# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
25
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
26
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
27
# POSSIBILITY OF SUCH DAMAGE.
28
#
29
# The views and conclusions contained in the software and
30
# documentation are those of the authors and should not be
31
# interpreted as representing official policies, either expressed
32
# or implied, of GRNET S.A.
33

    
34
"""
35
Common utils for burnin tests
36

37
"""
38

    
39
import sys
40
import traceback
41
# Use backported unittest functionality if Python < 2.7
42
try:
43
    import unittest2 as unittest
44
except ImportError:
45
    if sys.version_info < (2, 7):
46
        raise Exception("The unittest2 package is required for Python < 2.7")
47
    import unittest
48

    
49
from kamaki.clients.astakos import AstakosClient
50
from kamaki.clients.compute import ComputeClient
51

    
52
from synnefo_tools.burnin.logger import Log
53

    
54

    
55
# --------------------------------------------------------------------
56
# Global variables
57
logger = None  # Invalid constant name. pylint: disable-msg=C0103
58
SNF_TEST_PREFIX = "snf-test-"
59
CONNECTION_RETRY_LIMIT = 2
60

    
61

    
62
# --------------------------------------------------------------------
63
# BurninTestResult class
64
class BurninTestResult(unittest.TestResult):
65
    """Modify the TextTestResult class"""
66
    def __init__(self):
67
        super(BurninTestResult, self).__init__()
68

    
69
        # Test parameters
70
        self.failfast = True
71

    
72
    def startTest(self, test):  # noqa
73
        """Called when the test case test is about to be run"""
74
        super(BurninTestResult, self).startTest(test)
75
        # Access to a protected member. pylint: disable-msg=W0212
76
        logger.log(test.__class__.__name__, test._testMethodDoc)
77

    
78
    # Method could be a function. pylint: disable-msg=R0201
79
    def _test_failed(self, test, err):
80
        """Test failed"""
81
        # Access to a protected member. pylint: disable-msg=W0212
82
        err_msg = test._testMethodDoc + "... failed."
83
        logger.error(test.__class__.__name__, err_msg)
84
        (err_type, err_value, err_trace) = err
85
        trcback = traceback.format_exception(err_type, err_value, err_trace)
86
        logger.info(test.__class__.__name__, trcback)
87

    
88
    def addError(self, test, err):  # noqa
89
        """Called when the test case test raises an unexpected exception"""
90
        super(BurninTestResult, self).addError(test, err)
91
        self._test_failed(test, err)
92

    
93
    def addFailure(self, test, err):  # noqa
94
        """Called when the test case test signals a failure"""
95
        super(BurninTestResult, self).addFailure(test, err)
96
        self._test_failed(test, err)
97

    
98

    
99
# --------------------------------------------------------------------
100
# BurninTests class
101
# Too few public methods (0/2). pylint: disable-msg=R0903
102
class Clients(object):
103
    """Our kamaki clients"""
104
    auth_url = None
105
    token = None
106

    
107
    astakos = None
108
    retry = CONNECTION_RETRY_LIMIT
109

    
110
    compute = None
111
    compute_url = None
112

    
113

    
114
# Too many public methods (45/20). pylint: disable-msg=R0904
115
class BurninTests(unittest.TestCase):
116
    """Common class that all burnin tests should implement"""
117
    clients = Clients()
118
    opts = None
119

    
120
    @classmethod
121
    def setUpClass(cls):  # noqa
122
        """Initialize BurninTests"""
123
        cls.suite_name = cls.__name__
124
        logger.testsuite_start(cls.suite_name)
125

    
126
        # Set test parameters
127
        cls.longMessage = True
128

    
129
    def test_clients_setup(self):
130
        """Initializing astakos/cyclades/pithos clients"""
131
        # Update class attributes
132
        self.info("Astakos auth url is %s", self.clients.auth_url)
133
        self.clients.astakos = AstakosClient(
134
            self.clients.auth_url, self.clients.token)
135
        self.clients.astakos.CONNECTION_RETRY_LIMIT = self.clients.retry
136

    
137
        self.clients.compute_url = \
138
            self.clients.astakos.get_service_endpoints('compute')['publicURL']
139
        self.info("Cyclades url is %s", self.clients.compute_url)
140
        self.clients.compute = ComputeClient(
141
            self.clients.compute_url, self.clients.token)
142
        self.clients.compute.CONNECTION_RETRY_LIMIT = self.clients.retry
143

    
144
    def log(self, msg, *args):
145
        """Pass the section value to logger"""
146
        logger.log(self.suite_name, msg, *args)
147

    
148
    def info(self, msg, *args):
149
        """Pass the section value to logger"""
150
        logger.info(self.suite_name, msg, *args)
151

    
152
    def debug(self, msg, *args):
153
        """Pass the section value to logger"""
154
        logger.debug(self.suite_name, msg, *args)
155

    
156
    def warning(self, msg, *args):
157
        """Pass the section value to logger"""
158
        logger.warning(self.suite_name, msg, *args)
159

    
160
    def error(self, msg, *args):
161
        """Pass the section value to logger"""
162
        logger.error(self.suite_name, msg, *args)
163

    
164

    
165
# --------------------------------------------------------------------
166
# Initialize Burnin
167
def initialize(opts, testsuites):
168
    """Initalize burnin
169

170
    Initialize our logger and burnin state
171

172
    """
173
    # Initialize logger
174
    global logger  # Using global statement. pylint: disable-msg=C0103,W0603
175
    logger = Log(opts.log_folder, verbose=opts.verbose,
176
                 use_colors=opts.use_colors, in_parallel=False,
177
                 quiet=opts.quiet)
178

    
179
    # Initialize clients
180
    Clients.auth_url = opts.auth_url
181
    Clients.token = opts.token
182

    
183
    # Pass the rest options to BurninTests
184
    BurninTests.opts = opts
185

    
186
    # Choose tests to run
187
    if opts.tests != "all":
188
        testsuites = opts.tests
189
    if opts.exclude_tests is not None:
190
        testsuites = [tsuite for tsuite in testsuites
191
                      if tsuite not in opts.exclude_tests]
192

    
193
    return testsuites
194

    
195

    
196
# --------------------------------------------------------------------
197
# Run Burnin
198
def run(testsuites, failfast=False, final_report=False):
199
    """Run burnin testsuites"""
200
    global logger  # Using global. pylint: disable-msg=C0103,W0603,W0602
201

    
202
    success = True
203
    for tcase in testsuites:
204
        tsuite = unittest.TestLoader().loadTestsFromTestCase(tcase)
205
        results = tsuite.run(BurninTestResult())
206
        success = success and \
207
            was_successful(tcase.__name__, results.wasSuccessful())
208
        if failfast and not success:
209
            break
210

    
211
    # Are we going to print final report?
212
    if final_report:
213
        logger.print_logfile_to_stdout()
214
    # Clean up our logger
215
    del(logger)
216

    
217
    # Return
218
    return 0 if success else 1
219

    
220

    
221
# --------------------------------------------------------------------
222
# Helper functions
223
def was_successful(tsuite, success):
224
    """Handle whether a testsuite was succesful or not"""
225
    if success:
226
        logger.testsuite_success(tsuite)
227
        return True
228
    else:
229
        logger.testsuite_failure(tsuite)
230
        return False