« back to all changes in this revision
Viewing changes to tests/lib/test_mgmt/test_management.py
- browse files at revision 2088.9.1
- compare with another revision
- compare with revision 928.1.4
- download diff
- download tarball
- view history from revision 2088.9.1
- Committer: patrick crews
- Date: 2011-01-15 21:27:41 UTC
- mto: (2119.2.1 drizzle)
- mto: This revision was merged to the branch mainline in revision 2121.
- Revision ID: gleebix@gmail.com-20110115212741-htz3af0cib4fwdlv
Updated tree so that test-run.pl and test-run.py may live together in peace for a time
- files added:
- tests/lib/__init__.py
- tests/lib/drizzle_test_run
- tests/lib/server_mgmt
- tests/lib/sys_mgmt
- tests/lib/test_mgmt
- files modified:
- .bzrignore
added
removed
tests/lib/test_mgmt/test_management.py
1
#! /usr/bin/python
2
# -*- mode: c; c-basic-offset: 2; indent-tabs-mode: nil; -*-
3
# vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
4
#
5
# Copyright (C) 2010 Patrick Crews
6
#
7
""" dtr_test_management:
8
code related to the gathering / analysis / management of
9
the test cases
10
ie - collecting the list of tests in each suite, then
11
gathering additional, relevant information for the test-runner's dtr
12
mode. (traditional diff-based testing)
13
14
"""
15
16
# imports
17
import os
18
import re
19
import sys
20
import thread
21
22
class testManager:
23
"""Deals with scanning test directories, gathering test cases, and
24
collecting per-test information (opt files, etc) for use by the
25
test-runner
26
27
"""
28
29
def __init__( self, verbose, debug, engine, dotest, skiptest
30
, suitelist, suitepaths, system_manager, test_cases):
31
32
self.system_manager = system_manager
33
self.logging = system_manager.logging
34
if verbose:
35
self.logging.verbose("Initializing test manager...")
36
37
if dotest:
38
dotest = dotest.strip()
39
if skiptest:
40
skiptest = skiptest.strip()
41
self.skip_keys = [ 'system_manager'
42
]
43
self.test_list = []
44
self.total_test_count = 0
45
self.executed_tests = {} # We have a hash of 'status':[test_name..]
46
self.executing_tests = {}
47
self.verbose = verbose
48
self.debug = debug
49
self.engine = engine
50
self.dotest = dotest
51
self.skiptest = skiptest
52
self.suitelist = suitelist
53
54
self.code_tree = self.system_manager.code_tree
55
self.suitepaths = suitepaths + self.code_tree.suite_paths
56
self.testdir = self.code_tree.testdir
57
self.desired_tests = test_cases
58
self.mutex = thread.allocate_lock()
59
60
if self.debug:
61
self.logging.debug(self)
62
63
def add_test(self, new_test_case):
64
""" Add a new testCase to our self.test_list """
65
if self.debug: pass
66
self.test_list.append(new_test_case)
67
68
def gather_tests(self):
69
self.logging.info("Processing test suites...")
70
# BEGIN terrible hack to accomodate the fact that
71
# our 'main' suite is also our testdir : /
72
if self.suitelist is None:
73
self.suitepaths = [self.testdir]
74
self.suitelist = ['main']
75
# END horrible hack
76
for suite in self.suitelist:
77
suite_path = self.find_suite_path(suite)
78
if suite_path:
79
self.process_suite(suite_path)
80
else:
81
self.logging.error("Could not find suite: %s in any of paths: %s" %(suite, ", ".join(self.suitepaths)))
82
self.process_gathered_tests()
83
84
def process_gathered_tests(self):
85
""" We do some post-gathering analysis and whatnot
86
Report an error if there were desired_tests but no tests
87
were found. Otherwise just report what we found
88
89
"""
90
91
if self.desired_tests and not self.test_list:
92
# We wanted tests, but found none
93
# Probably need to make this smarter at some point
94
# To maybe make sure that we found all of the desired tests...
95
# However, this is a start / placeholder code
96
self.logging.error("Unable to locate any of the desired tests: %s" %(" ,".join(self.desired_tests)))
97
self.total_test_count = len(self.test_list)
98
self.logging.info("Found %d test(s) for execution" %(self.total_test_count))
99
100
if self.debug:
101
self.logging.debug("Found tests:")
102
self.logging.debug("%s" %(self.print_test_list()))
103
104
def find_suite_path(self, suitename):
105
""" We have a suitename, we need to locate the path to
106
the juicy suitedir in one of our suitepaths.
107
108
Theoretically, we could have multiple matches, but
109
such things should never be allowed, so we don't
110
code for it. We return the first match.
111
112
testdir can either be suitepath/suitename or
113
suitepath/suitename/tests. We test and return the
114
existing path. Return None if no match found
115
116
"""
117
# BEGIN horrible hack to accomodate bad location of main suite
118
if self.suitepaths == [self.testdir]:
119
# We treat this as the 'main' suite
120
return self.testdir
121
# END horrible hack
122
for suitepath in self.suitepaths:
123
suite_path = self.system_manager.find_path([ os.path.join(suitepath,suitename,'tests'),
124
os.path.join(suitepath,suitename) ], required = 0 )
125
if suite_path:
126
return suite_path
127
return suite_path
128
129
def process_suite(self,suite_dir):
130
"""Process a test suite.
131
This includes searching for tests in test_list and only
132
working with the named tests (all tests in suite is the default)
133
Further processing includes reading the disabled.def file
134
to know which tests to skip, processing the suite.opt file,
135
and processing the individual test cases for data relevant
136
to the rest of the test-runner
137
138
"""
139
if self.verbose:
140
self.logging.verbose("Processing suite: %s" %(suite))
141
142
def has_tests(self):
143
"""Return 1 if we have tests in our testlist, 0 otherwise"""
144
145
return len(self.test_list)
146
147
def get_testCase(self, requester):
148
"""return a testCase """
149
150
test_case = None
151
if self.has_tests():
152
test_case = self.test_list.pop(0)
153
self.record_test_executor(requester, test_case.fullname)
154
return test_case
155
156
def record_test_executor(self, requester, test_name):
157
""" We record the test case and executor name as this could be useful
158
We don't *know* this is needed, but we can always change this
159
later
160
161
"""
162
163
self.executing_tests[test_name] = requester
164
165
def record_test_result(self, test_case, test_status, output):
166
""" Accept the results of an executed testCase for further
167
processing.
168
169
"""
170
if test_status not in self.executed_tests:
171
self.executed_tests[test_status] = [test_case]
172
else:
173
self.executed_tests[test_status].append(test_case)
174
# report
175
self.logging.test_report(test_case.fullname, test_status, output)
176
177
178
def print_test_list(self):
179
test_names = []
180
for test in self.test_list:
181
test_names.append(test.fullname)
182
return "[ %s ]" %(", ".join(test_names))
183
184
def statistical_report(self):
185
""" Report out various testing statistics:
186
Failed/Passed %success
187
list of failed test cases
188
189
"""
190
191
self.logging.write_thick_line()
192
self.logging.info("Test execution complete")
193
self.logging.info("Summary report:")
194
self.report_executed_tests()
195
self.report_failing_tests()
196
197
def get_executed_test_count(self):
198
""" Return how many tests were executed """
199
total_count = 0
200
for test_list in self.executed_tests.values():
201
total_count = total_count + len(test_list)
202
return total_count
203
204
def report_executed_tests(self):
205
""" Report out tests by status """
206
total_executed_count = self.get_executed_test_count()
207
executed_ratio = (float(total_executed_count)/float(self.total_test_count))
208
executed_percent = executed_ratio * 100
209
self.logging.info("Executed %s/%s test cases, %.2f percent" %( total_executed_count
210
, self.total_test_count
211
, executed_percent))
212
213
for test_status in self.executed_tests.keys():
214
status_count = self.get_count_by_status(test_status)
215
test_percent = (float(status_count)/float(total_executed_count))*100
216
self.logging.info("STATUS: %s, %d/%d test cases, %.2f percent executed, %.2f percent found" %( test_status.upper()
217
, status_count
218
, total_executed_count
219
, test_percent
220
, test_percent*executed_ratio ))
221
222
223
def report_failing_tests(self):
224
failing_tests = []
225
if 'fail' in self.executed_tests:
226
for testcase in self.executed_tests['fail']:
227
failing_tests.append(testcase.fullname)
228
self.logging.info("Failing tests: %s" %(", ".join(failing_tests)))
229
230
def get_count_by_status(self, test_status):
231
""" Return how many tests are in a given test_status """
232
if test_status in self.executed_tests:
233
return len(self.executed_tests[test_status])
234
else:
235
return 0
236
Loggerhead is a web-based interface for Breezy
Version: 2.0.1