p53209761
/
test_ok2
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[svn r63888] finally. the event concept is basically gone. 

now we only have plugin hooks aka plugin calls

--HG--
branch : trunk
This commit is contained in:
hpk 2009-04-09 02:36:07 +02:00
parent 0c4fc99a6f
commit cd038ee708
12 changed files with 21 additions and 113 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
py/_com.py
View File

@ -1,21 +1,5 @@
"""
py lib plugins and events.
you can write plugins that extend the py lib API.
currently this is mostly used by py.test
registering a plugin
++++++++++++++++++++++++++++++++++
::
>>> class MyPlugin:
... def pyevent__plugin_registered(self, plugin):
... print "registering", plugin.__class__.__name__
...
>>> import py
>>> py._com.pyplugins.register(MyPlugin())
registering MyPlugin
py lib plugins and plugin call management
"""
import py
@ -92,7 +76,6 @@ class PyPlugins:
def import_module(self, modspec):
# XXX allow modspec to specify version / lookup
modpath = modspec
self.notify("importingmodule", modpath)
__import__(modpath)
def consider_env(self):
@ -155,13 +138,6 @@ class PyPlugins:
return MultiCall(self.listattr(methname, plugins=[plugin]),
*args, **kwargs).execute(firstresult=True)
def notify(self, eventname, *args, **kwargs):
#print "notifying", eventname, args, kwargs
MultiCall(self.listattr("pyevent__" + eventname),
*args, **kwargs).execute()
#print "calling anonymous hooks", args, kwargs
MultiCall(self.listattr("pyevent"), eventname, args, kwargs).execute()
class PluginAPI:
def __init__(self, apiclass, plugins=None):

1
py/initpkg.py
View File

@ -280,5 +280,4 @@ def autoimport(pkgname):
ENVKEY = pkgname.upper() + "_AUTOIMPORT"
if ENVKEY in os.environ:
for impname in os.environ[ENVKEY].split(","):
py._com.pyplugins.notify("autoimport", impname)
__import__(impname)

61
py/misc/testing/test_com.py
View File

@ -97,21 +97,6 @@ class TestPyPlugins:
assert not plugins.isregistered(my)
assert plugins.getplugins() == [my2]
def test_onregister(self):
plugins = PyPlugins()
l = []
class MyApi:
def pyevent__plugin_registered(self, plugin):
l.append(plugin)
def pyevent__plugin_unregistered(self, plugin):
l.remove(plugin)
myapi = MyApi()
plugins.register(myapi)
assert len(l) == 1
assert l[0] is myapi
plugins.unregister(myapi)
assert not l
def test_call_methods(self):
plugins = PyPlugins()
class api1:
@ -175,21 +160,6 @@ class TestPyPlugins:
l = list(plugins.listattr('x', reverse=True))
assert l == [43, 42, 41]
def test_notify_anonymous_ordered(self):
plugins = PyPlugins()
l = []
class api1:
def pyevent__hello(self):
l.append("hellospecific")
class api2:
def pyevent(self, name, args, kwargs):
if name == "hello":
l.append(name + "anonymous")
plugins.register(api1())
plugins.register(api2())
plugins.notify('hello')
assert l == ["hellospecific", "helloanonymous"]
def test_consider_env(self, monkeypatch):
plugins = PyPlugins()
monkeypatch.setitem(os.environ, 'PYLIB', "unknownconsider_env")
@ -201,14 +171,7 @@ class TestPyPlugins:
mod.pylib = ["xxx nomod"]
excinfo = py.test.raises(ImportError, "plugins.consider_module(mod)")
mod.pylib = "os"
class Events(list):
def pyevent__importingmodule(self, mod):
self.append(mod)
l = Events()
plugins.register(l)
plugins.consider_module(mod)
assert len(l) == 1
assert l[0] == (mod.pylib)
def test_api_and_defaults():
assert isinstance(py._com.pyplugins, PyPlugins)
@ -226,30 +189,6 @@ def test_subprocess_env(testdir, monkeypatch):
finally:
old.chdir()
class TestPyPluginsEvents:
def test_pyevent__named_dispatch(self):
plugins = PyPlugins()
l = []
class A:
def pyevent__name(self, x):
l.append(x)
plugins.register(A())
plugins.notify("name", 13)
assert l == [13]
def test_pyevent__anonymous_dispatch(self):
plugins = PyPlugins()
l = []
class A:
def pyevent(self, name, args, kwargs):
if name == "name":
l.extend([args, kwargs])
plugins.register(A())
plugins.notify("name", 13, x=15)
assert l == [(13, ), {'x':15}]
class TestPluginAPI:
def test_happypath(self):
plugins = PyPlugins()

2
py/misc/warn.py
View File

@ -62,7 +62,7 @@ class WarningPlugin(object):
filename = module
path = py.path.local(filename)
warning = Warning(msg, path, lineno)
self.bus.notify("WARNING", warning)
self.bus.call_each("pyevent__WARNING", warning)
# singleton api warner for py lib
apiwarner = WarningPlugin(py._com.pyplugins)

2
py/test/config.py
View File

@ -50,7 +50,7 @@ class Config(object):
def trace(self, msg):
if getattr(self.option, 'traceconfig', None):
self.bus.notify("trace", "config", msg)
self.api.pytest_trace(category="config", msg=msg)
def _processopt(self, opt):
if hasattr(opt, 'default') and opt.dest:

