launchpad-reviewers team mailing list archive
-
launchpad-reviewers team
-
Mailing list archive
-
Message #04154
[Merge] lp:~allenap/launchpad/lp-app-longpoll into lp:launchpad
Gavin Panella has proposed merging lp:~allenap/launchpad/lp-app-longpoll into lp:launchpad.
Requested reviews:
Launchpad code reviewers (launchpad-reviewers)
For more details, see:
https://code.launchpad.net/~allenap/launchpad/lp-app-longpoll/+merge/66872
New lp.app.longpoll package.
This provides a higher-level API on top of lp.services.messaging. It's
primary goal for now is to make it trivially easy to connect events in
the application to an interested page.
For example, an event in the application can be generated as simply
as:
emit(an_object, "event_name", {...})
and a page being prepared can arrange to receive these event via
long-poll with a single line of code:
subscribe(an_object, "event_name")
Behind the scenes a new, short-lived, queue will be created and
subscribed to the correct routing key for the event. The in-page JSON
cache will have the queue details added to it so that the client can
immediately long-poll for events.
This branch contains the server-side code. The client-side code is
being worked on by Raphael.
To support the emit/subscribe two-step a single adapter must be
registered. It must (multi-) adapt the object that issues events and
an event description to an ILongPollEvent. This branch contains a
suitable base-class, LongPollEvent. Suppose a Job emits events from
_set_status using emit(self, "status", status.name) the following
adapter would work:
class JobLongPollEvent(LongPollEvent):
adapts(IJob, Interface)
implements(ILongPollEvent)
@property
def event_key(self):
return generate_event_key(
"job", self.source.id, self.event)
This ensures that there's a stable and consistent event naming scheme.
--
https://code.launchpad.net/~allenap/launchpad/lp-app-longpoll/+merge/66872
Your team Launchpad code reviewers is requested to review the proposed merge of lp:~allenap/launchpad/lp-app-longpoll into lp:launchpad.
=== modified file 'lib/lp/app/configure.zcml'
--- lib/lp/app/configure.zcml 2011-05-27 21:25:58 +0000
+++ lib/lp/app/configure.zcml 2011-07-05 09:48:49 +0000
@@ -10,6 +10,7 @@
xmlns:lp="http://namespaces.canonical.com/lp";
i18n_domain="launchpad">
<include package=".browser"/>
+ <include package=".longpoll" />
<include package="lp.app.validators" />
<include package="lp.app.widgets" />
=== added directory 'lib/lp/app/longpoll'
=== added file 'lib/lp/app/longpoll/__init__.py'
--- lib/lp/app/longpoll/__init__.py 1970-01-01 00:00:00 +0000
+++ lib/lp/app/longpoll/__init__.py 2011-07-05 09:48:49 +0000
@@ -0,0 +1,47 @@
+# Copyright 2011 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Long-poll infrastructure."""
+
+__metaclass__ = type
+__all__ = [
+ "emit",
+ "subscribe",
+ ]
+
+from .interfaces import (
+ ILongPollEvent,
+ ILongPollSubscriber,
+ )
+from lazr.restful.utils import get_current_browser_request
+from zope.component import getMultiAdapter
+
+
+def subscribe(target, event, request=None):
+ """Convenience method to subscribe the current request.
+
+ :param target: Something that can be adapted to `ILongPollEvent`.
+ :param event: The name of the event to subscribe to.
+ :param request: The request for which to get an `ILongPollSubscriber`. It
+ a request is not specified the currently active request is used.
+ :return: The `ILongPollEvent` that has been subscribed to.
+ """
+ event = getMultiAdapter((target, event), ILongPollEvent)
+ if request is None:
+ request = get_current_browser_request()
+ subscriber = ILongPollSubscriber(request)
+ subscriber.subscribe(event)
+ return event
+
+
+def emit(source, event, data):
+ """Convenience method to emit a message for an event.
+
+ :param source: Something, along with `event`, that can be adapted to
+ `ILongPollEvent`.
+ :param event: A name/key of the event that is emitted.
+ :return: The `ILongPollEvent` that has been emitted.
+ """
+ event = getMultiAdapter((source, event), ILongPollEvent)
+ event.emit(data)
+ return event
=== added directory 'lib/lp/app/longpoll/adapters'
=== added file 'lib/lp/app/longpoll/adapters/__init__.py'
=== added file 'lib/lp/app/longpoll/adapters/event.py'
--- lib/lp/app/longpoll/adapters/event.py 1970-01-01 00:00:00 +0000
+++ lib/lp/app/longpoll/adapters/event.py 2011-07-05 09:48:49 +0000
@@ -0,0 +1,43 @@
+# Copyright 2011 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Long poll adapters."""
+
+__metaclass__ = type
+__all__ = [
+ "generate_event_key",
+ "LongPollEvent",
+ ]
+
+from lp.services.messaging.queue import RabbitRoutingKey
+
+
+def generate_event_key(*components):
+ """Generate a suitable event name."""
+ if len(components) == 0:
+ raise AssertionError(
+ "Event keys must contain at least one component.")
+ return "longpoll.event.%s" % ".".join(
+ str(component) for component in components)
+
+
+class LongPollEvent:
+ """Base-class for event adapters."""
+
+ #adapts(Interface, Interface)
+ #implements(ILongPollEvent)
+
+ def __init__(self, source, event):
+ self.source = source
+ self.event = event
+
+ @property
+ def event_key(self):
+ """See `ILongPollEvent`."""
+ raise NotImplementedError(self.__class__.event_key)
+
+ def emit(self, data):
+ """See `ILongPollEvent`."""
+ payload = {"event_key": self.event_key, "event_data": data}
+ routing_key = RabbitRoutingKey(self.event_key)
+ routing_key.send(payload)
=== added file 'lib/lp/app/longpoll/adapters/subscriber.py'
--- lib/lp/app/longpoll/adapters/subscriber.py 1970-01-01 00:00:00 +0000
+++ lib/lp/app/longpoll/adapters/subscriber.py 2011-07-05 09:48:49 +0000
@@ -0,0 +1,57 @@
+# Copyright 2011 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Long poll adapters."""
+
+__metaclass__ = type
+__all__ = [
+ "generate_subscribe_key",
+ "LongPollSubscriber",
+ ]
+
+from uuid import uuid4
+
+from lazr.restful.interfaces import IJSONRequestCache
+from zope.component import adapts
+from zope.interface import implements
+from zope.publisher.interfaces import IApplicationRequest
+
+from lp.app.longpoll.interfaces import ILongPollSubscriber
+from lp.services.messaging.queue import (
+ RabbitQueue,
+ RabbitRoutingKey,
+ )
+
+
+def generate_subscribe_key():
+ """Generate a suitable new, unique, subscribe key."""
+ return "longpoll.subscribe.%s" % uuid4()
+
+
+class LongPollSubscriber:
+
+ adapts(IApplicationRequest)
+ implements(ILongPollSubscriber)
+
+ def __init__(self, request):
+ self.request = request
+
+ @property
+ def subscribe_key(self):
+ objects = IJSONRequestCache(self.request).objects
+ if "longpoll" in objects:
+ return objects["longpoll"]["key"]
+ return None
+
+ def subscribe(self, event):
+ cache = IJSONRequestCache(self.request)
+ if "longpoll" not in cache.objects:
+ cache.objects["longpoll"] = {
+ # TODO: Add something descriptive into the key.
+ "key": generate_subscribe_key(),
+ "subscriptions": [],
+ }
+ subscribe_queue = RabbitQueue(self.subscribe_key)
+ routing_key = RabbitRoutingKey(event.event_key)
+ routing_key.associateConsumer(subscribe_queue)
+ cache.objects["longpoll"]["subscriptions"].append(event.event_key)
=== added directory 'lib/lp/app/longpoll/adapters/tests'
=== added file 'lib/lp/app/longpoll/adapters/tests/__init__.py'
=== added file 'lib/lp/app/longpoll/adapters/tests/test_event.py'
--- lib/lp/app/longpoll/adapters/tests/test_event.py 1970-01-01 00:00:00 +0000
+++ lib/lp/app/longpoll/adapters/tests/test_event.py 2011-07-05 09:48:49 +0000
@@ -0,0 +1,84 @@
+# Copyright 2011 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Long-poll event adapter tests."""
+
+__metaclass__ = type
+
+from zope.interface import implements
+
+from canonical.testing.layers import (
+ BaseLayer,
+ LaunchpadFunctionalLayer,
+ )
+from lp.app.longpoll.adapters.event import (
+ generate_event_key,
+ LongPollEvent,
+ )
+from lp.app.longpoll.interfaces import ILongPollEvent
+from lp.services.messaging.queue import RabbitMessageBase
+from lp.testing import TestCase
+from lp.testing.matchers import Contains
+
+
+class FakeEvent(LongPollEvent):
+
+ implements(ILongPollEvent)
+
+ @property
+ def event_key(self):
+ return "event-key-%s-%s" % (self.source, self.event)
+
+
+class TestLongPollEvent(TestCase):
+
+ layer = LaunchpadFunctionalLayer
+
+ def test_interface(self):
+ event = FakeEvent("source", "event")
+ self.assertProvides(event, ILongPollEvent)
+
+ def test_event_key(self):
+ # event_key is not implemented in LongPollEvent; subclasses must
+ # provide it.
+ event = LongPollEvent("source", "event")
+ self.assertRaises(NotImplementedError, getattr, event, "event_key")
+
+ def test_emit(self):
+ # LongPollEvent.emit() sends the given data to `event_key`.
+ event = FakeEvent("source", "event")
+ event_data = {"hello": 1234}
+ event.emit(event_data)
+ expected_message = {
+ "event_key": event.event_key,
+ "event_data": event_data,
+ }
+ pending_messages = [
+ message for (call, message) in
+ RabbitMessageBase.class_locals.messages]
+ self.assertThat(pending_messages, Contains(expected_message))
+
+
+class TestModule(TestCase):
+
+ layer = BaseLayer
+
+ def test_generate_event_key_no_components(self):
+ self.assertRaises(
+ AssertionError, generate_event_key)
+
+ def test_generate_event_key(self):
+ self.assertEqual(
+ "longpoll.event.event-name",
+ generate_event_key("event-name"))
+ self.assertEqual(
+ "longpoll.event.source-name.event-name",
+ generate_event_key("source-name", "event-name"))
+ self.assertEqual(
+ "longpoll.event.type-name.source-name.event-name",
+ generate_event_key("type-name", "source-name", "event-name"))
+
+ def test_generate_event_key_stringifies_components(self):
+ self.assertEqual(
+ "longpoll.event.job.1234.COMPLETED",
+ generate_event_key("job", 1234, "COMPLETED"))
=== added file 'lib/lp/app/longpoll/adapters/tests/test_subscriber.py'
--- lib/lp/app/longpoll/adapters/tests/test_subscriber.py 1970-01-01 00:00:00 +0000
+++ lib/lp/app/longpoll/adapters/tests/test_subscriber.py 2011-07-05 09:48:49 +0000
@@ -0,0 +1,134 @@
+# Copyright 2011 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Long-poll subscriber adapter tests."""
+
+__metaclass__ = type
+
+from itertools import count
+
+from lazr.restful.interfaces import IJSONRequestCache
+from testtools.matchers import (
+ Not,
+ StartsWith,
+ )
+from zope.interface import implements
+
+from canonical.launchpad.webapp.servers import LaunchpadTestRequest
+from canonical.testing.layers import (
+ BaseLayer,
+ LaunchpadFunctionalLayer,
+ )
+from lp.app.longpoll.adapters.subscriber import (
+ generate_subscribe_key,
+ LongPollSubscriber,
+ )
+from lp.app.longpoll.interfaces import (
+ ILongPollEvent,
+ ILongPollSubscriber,
+ )
+from lp.services.messaging.queue import (
+ RabbitQueue,
+ RabbitRoutingKey,
+ )
+from lp.testing import TestCase
+from lp.testing.matchers import Contains
+
+
+class FakeEvent:
+
+ implements(ILongPollEvent)
+
+ event_key_indexes = count(1)
+
+ def __init__(self):
+ self.event_key = "event-key-%d" % next(self.event_key_indexes)
+
+
+class TestLongPollSubscriber(TestCase):
+
+ layer = LaunchpadFunctionalLayer
+
+ def test_interface(self):
+ request = LaunchpadTestRequest()
+ subscriber = LongPollSubscriber(request)
+ self.assertProvides(subscriber, ILongPollSubscriber)
+
+ def test_subscribe_key(self):
+ request = LaunchpadTestRequest()
+ subscriber = LongPollSubscriber(request)
+ # A subscribe key is not generated yet.
+ self.assertIs(subscriber.subscribe_key, None)
+ # It it only generated on the first subscription.
+ subscriber.subscribe(FakeEvent())
+ subscribe_key = subscriber.subscribe_key
+ self.assertIsInstance(subscribe_key, str)
+ self.assertNotEqual(0, len(subscribe_key))
+ # It remains the same for later subscriptions.
+ subscriber.subscribe(FakeEvent())
+ self.assertEqual(subscribe_key, subscriber.subscribe_key)
+
+ def test_adapter(self):
+ request = LaunchpadTestRequest()
+ subscriber = ILongPollSubscriber(request)
+ self.assertIsInstance(subscriber, LongPollSubscriber)
+ # A difference subscriber is returned on subsequent adaptions, but it
+ # has the same subscribe_key.
+ subscriber2 = ILongPollSubscriber(request)
+ self.assertIsNot(subscriber, subscriber2)
+ self.assertEqual(subscriber.subscribe_key, subscriber2.subscribe_key)
+
+ def test_subscribe_queue(self):
+ # LongPollSubscriber.subscribe() creates a new queue with a new unique
+ # name that is bound to the event's event_key.
+ request = LaunchpadTestRequest()
+ event = FakeEvent()
+ subscriber = ILongPollSubscriber(request)
+ subscriber.subscribe(event)
+ message = '{"hello": 1234}'
+ routing_key = RabbitRoutingKey(event.event_key)
+ routing_key.send_now(message)
+ subscribe_queue = RabbitQueue(subscriber.subscribe_key)
+ self.assertEqual(
+ message, subscribe_queue.receive(timeout=5))
+
+ def test_json_cache_not_populated_on_init(self):
+ # LongPollSubscriber does not put the name of the new queue into the
+ # JSON cache.
+ request = LaunchpadTestRequest()
+ cache = IJSONRequestCache(request)
+ self.assertThat(cache.objects, Not(Contains("longpoll")))
+ ILongPollSubscriber(request)
+ self.assertThat(cache.objects, Not(Contains("longpoll")))
+
+ def test_json_cache_populated_on_subscribe(self):
+ # To aid with debugging the event_key of subscriptions are added to
+ # the JSON cache.
+ request = LaunchpadTestRequest()
+ cache = IJSONRequestCache(request)
+ event1 = FakeEvent()
+ ILongPollSubscriber(request).subscribe(event1) # Side-effects!
+ self.assertThat(cache.objects, Contains("longpoll"))
+ self.assertThat(cache.objects["longpoll"], Contains("key"))
+ self.assertThat(cache.objects["longpoll"], Contains("subscriptions"))
+ self.assertEqual(
+ [event1.event_key],
+ cache.objects["longpoll"]["subscriptions"])
+ # More events can be subscribed.
+ event2 = FakeEvent()
+ ILongPollSubscriber(request).subscribe(event2)
+ self.assertEqual(
+ [event1.event_key, event2.event_key],
+ cache.objects["longpoll"]["subscriptions"])
+
+
+class TestModule(TestCase):
+
+ layer = BaseLayer
+
+ def test_generate_subscribe_key(self):
+ subscribe_key = generate_subscribe_key()
+ expected_prefix = "longpoll.subscribe."
+ self.assertThat(subscribe_key, StartsWith(expected_prefix))
+ # The key contains a 36 character UUID.
+ self.assertEqual(len(expected_prefix) + 36, len(subscribe_key))
=== added file 'lib/lp/app/longpoll/configure.zcml'
--- lib/lp/app/longpoll/configure.zcml 1970-01-01 00:00:00 +0000
+++ lib/lp/app/longpoll/configure.zcml 2011-07-05 09:48:49 +0000
@@ -0,0 +1,10 @@
+<!-- Copyright 2011 Canonical Ltd. This software is licensed under the
+ GNU Affero General Public License version 3 (see the file LICENSE).
+-->
+<configure
+ xmlns="http://namespaces.zope.org/zope";
+ xmlns:browser="http://namespaces.zope.org/browser";
+ xmlns:i18n="http://namespaces.zope.org/i18n";
+ i18n_domain="launchpad">
+ <adapter factory=".adapters.subscriber.LongPollSubscriber" />
+</configure>
=== added file 'lib/lp/app/longpoll/interfaces.py'
--- lib/lp/app/longpoll/interfaces.py 1970-01-01 00:00:00 +0000
+++ lib/lp/app/longpoll/interfaces.py 2011-07-05 09:48:49 +0000
@@ -0,0 +1,53 @@
+# Copyright 2011 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Long-poll infrastructure interfaces."""
+
+__metaclass__ = type
+__all__ = [
+ "ILongPollEvent",
+ "ILongPollSubscriber",
+ ]
+
+
+from zope.interface import (
+ Attribute,
+ Interface,
+ )
+
+
+class ILongPollEvent(Interface):
+
+ source = Attribute(
+ "The event source.")
+
+ event = Attribute(
+ "An object indicating the type of event.")
+
+ event_key = Attribute(
+ "The key with which events will be emitted. Should be predictable "
+ "and stable.")
+
+ def emit(data):
+ """Emit the given data to `event_key`.
+
+ The data will be wrapped up into a `dict` with the keys `event_key`
+ and `event_data`, where `event_key` is a copy of `self.event_key` and
+ `event_data` is the `data` argument.
+
+ :param data: Any data structure that can be dumped as JSON.
+ """
+
+
+class ILongPollSubscriber(Interface):
+
+ subscribe_key = Attribute(
+ "The key which the subscriber must know in order to be able "
+ "to long-poll for subscribed events. Should be infeasible to "
+ "guess, a UUID for example.")
+
+ def subscribe(event):
+ """Subscribe to the given event.
+
+ :type event: ILongPollEvent
+ """
=== added directory 'lib/lp/app/longpoll/tests'
=== added file 'lib/lp/app/longpoll/tests/__init__.py'
--- lib/lp/app/longpoll/tests/__init__.py 1970-01-01 00:00:00 +0000
+++ lib/lp/app/longpoll/tests/__init__.py 2011-07-05 09:48:49 +0000
@@ -0,0 +1,2 @@
+# Copyright 2011 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
=== added file 'lib/lp/app/longpoll/tests/test_longpoll.py'
--- lib/lp/app/longpoll/tests/test_longpoll.py 1970-01-01 00:00:00 +0000
+++ lib/lp/app/longpoll/tests/test_longpoll.py 2011-07-05 09:48:49 +0000
@@ -0,0 +1,97 @@
+# Copyright 2011 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Long-poll subscriber adapter tests."""
+
+__metaclass__ = type
+
+from zope.component import adapts
+from zope.interface import (
+ Attribute,
+ implements,
+ Interface,
+ )
+
+from canonical.launchpad.webapp.servers import LaunchpadTestRequest
+from canonical.testing.layers import LaunchpadFunctionalLayer
+from lp.app.longpoll import (
+ emit,
+ subscribe,
+ )
+from lp.app.longpoll.interfaces import (
+ ILongPollEvent,
+ ILongPollSubscriber,
+ )
+from lp.services.messaging.queue import (
+ RabbitQueue,
+ RabbitRoutingKey,
+ )
+from lp.testing import TestCase
+from lp.testing.fixture import ZopeAdapterFixture
+
+
+class IFakeObject(Interface):
+
+ ident = Attribute("ident")
+
+
+class FakeObject:
+
+ implements(IFakeObject)
+
+ def __init__(self, ident):
+ self.ident = ident
+
+
+class FakeEvent:
+
+ adapts(IFakeObject, Interface)
+ implements(ILongPollEvent)
+
+ def __init__(self, source, event):
+ self.source = source
+ self.event = event
+
+ @property
+ def event_key(self):
+ return "event-key-%s-%s" % (
+ self.source.ident, self.event)
+
+ def emit(self, data):
+ # Don't cargo-cult this; see .adapters.event.LongPollEvent instead.
+ RabbitRoutingKey(self.event_key).send_now(data)
+
+
+class TestModule(TestCase):
+
+ layer = LaunchpadFunctionalLayer
+
+ def test_subscribe(self):
+ request = LaunchpadTestRequest()
+ an_object = FakeObject(12345)
+ with ZopeAdapterFixture(FakeEvent):
+ event = subscribe(an_object, "foo", request=request)
+ self.assertIsInstance(event, FakeEvent)
+ self.assertEqual("event-key-12345-foo", event.event_key)
+ # Emitting an event-key-12345-foo event will put something on the
+ # subscriber's queue.
+ event_data = {"1234": 5678}
+ event.emit(event_data)
+ subscriber = ILongPollSubscriber(request)
+ subscribe_queue = RabbitQueue(subscriber.subscribe_key)
+ message = subscribe_queue.receive(timeout=5)
+ self.assertEqual(event_data, message)
+
+ def test_emit(self):
+ an_object = FakeObject(12345)
+ with ZopeAdapterFixture(FakeEvent):
+ event = emit(an_object, "bar", {})
+ routing_key = RabbitRoutingKey(event.event_key)
+ subscribe_queue = RabbitQueue("whatever")
+ routing_key.associateConsumer(subscribe_queue)
+ # Emit the event again; the subscribe queue was not associated
+ # with the event before now.
+ event_data = {"8765": 4321}
+ event = emit(an_object, "bar", event_data)
+ message = subscribe_queue.receive(timeout=5)
+ self.assertEqual(event_data, message)
=== modified file 'lib/lp/testing/fixture.py'
--- lib/lp/testing/fixture.py 2011-06-10 02:30:43 +0000
+++ lib/lp/testing/fixture.py 2011-07-05 09:48:49 +0000
@@ -5,7 +5,9 @@
__metaclass__ = type
__all__ = [
+ 'ZopeAdapterFixture',
'ZopeEventHandlerFixture',
+ 'ZopeViewReplacementFixture',
]
from fixtures import Fixture
@@ -22,6 +24,22 @@
)
+class ZopeAdapterFixture(Fixture):
+ """A fixture to register and unregister an adapter."""
+
+ def __init__(self, *args, **kwargs):
+ self._args, self._kwargs = args, kwargs
+
+ def setUp(self):
+ super(ZopeAdapterFixture, self).setUp()
+ site_manager = getGlobalSiteManager()
+ site_manager.registerAdapter(
+ *self._args, **self._kwargs)
+ self.addCleanup(
+ site_manager.unregisterAdapter,
+ *self._args, **self._kwargs)
+
+
class ZopeEventHandlerFixture(Fixture):
"""A fixture that provides and then unprovides a Zope event handler."""
=== added file 'lib/lp/testing/tests/test_fixture.py'
--- lib/lp/testing/tests/test_fixture.py 1970-01-01 00:00:00 +0000
+++ lib/lp/testing/tests/test_fixture.py 2011-07-05 09:48:49 +0000
@@ -0,0 +1,61 @@
+# Copyright 2011 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Tests for lp.testing.fixture."""
+
+__metaclass__ = type
+
+from zope.component import (
+ adapts,
+ queryAdapter,
+ )
+from zope.interface import (
+ implements,
+ Interface,
+ )
+
+from canonical.testing.layers import BaseLayer
+from lp.testing import TestCase
+from lp.testing.fixture import ZopeAdapterFixture
+
+
+class IFoo(Interface):
+ pass
+
+
+class IBar(Interface):
+ pass
+
+
+class Foo:
+ implements(IFoo)
+
+
+class Bar:
+ implements(IBar)
+
+
+class FooToBar:
+
+ adapts(IFoo)
+ implements(IBar)
+
+ def __init__(self, foo):
+ self.foo = foo
+
+
+class TestZopeAdapterFixture(TestCase):
+
+ layer = BaseLayer
+
+ def test_register_and_unregister(self):
+ context = Foo()
+ # No adapter from Foo to Bar is registered.
+ self.assertIs(None, queryAdapter(context, IBar))
+ with ZopeAdapterFixture(FooToBar):
+ # Now there is an adapter from Foo to Bar.
+ adapter = queryAdapter(context, IBar)
+ self.assertIsNot(None, adapter)
+ self.assertIsInstance(adapter, FooToBar)
+ # Again, it's no longer registered.
+ self.assertIs(None, queryAdapter(context, IBar))
