From 1d3519bfe092278238b14ccbb46ec9789f2406d1 Mon Sep 17 00:00:00 2001 From: hpk Date: Sun, 28 Jan 2007 14:25:03 +0100 Subject: [PATCH] [svn r37465] * all Sessions now have fixoptions() * added some tests for implied and conflicting options * make all Session objects inherit from the base session * small cleanups with respect to test setup * separate tests a bit --HG-- branch : trunk --- py/test/config.py | 27 +------- py/test/rsession/rsession.py | 13 +++- py/test/session.py | 15 +++++ py/test/terminal/remote.py | 8 +-- py/test/testing/setupdata.py | 11 ++- py/test/testing/test_config.py | 4 +- py/test/testing/test_remote.py | 53 +++++++++++++++ py/test/testing/test_session.py | 114 +++++++++----------------------- 8 files changed, 128 insertions(+), 117 deletions(-) create mode 100644 py/test/testing/test_remote.py diff --git a/py/test/config.py b/py/test/config.py index 565634831..c8a68c123 100644 --- a/py/test/config.py +++ b/py/test/config.py @@ -48,7 +48,6 @@ class Config(object): args = [str(x) for x in args] cmdlineoption, args = self._parser.parse_args(args) self.option.__dict__.update(vars(cmdlineoption)) - fixoptions(self.option) # XXX fixing should be moved to sessions if not args: args.append(py.std.os.getcwd()) self.topdir = gettopdir(args) @@ -130,9 +129,7 @@ class Config(object): """ return an initialized session object. """ cls = self._getsessionclass() session = cls(self) - # XXX: all sessions should have one - if hasattr(session, 'fixoptions'): - session.fixoptions() + session.fixoptions() return session def _getsessionclass(self): @@ -272,28 +269,6 @@ def checkmarshal(name, value): except ValueError: raise ValueError("%s=%r is not marshallable" %(name, value)) -def fixoptions(option): - """ sanity checks and making option values canonical. """ - - # implied options - if option.usepdb: - if not option.nocapture: - #print "--pdb implies --nocapture" - option.nocapture = True - - if option.runbrowser and not option.startserver: - #print "--runbrowser implies --startserver" - option.startserver = True - - # conflicting options - if option.looponfailing and option.usepdb: - raise ValueError, "--looponfailing together with --pdb not supported." - if option.looponfailing and option.dist: - raise ValueError, "--looponfailing together with --dist not supported." - if option.executable and option.usepdb: - raise ValueError, "--exec together with --pdb not supported." - - def gettopdir(args): """ return the top directory for the given paths. if the common base dir resides in a python package diff --git a/py/test/rsession/rsession.py b/py/test/rsession/rsession.py index 58035c30e..ea4762527 100644 --- a/py/test/rsession/rsession.py +++ b/py/test/rsession/rsession.py @@ -16,15 +16,23 @@ from py.__.test.rsession.hostmanage import HostInfo, HostOptions, HostManager from py.__.test.rsession.local import local_loop, plain_runner, apigen_runner,\ box_runner from py.__.test.rsession.reporter import LocalReporter, RemoteReporter +from py.__.test.session import Session -class AbstractSession(object): +class AbstractSession(Session): """ An abstract session executes collectors/items through a runner. """ def __init__(self, config, optimise_localhost=True): - self.config = config + super(AbstractSession, self).__init__(config=config) self.optimise_localhost = optimise_localhost + + def fixoptions(self): + option = self.config.option + if option.runbrowser and not option.startserver: + #print "--runbrowser implies --startserver" + option.startserver = True + super(AbstractSession, self).fixoptions() def getpkgdir(path): path = py.path.local(path) @@ -122,6 +130,7 @@ class RSession(AbstractSession): """ Remote version of session """ def fixoptions(self): + super(RSession, self).fixoptions() config = self.config try: config.getvalue('disthosts') diff --git a/py/test/session.py b/py/test/session.py index 9d5551fea..346c56bd3 100644 --- a/py/test/session.py +++ b/py/test/session.py @@ -23,6 +23,21 @@ class Session(object): if not self.config.option.nomagic: py.magic.revoke(assertion=1) + def fixoptions(self): + """ check, fix and determine conflicting options. """ + option = self.config.option + # implied options + if option.usepdb: + if not option.nocapture: + option.nocapture = True + # conflicting options + if option.looponfailing and option.usepdb: + raise ValueError, "--looponfailing together with --pdb not supported." + if option.looponfailing and option.dist: + raise ValueError, "--looponfailing together with --dist not supported." + if option.executable and option.usepdb: + raise ValueError, "--exec together with --pdb not supported." + def start(self, colitem): """ hook invoked before each colitem.run() invocation. """ diff --git a/py/test/terminal/remote.py b/py/test/terminal/remote.py index dbb1db891..1dd3ab436 100644 --- a/py/test/terminal/remote.py +++ b/py/test/terminal/remote.py @@ -1,7 +1,7 @@ from __future__ import generators import py +from py.__.test.session import Session from py.__.test.terminal.out import getout -import sys def checkpyfilechange(rootdir, statcache={}): """ wait until project files are changed. """ @@ -51,9 +51,9 @@ def getfailureitems(failures): l.append(current) return l -class RemoteTerminalSession(object): +class RemoteTerminalSession(Session): def __init__(self, config, file=None): - self.config = config + super(RemoteTerminalSession, self).__init__(config=config) self._setexecutable() if file is None: file = py.std.sys.stdout @@ -123,8 +123,6 @@ def slaverun_TerminalSession(channel): topdir, repr, failures = channel.receive() print "SLAVE: received configuration, using topdir:", topdir config = py.test.config - import sys - sys.stdout.flush() config.initdirect(topdir, repr, failures) config.option.session = None config.option.looponfailing = False diff --git a/py/test/testing/setupdata.py b/py/test/testing/setupdata.py index 293ef7fe2..167add692 100644 --- a/py/test/testing/setupdata.py +++ b/py/test/testing/setupdata.py @@ -1,13 +1,20 @@ import py +def setup_module(mod): + mod.datadir = setupdatadir() + mod.tmpdir = py.test.ensuretemp(mod.__name__) + def setupdatadir(): datadir = py.test.ensuretemp("datadir") - if not datadir.listdir(): - for name, content in namecontent: + names = [x.basename for x in datadir.listdir()] + for name, content in namecontent: + if name not in names: datadir.join(name).write(content) return datadir namecontent = [ +('syntax_error.py', "this is really not python\n"), + ('disabled_module.py', py.code.Source(''' disabled = True diff --git a/py/test/testing/test_config.py b/py/test/testing/test_config.py index 56c6ed42b..8cdc6c1f8 100644 --- a/py/test/testing/test_config.py +++ b/py/test/testing/test_config.py @@ -231,7 +231,8 @@ class TestSessionAndOptions: def test_sessionname_lookup_custom(self): self.tmpdir.join("conftest.py").write(py.code.Source(""" - class MySession: + from py.__.test.session import Session + class MySession(Session): def __init__(self, config): self.config = config """)) @@ -357,3 +358,4 @@ class TestConfigColitems: col3 = config._getcollector(config.topdir.dirpath()) py.test.raises(ValueError, "config.get_collector_trail(col3)") + diff --git a/py/test/testing/test_remote.py b/py/test/testing/test_remote.py new file mode 100644 index 000000000..ea8fe2b97 --- /dev/null +++ b/py/test/testing/test_remote.py @@ -0,0 +1,53 @@ +import py +from py.__.test.testing.setupdata import setup_module + +class TestRemote: + def test_exec(self): + o = tmpdir.ensure('remote', dir=1) + tfile = o.join('test_exec.py') + tfile.write(py.code.Source(""" + def test_1(): + assert 1 == 0 + """)) + print py.std.sys.executable + config = py.test.config._reparse( + ['--exec=' + py.std.sys.executable, + o]) + cls = config._getsessionclass() + out = [] # out = py.std.Queue.Queue() + session = cls(config, out.append) + session.main() + for s in out: + if s.find('1 failed') != -1: + break + else: + py.test.fail("did not see test_1 failure") + + def test_looponfailing(self): + o = tmpdir.ensure('looponfailing', dir=1) + tfile = o.join('test_looponfailing.py') + tfile.write(py.code.Source(""" + def test_1(): + assert 1 == 0 + """)) + print py.std.sys.executable + config = py.test.config._reparse(['--looponfailing', str(o)]) + cls = config._getsessionclass() + out = py.std.Queue.Queue() + session = cls(config, out.put) + pool = py._thread.WorkerPool() + reply = pool.dispatch(session.main) + while 1: + s = out.get(timeout=1.0) + if s.find('1 failed') != -1: + break + print s + else: + py.test.fail("did not see test_1 failure") + # XXX we would like to have a cleaner way to finish + try: + reply.get(timeout=0.5) + except IOError, e: + assert str(e).lower().find('timeout') != -1 + + diff --git a/py/test/testing/test_session.py b/py/test/testing/test_session.py index f423bbecc..c6fffb68e 100644 --- a/py/test/testing/test_session.py +++ b/py/test/testing/test_session.py @@ -1,9 +1,38 @@ import py -from setupdata import setupdatadir +from setupdata import setup_module # sets up global 'tmpdir' -def setup_module(mod): - mod.datadir = setupdatadir() - mod.tmpdir = py.test.ensuretemp(mod.__name__) +implied_options = { + '--pdb': 'usepdb and nocapture', + '-v': 'verbose', + '-l': 'showlocals', + '--runbrowser': 'startserver and runbrowser', +} + +conflict_options = ('--looponfailing --pdb', + '--dist --pdb', + '--exec=%s --pdb' % py.std.sys.executable, + ) + +def test_conflict_options(): + for spec in conflict_options: + opts = spec.split() + yield check_conflict_option, opts + +def check_conflict_option(opts): + print "testing if options conflict:", " ".join(opts) + config = py.test.config._reparse(opts + [datadir/'filetest.py']) + py.test.raises((ValueError, SystemExit), """ + config.initsession() + """) + +def test_implied_options(): + for key, expr in implied_options.items(): + yield check_implied_option, [key], expr + +def check_implied_option(opts, expr): + config = py.test.config._reparse(opts + [datadir/'filetest.py']) + session = config.initsession() + assert eval(expr, session.config.option.__dict__) def test_default_session_options(): for opts in ([], ['-l'], ['-s'], ['--tb=no'], ['--tb=short'], @@ -63,15 +92,7 @@ class TestKeywordSelection: l = session.getitemoutcomepairs(py.test.Item.Skipped) assert l[0][0].name == 'test_1' -#f = open('/tmp/logfile', 'wa') class TestTerminalSession: - - def setup_class(cls): - (datadir / 'syntax_error.py').write("\nthis is really not python\n") - - def teardown_class(cls): - (datadir / 'syntax_error.py').remove() - def mainsession(self, *args): from py.__.test.terminal.terminal import TerminalSession self.file = py.std.StringIO.StringIO() @@ -291,72 +312,3 @@ class TestTerminalSession: expected_output = '\nE ' + line_to_report + '\n' print 'Looking for:', expected_output assert expected_output in out - - -class TestRemote: - def XXXtest_rootdir_is_package(self): - d = tmpdir.ensure('rootdirtest1', dir=1) - d.ensure('__init__.py') - x1 = d.ensure('subdir', '__init__.py') - x2 = d.ensure('subdir2', '__init__.py') - x3 = d.ensure('subdir3', 'noinit', '__init__.py') - assert getrootdir([x1]) == d - assert getrootdir([x2]) == d - assert getrootdir([x1,x2]) == d - assert getrootdir([x3,x2]) == d - assert getrootdir([x2,x3]) == d - - def XXXtest_rootdir_is_not_package(self): - one = tmpdir.ensure('rootdirtest1', 'hello') - rootdir = getrootdir([one]) - assert rootdir == one.dirpath() - - def test_exec(self): - o = tmpdir.ensure('remote', dir=1) - tfile = o.join('test_exec.py') - tfile.write(py.code.Source(""" - def test_1(): - assert 1 == 0 - """)) - print py.std.sys.executable - config = py.test.config._reparse( - ['--exec=' + py.std.sys.executable, - o]) - cls = config._getsessionclass() - out = [] # out = py.std.Queue.Queue() - session = cls(config, out.append) - session.main() - for s in out: - if s.find('1 failed') != -1: - break - else: - py.test.fail("did not see test_1 failure") - - def test_looponfailing(self): - o = tmpdir.ensure('looponfailing', dir=1) - tfile = o.join('test_looponfailing.py') - tfile.write(py.code.Source(""" - def test_1(): - assert 1 == 0 - """)) - print py.std.sys.executable - config = py.test.config._reparse(['--looponfailing', str(o)]) - cls = config._getsessionclass() - out = py.std.Queue.Queue() - session = cls(config, out.put) - pool = py._thread.WorkerPool() - reply = pool.dispatch(session.main) - while 1: - s = out.get(timeout=1.0) - if s.find('1 failed') != -1: - break - print s - else: - py.test.fail("did not see test_1 failure") - # XXX we would like to have a cleaner way to finish - try: - reply.get(timeout=0.5) - except IOError, e: - assert str(e).lower().find('timeout') != -1 - -