8
py/test/dist/dsession.py vendored
View File

@ -96,11 +96,9 @@ class DSession(Session):
loopstate.dowork = True
callname, args, kwargs = eventcall
call = getattr(self.config.api, callname, None)
if call is not None:
if callname is not None:
call = getattr(self.config.api, callname)
call(*args, **kwargs)
else:
self.bus.notify(callname, *args, **kwargs)
# termination conditions
if ((loopstate.testsfailed and self.config.option.exitfirst) or
@ -177,7 +175,7 @@ class DSession(Session):
if isinstance(next, py.test.collect.Item):
senditems.append(next)
else:
self.bus.notify("collectstart", next)
self.config.api.pytest_collectstart(collector=next)
self.queueevent("pytest_collectreport", basic_collect_report(next))
if self.config.option.dist == "each":
self.senditems_each(senditems)

6
py/test/dist/testing/test_dsession.py vendored
View File

@ -129,9 +129,9 @@ class TestDSession:
# check that RescheduleEvents are not immediately
# rescheduled if there are no nodes
assert loopstate.dowork == False
session.queueevent("anonymous")
session.queueevent(None)
session.loop_once(loopstate)
session.queueevent("anonymous")
session.queueevent(None)
session.loop_once(loopstate)
assert node.sent == [[item]]
session.queueevent("pytest_itemtestreport", run(item, node))
@ -204,7 +204,7 @@ class TestDSession:
session.addnode(node)
loopstate = session._initloopstate([item])
session.queueevent("NOP")
session.queueevent(None)
session.loop_once(loopstate)
assert node.sent == [[item]]

2
py/test/looponfail/remote.py
View File

@ -129,7 +129,7 @@ def slave_runsession(channel, config, fullwidth, hasmarkup):
try:
colitem = py.test.collect.Collector._fromtrail(trail, config)
except AssertionError, e:
#XXX session.bus.notify of "test disappeared"
#XXX send info for "test disappeared" or so
continue
colitems.append(colitem)
else:

1
py/test/plugin/pytest__pytest.py
View File

@ -49,6 +49,7 @@ class CallRecorder:
def finalize(self):
for recorder in self._recorders.values():
self._pyplugins.unregister(recorder)
self._recorders.clear()
def recordsmethod(self, name):
for apiclass in self._recorders:

18
py/test/plugin/pytest_pytester.py
View File

@ -385,21 +385,20 @@ class EventRecorder(object):
def clear(self):
self.events[:] = []
self.callrecorder.calls[:] = []
def unregister(self):
self.pyplugins.unregister(self)
self.callrecorder.finalize()
@py.test.mark.xfail
def test_eventrecorder():
def test_eventrecorder(testdir):
bus = py._com.PyPlugins()
recorder = EventRecorder(bus)
bus.notify("anonymous")
assert recorder.events
recorder = testdir.geteventrecorder(bus)
assert not recorder.getfailures()
rep = runner.ItemTestReport(None, None)
rep.passed = False
rep.failed = True
bus.notify("pytest_itemtestreport", rep)
bus.call_each("pytest_itemtestreport", rep=rep)
failures = recorder.getfailures()
assert failures == [rep]
failures = recorder.getfailures()
@ -408,12 +407,12 @@ def test_eventrecorder():
rep = runner.ItemTestReport(None, None)
rep.passed = False
rep.skipped = True
bus.notify("pytest_itemtestreport", rep)
bus.call_each("pytest_itemtestreport", rep=rep)
rep = runner.CollectReport(None, None)
rep.passed = False
rep.failed = True
bus.notify("pytest_itemtestreport", rep)
bus.call_each("pytest_itemtestreport", rep=rep)
passed, skipped, failed = recorder.listoutcomes()
assert not passed and skipped and failed
@ -427,8 +426,7 @@ def test_eventrecorder():
recorder.clear()
assert not recorder.events
assert not recorder.getfailures()
bus.notify(pytest_itemtestreport, rep)
assert not recorder.events
bus.call_each("pytest_itemtestreport", rep=rep)
assert not recorder.getfailures()
class LineComp:

4
py/test/plugin/pytest_runner.py
View File

@ -11,9 +11,9 @@ class RunnerPlugin:
def pytest_item_setup_and_runtest(self, item):
setupstate = item.config._setupstate
call = item.config.guardedcall(lambda: setupstate.prepare(item))
rep = ItemSetupReport(item, call.excinfo, call.outerr)
if call.excinfo:
item.config.pytestplugins.notify(pytest_itemsetupreport, rep)
rep = ItemSetupReport(item, call.excinfo, call.outerr)
item.config.api.pytest_itemsetupreport(rep=rep)
else:
call = item.config.guardedcall(lambda: item.runtest())
item.config.api.pytest_item_runtest_finished(

3
py/test/pytestplugin.py
View File

@ -79,9 +79,6 @@ class PytestPlugins(object):
#print "plugins.call_each", args[0], args[1:], kwargs
return self.pyplugins.call_each(*args, **kwargs)
def notify(self, eventname, *args, **kwargs):
return self.pyplugins.notify(eventname, *args, **kwargs)
def notify_exception(self, excinfo=None):
if excinfo is None:
excinfo = py.code.ExceptionInfo()