diff options
| author | Michael Merickel <michael@merickel.org> | 2018-10-14 21:11:41 -0500 |
|---|---|---|
| committer | Michael Merickel <michael@merickel.org> | 2018-10-14 21:11:41 -0500 |
| commit | 3670c2cdb732d378ba6d38e72e7cd875ff726aa9 (patch) | |
| tree | 5213452a778c992d42602efe7d3b3655a349abd5 /tests/test_config/test_adapters.py | |
| parent | 2b024920847481592b1a13d4006d2a9fa8881d72 (diff) | |
| download | pyramid-3670c2cdb732d378ba6d38e72e7cd875ff726aa9.tar.gz pyramid-3670c2cdb732d378ba6d38e72e7cd875ff726aa9.tar.bz2 pyramid-3670c2cdb732d378ba6d38e72e7cd875ff726aa9.zip | |
move tests out of the package
Diffstat (limited to 'tests/test_config/test_adapters.py')
| -rw-r--r-- | tests/test_config/test_adapters.py | 365 |
1 files changed, 365 insertions, 0 deletions
diff --git a/tests/test_config/test_adapters.py b/tests/test_config/test_adapters.py new file mode 100644 index 000000000..ab5d6ef61 --- /dev/null +++ b/tests/test_config/test_adapters.py @@ -0,0 +1,365 @@ +import unittest + +from pyramid.compat import PY2 +from pyramid.tests.test_config import IDummy + +class AdaptersConfiguratorMixinTests(unittest.TestCase): + def _makeOne(self, *arg, **kw): + from pyramid.config import Configurator + config = Configurator(*arg, **kw) + return config + + def test_add_subscriber_defaults(self): + from zope.interface import implementer + from zope.interface import Interface + class IEvent(Interface): + pass + @implementer(IEvent) + class Event: + pass + L = [] + def subscriber(event): + L.append(event) + config = self._makeOne(autocommit=True) + config.add_subscriber(subscriber) + event = Event() + config.registry.notify(event) + self.assertEqual(len(L), 1) + self.assertEqual(L[0], event) + config.registry.notify(object()) + self.assertEqual(len(L), 2) + + def test_add_subscriber_iface_specified(self): + from zope.interface import implementer + from zope.interface import Interface + class IEvent(Interface): + pass + @implementer(IEvent) + class Event: + pass + L = [] + def subscriber(event): + L.append(event) + config = self._makeOne(autocommit=True) + config.add_subscriber(subscriber, IEvent) + event = Event() + config.registry.notify(event) + self.assertEqual(len(L), 1) + self.assertEqual(L[0], event) + config.registry.notify(object()) + self.assertEqual(len(L), 1) + + def test_add_subscriber_dottednames(self): + import pyramid.tests.test_config + from pyramid.interfaces import INewRequest + config = self._makeOne(autocommit=True) + config.add_subscriber('pyramid.tests.test_config', + 'pyramid.interfaces.INewRequest') + handlers = list(config.registry.registeredHandlers()) + self.assertEqual(len(handlers), 1) + handler = handlers[0] + self.assertEqual(handler.handler, pyramid.tests.test_config) + self.assertEqual(handler.required, (INewRequest,)) + + def test_add_object_event_subscriber(self): + from zope.interface import implementer + from zope.interface import Interface + class IEvent(Interface): + pass + @implementer(IEvent) + class Event: + object = 'foo' + event = Event() + L = [] + def subscriber(object, event): + L.append(event) + config = self._makeOne(autocommit=True) + config.add_subscriber(subscriber, (Interface, IEvent)) + config.registry.subscribers((event.object, event), None) + self.assertEqual(len(L), 1) + self.assertEqual(L[0], event) + config.registry.subscribers((event.object, IDummy), None) + self.assertEqual(len(L), 1) + + def test_add_subscriber_with_specific_type_and_predicates_True(self): + from zope.interface import implementer + from zope.interface import Interface + class IEvent(Interface): + pass + @implementer(IEvent) + class Event: + pass + L = [] + def subscriber(event): + L.append(event) + config = self._makeOne(autocommit=True) + predlist = config.get_predlist('subscriber') + jam_predicate = predicate_maker('jam') + jim_predicate = predicate_maker('jim') + predlist.add('jam', jam_predicate) + predlist.add('jim', jim_predicate) + config.add_subscriber(subscriber, IEvent, jam=True, jim=True) + event = Event() + event.jam = True + event.jim = True + config.registry.notify(event) + self.assertEqual(len(L), 1) + self.assertEqual(L[0], event) + config.registry.notify(object()) + self.assertEqual(len(L), 1) + + def test_add_subscriber_with_default_type_predicates_True(self): + from zope.interface import implementer + from zope.interface import Interface + class IEvent(Interface): + pass + @implementer(IEvent) + class Event: + pass + L = [] + def subscriber(event): + L.append(event) + config = self._makeOne(autocommit=True) + predlist = config.get_predlist('subscriber') + jam_predicate = predicate_maker('jam') + jim_predicate = predicate_maker('jim') + predlist.add('jam', jam_predicate) + predlist.add('jim', jim_predicate) + config.add_subscriber(subscriber, jam=True, jim=True) + event = Event() + event.jam = True + event.jim = True + config.registry.notify(event) + self.assertEqual(len(L), 1) + self.assertEqual(L[0], event) + config.registry.notify(object()) + self.assertEqual(len(L), 1) + + def test_add_subscriber_with_specific_type_and_predicates_False(self): + from zope.interface import implementer + from zope.interface import Interface + class IEvent(Interface): + pass + @implementer(IEvent) + class Event: + pass + L = [] + def subscriber(event): L.append(event) + config = self._makeOne(autocommit=True) + predlist = config.get_predlist('subscriber') + jam_predicate = predicate_maker('jam') + jim_predicate = predicate_maker('jim') + predlist.add('jam', jam_predicate) + predlist.add('jim', jim_predicate) + config.add_subscriber(subscriber, IEvent, jam=True, jim=True) + event = Event() + event.jam = True + event.jim = False + config.registry.notify(event) + self.assertEqual(len(L), 0) + + def test_add_subscriber_with_default_type_predicates_False(self): + from zope.interface import implementer + from zope.interface import Interface + class IEvent(Interface): + pass + @implementer(IEvent) + class Event: + pass + L = [] + def subscriber(event): L.append(event) + config = self._makeOne(autocommit=True) + predlist = config.get_predlist('subscriber') + jam_predicate = predicate_maker('jam') + jim_predicate = predicate_maker('jim') + predlist.add('jam', jam_predicate) + predlist.add('jim', jim_predicate) + config.add_subscriber(subscriber, jam=True, jim=True) + event = Event() + event.jam = False + event.jim = True + config.registry.notify(event) + self.assertEqual(len(L), 0) + + def test_add_subscriber_predicate(self): + config = self._makeOne() + L = [] + def add_predicate(type, name, factory, weighs_less_than=None, + weighs_more_than=None): + self.assertEqual(type, 'subscriber') + self.assertEqual(name, 'name') + self.assertEqual(factory, 'factory') + self.assertEqual(weighs_more_than, 1) + self.assertEqual(weighs_less_than, 2) + L.append(1) + config._add_predicate = add_predicate + config.add_subscriber_predicate('name', 'factory', 1, 2) + self.assertTrue(L) + + def test_add_response_adapter(self): + from pyramid.interfaces import IResponse + config = self._makeOne(autocommit=True) + class Adapter(object): + def __init__(self, other): + self.other = other + config.add_response_adapter(Adapter, str) + result = config.registry.queryAdapter('foo', IResponse) + self.assertTrue(result.other, 'foo') + + def test_add_response_adapter_self(self): + from pyramid.interfaces import IResponse + config = self._makeOne(autocommit=True) + class Adapter(object): + pass + config.add_response_adapter(None, Adapter) + adapter = Adapter() + result = config.registry.queryAdapter(adapter, IResponse) + self.assertTrue(result is adapter) + + def test_add_response_adapter_dottednames(self): + from pyramid.interfaces import IResponse + config = self._makeOne(autocommit=True) + if PY2: + str_name = '__builtin__.str' + else: + str_name = 'builtins.str' + config.add_response_adapter('pyramid.response.Response', str_name) + result = config.registry.queryAdapter('foo', IResponse) + self.assertTrue(result.body, b'foo') + + def test_add_traverser_dotted_names(self): + from pyramid.interfaces import ITraverser + config = self._makeOne(autocommit=True) + config.add_traverser( + 'pyramid.tests.test_config.test_adapters.DummyTraverser', + 'pyramid.tests.test_config.test_adapters.DummyIface') + iface = DummyIface() + traverser = config.registry.getAdapter(iface, ITraverser) + self.assertEqual(traverser.__class__, DummyTraverser) + self.assertEqual(traverser.root, iface) + + def test_add_traverser_default_iface_means_Interface(self): + from pyramid.interfaces import ITraverser + config = self._makeOne(autocommit=True) + config.add_traverser(DummyTraverser) + traverser = config.registry.getAdapter(None, ITraverser) + self.assertEqual(traverser.__class__, DummyTraverser) + + def test_add_traverser_nondefault_iface(self): + from pyramid.interfaces import ITraverser + config = self._makeOne(autocommit=True) + config.add_traverser(DummyTraverser, DummyIface) + iface = DummyIface() + traverser = config.registry.getAdapter(iface, ITraverser) + self.assertEqual(traverser.__class__, DummyTraverser) + self.assertEqual(traverser.root, iface) + + def test_add_traverser_introspectables(self): + config = self._makeOne() + config.add_traverser(DummyTraverser, DummyIface) + actions = config.action_state.actions + self.assertEqual(len(actions), 1) + intrs = actions[0]['introspectables'] + self.assertEqual(len(intrs), 1) + intr = intrs[0] + self.assertEqual(intr.type_name, 'traverser') + self.assertEqual(intr.discriminator, ('traverser', DummyIface)) + self.assertEqual(intr.category_name, 'traversers') + self.assertEqual(intr.title, 'traverser for %r' % DummyIface) + self.assertEqual(intr['adapter'], DummyTraverser) + self.assertEqual(intr['iface'], DummyIface) + + def test_add_resource_url_adapter_dotted_names(self): + from pyramid.interfaces import IResourceURL + config = self._makeOne(autocommit=True) + config.add_resource_url_adapter( + 'pyramid.tests.test_config.test_adapters.DummyResourceURL', + 'pyramid.tests.test_config.test_adapters.DummyIface', + ) + iface = DummyIface() + adapter = config.registry.getMultiAdapter((iface, iface), + IResourceURL) + self.assertEqual(adapter.__class__, DummyResourceURL) + self.assertEqual(adapter.resource, iface) + self.assertEqual(adapter.request, iface) + + def test_add_resource_url_default_resource_iface_means_Interface(self): + from pyramid.interfaces import IResourceURL + config = self._makeOne(autocommit=True) + config.add_resource_url_adapter(DummyResourceURL) + iface = DummyIface() + adapter = config.registry.getMultiAdapter((iface, iface), + IResourceURL) + self.assertEqual(adapter.__class__, DummyResourceURL) + self.assertEqual(adapter.resource, iface) + self.assertEqual(adapter.request, iface) + + def test_add_resource_url_nodefault_resource_iface(self): + from zope.interface import Interface + from pyramid.interfaces import IResourceURL + config = self._makeOne(autocommit=True) + config.add_resource_url_adapter(DummyResourceURL, DummyIface) + iface = DummyIface() + adapter = config.registry.getMultiAdapter((iface, iface), + IResourceURL) + self.assertEqual(adapter.__class__, DummyResourceURL) + self.assertEqual(adapter.resource, iface) + self.assertEqual(adapter.request, iface) + bad_result = config.registry.queryMultiAdapter( + (Interface, Interface), + IResourceURL, + ) + self.assertEqual(bad_result, None) + + def test_add_resource_url_adapter_introspectables(self): + config = self._makeOne() + config.add_resource_url_adapter(DummyResourceURL, DummyIface) + actions = config.action_state.actions + self.assertEqual(len(actions), 1) + intrs = actions[0]['introspectables'] + self.assertEqual(len(intrs), 1) + intr = intrs[0] + self.assertEqual(intr.type_name, 'resource url adapter') + self.assertEqual(intr.discriminator, + ('resource url adapter', DummyIface)) + self.assertEqual(intr.category_name, 'resource url adapters') + self.assertEqual( + intr.title, + "resource url adapter for resource iface " + "<class 'pyramid.tests.test_config.test_adapters.DummyIface'>" + ) + self.assertEqual(intr['adapter'], DummyResourceURL) + self.assertEqual(intr['resource_iface'], DummyIface) + +class Test_eventonly(unittest.TestCase): + def _callFUT(self, callee): + from pyramid.config.adapters import eventonly + return eventonly(callee) + + def test_defaults(self): + def acallable(event, a=1, b=2): pass + self.assertTrue(self._callFUT(acallable)) + +class DummyTraverser(object): + def __init__(self, root): + self.root = root + +class DummyIface(object): + pass + +class DummyResourceURL(object): + def __init__(self, resource, request): + self.resource = resource + self.request = request + +def predicate_maker(name): + class Predicate(object): + def __init__(self, val, config): + self.val = val + def phash(self): + return 'phash' + text = phash + def __call__(self, event): + return getattr(event, name, None) == self.val + return Predicate + |
