[py-svn] r10525 - in py/branch/py-collect/test: . terminal testing tkinter
hpk at codespeak.net
hpk at codespeak.net
Mon Apr 11 00:45:46 CEST 2005
Author: hpk
Date: Mon Apr 11 00:45:46 2005
New Revision: 10525
Added:
py/branch/py-collect/test/terminal/remote.py
- copied, changed from r10519, py/branch/py-collect/test/run.py
Removed:
py/branch/py-collect/test/run.py
Modified:
py/branch/py-collect/test/config.py
py/branch/py-collect/test/testing/test_session.py
py/branch/py-collect/test/tkinter/tkgui.py
Log:
- shuffle things some more
- add some tests for --session parsing
Modified: py/branch/py-collect/test/config.py
==============================================================================
--- py/branch/py-collect/test/config.py (original)
+++ py/branch/py-collect/test/config.py Mon Apr 11 00:45:46 2005
@@ -99,13 +99,8 @@
""" return Session class determined from cmdline options
and looked up in initial config modules.
"""
- if self.option.session:
- name = self.option.session
- elif self.option.tkinter:
- name = 'tkinter'
- else:
- name = 'terminal'
- name = name.capitalize() + "Session"
+ name = self.option.session
+ name += 'Session'
return self.getinitialvalue(name)
Config._reset()
@@ -129,6 +124,15 @@
assert exe.check()
option.executable = exe
+ # setting a correct frontend session
+ if option.session:
+ name = option.session
+ elif option.tkinter:
+ name = 'tkinter'
+ else:
+ name = 'terminal'
+ name = name.capitalize()
+ option.session = name
def bootstrapconfig(configpaths):
""" return 'current' config object, after initializing
Deleted: /py/branch/py-collect/test/run.py
==============================================================================
--- /py/branch/py-collect/test/run.py Mon Apr 11 00:45:46 2005
+++ (empty file)
@@ -1,152 +0,0 @@
-from __future__ import generators
-import py
-from py.__impl__.execnet.channel import ChannelFile, receive2file
-from py.__impl__.test.config import configbasename
-from py.__impl__.test.collect import getfscollector
-from py.__impl__.test.session import map2colitems
-import sys
-
-def checkpyfilechange(rootdir, statcache):
- """ wait until project files are changed. """
- fil = py.path.checker(fnmatch='*.py')
- rec = py.path.checker(dotfile=0)
- changed = False
- for path in rootdir.visit(fil, rec):
- oldstat = statcache.get(path, None)
- try:
- statcache[path] = curstat = path.stat()
- except py.error.ENOENT:
- if oldstat:
- del statcache[path]
- print "# WARN: race condition on", path
- else:
- if oldstat:
- if oldstat.st_mtime != curstat.st_mtime or \
- oldstat.st_size != curstat.st_size:
- changed = True
- print "# MODIFIED", path
- else:
- changed = True
- return changed
-
-class FailingCollector(py.test.collect.Collector):
- def __init__(self, faileditems):
- self._faileditems = faileditems
-
- def __iter__(self):
- for x in self._faileditems:
- yield x
-
-
-class StdouterrProxy:
- def __init__(self, gateway):
- self.gateway = gateway
- def setup(self):
- channel = self.gateway.remote_exec("""
- import sys
- out, err = channel.newchannel(), channel.newchannel()
- channel.send(out)
- channel.send(err)
- sys.stdout, sys.stderr = out.open('w'), err.open('w')
- """)
- self.stdout = channel.receive()
- self.stderr = channel.receive()
- channel.waitclose(1.0)
- py.std.threading.Thread(target=receive2file,
- args=(self.stdout, sys.stdout)).start()
- py.std.threading.Thread(target=receive2file,
- args=(self.stderr, sys.stderr)).start()
- def teardown(self):
- self.stdout.close()
- self.stderr.close()
-
-def waitfinish(channel):
- try:
- while 1:
- try:
- channel.waitclose(0.1)
- except (IOError, py.error.Error):
- continue
- else:
- failures = channel.receive()
- return failures
- break
- finally:
- #print "closing down channel and gateway"
- channel.close()
- channel.gateway.exit()
-
-def failure_master(executable, args, failures):
- gw = py.execnet.PopenGateway(executable)
- outproxy = StdouterrProxy(gw)
- outproxy.setup()
- try:
- channel = gw.remote_exec("""
- from py.__impl__.test.run import failure_slave
- failure_slave(channel)
- """)
- channel.send((args, failures))
- return waitfinish(channel)
- finally:
- outproxy.teardown()
-
-def failure_slave(channel):
- """ we run this on the other side. """
- args, failures = channel.receive()
- sessionclass, config = py.test.init(args, ignoreremote=True)
- if failures:
- cols = getfailureitems(failures)
- else:
- cols = config.paths
- session = sessionclass(config)
- session.shouldclose = channel.isclosed
- failures = session.main(cols)
- channel.send(failures)
-
-def getfailureitems(failures):
- l = []
- for rootpath, names in failures:
- root = py.path.local(rootpath)
- if root.check(dir=1):
- current = py.test.collect.Directory(root).Directory(root)
- elif root.check(file=1):
- current = py.test.collect.Module(root).Module(root)
- # root is fspath of names[0] -> pop names[0]
- # slicing works with empty lists
- names = names[1:]
- while names:
- name = names.pop(0)
- try:
- current = current.join(name)
- except NameError:
- print "WARNING: could not find %s on %r" %(name, current)
- break
- else:
- l.append(current)
- return l
-
-class FrontendSession:
- def __init__(self, option, args):
- self.config.option = option
- self.args = args
-
- def main(self, *paths):
- statcache = {}
- # XXX figure out a better rootdir?
- rootdir = py.path.local()
- failures = []
- while 1:
- if self.config.option.looponfailing:
- while not checkpyfilechange(rootdir, statcache):
- py.std.time.sleep(0.4)
- failures = failure_master(self.config.option.executable or sys.executable,
- self.args, failures)
- if not self.config.option.session:
- break
- print "#" * 60
- print "# session mode: %d failures remaining" % len(failures)
- for root, names in failures:
- name = "/".join(names) # XXX
- print "Failure at: %r" % (name,)
- print "# watching py files below %s" % rootdir
- print "# ", "^" * len(str(rootdir))
Copied: py/branch/py-collect/test/terminal/remote.py (from r10519, py/branch/py-collect/test/run.py)
==============================================================================
--- py/branch/py-collect/test/run.py (original)
+++ py/branch/py-collect/test/terminal/remote.py Mon Apr 11 00:45:46 2005
@@ -6,7 +6,7 @@
from py.__impl__.test.session import map2colitems
import sys
-def checkpyfilechange(rootdir, statcache):
+def checkpyfilechange(rootdir, statcache={}):
""" wait until project files are changed. """
fil = py.path.checker(fnmatch='*.py')
rec = py.path.checker(dotfile=0)
@@ -125,19 +125,14 @@
l.append(current)
return l
-class FrontendSession:
- def __init__(self, option, args):
- self.config.option = option
- self.args = args
-
- def main(self, *paths):
- statcache = {}
+class TerminalFrontendSession:
+ def main(self, *args):
# XXX figure out a better rootdir?
rootdir = py.path.local()
failures = []
while 1:
if self.config.option.looponfailing:
- while not checkpyfilechange(rootdir, statcache):
+ while not checkpyfilechange(rootdir):
py.std.time.sleep(0.4)
failures = failure_master(self.config.option.executable or sys.executable,
self.args, failures)
Modified: py/branch/py-collect/test/testing/test_session.py
==============================================================================
--- py/branch/py-collect/test/testing/test_session.py (original)
+++ py/branch/py-collect/test/testing/test_session.py Mon Apr 11 00:45:46 2005
@@ -13,6 +13,14 @@
assert len(l) == 2
l = session.getresults(py.test.Item.Passed)
assert not l
+
+ def test_session_parsing(self):
+ config, args = py.test.Config.parse(['--session=terminal'])
+ assert config.getsessionclass() is py.test.TerminalSession
+ config, args = py.test.Config.parse(['--session=tkinter'])
+ assert config.getsessionclass() is py.test.TkinterSession
+ config, args = py.test.Config.parse(['--tkinter'])
+ assert config.getsessionclass() is py.test.TkinterSession
#f = open('/tmp/logfile', 'wa')
class TestTerminalSession:
Modified: py/branch/py-collect/test/tkinter/tkgui.py
==============================================================================
--- py/branch/py-collect/test/tkinter/tkgui.py (original)
+++ py/branch/py-collect/test/tkinter/tkgui.py Mon Apr 11 00:45:46 2005
@@ -328,7 +328,7 @@
from py.__impl__.test.tkinter import repository
import os
import time
- from py.__impl__.test.run import getfailureitems
+ from py.__impl__.test.terminal.remote import getfailureitems
rerun_cols, args = channel.receive()
col = getfailureitems(rerun_cols)
More information about the pytest-commit
mailing list