autonose/ui/basic.py @@ -6,10 +6,9 @@ import os import sys import time import logging -import subprocess import scanner -from watcher import Watcher +import watcher log = logging.getLogger(__name__) debug = log.debug @@ -23,63 +22,72 @@ class Main(mandy.Command): self.opt('clear', bool, default=False, opposite=False, desc='reset all dependency information') self.opt('once', bool, default=False, opposite=False, desc='run all outdated tests and then exit') self.opt('debug', bool, default=False, opposite=False, desc='show debug output') - self.opt('changelog', bool, default=False, opposite=False, desc='show more info about what files have changed') + self.opt('info', bool, default=False, opposite=False, desc='show more info about what files have changed') self.opt('wait', int, default=5, desc='sleep time (between filesystem scans)') self.opt('config', str, default=None, desc='nosetests config file') self.opt('curses', bool, default=False, desc='use the curses interface') def run(self, opts): self.opts = opts - if opts.debug: - logging.basicConfig(level=logging.DEBUG) - else: - logging.getLogger('autonose').addHandler(NullHandler()) - - if opts.debug: - self.info() - sleep_time = opts.wait + self.init_logging() + self.init_nose_args() + self.init_ui() + self.save_init_modules() + self.run_loop() + + def run_loop(self): first_run = True - config_file = opts.config - self.ui = None - if opts.curses: - from ui.terminal import Terminal - self.ui = Terminal() - self.nose_args = ['--autorun'] - if config_file is not None: - self.nose_args.append('--config=%s' % (config_file)) - if opts.debug: - self.nose_args.append('--debug=autonose') - elif opts.changelog: - self.nose_args.append('--debug=autonose.shared.state.summary') try: while True: state = scanner.scan() if state.anything_changed() or first_run: first_run = False + watcher.global_state = state self.run_with_state(state) - if opts.once: + if self.opts.once: break - debug("sleeping (%s)..." % (sleep_time,)) - time.sleep(sleep_time) + debug("sleeping (%s)..." % (self.opts.wait,)) + time.sleep(self.opts.wait) finally: if self.ui is not None: self.ui.finalize() + def init_logging(self): + if self.opts.debug: + logging.basicConfig(level=logging.DEBUG) + self.info() + elif self.opts.info: + logging.basicConfig(level=logging.INFO) + else: + logging.getLogger().addHandler(NullHandler()) + + def init_nose_args(self): + self.nose_args = ['nosetests','--autorun'] + if self.opts.config is not None: + self.nose_args.append('--config=%s' % (self.opts.config)) + + def save_init_modules(self): + self._sys_modules = set(sys.modules.keys()) + + def restore_init_modules(self): + for modname in set(sys.modules.keys()).difference(self._sys_modules): + del(sys.modules[modname]) + + def init_ui(self): + self.ui = None + if self.opts.curses: + from ui.terminal import Terminal + self.ui = Terminal() + else: + from ui.basic import Basic + self.ui = Basic() + def run_with_state(self, state): debug("running with %s affected files..." % (len(state.affected))) - self.clear() - self.timestamp() - #todo: run this in-process (currently it doesn't seem to reload tested modules properly...) - subprocess.call(['nosetests'] + self.nose_args) + self.restore_init_modules() + self.ui.begin_new_run(time.localtime()) + nose.run(argv=self.nose_args) - def timestamp(self): - print >> sys.stderr, "# Running tests at %s " % (time.strftime("%H:%m:%S")) - print >> sys.stderr, "" - - def clear(self): - print "\n" * 80 - subprocess.call('clear') - def info(self): state = scanner.scan() print '-'*80 autonose/runner.py @@ -1,13 +1,16 @@ import curses +import sys +import time + +import os +import termstyle + +import sys def _start(): window = curses.initscr() curses.cbreak() curses.noecho() - try: - curses.start_color() - except: - pass window.keypad(1) return window @@ -17,29 +20,80 @@ def _stop(window): curses.echo() curses.endwin() +class OutputProxy(object): + def __init__(self, proxy): + self.proxy = proxy + self._line = '' + + def write(self, s): + self._line += s + if '\n' in self._line: + map(self.proxy.writeln, self._line.split('\n')) + self._line = '' + + def writeln(self, s=''): + if not s.endswith('\n'): + s = s + '\n' + self.write(s) + + def __getattr__(self, attr): + return DoNothing() + +class DoNothing(object): + def __nonzero__(self): return False + def __call__(self, *args): pass + class Terminal(object): def __init__(self): + sys.stdout = sys.stderr = OutputProxy(self) + self._logfile = open('.curses-log','w') + self.last_run_time = time.localtime() self.testinfo = {} self.window = _start() - self.window.insstr('hi there!', 0) + self.current_line = 0 + self._size() + self._redraw() + + def begin_new_run(self, current_time=None): + if current_time is None: + current_time = self.last_run_time + else: + self.last_run_time = current_time + self.status.erase() + self.status.bkgd(' ', curses.A_REVERSE | curses.A_BOLD) + self.status.insstr("Run started at: %s " % (time.strftime("%H:%m:%S", self.last_run_time)), 0) self.window.refresh() - def updateStatus(self, time): - pass - - def updateTest(self, fileStamp, output): + def update_test(self, fileStamp, output): if fileStamp.path in self.testinfo: # update pass + def writeln(self, s): + self.window.addstr(self.current_line, 0, s) + self._logfile.write(s + '\n') + self.current_line += 1 + self._redraw() + def finalize(self): _stop(self.window) - - + self._logfile.close() #non-API methods - def update_ui(self): - pass + + def _resize(self): + self.status.erase() + self.window.erase() + self._size() + self._redraw() + + def _redraw(self): + self.begin_new_run() + self.window.refresh() + + def _size(self): + height, width = self.window.getmaxyx() + self.status = self.window.subwin(1, 0, height-1, 0) def scroll_next(self): pass autonose/ui/terminal.py @@ -10,8 +10,12 @@ import scanner log = logging.getLogger(__name__) debug = log.debug +info = log.info from shared.test_result import TestResult, TestResultSet, success, skip, error, fail +def get_paths(*items): return '\n'.join([x.path for x in items]) + +global_state = None class Watcher(nose.plugins.Plugin): name = 'autonose' @@ -25,11 +29,14 @@ class Watcher(nose.plugins.Plugin): def _setup(self): if self.state is None: - self.state = scanner.scan() + if global_state is not None: + self.state = global_state + else: + self.state = scanner.scan() self.start_time = time.time() self.files_to_run = set(self.state.affected).union(set(self.state.bad)) - debug("changed files: %s" % (self.state.affected,)) - debug("bad files: %s" % (self.state.bad,)) + info("changed files: %s" % (get_paths(*self.state.affected),)) + info("bad files: %s" % (get_paths(*self.state.bad),)) def options(self, parser, env=os.environ): parser.add_option( @@ -59,7 +66,6 @@ class Watcher(nose.plugins.Plugin): def _test_file(self, test): file_path = test.address()[0] - print file_path if not os.path.exists(file_path): raise RuntimeError("test.address does not contain a valid file: %s" % (test.address(),)) return file_util.relative(file_util.source(file_path)) @@ -101,4 +107,4 @@ class Watcher(nose.plugins.Plugin): def finalize(self, result=None): debug(self.state) scanner.save() - + autonose/watcher.py @@ -1,6 +1,7 @@ [nosetests] verbosity=2 with-doctest=1 +rednose=1 ; pdb=1 ; pdb-failures=1 nose.cfg ea7060928117c7564fdbfe3a07554b1ee444382d Tim Cuthbertson tim3d.junk@gmail.com http://github.com/gfxmonk/autonose/commit/cb4609cf0ed6488119c9ecda7d563951c65f808b cb4609cf0ed6488119c9ecda7d563951c65f808b 2009-05-23T20:40:56-07:00 2009-05-23T20:40:56-07:00 added basic UI for console operation now running nosetests in-process; reverting loaded modules after each run to prevent undesired caching 13ffe814c2f6f37a1f447d5ed8510dd9646799bd Tim Cuthbertson tim3d.junk@gmail.com