← Back to branch summary

~ubuntu-branches/ubuntu/vivid/drizzle/vivid

« back to all changes in this revision

Viewing changes to tests/lib/dbqp_modes/crashme/crashme_test_management.py

  • Committer: Package Import Robot
  • Author(s): Dmitrijs Ledkovs
  • Date: 2013-10-29 15:43:40 UTC
  • mfrom: (20.1.2 sid)
  • Revision ID: package-import@ubuntu.com-20131029154340-j36v7gxq9tm1gi5f
Tags: 1:7.2.3-2ubuntu1
* Merge from debian, remaining changes:
  - Link against boost_system because of boost_thread.
  - Add required libs to message/include.am
  - Add upstart job and adjust init script to be upstart compatible.
  - Disable -floop-parallelize-all due to gcc-4.8/4.9 compiler ICE
    http://gcc.gnu.org/bugzilla/show_bug.cgi?id=57732

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/lib/dbqp_modes/crashme/crashme_test_management.py
1
 
#! /usr/bin/env python
2
 
# -*- mode: python; indent-tabs-mode: nil; -*-
3
 
# vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
4
 
#
5
 
# Copyright (C) 2011 Patrick Crews
6
 
#
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.
11
 
#
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.
16
 
#
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
20
 
 
21
 
""" crashme_test_management:
22
 
    code related to the gathering / analysis / management of 
23
 
    the test cases
24
 
    ie - collecting the list of tests in each suite, then
25
 
    gathering additional, relevant information for crashme mode
26
 
 
27
 
"""
28
 
 
29
 
# imports
30
 
import os
31
 
import re
32
 
import sys
33
 
from ConfigParser import RawConfigParser
34
 
 
35
 
import lib.test_mgmt.test_management as test_management
36
 
 
37
 
 
38
 
    
39
 
class testCase:
40
 
    """Holds info on a single crashme test
41
 
 
42
 
    """
43
 
    def __init__( self, system_manager, name=None
44
 
                , fullname = None, server_requirements=[[]]
45
 
                , comment=None, test_command=None, cnf_path=None
46
 
                , suitename = 'crashme_tests'
47
 
                , debug=False ):
48
 
        self.system_manager = system_manager
49
 
        self.logging = self.system_manager.logging
50
 
        self.skip_keys = ['system_manager','logging']
51
 
        self.name = name
52
 
        self.fullname = fullname
53
 
        self.suitename = suitename
54
 
        self.master_sh = None
55
 
        self.comment = comment
56
 
        self.server_requirements = server_requirements
57
 
        self.test_command = test_command
58
 
        self.cnf_path = cnf_path
59
 
        
60
 
        if debug:
61
 
            self.system_manager.logging.debug_class(self)
62
 
 
63
 
    def should_run(self):
64
 
        if self.skip_flag or self.disable:
65
 
            return 0
66
 
        else:
67
 
            return 1
68
 
 
69
 
 
70
 
        
71
 
        
72
 
          
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
76
 
       test-runner
77
 
 
78
 
    """
79
 
 
80
 
    def __init__( self, variables, system_manager):
81
 
        super(testManager, self).__init__( variables, system_manager)
82
 
        self.suitepaths = [os.path.join(self.testdir,'crashme_tests')]
83
 
        if variables['suitelist'] is None:
84
 
            self.suitelist = ['main']
85
 
        else:
86
 
            self.suitelist = variables['suitelist']
87
 
 
88
 
    def process_suite(self,suite_dir):
89
 
        """Process a test suite.
90
 
           Look for crashme tests, which are nice clean conf files
