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
""" sqlbench_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 sqlbench mode
33
from ConfigParser import RawConfigParser
35
import lib.test_mgmt.test_management as test_management
40
"""Holds info on a single sqlbench test
43
def __init__( self, system_manager, name=None
44
, fullname = None, server_requirements=[[]]
45
, comment=None, test_command=None, cnf_path=None
46
, suitename = 'sqlbench_tests'
48
self.system_manager = system_manager
49
self.logging = self.system_manager.logging
50
self.skip_keys = ['system_manager','logging']
52
self.fullname = fullname
53
self.suitename = suitename
55
self.comment = comment
56
self.server_requirements = server_requirements
57
self.test_command = test_command
58
self.cnf_path = cnf_path
61
self.system_manager.logging.debug_class(self)
64
if self.skip_flag or self.disable:
73
class testManager(test_management.testManager):
74
"""Deals with scanning test directories, gathering test cases, and
75
collecting per-test information (opt files, etc) for use by the
80
def __init__(self, verbose, debug, default_engine, dotest, skiptest
81
, reorder, suitelist, suitepaths, system_manager
83
super(testManager, self).__init__( verbose, debug, default_engine
84
, dotest, skiptest, reorder
85
, suitelist, suitepaths
86
, system_manager, test_cases, mode)
87
self.suitepaths = [os.path.join(self.testdir,'sqlbench_tests')]
89
self.suitelist = ['main']
91
self.suitelist = suitelist
93
def process_suite(self,suite_dir):
94
"""Process a test suite.
95
Look for sqlbench tests, which are nice clean conf files
99
# We know this based on how we organize sqlbench test conf files
100
suite_name = os.path.basename(suite_dir)
102
self.system_manager.logging.verbose("Processing suite: %s" %(suite_name))
103
testlist = [os.path.join(suite_dir,test_file) for test_file in sorted(os.listdir(suite_dir)) if test_file.endswith('.cnf')]
105
# Search for specific test names
106
if self.desired_tests: # We have specific, named tests we want from the suite(s)
108
for test in self.desired_tests:
109
if test.endswith('.cnf'):
113
test = os.path.join(suite_dir,test)
115
tests_to_use.append(test)
116
testlist = tests_to_use
117
for test_case in testlist:
118
self.add_test(self.process_test_file(suite_name, test_case))
121
def process_test_file(self, suite_name, testfile):
122
""" We convert the info in a testfile into a testCase object """
124
config_reader = RawConfigParser()
125
config_reader.read(testfile)
126
# test_name = filename - .cnf...simpler
127
test_name = os.path.basename(testfile).replace('.cnf','')
128
test_comment = config_reader.get('test_info','comment')
129
server_requirements = self.process_server_reqs(config_reader.get('test_servers','servers'))
130
test_command = config_reader.get('test_command','command')
131
return testCase( self.system_manager
133
, fullname = "%s.%s" %(suite_name, test_name)
134
, server_requirements = server_requirements
135
, test_command = test_command
136
, cnf_path = testfile
137
, debug = self.debug )
141
def process_server_reqs(self,data_string):
142
""" We read in the list of lists as a string, so we need to
143
handle this / break it down into proper chunks
147
# We expect to see a list of lists and throw away the
149
option_sets = data_string[1:-1].strip().split(',')
150
for option_set in option_sets:
151
server_reqs.append([option_set[1:-1].strip()])
154
def record_test_result(self, test_case, test_status, output, exec_time):
155
""" Accept the results of an executed testCase for further
159
if test_status not in self.executed_tests:
160
self.executed_tests[test_status] = [test_case]
162
self.executed_tests[test_status].append(test_case)
164
self.logging.test_report( test_case.fullname, test_status
165
, str(exec_time), output
166
, report_output= True)
168
class crashmeTestManager(testManager):
169
"""Deals with scanning test directories, gathering test cases, and
170
collecting per-test information (opt files, etc) for use by the
175
def __init__(self, verbose, debug, default_engine, dotest, skiptest
176
, reorder, suitelist, suitepaths, system_manager
178
super(testManager, self).__init__( verbose, debug, default_engine
179
, dotest, skiptest, reorder
180
, suitelist, suitepaths
181
, system_manager, test_cases, mode)
182
self.suitepaths = [os.path.join(self.testdir,'crashme_tests')]
183
if suitelist is None:
184
self.suitelist = ['main']
186
self.suitelist = suitelist
188
def record_test_result(self, test_case, test_status, output, exec_time):
189
""" Accept the results of an executed testCase for further
193
if test_status not in self.executed_tests:
194
self.executed_tests[test_status] = [test_case]
196
self.executed_tests[test_status].append(test_case)
198
self.logging.test_report( test_case.fullname, test_status
199
, str(exec_time), output
200
, report_output= True)
added
removed
tests/lib/sqlbench/sqlbench_test_management.py
