Lines Matching refs:self
58 def __init__(self, filename, mode='a', encoding=None, delay=0, max_tries=5):
59 self.max_tries = max_tries
60 self.tries = 0
61 WatchedFileHandler.__init__(self, filename, mode, encoding, delay)
63 def emit(self, record):
66 WatchedFileHandler.emit(self, record)
67 self.tries = 0
70 if self.tries == self.max_tries:
72 self.stream.close()
73 self.stream = self._open()
74 self.tries += 1
80 def __init__(self):
81 self.starttime = None
82 self.returncode = None
83 self.runtime = ''
84 self.stdout = []
85 self.stderr = []
86 self.result = ''
88 def done(self, proc, killed):
93 m, s = divmod(time() - self.starttime, 60)
94 self.runtime = '%02d:%02d' % (m, s)
95 self.returncode = proc.returncode
97 self.result = 'KILLED'
99 elif self.returncode == 0:
100 self.result = 'PASS'
102 elif self.returncode != 0:
103 self.result = 'FAIL'
112 def __init__(self, stream):
113 self.stream = stream
114 self._buf = ''
115 self.lines = []
117 def fileno(self):
118 return self.stream.fileno()
120 def read(self, drain=0):
124 while self._read() is not None:
128 def _read(self):
135 fd = self.fileno()
140 self._buf += buf
143 buf = self._buf + buf
145 self._buf = rest
148 self.lines += [(now, r) for r in rows]
154 def __init__(self, pathname, outputdir=None, timeout=None, user=None):
155 self.pathname = pathname
156 self.outputdir = outputdir or 'BASEDIR'
157 self.timeout = timeout or 60
158 self.user = user or ''
159 self.killed = False
160 self.result = Result()
162 def __str__(self):
164 self.pathname, self.outputdir, self.timeout, self.user)
166 def kill_cmd(self, proc):
171 self.killed = True
172 do_sudo = len(self.user) != 0
185 def update_cmd_privs(self, cmd, user):
199 def collect_output(self, proc):
218 def run(self, options):
225 print(self)
228 privcmd = self.update_cmd_privs(self.pathname, self.user)
231 if not os.path.isdir(self.outputdir):
232 os.makedirs(self.outputdir, mode=0o777)
238 self.result.starttime = time()
241 t = Timer(int(self.timeout), self.kill_cmd, [proc])
243 self.result.stdout, self.result.stderr = self.collect_output(proc)
245 self.kill_cmd(proc)
250 self.result.done(proc, self.killed)
252 def skip(self):
259 self.result.stdout = self.result.stderr = []
260 self.result.starttime = time()
261 m, s = divmod(time() - self.result.starttime, 60)
262 self.result.runtime = '%02d:%02d' % (m, s)
263 self.result.result = 'SKIP'
265 def log(self, logger, options):
276 user = ' (run as %s)' % (self.user if len(self.user) else logname)
277 msga = 'Test: %s%s ' % (self.pathname, user)
278 msgb = '[%s] [%s]' % (self.result.runtime, self.result.result)
286 elif self.result.result != 'PASS':
291 lines = sorted(self.result.stdout + self.result.stderr,
297 if len(self.result.stdout):
298 with open(os.path.join(self.outputdir, 'stdout'), 'w') as out:
299 for _, line in self.result.stdout:
301 if len(self.result.stderr):
302 with open(os.path.join(self.outputdir, 'stderr'), 'w') as err:
303 for _, line in self.result.stderr:
305 if len(self.result.stdout) and len(self.result.stderr):
306 with open(os.path.join(self.outputdir, 'merged'), 'w') as merged:
315 def __init__(self, pathname, outputdir=None, timeout=None, user=None,
317 super(Test, self).__init__(pathname, outputdir, timeout, user)
318 self.pre = pre or ''
319 self.pre_user = pre_user or ''
320 self.post = post or ''
321 self.post_user = post_user or ''
323 def __str__(self):
325 if len(self.pre_user):
326 pre_user = ' (as %s)' % (self.pre_user)
327 if len(self.post_user):
328 post_user = ' (as %s)' % (self.post_user)
330 "%s%s\nUser: %s\n" % (self.pathname, self.outputdir,
331 self.timeout, self.pre, pre_user, self.post, post_user,
332 self.user)
334 def verify(self, logger):
339 files = [self.pre, self.pathname, self.post]
340 users = [self.pre_user, self.user, self.post_user]
351 self.pathname)
356 def run(self, logger, options):
361 pretest = Cmd(self.pre, outputdir=os.path.join(self.outputdir,
362 os.path.basename(self.pre)), timeout=self.timeout,
363 user=self.pre_user)
364 test = Cmd(self.pathname, outputdir=self.outputdir,
365 timeout=self.timeout, user=self.user)
366 posttest = Cmd(self.post, outputdir=os.path.join(self.outputdir,
367 os.path.basename(self.post)), timeout=self.timeout,
368 user=self.post_user)
391 def __init__(self, pathname, outputdir=None, timeout=None, user=None,
394 super(TestGroup, self).__init__(pathname, outputdir, timeout, user,
396 self.tests = tests or []
398 def __str__(self):
400 if len(self.pre_user):
401 pre_user = ' (as %s)' % (self.pre_user)
402 if len(self.post_user):
403 post_user = ' (as %s)' % (self.post_user)
405 "Pre: %s%s\nPost: %s%s\nUser: %s\n" % (self.pathname,
406 self.outputdir, self.tests, self.timeout, self.pre, pre_user,
407 self.post, post_user, self.user)
409 def verify(self, logger):
417 if len(self.pre) and not os.path.isabs(self.pre):
418 self.pre = os.path.join(self.pathname, self.pre)
419 if len(self.post) and not os.path.isabs(self.post):
420 self.post = os.path.join(self.pathname, self.post)
422 auxfiles = [self.pre, self.post]
423 users = [self.pre_user, self.user, self.post_user]
426 if self.pathname != os.path.dirname(f):
429 "directory." % (self.pathname, f))
435 (self.pathname, f))
441 self.pathname)
445 for test in self.tests:
446 if not verify_file(os.path.join(self.pathname, test)):
447 del self.tests[self.tests.index(test)]
450 self.pathname))
452 return len(self.tests) != 0
454 def run(self, logger, options):
460 pretest = Cmd(self.pre, outputdir=os.path.join(self.outputdir,
461 os.path.basename(self.pre)), timeout=self.timeout,
462 user=self.pre_user)
463 posttest = Cmd(self.post, outputdir=os.path.join(self.outputdir,
464 os.path.basename(self.post)), timeout=self.timeout,
465 user=self.post_user)
473 for fname in self.tests:
474 test = Cmd(os.path.join(self.pathname, fname),
475 outputdir=os.path.join(self.outputdir, fname),
476 timeout=self.timeout, user=self.user)
492 def __init__(self, options):
493 self.tests = {}
494 self.testgroups = {}
495 self.starttime = time()
496 self.timestamp = datetime.now().strftime('%Y%m%dT%H%M%S')
497 self.outputdir = os.path.join(options.outputdir, self.timestamp)
498 self.logger = self.setup_logging(options)
499 self.defaults = [
510 def __str__(self):
511 s = 'TestRun:\n outputdir: %s\n' % self.outputdir
513 for key in sorted(self.tests.keys()):
514 s += '%s%s' % (self.tests[key].__str__(), '\n')
516 for key in sorted(self.testgroups.keys()):
517 s += '%s%s' % (self.testgroups[key].__str__(), '\n')
520 def addtest(self, pathname, options):
530 if test.verify(self.logger):
531 self.tests[pathname] = test
533 def addtestgroup(self, dirname, filenames, options):
539 if dirname not in self.testgroups:
549 self.testgroups[dirname] = testgroup
550 self.testgroups[dirname].tests = sorted(filenames)
552 testgroup.verify(self.logger)
554 def read(self, logger, options):
569 setattr(self, opt, config.get('DEFAULT', opt))
570 self.outputdir = os.path.join(self.outputdir, self.timestamp)
586 self.testgroups[section] = testgroup
596 self.tests[section] = test
598 def write(self, options):
609 self.defaults])
612 for test in sorted(self.tests.keys()):
615 for testgroup in sorted(self.testgroups.keys()):
617 config.set(testgroup, 'tests', self.testgroups[testgroup].tests)
625 def complete_outputdirs(self, options):
639 tmp_dict = dict(list(self.tests.items()) + list(self.testgroups.items()))
641 base = self.outputdir
655 def setup_logging(self, options):
674 os.makedirs(self.outputdir, mode=0o777)
678 filename = os.path.join(self.outputdir, 'log')
694 def run(self, options):
699 os.chdir(self.outputdir)
701 fail('Could not change to directory %s' % self.outputdir)
702 for test in sorted(self.tests.keys()):
703 self.tests[test].run(self.logger, options)
704 for testgroup in sorted(self.testgroups.keys()):
705 self.testgroups[testgroup].run(self.logger, options)
707 def summary(self):
716 m, s = divmod(time() - self.starttime, 60)
721 print('Log directory:\t%s' % self.outputdir)