Changeset ffa6a59 in nscp
- Timestamp:
- 01/18/12 00:50:16 (17 months ago)
- Branches:
- master, 0.4.0, 0.4.1, 0.4.2
- Children:
- 441a022
- Parents:
- 89838be
- Files:
-
- 2 added
- 3 edited
-
changelog (modified) (1 diff)
-
scripts/python/__init__.py (added)
-
scripts/python/lib/test_helper.py (modified) (7 diffs)
-
scripts/python/test_all.py (added)
-
scripts/python/test_w32_system.py (modified) (3 diffs)
Legend:
- Unmodified
- Added
- Removed
-
changelog
r89838be rffa6a59 6 6 * Fix RtlStringFromGUID problem on NT4 7 7 8 2012-01-18 MickeM 9 * Improved unit tests so they can be run all in sequence 10 * Fixed various bugs in the unit tests 11 * Improved the syntax and output of running unit tests 12 * Refactored unit tests to be slightly nicer 13 8 14 2012-01-16 MickeM 9 15 * Added option to both execute and query item (with a reload in the middle for installing/running unittests) -
scripts/python/lib/test_helper.py
r89838be rffa6a59 26 26 def destroy_test_manager(): 27 27 global test_manager 28 if test_manager: 29 test_manager.destroy() 28 30 test_manager = None 29 31 … … 109 111 110 112 def run_test(self): 111 result = TestResult( )113 result = TestResult('run_test') 112 114 result.add_message(False, 'TODO add implementation') 113 115 return result … … 139 141 return False 140 142 141 142 class TestResult: 143 class Entry: 144 status = False 145 desc = 'Unassigned result' 146 level = 0 147 error = None 148 def __init__(self, status, desc, error): 149 self.status = status 150 self.desc = desc 151 self.error = error 152 153 def log(self): 143 class TestResultEntry: 144 status = False 145 desc = 'Unassigned result' 146 error = None 147 def __init__(self, status, desc, error): 148 self.status = status 149 self.desc = desc 150 self.error = error 151 152 def log(self, prefix = '', indent = 0): 153 if self.status: 154 log('%s%s%s'%(prefix, ''.rjust(indent, ' '), self)) 155 else: 156 log_error('%s%s%s'%(prefix, ''.rjust(indent, ' '), self)) 157 158 def is_ok(self): 159 return self.status 160 161 def count(self): 162 if self.status: 163 return (1, 1) 164 return (1, 0) 165 166 def contains(self, other): 167 if self == other: 168 return True 169 return False 170 171 def __str__(self): 172 if self.status: 173 return 'OK: %s'%self.desc 174 else: 175 return 'ERROR: %s (%s)'%(self.desc, self.error) 176 177 class TestResultCollection(TestResultEntry): 178 179 status = True 180 title = None 181 children = [] 182 def __init__(self, title, list = None): 183 self.title = title 184 self.children = [] 185 if list: 186 self.extend(list) 187 188 def log(self, prefix = '', indent = 0): 189 start = '%s%s'%(prefix, ''.rjust(indent, ' ')) 190 if self.status: 191 log('%s%s'%(start, self)) 192 else: 193 log_error('%s%s'%(start, self)) 194 for c in self.children: 195 c.log(prefix, indent+1) 196 197 def is_ok(self): 198 return self.status 199 200 def count(self): 201 total_count = 1 202 ok_count = 0 203 if self.status: 204 ok_count = 1 205 206 for c in self.children: 207 (total, ok) = c.count() 208 total_count = total_count + total 209 ok_count = ok_count + ok 210 211 return (total_count, ok_count) 212 213 def contains(self, other): 214 for c in self.children: 215 if c.contains(other): 216 return True 217 return False 218 219 def __str__(self): 220 if self.status: 221 return 'OK: %s'%self.title 222 else: 223 (total, ok) = c.count() 224 return 'ERROR: %s (%d/%d)'%(self.title, ok, total) 225 226 def extend(self, lst): 227 if isinstance(lst, list): 154 228 if self.status: 155 log('%s%s'%(''.rjust(self.level, ' '), self)) 156 else: 157 log_error('%s%s'%(''.rjust(self.level, ' '), self)) 158 159 def is_ok(self): 160 return self.status 161 162 def indent(self): 163 self.level = self.level + 1 164 165 def __str__(self): 166 if self.status: 167 return 'OK: %s'%self.desc 168 else: 169 return 'ERROR: %s (%s)'%(self.desc, self.error) 170 171 results = [] 172 173 def __init__(self): 174 self.results = [] 229 for c in lst: 230 if not c.is_ok(): 231 self.status = False 232 233 for c in lst: 234 if c.contains(self): 235 log_error('Attempting to add a list with me in it') 236 return 237 self.children.extend(lst) 238 else: 239 self.append(lst) 240 241 def append(self, entry): 242 if not entry: 243 log_error('Attempting to add invalid entry (None)') 244 elif entry == self: 245 log_error('Attempting to add self to self') 246 else: 247 if self.status and not entry.is_ok(): 248 self.status = False 249 self.children.append(entry) 250 251 class TestResult(TestResultCollection): 252 253 def __init__(self, title = 'DUMMY TITLE'): 254 TestResultCollection.__init__(self, title) 175 255 176 256 def add_message(self, status, message, error = None): 177 e = TestResult .Entry(status, message, error)257 e = TestResultEntry(status, message, error) 178 258 e.log() 179 self.a dd_entry(e)259 self.append(e) 180 260 181 261 def assert_equals(self, s1, s2, msg): … … 189 269 else: 190 270 self.add_message(s1 in s2 or s2 in s1, msg, '"%s" (contains) "%s"'%(s1, s2)) 191 192 271 193 272 def add_entry(self, e): 194 self. results.append(e)195 273 self.append(e) 274 196 275 def add(self, result): 197 try: 198 for e in result.results: 199 e.indent() 200 self.results.extend(result.results) 201 except: 202 log_error('Failed to process results...') 203 204 def log(self): 205 okcount = 0 206 count = len(self.results) 207 for e in self.results: 208 e.log() 209 if e.is_ok(): 210 okcount = okcount + 1 211 if okcount == count: 212 log("OK: %d test(s) successfull"%count) 213 else: 214 log("ERROR: %d of %d test(s) succedded (%d failed)"%(okcount, count, count-okcount)) 215 return self 216 217 def is_ok(self): 218 for e in self.results: 219 if not e.is_ok(): 220 return False 221 return True 222 223 def __str__(self): 224 s = '' 225 for e in self.results: 226 s += '%s, '%e 227 return s 276 self.extend(result) 228 277 229 278 def return_nagios(self): 230 okcount = 0 231 count = len(self.results) 232 for e in self.results: 233 if e.is_ok(): 234 okcount = okcount + 1 235 self.log() 236 if okcount == count: 237 return (status.OK, "OK: %d test(s) successfull"%count) 238 else: 239 return (status.CRITICAL, "ERROR: %d/%d test(s) failed"%(count-okcount, count)) 279 (total, ok) = self.count() 280 log(' | Test result log (only summary will be returned to query)') 281 self.log(' | ') 282 if total == ok: 283 return (status.OK, "OK: %d test(s) successfull"%(total)) 284 else: 285 return (status.CRITICAL, "ERROR: %d/%d test(s) failed"%(total-ok, total)) 240 286 241 287 class TestManager: … … 253 299 self.plugin_alias = plugin_alias 254 300 self.script_alias = script_alias 301 self.suites = [] 255 302 256 303 def add(self, suite): 257 if isinstance(suite, (list)):304 if isinstance(suite, list): 258 305 for s in suite: 259 self.suites.append(s) 260 else: 261 self.suites.append(suites) 306 self.add(s) 307 else: 308 if not suite in self.suites: 309 self.suites.append(suite) 262 310 263 311 def run_suite(self, suite): 264 result = TestResult( )312 result = TestResult('Running suite: %s'%suite.title()) 265 313 for c in list: 266 314 result.add(run_test(plugin_id, prefix, c)) … … 268 316 269 317 def run(self, arguments = []): 270 result = TestResult( )318 result = TestResult('Test result for %d suites'%len(self.suites)) 271 319 for suite in self.suites: 272 320 instance = suite.getInstance() 273 321 instance.setup(self.plugin_id, self.prefix) 274 tmp = TestResult()275 tmp.add(instance.run_test())276 result.a dd_message(tmp.is_ok(), 'Running suite: %s'%instance.title())277 result.add (tmp)322 suite_result = TestResult('Running suite: %s'%instance.title()) 323 suite_result.append(instance.run_test()) 324 result.append(suite_result) 325 result.add_message(suite_result.is_ok(), 'Result from suite: %s'%instance.title()) 278 326 instance.teardown() 279 327 return result … … 283 331 instance = suite.getInstance() 284 332 instance.init(self.plugin_id) 333 334 def destroy(self): 335 self.suites = [] 336 self.prefix = '' 337 self.plugin_id = None 338 self.plugin_alias = None 339 self.script_alias = None 285 340 286 341 def install(self, arguments = []): -
scripts/python/test_w32_system.py
r330af36 rffa6a59 40 40 41 41 def test_one_proc_int(self, proc, actual, asked): 42 result = TestResult( )42 result = TestResult('Checking one state %d/%d'%(actual, asked)) 43 43 for s in ['eq', 'gt', 'lt', 'ne']: 44 44 (retcode, retmessage, retperf) = core.simple_query('CheckProcState', ['ShowAll', 'critCount=%s:%d'%(s, asked), '%s=started'%proc]) … … 48 48 49 49 def test_one_proc(self): 50 result = TestResult( )50 result = TestResult('Checking CheckProcState') 51 51 52 52 for j in range(0,3): 53 result.a dd(self.test_one_proc_int('notepad.exe', 0, j))53 result.append(self.test_one_proc_int('notepad.exe', 0, j)) 54 54 55 55 pids = [] … … 60 60 pids.append(handle.pid) 61 61 for j in range(0,3): 62 result.a dd(self.test_one_proc_int('notepad.exe', i, j))62 result.append(self.test_one_proc_int('notepad.exe', i, j)) 63 63 64 64 for p in pids: 65 65 subprocess.Popen("taskkill /F /T /PID %i"%p , shell=True) 66 67 return result 66 68 67 69 def run_test(self): 68 result = TestResult( )69 result. add(self.test_one_proc())70 result = TestResult('Testing process checks on windows 32 bit systems') 71 result.extend(self.test_one_proc()) 70 72 return result 71 73
Note: See TracChangeset
for help on using the changeset viewer.








