[py-svn] r62002 - in py/branch/pytestplugin/py/test: . dsession/testing looponfail looponfail/testing plugin testing
hpk at codespeak.net
hpk at codespeak.net
Wed Feb 18 13:08:35 CET 2009
Author: hpk
Date: Wed Feb 18 13:08:33 2009
New Revision: 62002
Modified:
py/branch/pytestplugin/py/test/dsession/testing/test_dsession.py
py/branch/pytestplugin/py/test/dsession/testing/test_functional_dsession.py
py/branch/pytestplugin/py/test/event.py
py/branch/pytestplugin/py/test/looponfail/remote.py
py/branch/pytestplugin/py/test/looponfail/testing/test_remote.py
py/branch/pytestplugin/py/test/looponfail/testing/test_util.py
py/branch/pytestplugin/py/test/looponfail/util.py
py/branch/pytestplugin/py/test/plugin/pytest_eventlog.py
py/branch/pytestplugin/py/test/plugin/pytest_pytester.py
py/branch/pytestplugin/py/test/testing/suptest.py
py/branch/pytestplugin/py/test/testing/test_collect.py
py/branch/pytestplugin/py/test/testing/test_event.py
Log:
getting rid of pytest-event speciality of subscribing with "value" only callbacks.
now all anonymous callbacks receive (name, obj) pairs as passed to notify(name=obj)
Modified: py/branch/pytestplugin/py/test/dsession/testing/test_dsession.py
==============================================================================
--- py/branch/pytestplugin/py/test/dsession/testing/test_dsession.py (original)
+++ py/branch/pytestplugin/py/test/dsession/testing/test_dsession.py Wed Feb 18 13:08:33 2009
@@ -24,6 +24,53 @@
while queue.qsize():
print queue.get()
+class EventRecorder(object):
+ def __init__(self, bus):
+ self.events = []
+ self.bus = bus
+ bus.subscribe(self.record)
+
+ def record(self, (name, event)):
+ self.events.append((name, event))
+
+ def getfirstnamed(self, name):
+ for evname, event in self.events:
+ if evname == name:
+ return event
+
+ def getfailures(self):
+ l = []
+ for evname, ev in self.events:
+ if evname in ('itemtestreport', 'collectionreport'):
+ if ev.failed:
+ l.append(ev)
+ return l
+
+ def clear(self):
+ self.events[:] = []
+
+ def unsubscribe(self):
+ self.bus.unsubscribe(self.record)
+
+def test_eventrecorder():
+ bus = event.EventBus()
+ recorder = EventRecorder(bus)
+ bus.notify(anonymous=event.NOP())
+ assert recorder.events
+ assert not recorder.getfailures()
+ rep = event.ItemTestReport(None, None)
+ rep.failed = True
+ bus.notify(itemtestreport=rep)
+ failures = recorder.getfailures()
+ assert failures == [rep]
+ recorder.clear()
+ assert not recorder.events
+ assert not recorder.getfailures()
+ recorder.unsubscribe()
+ bus.notify(itemtestreport=rep)
+ assert not recorder.events
+ assert not recorder.getfailures()
+
class TestDSession(InlineCollection):
def test_fixoptions(self):
config = self.parseconfig("--exec=xxx")
@@ -175,13 +222,12 @@
host = session.item2host[item1]
ev = event.HostDown(host, None)
session.queueput(hostdown=ev)
-
- events = [] ; session.bus.subscribe(events.append)
+ evrec = EventRecorder(session.bus)
loopstate = LoopState([])
session.loop_once(loopstate)
assert loopstate.colitems == [item2] # do not reschedule crash item
- testrep = [x for x in events if isinstance(x, event.ItemTestReport)][0]
+ testrep = evrec.getfirstnamed("itemtestreport")
assert testrep.failed
assert testrep.colitem == item1
assert str(testrep.longrepr).find("crashed") != -1
@@ -205,10 +251,10 @@
session = DSession(item._config)
ev = event.NOP()
- events = [] ; session.bus.subscribe(events.append)
+ evrec = EventRecorder(session.bus)
session.queueput(anonymous=ev)
session.loop_once(LoopState([]))
- assert events[0] == ev
+ assert evrec.getfirstnamed('anonymous') == ev
def runthrough(self, item):
session = DSession(item._config)
@@ -275,15 +321,14 @@
session.addhost(host)
loopstate = LoopState([])
loopstate.shuttingdown = True
- l = []
- session.bus.subscribe(l.append)
+ evrec = EventRecorder(session.bus)
session.queueput(itemtestreport=run(item))
session.loop_once(loopstate)
- assert not l
+ assert not evrec.events
ev = event.HostDown(host)
session.queueput(hostdown=ev)
session.loop_once(loopstate)
- assert l == [ev]
+ assert evrec.getfirstnamed('hostdown') == ev
def test_filteritems(self):
modcol = self.getmodulecol("""
@@ -298,11 +343,11 @@
dsel = session.filteritems([modcol])
assert dsel == [modcol]
items = modcol.collect()
- events = [] ; session.bus.subscribe(events.append)
+ evrec = EventRecorder(session.bus)
remaining = session.filteritems(items)
assert remaining == []
- ev = events[-1]
+ evname, ev = evrec.events[-1]
assert isinstance(ev, event.Deselected)
assert ev.items == items
@@ -310,7 +355,7 @@
remaining = session.filteritems(items)
assert remaining == [items[0]]
- ev = events[-1]
+ evname, ev = evrec.events[-1]
assert isinstance(ev, event.Deselected)
assert ev.items == [items[1]]
Modified: py/branch/pytestplugin/py/test/dsession/testing/test_functional_dsession.py
==============================================================================
--- py/branch/pytestplugin/py/test/dsession/testing/test_functional_dsession.py (original)
+++ py/branch/pytestplugin/py/test/dsession/testing/test_functional_dsession.py Wed Feb 18 13:08:33 2009
@@ -16,7 +16,7 @@
events = []
while 1:
try:
- ev = queue.get(timeout=timeout)
+ evname, ev = queue.get(timeout=timeout)
except py.std.Queue.Empty:
print "seen events", events
raise IOError("did not see %r events" % (eventtype))
Modified: py/branch/pytestplugin/py/test/event.py
==============================================================================
--- py/branch/pytestplugin/py/test/event.py (original)
+++ py/branch/pytestplugin/py/test/event.py Wed Feb 18 13:08:33 2009
@@ -7,13 +7,6 @@
from py.__.test.outcome import Skipped
import py.__.misc.event
-
-class CallWithValue(object):
- def __init__(self, callback):
- self.callback = callback
- def __call__(self, (name, obj)):
- return self.callback(obj)
-
class EventBus(object):
""" Bus for distributing events. """
def __init__(self, bus=None):
@@ -24,9 +17,7 @@
def issubscribed(self, callback=None, **kw):
if callback is not None:
- for x in self._bus._getsubscribers(''):
- if isinstance(x, CallWithValue) and x.callback == callback:
- return True
+ kw[''] = callback
for name, value in kw.items():
if value in self._bus._getsubscribers(name):
return True
@@ -38,7 +29,6 @@
d = kw.copy()
if callback is not None:
d[''] = callback
- callback = CallWithValue(callback)
self._memo.append(d)
#print "appending", d
self._bus.subscribe(callback, **kw)
@@ -62,12 +52,7 @@
def unsubscribe(self, callback=None, **kw):
""" unsubscribe given callback from bus events. """
- for x in self._bus._getsubscribers(''):
- if isinstance(x, CallWithValue) and x.callback == callback:
- self._bus.unsubscribe(x, **kw)
- break
- else:
- self._bus.unsubscribe(**kw)
+ return self._bus.unsubscribe(callback, **kw)
def notify(self, **kw):
self._bus.notify(**kw)
Modified: py/branch/pytestplugin/py/test/looponfail/remote.py
==============================================================================
--- py/branch/pytestplugin/py/test/looponfail/remote.py (original)
+++ py/branch/pytestplugin/py/test/looponfail/remote.py Wed Feb 18 13:08:33 2009
@@ -154,7 +154,7 @@
#session.bus.subscribe(sendevent)
failreports = []
- def recordfailures(ev):
+ def recordfailures((evname, ev)):
if isinstance(ev, event.BaseReport):
if ev.failed:
failreports.append(ev)
Modified: py/branch/pytestplugin/py/test/looponfail/testing/test_remote.py
==============================================================================
--- py/branch/pytestplugin/py/test/looponfail/testing/test_remote.py (original)
+++ py/branch/pytestplugin/py/test/looponfail/testing/test_remote.py Wed Feb 18 13:08:33 2009
@@ -1,26 +1,10 @@
import py
from py.__.test.testing import suptest
from py.__.test.looponfail.remote import LooponfailingSession, LoopState, RemoteControl
-from py.__.test import event
-def getevent(l, evtype):
- result = getevents(l, evtype)
- if not result:
- raise ValueError("event %r not found in %r" %(evtype, l))
- return result[0]
-
-def getevents(l, evtype):
- result = []
- for ev in l:
- if isinstance(ev, evtype):
- result.append(ev)
- return result
-
-
class TestRemoteControl(suptest.InlineCollection):
def test_nofailures(self):
item = self.getitem("def test_func(): pass\n")
- events = []
control = RemoteControl(item._config)
control.setup()
failures = control.runsession()
Modified: py/branch/pytestplugin/py/test/looponfail/testing/test_util.py
==============================================================================
--- py/branch/pytestplugin/py/test/looponfail/testing/test_util.py (original)
+++ py/branch/pytestplugin/py/test/looponfail/testing/test_util.py Wed Feb 18 13:08:33 2009
@@ -1,5 +1,5 @@
import py
-from py.__.test.looponfail.util import StatRecorder, EventRecorder
+from py.__.test.looponfail.util import StatRecorder
from py.__.test import event
def test_filechange():
@@ -62,30 +62,4 @@
tmp.ensure("newfile.py")
reply.get(timeout=0.5)
wp.shutdown()
-
-def test_eventrecorder():
- bus = event.EventBus()
- recorder = EventRecorder(bus)
- bus.notify(anonymous=event.NOP())
- assert recorder.events
- assert not recorder.getfailures()
- rep = event.ItemTestReport(None, None)
- rep.failed = True
- bus.notify(itemtestreport=rep)
- failures = recorder.getfailures()
- assert failures == [rep]
- recorder.clear()
- assert not recorder.events
- assert not recorder.getfailures()
- recorder.unsubscribe()
- bus.notify(itemtestreport=rep)
- assert not recorder.events
- assert not recorder.getfailures()
-
-
-
-
-
-
-
-
+
Modified: py/branch/pytestplugin/py/test/looponfail/util.py
==============================================================================
--- py/branch/pytestplugin/py/test/looponfail/util.py (original)
+++ py/branch/pytestplugin/py/test/looponfail/util.py Wed Feb 18 13:08:33 2009
@@ -52,19 +52,3 @@
self.statcache = newstat
return changed
-
-class EventRecorder(object):
- def __init__(self, bus):
- self.events = []
- self.bus = bus
- self.bus.subscribe(self.events.append)
-
- def getfailures(self):
- return [ev for ev in self.events
- if isinstance(ev, event.BaseReport) and \
- ev.failed]
- def clear(self):
- self.events[:] = []
-
- def unsubscribe(self):
- self.bus.unsubscribe(self.events.append)
Modified: py/branch/pytestplugin/py/test/plugin/pytest_eventlog.py
==============================================================================
--- py/branch/pytestplugin/py/test/plugin/pytest_eventlog.py (original)
+++ py/branch/pytestplugin/py/test/plugin/pytest_eventlog.py Wed Feb 18 13:08:33 2009
@@ -16,7 +16,7 @@
self.eventlogfile.close()
del self.eventlogfile
- def pytest_event(self, event):
+ def pytest_event(self, (name, event)):
if hasattr(self, 'eventlogfile'):
f = self.eventlogfile
print >>f, event
Modified: py/branch/pytestplugin/py/test/plugin/pytest_pytester.py
==============================================================================
--- py/branch/pytestplugin/py/test/plugin/pytest_pytester.py (original)
+++ py/branch/pytestplugin/py/test/plugin/pytest_pytester.py Wed Feb 18 13:08:33 2009
@@ -143,7 +143,7 @@
self.config = config
self.session = session
self.cls2events = d = {}
- def app(event):
+ def app((name, event)):
print "[event]", event
for cls in py.std.inspect.getmro(event.__class__):
if cls is not object:
Modified: py/branch/pytestplugin/py/test/testing/suptest.py
==============================================================================
--- py/branch/pytestplugin/py/test/testing/suptest.py (original)
+++ py/branch/pytestplugin/py/test/testing/suptest.py Wed Feb 18 13:08:33 2009
@@ -13,12 +13,12 @@
from py.__.test import event
from fnmatch import fnmatch
-def eventappender(session):
+def eventappender(bus):
l = []
- def app(ev):
+ def app((name, ev)):
print ev
l.append(ev)
- session.bus.subscribe(app)
+ bus.subscribe(app)
return l
def initsorter_from_cmdline(args=None):
@@ -52,7 +52,7 @@
self.config = config
self.session = session
self.cls2events = d = {}
- def app(event):
+ def app((name, event)):
print "[event]", event
for cls in py.std.inspect.getmro(event.__class__):
if cls is not object:
Modified: py/branch/pytestplugin/py/test/testing/test_collect.py
==============================================================================
--- py/branch/pytestplugin/py/test/testing/test_collect.py (original)
+++ py/branch/pytestplugin/py/test/testing/test_collect.py Wed Feb 18 13:08:33 2009
@@ -454,7 +454,7 @@
print "using tempdir", tmp
config = py.test.config._reparse([tmp])
session = config.initsession()
- l = suptest.eventappender(session)
+ l = suptest.eventappender(config.bus)
items = list(session.genitems(getcolitems(config)))
return items, l
Modified: py/branch/pytestplugin/py/test/testing/test_event.py
==============================================================================
--- py/branch/pytestplugin/py/test/testing/test_event.py (original)
+++ py/branch/pytestplugin/py/test/testing/test_event.py Wed Feb 18 13:08:33 2009
@@ -6,26 +6,23 @@
class TestEventBus:
- def test_simple(self):
+ def test_anonymous_subscription(self):
bus = EventBus()
l = []
bus.subscribe(l.append)
bus.notify(x=1)
- bus.notify(x=2)
- bus.notify(x=3)
- assert l == [1,2,3]
+ assert len(l) == 1
+ assert l[0] == ('x', 1)
def test_multi_sub(self):
bus = EventBus()
- l1 = []
- l2 = []
+ l1, l2 = [], []
bus.subscribe(l1.append)
bus.subscribe(l2.append)
bus.notify(x=1)
bus.notify(x=2)
- bus.notify(x=3)
- assert l1 == [1,2,3]
- assert l2 == [1,2,3]
+ assert l1 == l2
+ assert l1 == [('x', 1), ('x', 2)]
def test_remove(self):
bus = EventBus()
@@ -34,7 +31,7 @@
bus.notify(x=1)
bus.unsubscribe(l.append)
bus.notify(x=22)
- assert l == [1]
+ assert l == [('x', 1)]
def test_subscribe_methods(self):
bus = EventBus()
More information about the pytest-commit
mailing list