[py-svn] r61787 - in py/branch/pytestplugin/py: . doc misc misc/testing
hpk at codespeak.net
hpk at codespeak.net
Thu Feb 12 15:59:42 CET 2009
Author: hpk
Date: Thu Feb 12 15:59:41 2009
New Revision: 61787
Added:
py/branch/pytestplugin/py/doc/plugin.txt (contents, props changed)
py/branch/pytestplugin/py/misc/bus.py (contents, props changed)
py/branch/pytestplugin/py/misc/plugin.py (contents, props changed)
py/branch/pytestplugin/py/misc/testing/test_bus.py (contents, props changed)
py/branch/pytestplugin/py/misc/testing/test_plugin.py (contents, props changed)
Modified:
py/branch/pytestplugin/py/__init__.py
Log:
experimentation towards a more generalized plugin and event system,
suitable not only for py.test but also for the py lib
Modified: py/branch/pytestplugin/py/__init__.py
==============================================================================
--- py/branch/pytestplugin/py/__init__.py (original)
+++ py/branch/pytestplugin/py/__init__.py Thu Feb 12 15:59:41 2009
@@ -54,6 +54,11 @@
# EXPORTED API
exportdefs = {
+
+ # py lib events and plugins
+ 'bus' : ('./misc/bus.py', 'bus'),
+ 'plugin' : ('./misc/plugin.py', 'plugin'),
+
# py lib cmdline tools
'cmdline.pytest' : ('./cmdline/pytest.py', 'main',),
'cmdline.pyrest' : ('./cmdline/pyrest.py', 'main',),
Added: py/branch/pytestplugin/py/doc/plugin.txt
==============================================================================
--- (empty file)
+++ py/branch/pytestplugin/py/doc/plugin.txt Thu Feb 12 15:59:41 2009
@@ -0,0 +1,68 @@
+py lib plugins
+----------------
+
+you can write plugins that extend the py lib API
+or implement certain hooks.
+
+registering a plugin
+++++++++++++++++++++++++++++++++++
+
+::
+ >>> class MyPlugin:
+ def pyevent_pluginregister(self, plugin):
+ print "registering", plugin
+ ...
+ >>> import py
+ >>> py.plugin.register(MyPlugin())
+ registering <MyPlugin at 0x129031>
+
+registering a plugin at import time
+++++++++++++++++++++++++++++++++++++++++++
+
+::
+ import py
+ class EventLogger:
+ def pyevent_PluginAdded(self, event):
+ print "adding plugin", event.pluginobj
+ py.plugin.register(EventLogger())
+
+explanation: on import this will add and activate a plugin that
+will subsequently print information on events
+whose class name is "PluginAdded".
+
+logging all events caused by py lib
+++++++++++++++++++++++++++++++++++++++++
+
+you can log all py lib generated events by setting
+an environment variable, for example on linux::
+
+ export PYEVENT_LOG="/tmp/mylog"
+
+automatically importing plugins/modules
++++++++++++++++++++++++++++++++++++++++++
+
+you can set the environment variable ``PY_AUTOIMPORT=myplugin``
+to ensure that a module gets imported automatically at
+py lib import time.
+
+
+py lib events
+----------------
+
+sending events
++++++++++++++++++
+
+example sending events::
+
+ class myobj:
+ pass
+ py.bus.notify(myeventname=myobj)
+
+subscribing to all events
++++++++++++++++++++++++++++++
+
+example subscribing to all send objects::
+
+ l = []
+ py.bus.subscribe(l.append) # anonymous subscription
+ py.bus.subscribe(myeventname=l.append) # named subscription
Added: py/branch/pytestplugin/py/misc/bus.py
==============================================================================
--- (empty file)
+++ py/branch/pytestplugin/py/misc/bus.py Thu Feb 12 15:59:41 2009
@@ -0,0 +1,36 @@
+
+class EventBus(object):
+ """ Event Bus for distributing named and unnamed events. """
+ def __init__(self):
+ self._kwsubscribers = {}
+ self._subscribers = []
+
+ def subscribe(self, callback=None, **kwcallback):
+ """ subscribe callback to bus events. """
+ if callback:
+ self._subscribers.append(callback)
+ assert not kwcallback
+ else:
+ assert len(kwcallback) == 1
+ name, call = kwcallback.popitem()
+ self._kwsubscribers.setdefault(name, []).append(call)
+
+ def unsubscribe(self, callback=None, **kwcallback):
+ """ unsubscribe given callback from bus events. """
+ try:
+ if callback is not None:
+ self._subscribers.remove(callback)
+ else:
+ for name, call in kwcallback.items():
+ self._kwsubscribers[name].remove(call)
+ except ValueError, e:
+ raise KeyError(*e.args)
+
+ def notify(self, **kw):
+ for name in kw:
+ for callback in self._kwsubscribers.get(name, ()):
+ callback(kw[name])
+ for callback in self._subscribers:
+ callback((name, kw[name]))
+
+bus = EventBus()
Added: py/branch/pytestplugin/py/misc/plugin.py
==============================================================================
--- (empty file)
+++ py/branch/pytestplugin/py/misc/plugin.py Thu Feb 12 15:59:41 2009
@@ -0,0 +1,37 @@
+import py
+from py.__.misc.bus import bus
+
+class PluginManager:
+ def __init__(self):
+ self.list = []
+ bus.subscribe(self._forwardevent)
+
+ def _forwardevent(self, (name, obj)):
+ self.calleach("pyevent_" + name, obj)
+
+ def register(self, plugin):
+ assert not isinstance(plugin, str)
+ self.list.append(plugin)
+ py.bus.notify(pluginregistered=plugin)
+
+ def iterattr(self, attrname):
+ for plugin in self.list:
+ try:
+ yield getattr(plugin, attrname)
+ except AttributeError:
+ continue
+
+ def calleach(self, methname, *args, **kwargs):
+ ret = []
+ for call in self.iterattr(methname):
+ ret.append(call(*args, **kwargs))
+ return ret
+
+ def callfirst(self, methname, *args, **kwargs):
+ ret = []
+ for call in self.iterattr(methname):
+ ret.append(call(*args, **kwargs))
+ break
+ return ret
+
+plugin = PluginManager()
Added: py/branch/pytestplugin/py/misc/testing/test_bus.py
==============================================================================
--- (empty file)
+++ py/branch/pytestplugin/py/misc/testing/test_bus.py Thu Feb 12 15:59:41 2009
@@ -0,0 +1,72 @@
+
+import py
+from py.__.misc.bus import EventBus
+
+class TestEventBus:
+ def test_kw_register_notify_unregister(self):
+ bus = EventBus()
+ l = []
+ bus.subscribe(hello=l.append)
+ bus.notify(some=1)
+ assert not l
+ bus.notify(hello=5)
+ assert len(l) == 1
+ l.remove(5)
+ bus.unsubscribe(hello=l.append)
+ bus.notify(hello=7)
+ assert not l
+
+ def test_kw_unregister_non_existing(self):
+ bus = EventBus()
+ py.test.raises(KeyError, "bus.unsubscribe(hello=42)")
+ bus.subscribe(hello=10)
+ bus.unsubscribe(hello=10)
+ py.test.raises(KeyError, "bus.unsubscribe(hello=42)")
+
+ def test_kw_multiregister(self):
+ bus = EventBus()
+ l1 = []
+ l2 = []
+ bus.subscribe(hello=l1.append)
+ bus.subscribe(hello=l2.append)
+ bus.subscribe(world=l2.append)
+ bus.notify(some=1)
+ assert not l1 and not l2
+ bus.notify(hello=5)
+ assert len(l1) == 1
+ assert len(l2) == 1
+ bus.notify(world=42)
+ assert len(l2) == 2
+ bus.unsubscribe(hello=l2.append)
+ bus.unsubscribe(world=l2.append)
+ bus.unsubscribe(hello=l1.append)
+
+ def test_simple_anonymous(self):
+ bus = EventBus()
+ l = []
+ bus.subscribe(l.append)
+ bus.notify(hello=1)
+ assert l == [("hello", 1)]
+ bus.unsubscribe(l.append)
+ bus.notify(hello=1)
+ assert l == [("hello", 1)]
+
+ def test_multi_anonymous(self):
+ bus = EventBus()
+ l1 = []
+ l2 = []
+ bus.subscribe(l1.append)
+ bus.subscribe(l2.append)
+ bus.notify(event=1)
+ bus.notify(event2=2)
+ bus.notify(event3=3)
+ assert l1 == [("event", 1), ("event2", 2), ("event3", 3)]
+ assert l2 == [("event", 1), ("event2", 2), ("event3", 3)]
+ bus.unsubscribe(l1.append)
+ bus.unsubscribe(l2.append)
+ py.test.raises(KeyError, "bus.unsubscribe(l1.append)")
+ py.test.raises(KeyError, "bus.unsubscribe(l2.append)")
+
+
+def test_api():
+ assert isinstance(py.bus, EventBus)
Added: py/branch/pytestplugin/py/misc/testing/test_plugin.py
==============================================================================
--- (empty file)
+++ py/branch/pytestplugin/py/misc/testing/test_plugin.py Thu Feb 12 15:59:41 2009
@@ -0,0 +1,64 @@
+
+import py
+from py.__.misc.plugin import PluginManager
+
+class TestPluginManager:
+ def test_register(self):
+ pm = PluginManager()
+ class MyPlugin:
+ pass
+ pm.register(MyPlugin)
+ assert MyPlugin in pm.list
+
+ def test_onregister(self):
+ pm = PluginManager()
+ l = []
+ class MyApi:
+ def pyevent_pluginregistered(self, plugin):
+ l.append(plugin)
+ myapi = MyApi()
+ pm.register(myapi)
+ assert len(l) == 1
+ assert l[0] is myapi
+
+ def test_calleach(self):
+ pm = PluginManager()
+ class api1:
+ def m(self):
+ return 42
+ class api2:
+ def m(self):
+ return 41
+ pm.register(api1())
+ pm.register(api2())
+ l = pm.calleach('m')
+ l.sort()
+ assert l == [41, 42]
+
+ def test_callfirst(self):
+ pm = PluginManager()
+ class api1:
+ def m(self):
+ return 42
+ class api2:
+ def m(self):
+ return 41
+ pm.register(api1())
+ pm.register(api2())
+ l = pm.callfirst('m')
+ assert l == [42]
+
+ def test_iterattr(self):
+ pm = PluginManager()
+ class api1:
+ x = 42
+ class api2:
+ x = 41
+ pm.register(api1())
+ pm.register(api2())
+ l = list(pm.iterattr('x'))
+ l.sort()
+ assert l == [41, 42]
+
+def test_api():
+ assert isinstance(py.plugin, PluginManager)
More information about the pytest-commit
mailing list