91
 
        
92
 
        """
93
 
 
94
 
        # We know this based on how we organize crashme test conf files
95
 
        suite_name = os.path.basename(suite_dir) 
96
 
        self.system_manager.logging.verbose("Processing suite: %s" %(suite_name))
97
 
        testlist = [os.path.join(suite_dir,test_file) for test_file in sorted(os.listdir(suite_dir)) if test_file.endswith('.cnf')]
98
 
 
99
 
        # Search for specific test names
100
 
        if self.desired_tests: # We have specific, named tests we want from the suite(s)
101
 
           tests_to_use = []
102
 
           for test in self.desired_tests:
103
 
               if test.endswith('.cnf'): 
104
 
                   pass
105
 
               else:
106
 
                   test = test+'.cnf'
107
 
               test = os.path.join(suite_dir,test)
108
 
               if test in testlist:
109
 
                   tests_to_use.append(test)
110
 
           testlist = tests_to_use
111
 
        for test_case in testlist:
112
 
            self.add_test(self.process_test_file(suite_name, test_case))
113
 
 
114
 
 
115
 
    def process_test_file(self, suite_name, testfile):
116
 
        """ We convert the info in a testfile into a testCase object """
117
 
 
118
 
        config_reader = RawConfigParser()
119
 
        config_reader.read(testfile)
120
 
        # test_name = filename - .cnf...simpler
121
 
        test_name = os.path.basename(testfile).replace('.cnf','')
122
 
        test_comment = config_reader.get('test_info','comment')
123
 
        server_requirements = self.process_server_reqs(config_reader.get('test_servers','servers'))
124
 
        test_command = config_reader.get('test_command','command')
125
 
        return testCase( self.system_manager
126
 
                       , name = test_name
127
 
                       , fullname = "%s.%s" %(suite_name, test_name)
128
 
                       , server_requirements = server_requirements
129
 
                       , test_command = test_command
130
 
                       , cnf_path = testfile
131
 
                       , debug = self.debug )
132
 
 
133
 
        #sys.exit(0)
134
 
 
135
 
    def process_server_reqs(self,data_string):
136
 
        """ We read in the list of lists as a string, so we need to 
137
 
            handle this / break it down into proper chunks
138
 
 
139
 
        """
140
 
        server_reqs = []
141
 
        # We expect to see a list of lists and throw away the 
142
 
        # enclosing brackets
143
 
        option_sets = data_string[1:-1].strip().split(',')
144
 
        for option_set in option_sets:
145
 
            server_reqs.append([option_set[1:-1].strip()])
146
 
        return server_reqs
147
 
 
148
 
    def record_test_result(self, test_case, test_status, output, exec_time):
149
 
        """ Accept the results of an executed testCase for further
150
 
            processing.
151
 
 
152
 
        """
153
 
        if test_status not in self.executed_tests:
154
 
            self.executed_tests[test_status] = [test_case]
155
 
        else:
156
 
            self.executed_tests[test_status].append(test_case)
157
 
        # report
158
 
        self.logging.test_report( test_case.fullname, test_status
159
 
                                , str(exec_time), output
160
 
                                , report_output= True)
161
 
 
162
 
class crashmeTestManager(testManager):
163
 
    """Deals with scanning test directories, gathering test cases, and 
164
 
       collecting per-test information (opt files, etc) for use by the
165
 
       test-runner
166
 
 
167
 
    """
168
 
 
169
 
    def __init__( self, variables, system_manager):
170
 
        super(testManager, self).__init__( variables, system_manager)
171
 
        self.suitepaths = [os.path.join(self.testdir,'crashme_tests')]
172
 
        if variables['suitelist'] is None:
173
 
            self.suitelist = ['main']
174
 
        else:
175
 
            self.suitelist = variables['suitelist']
176
 
 
177
 
    def record_test_result(self, test_case, test_status, output, exec_time):
178
 
        """ Accept the results of an executed testCase for further
179
 
            processing.
180
 
 
181
 
        """
182
 
        if test_status not in self.executed_tests:
183
 
            self.executed_tests[test_status] = [test_case]
184
 
        else:
185
 
            self.executed_tests[test_status].append(test_case)
186
 
        # report
187
 
        self.logging.test_report( test_case.fullname, test_status
188
 
                                , str(exec_time), output
189
 
                                , report_output= True)