2
# -*- mode: python; indent-tabs-mode: nil; -*-
3
# vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
5
# Copyright (C) 2011 Patrick Crews
7
## This program is free software; you can redistribute it and/or modify
8
# it under the terms of the GNU General Public License as published by
9
# the Free Software Foundation; either version 2 of the License, or
10
# (at your option) any later version.
12
# This program is distributed in the hope that it will be useful,
13
# but WITHOUT ANY WARRANTY; without even the implied warranty of
14
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15
# GNU General Public License for more details.
17
# You should have received a copy of the GNU General Public License
18
# along with this program; if not, write to the Free Software
19
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
21
""" crashme_test_management:
22
code related to the gathering / analysis / management of
24
ie - collecting the list of tests in each suite, then
25
gathering additional, relevant information for crashme mode
35
import lib.test_mgmt.test_management as test_management
40
"""Holds info on a single crashme test
47
, server_requirements=[[]]
52
, suitename = 'native_tests'
54
self.system_manager = system_manager
55
self.logging = self.system_manager.logging
56
self.skip_keys = ['system_manager','logging']
58
self.fullname = fullname
59
self.suitename = suitename
61
self.comment = comment
62
self.server_requirements = server_requirements
63
self.cnf_path = cnf_path
64
self.server_requests = request_dict
65
self.test_path = test_path
67
self.system_manager.logging.debug_class(self)
70
if self.skip_flag or self.disable:
79
class testManager(test_management.testManager):
80
"""Deals with scanning test directories, gathering test cases, and
81
collecting per-test information (opt files, etc) for use by the
86
def __init__( self, variables, system_manager):
87
super(testManager, self).__init__( variables, system_manager)
88
server_type = variables['defaultservertype']
89
if server_type == 'mysql' or server_type =='galera':
90
server_type = 'percona'
91
self.suitepaths = [os.path.join(self.testdir,'%s_tests' %(server_type))]
92
if variables['suitelist'] is None:
93
self.suitelist = ['main']
95
self.suitelist = variables['suitelist']
97
def process_suite(self,suite_dir):
98
"""Process a test suite.
99
Look for tests, which are nice clean python unittest files
103
# We know this based on how we organize native test conf files
104
suite_name = os.path.basename(suite_dir)
105
self.system_manager.logging.verbose("Processing suite: %s" %(suite_name))
106
testlist = [os.path.join(suite_dir,test_file) for test_file in sorted(os.listdir(suite_dir)) if test_file.endswith('_test.py')]
108
# Search for specific test names
109
if self.desired_tests: # We have specific, named tests we want from the suite(s)
111
for test in self.desired_tests:
112
if test.endswith('.py'):
116
test = os.path.join(suite_dir,test)
118
tests_to_use.append(test)
119
testlist = tests_to_use
120
for test_case in testlist:
121
self.add_test(self.process_test_file( suite_name
125
def get_server_reqs(self, module_file):
126
""" Code to handle extraction of server_requests & requirements
127
from unittest test modules
131
module_name = os.path.basename(module_file).replace('.py','')
132
my_module = imp.load_source(module_name, module_file)
133
server_requirements = None
134
server_requests = None
136
server_requirements = my_module.server_requirements
137
except AttributeError, NameError: pass
139
server_requests = my_module.server_requests
140
except AttributeError, NameError: pass
141
return server_requirements, server_requests
144
def process_test_file(self, suite_name, testfile):
145
""" We convert the info in a testfile into a testCase object """
147
# test_name = filename - .py...simpler
148
test_name = os.path.basename(testfile).replace('.py','')
150
server_requirements, server_requests = self.get_server_reqs(testfile)
152
return testCase( self.system_manager
154
, fullname = "%s.%s" %(suite_name, test_name)
155
, server_requirements = server_requirements
157
, request_dict = server_requests
158
, test_path = testfile
159
, debug = self.debug )
162
def record_test_result(self, test_case, test_status, output, exec_time):
163
""" Accept the results of an executed testCase for further
167
if test_status not in self.executed_tests:
168
self.executed_tests[test_status] = [test_case]
170
self.executed_tests[test_status].append(test_case)
172
self.logging.test_report( test_case.fullname, test_status
173
, str(exec_time), output
174
, report_output= True)