summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorChris McDonough <chrism@agendaless.com>2009-11-16 18:58:34 +0000
committerChris McDonough <chrism@agendaless.com>2009-11-16 18:58:34 +0000
commit58fdd1e948b7223cbcaf4fdceb159de200356d79 (patch)
tree162985896bd8ff3e88069029caf55ed5b5e0ec1b
parent131f5f3046eb71145ebeb2a05b90bd89ec829dd9 (diff)
downloadpyramid-58fdd1e948b7223cbcaf4fdceb159de200356d79.tar.gz
pyramid-58fdd1e948b7223cbcaf4fdceb159de200356d79.tar.bz2
pyramid-58fdd1e948b7223cbcaf4fdceb159de200356d79.zip
Merge imperativeconfig branch.
-rw-r--r--repoze/bfg/configuration.py18
-rw-r--r--repoze/bfg/registry.py408
-rw-r--r--repoze/bfg/testing.py9
-rw-r--r--repoze/bfg/tests/test_configuration.py21
-rw-r--r--repoze/bfg/tests/test_integration.py50
-rw-r--r--repoze/bfg/tests/test_registry.py120
-rw-r--r--repoze/bfg/tests/test_zcml.py500
-rw-r--r--repoze/bfg/zcml.py457
8 files changed, 806 insertions, 777 deletions
diff --git a/repoze/bfg/configuration.py b/repoze/bfg/configuration.py
index 077b72438..2b74017dc 100644
--- a/repoze/bfg/configuration.py
+++ b/repoze/bfg/configuration.py
@@ -21,6 +21,7 @@ from repoze.bfg.interfaces import ISettings
from repoze.bfg.authorization import ACLAuthorizationPolicy
from repoze.bfg.log import make_stream_logger
from repoze.bfg.registry import Registry
+from repoze.bfg.registry import DefaultRootFactory
from repoze.bfg.settings import Settings
from repoze.bfg.settings import get_options
from repoze.bfg.threadlocal import get_current_registry
@@ -62,11 +63,10 @@ def make_registry(root_factory, package=None, filename='configure.zcml',
if root_factory is None:
root_factory = DefaultRootFactory
- # register the *default* root factory so apps can find it later
- registry.registerUtility(root_factory, IDefaultRootFactory)
-
mapper = RoutesRootFactory(root_factory)
registry.registerUtility(mapper, IRoutesMapper)
+ # register the *default* root factory so apps can find it later
+ registry.registerUtility(root_factory, IDefaultRootFactory)
if authentication_policy:
debug_logger.warn(
@@ -113,6 +113,8 @@ def make_registry(root_factory, package=None, filename='configure.zcml',
lock.release()
manager.pop()
+ mapper = registry.getUtility(IRoutesMapper)
+
if mapper.has_routes():
# if the user had any <route/> statements in his configuration,
# use the RoutesRootFactory as the IRootFactory; otherwise use the
@@ -124,16 +126,6 @@ def make_registry(root_factory, package=None, filename='configure.zcml',
return registry
-class DefaultRootFactory:
- __parent__ = None
- __name__ = None
- def __init__(self, request):
- matchdict = getattr(request, 'matchdict', {})
- # provide backwards compatibility for applications which
- # used routes (at least apps without any custom "context
- # factory") in BFG 0.9.X and before
- self.__dict__.update(matchdict)
-
def zcml_configure(name, package):
""" Given a ZCML filename as ``name`` and a Python package as
``package`` which the filename should be relative to, load the
diff --git a/repoze/bfg/registry.py b/repoze/bfg/registry.py
index e93d9dcaf..dfd940af5 100644
--- a/repoze/bfg/registry.py
+++ b/repoze/bfg/registry.py
@@ -1,11 +1,50 @@
+import re
+import sys
+
from zope.component.registry import Components
+from zope.configuration.exceptions import ConfigurationError
+
+from zope.interface import Interface
+from zope.interface import implementedBy
+from zope.interface.interfaces import IInterface
+
+import martian
+
+from repoze.bfg.interfaces import IAuthenticationPolicy
+from repoze.bfg.interfaces import IAuthorizationPolicy
+from repoze.bfg.interfaces import IForbiddenView
+from repoze.bfg.interfaces import IMultiView
+from repoze.bfg.interfaces import INotFoundView
+from repoze.bfg.interfaces import IPackageOverrides
+from repoze.bfg.interfaces import IRendererFactory
+from repoze.bfg.interfaces import IRequest
+from repoze.bfg.interfaces import IRouteRequest
+from repoze.bfg.interfaces import IRoutesMapper
+from repoze.bfg.interfaces import ISecuredView
+from repoze.bfg.interfaces import IView
+from repoze.bfg.interfaces import IViewPermission
+
+from repoze.bfg.request import route_request_iface
+from repoze.bfg.resource import PackageOverrides
+from repoze.bfg.static import StaticRootFactory
+from repoze.bfg.traversal import find_interface
+from repoze.bfg.view import MultiView
+from repoze.bfg.view import derive_view
+from repoze.bfg.view import static as static_view
+from repoze.bfg.urldispatch import RoutesRootFactory
+
class Registry(Components, dict):
# for optimization purposes, if no listeners are listening, don't try
# to notify them
has_listeners = False
+ def __init__(self, name='', bases=()):
+ Components.__init__(self, name=name, bases=bases)
+ mapper = RoutesRootFactory(DefaultRootFactory)
+ self.registerUtility(mapper, IRoutesMapper)
+
def registerSubscriptionAdapter(self, *arg, **kw):
result = Components.registerSubscriptionAdapter(self, *arg, **kw)
self.has_listeners = True
@@ -19,6 +58,371 @@ class Registry(Components, dict):
def notify(self, *events):
if self.has_listeners:
# iterating over subscribers assures they get executed
- for ignored in self.subscribers(events, None):
- """ """
+ [ _ for _ in self.subscribers(events, None) ]
+
+ def view(self, permission=None, for_=None, view=None, name="",
+ request_type=None, route_name=None, request_method=None,
+ request_param=None, containment=None, attr=None,
+ renderer=None, wrapper=None, xhr=False, accept=None,
+ header=None, path_info=None, _info=u''):
+
+ if not view:
+ if renderer:
+ def view(context, request):
+ return {}
+ else:
+ raise ConfigurationError('"view" was not specified and '
+ 'no "renderer" specified')
+
+ if request_type in ('GET', 'HEAD', 'PUT', 'POST', 'DELETE'):
+ # b/w compat for 1.0
+ request_method = request_type
+ request_type = None
+
+ if request_type is None:
+ if route_name is None:
+ request_type = IRequest
+ else:
+ request_type = self.queryUtility(IRouteRequest, name=route_name)
+ if request_type is None:
+ request_type = route_request_iface(route_name)
+ self.registerUtility(request_type, IRouteRequest,
+ name=route_name)
+
+ score, predicates = _make_predicates(
+ xhr=xhr, request_method=request_method, path_info=path_info,
+ request_param=request_param, header=header, accept=accept,
+ containment=containment)
+
+ derived_view = derive_view(view, permission, predicates, attr,
+ renderer, wrapper, name)
+ r_for_ = for_
+ r_request_type = request_type
+ if r_for_ is None:
+ r_for_ = Interface
+ if not IInterface.providedBy(r_for_):
+ r_for_ = implementedBy(r_for_)
+ if not IInterface.providedBy(r_request_type):
+ r_request_type = implementedBy(r_request_type)
+ old_view = self.adapters.lookup((r_for_, r_request_type),
+ IView,name=name)
+ if old_view is None:
+ if hasattr(derived_view, '__call_permissive__'):
+ self.registerAdapter(derived_view, (for_, request_type),
+ ISecuredView, name, info=_info)
+ if hasattr(derived_view, '__permitted__'):
+ # bw compat
+ self.registerAdapter(
+ derived_view.__permitted__,
+ (for_, request_type), IViewPermission,
+ name, info=_info)
+ else:
+ self.registerAdapter(derived_view, (for_, request_type),
+ IView, name, info=_info)
+ else:
+ # XXX we could try to be more efficient here and register
+ # a non-secured view for a multiview if none of the
+ # multiview's consituent views have a permission
+ # associated with them, but this code is getting pretty
+ # rough already
+ if IMultiView.providedBy(old_view):
+ multiview = old_view
+ else:
+ multiview = MultiView(name)
+ multiview.add(old_view, sys.maxint)
+ multiview.add(derived_view, score)
+ for i in (IView, ISecuredView):
+ # unregister any existing views
+ self.adapters.unregister((r_for_, r_request_type), i,
+ name=name)
+ self.registerAdapter(multiview, (for_, request_type),
+ IMultiView, name, info=_info)
+ # b/w compat
+ self.registerAdapter(multiview.__permitted__,
+ (for_, request_type), IViewPermission,
+ name, info=_info)
+
+ def route(self, name, path, view=None, view_for=None,
+ permission=None, factory=None, request_type=None, for_=None,
+ header=None, xhr=False, accept=None, path_info=None,
+ request_method=None, request_param=None,
+ view_permission=None, view_request_type=None,
+ view_request_method=None, view_request_param=None,
+ view_containment=None, view_attr=None,
+ renderer=None, view_renderer=None, view_header=None,
+ view_accept=None, view_xhr=False,
+ view_path_info=None, _info=u''):
+ # the strange ordering of the request kw args above is for b/w
+ # compatibility purposes.
+ # these are route predicates; if they do not match, the next route
+ # in the routelist will be tried
+ _, predicates = _make_predicates(xhr=xhr,
+ request_method=request_method,
+ path_info=path_info,
+ request_param=request_param,
+ header=header,
+ accept=accept)
+
+ if request_type in ('GET', 'HEAD', 'PUT', 'POST', 'DELETE'):
+ # b/w compat for 1.0
+ view_request_method = request_type
+ request_type = None
+
+ request_iface = self.queryUtility(IRouteRequest, name=name)
+ if request_iface is None:
+ request_iface = route_request_iface(name)
+ self.registerUtility(request_iface, IRouteRequest, name=name)
+
+ if view:
+ view_for = view_for or for_
+ view_request_type = view_request_type or request_type
+ view_permission = view_permission or permission
+ view_renderer = view_renderer or renderer
+ self.view(
+ permission=view_permission,
+ for_=view_for,
+ view=view,
+ name='',
+ request_type=view_request_type,
+ route_name=name,
+ request_method=view_request_method,
+ request_param=view_request_param,
+ containment=view_containment,
+ attr=view_attr,
+ renderer=view_renderer,
+ header=view_header,
+ accept=view_accept,
+ xhr=view_xhr,
+ path_info=view_path_info,
+ info=_info,
+ )
+
+ mapper = self.getUtility(IRoutesMapper)
+ mapper.connect(path, name, factory, predicates=predicates)
+
+ def scan(self, package, _info=u'', martian=martian):
+ # martian overrideable only for unit tests
+ multi_grokker = BFGMultiGrokker()
+ multi_grokker.register(BFGViewGrokker())
+ module_grokker = martian.ModuleGrokker(grokker=multi_grokker)
+ martian.grok_dotted_name(
+ package.__name__, grokker=module_grokker,
+ _info=_info, _registry=self,
+ exclude_filter=lambda name: name.startswith('.'))
+
+ def authentication_policy(self, policy, _info=u''):
+ self.registerUtility(policy, IAuthenticationPolicy, info=_info)
+
+ def authorization_policy(self, policy, _info=u''):
+ self.registerUtility(policy, IAuthorizationPolicy, info=_info)
+
+ def renderer(self, factory, name, _info=u''):
+ self.registerUtility(factory, IRendererFactory, name=name, info=_info)
+
+ def resource(self, to_override, override_with, _override=None,
+ _info=u''):
+ if to_override == override_with:
+ raise ConfigurationError('You cannot override a resource with '
+ 'itself')
+
+ package = to_override
+ path = ''
+ if ':' in to_override:
+ package, path = to_override.split(':', 1)
+
+ override_package = override_with
+ override_prefix = ''
+ if ':' in override_with:
+ override_package, override_prefix = override_with.split(':', 1)
+
+ if path and path.endswith('/'):
+ if override_prefix and (not override_prefix.endswith('/')):
+ raise ConfigurationError(
+ 'A directory cannot be overridden with a file (put a slash '
+ 'at the end of override_with if necessary)')
+
+ if override_prefix and override_prefix.endswith('/'):
+ if path and (not path.endswith('/')):
+ raise ConfigurationError(
+ 'A file cannot be overridden with a directory (put a slash '
+ 'at the end of to_override if necessary)')
+
+ __import__(package)
+ __import__(override_package)
+ package = sys.modules[package]
+ override_package = sys.modules[override_package]
+
+ if _override is not None:
+ _override(package, path, override_package, override_prefix)
+ else:
+ self._override(package, path, override_package, override_prefix)
+
+ def _override(self, package, path, override_package, override_prefix,
+ _info=u'', PackageOverrides=PackageOverrides):
+ pkg_name = package.__name__
+ override_pkg_name = override_package.__name__
+ override = self.queryUtility(IPackageOverrides, name=pkg_name)
+ if override is None:
+ override = PackageOverrides(package)
+ self.registerUtility(override, IPackageOverrides, name=pkg_name,
+ info=_info)
+ override.insert(path, override_pkg_name, override_prefix)
+
+
+ def notfound(self, view=None, attr=None, renderer=None, wrapper=None,
+ _info=u''):
+ self._view_utility(view, attr, renderer, wrapper, INotFoundView,
+ _info=_info)
+
+ def forbidden(self, view=None, attr=None, renderer=None, wrapper=None,
+ _info=u''):
+ self._view_utility(view, attr, renderer, wrapper,
+ IForbiddenView, _info=_info)
+
+ def view_utility(self, view, attr, renderer, wrapper, iface, _info=u''):
+ if not view:
+ if renderer:
+ def view(context, request):
+ return {}
+ else:
+ raise ConfigurationError('"view" attribute was not specified and '
+ 'no renderer specified')
+
+ derived_view = derive_view(view, attr=attr, renderer_name=renderer,
+ wrapper_viewname=wrapper)
+ self.registerUtility(derived_view, iface, '', info=_info)
+
+ def static(self, name, path, cache_max_age=3600, _info=u''):
+ view = static_view(path, cache_max_age=cache_max_age)
+ self.route(name, "%s*subpath" % name, view=view,
+ view_for=StaticRootFactory, factory=StaticRootFactory(path),
+ _info=_info)
+
+
+def _make_predicates(xhr=None, request_method=None, path_info=None,
+ request_param=None, header=None, accept=None,
+ containment=None):
+ # Predicates are added to the predicate list in (presumed)
+ # computation expense order. All predicates associated with a
+ # view must evaluate true for the view to "match" a request.
+ # Elsewhere in the code, we evaluate them using a generator
+ # expression. The fastest predicate should be evaluated first,
+ # then the next fastest, and so on, as if one returns false, the
+ # remainder of the predicates won't need to be evaluated.
+
+ # Each predicate is associated with a weight value. The weight
+ # symbolizes the relative potential "importance" of the predicate
+ # to all other predicates. A larger weight indicates greater
+ # importance. These weights are subtracted from an aggregate
+ # 'weight' variable. The aggregate weight is then divided by the
+ # length of the predicate list to compute a "score" for this view.
+ # The score represents the ordering in which a "multiview" ( a
+ # collection of views that share the same context/request/name
+ # triad but differ in other ways via predicates) will attempt to
+ # call its set of views. Views with lower scores will be tried
+ # first. The intent is to a) ensure that views with more
+ # predicates are always evaluated before views with fewer
+ # predicates and b) to ensure a stable call ordering of views that
+ # share the same number of predicates.
+
+ # Views which do not have any predicates get a score of
+ # sys.maxint, meaning that they will be tried very last.
+
+ predicates = []
+ weight = sys.maxint
+
+ if xhr:
+ def xhr_predicate(context, request):
+ return request.is_xhr
+ weight = weight - 10
+ predicates.append(xhr_predicate)
+
+ if request_method is not None:
+ def request_method_predicate(context, request):
+ return request.method == request_method
+ weight = weight - 20
+ predicates.append(request_method_predicate)
+
+ if path_info is not None:
+ try:
+ path_info_val = re.compile(path_info)
+ except re.error, why:
+ raise ConfigurationError(why[0])
+ def path_info_predicate(context, request):
+ return path_info_val.match(request.path_info) is not None
+ weight = weight - 30
+ predicates.append(path_info_predicate)
+
+ if request_param is not None:
+ request_param_val = None
+ if '=' in request_param:
+ request_param, request_param_val = request_param.split('=', 1)
+ def request_param_predicate(context, request):
+ if request_param_val is None:
+ return request_param in request.params
+ return request.params.get(request_param) == request_param_val
+ weight = weight - 40
+ predicates.append(request_param_predicate)
+
+ if header is not None:
+ header_name = header
+ header_val = None
+ if ':' in header:
+ header_name, header_val = header.split(':', 1)
+ try:
+ header_val = re.compile(header_val)
+ except re.error, why:
+ raise ConfigurationError(why[0])
+ def header_predicate(context, request):
+ if header_val is None:
+ return header_name in request.headers
+ val = request.headers.get(header_name)
+ return header_val.match(val) is not None
+ weight = weight - 50
+ predicates.append(header_predicate)
+
+ if accept is not None:
+ def accept_predicate(context, request):
+ return accept in request.accept
+ weight = weight - 60
+ predicates.append(accept_predicate)
+
+ if containment is not None:
+ def containment_predicate(context, request):
+ return find_interface(context, containment) is not None
+ weight = weight - 70
+ predicates.append(containment_predicate)
+
+ # this will be == sys.maxint if no predicates
+ score = weight / (len(predicates) + 1)
+ return score, predicates
+
+class BFGViewMarker(object):
+ pass
+
+class BFGMultiGrokker(martian.core.MultiInstanceOrClassGrokkerBase):
+ def get_bases(self, obj):
+ if hasattr(obj, '__bfg_view_settings__'):
+ return [BFGViewMarker]
+ return []
+
+class BFGViewGrokker(martian.InstanceGrokker):
+ martian.component(BFGViewMarker)
+ def grok(self, name, obj, **kw):
+ config = getattr(obj, '__bfg_view_settings__', [])
+ for settings in config:
+ registry = kw['_registry']
+ info = kw['_info']
+ registry.view(view=obj, _info=info, **settings)
+ return bool(config)
+
+class DefaultRootFactory:
+ __parent__ = None
+ __name__ = None
+ def __init__(self, request):
+ matchdict = getattr(request, 'matchdict', {})
+ # provide backwards compatibility for applications which
+ # used routes (at least apps without any custom "context
+ # factory") in BFG 0.9.X and before
+ self.__dict__.update(matchdict)
diff --git a/repoze/bfg/testing.py b/repoze/bfg/testing.py
index 2df81c5a3..7167ec7b0 100644
--- a/repoze/bfg/testing.py
+++ b/repoze/bfg/testing.py
@@ -232,13 +232,8 @@ def registerRoute(path, name, factory=None):
.. note:: This API was added in :mod:`repoze.bfg` version 1.1.
"""
from repoze.bfg.interfaces import IRoutesMapper
- from zope.component import queryUtility
- from repoze.bfg.urldispatch import RoutesRootFactory
- mapper = queryUtility(IRoutesMapper)
- if mapper is None:
- mapper = RoutesRootFactory(DummyRootFactory)
- sm = getSiteManager()
- sm.registerUtility(mapper, IRoutesMapper)
+ from zope.component import getUtility
+ mapper = getUtility(IRoutesMapper)
mapper.connect(path, name, factory)
def registerRoutesMapper(root_factory=None):
diff --git a/repoze/bfg/tests/test_configuration.py b/repoze/bfg/tests/test_configuration.py
index 669ee2ecf..026002253 100644
--- a/repoze/bfg/tests/test_configuration.py
+++ b/repoze/bfg/tests/test_configuration.py
@@ -217,27 +217,6 @@ class MakeRegistryTests(unittest.TestCase):
self.assertEqual(dummylock.acquired, True)
self.assertEqual(dummylock.released, True)
-class TestDefaultRootFactory(unittest.TestCase):
- def _getTargetClass(self):
- from repoze.bfg.configuration import DefaultRootFactory
- return DefaultRootFactory
-
- def _makeOne(self, environ):
- return self._getTargetClass()(environ)
-
- def test_no_matchdict(self):
- environ = {}
- root = self._makeOne(environ)
- self.assertEqual(root.__parent__, None)
- self.assertEqual(root.__name__, None)
-
- def test_matchdict(self):
- request = DummyRequest()
- request.matchdict = {'a':1, 'b':2}
- root = self._makeOne(request)
- self.assertEqual(root.a, 1)
- self.assertEqual(root.b, 2)
-
class DummyRequest:
pass
diff --git a/repoze/bfg/tests/test_integration.py b/repoze/bfg/tests/test_integration.py
index f05601031..c1779cdb6 100644
--- a/repoze/bfg/tests/test_integration.py
+++ b/repoze/bfg/tests/test_integration.py
@@ -43,10 +43,7 @@ class WGSIAppPlusBFGViewTests(unittest.TestCase):
from repoze.bfg.tests import test_integration
scan(context, test_integration)
actions = context.actions
- self.assertEqual(len(actions), 1)
- action = actions[0]
- register = action['callable']
- register()
+ context.actions[-1]['callable']()
sm = getSiteManager()
view = sm.adapters.lookup((INothing, IRequest), IView, name='')
self.assertEqual(view, wsgiapptest)
@@ -108,47 +105,8 @@ class TestGrokkedApp(unittest.TestCase):
actions = zcml_configure('configure.zcml', package)
actions.sort()
-
- num = 23
-
- action_types = [(actions[x][0][1],
- actions[x][0][3],
- actions[x][0][4]) for x in range(len(actions[:num]))]
-
- for typ in action_types:
- self.assertEqual(typ, (None, IRequest, IView))
-
- action_names = [actions[x][0][2] for x in range(len(actions[:num]))]
- action_names.sort()
-
- self.assertEqual(
- action_names, [
- '',
- '',
- 'another',
- 'another',
- 'another_grokked_class',
- 'another_grokked_instance',
- 'another_oldstyle_grokked_class',
- 'another_stacked1',
- 'another_stacked2',
- 'another_stacked_class1',
- 'another_stacked_class2',
- 'basemethod',
- 'grokked_class',
- 'grokked_instance',
- 'method1',
- 'method2',
- 'oldstyle_grokked_class',
- 'stacked1',
- 'stacked2',
- 'stacked_class1',
- 'stacked_class2',
- 'stacked_method1',
- 'stacked_method2',
- ]
- )
-
+ scan_action = actions[0][1]
+ scan_action()
ctx = DummyContext()
req = DummyRequest()
@@ -243,7 +201,7 @@ class DummyZCMLContext:
self.actions = []
self.info = None
- def action(self, discriminator, callable, args):
+ def action(self, discriminator=None, callable=None, args=None):
self.actions.append(
{'discriminator':discriminator,
'callable':callable,
diff --git a/repoze/bfg/tests/test_registry.py b/repoze/bfg/tests/test_registry.py
index 57e8134c9..3eada7b1c 100644
--- a/repoze/bfg/tests/test_registry.py
+++ b/repoze/bfg/tests/test_registry.py
@@ -1,4 +1,5 @@
import unittest
+from repoze.bfg.testing import cleanUp
class TestRegistry(unittest.TestCase):
def _getTargetClass(self):
@@ -37,3 +38,122 @@ class TestRegistry(unittest.TestCase):
registry.registerSubscriptionAdapter(EventHandler, [IFoo], Interface)
self.assertEqual(registry.has_listeners, True)
+ def test__override_not_yet_registered(self):
+ from repoze.bfg.interfaces import IPackageOverrides
+ package = DummyPackage('package')
+ opackage = DummyPackage('opackage')
+ registry = self._makeOne()
+ registry._override(package, 'path', opackage, 'oprefix',
+ PackageOverrides=DummyOverrides)
+ overrides = registry.queryUtility(IPackageOverrides, name='package')
+ self.assertEqual(overrides.inserted, [('path', 'opackage', 'oprefix')])
+ self.assertEqual(overrides.package, package)
+
+ def test__override_already_registered(self):
+ from repoze.bfg.interfaces import IPackageOverrides
+ package = DummyPackage('package')
+ opackage = DummyPackage('opackage')
+ overrides = DummyOverrides(package)
+ registry = self._makeOne()
+ registry.registerUtility(overrides, IPackageOverrides, name='package')
+ registry._override(package, 'path', opackage, 'oprefix',
+ PackageOverrides=DummyOverrides)
+ self.assertEqual(overrides.inserted, [('path', 'opackage', 'oprefix')])
+ self.assertEqual(overrides.package, package)
+
+class TestBFGViewGrokker(unittest.TestCase):
+ def setUp(self):
+ cleanUp()
+
+ def tearDown(self):
+ cleanUp()
+
+ def _getTargetClass(self):
+ from repoze.bfg.registry import BFGViewGrokker
+ return BFGViewGrokker
+
+ def _makeOne(self, *arg, **kw):
+ return self._getTargetClass()(*arg, **kw)
+
+ def test_grok_is_bfg_view(self):
+ from zope.component import getSiteManager
+ from repoze.bfg.interfaces import IRequest
+ from repoze.bfg.interfaces import IView
+ from zope.interface import Interface
+ grokker = self._makeOne()
+ class obj:
+ def __init__(self, context, request):
+ pass
+ def __call__(self):
+ return 'OK'
+ settings = dict(permission='foo', for_=Interface, name='foo.html',
+ request_type=IRequest, route_name=None,
+ request_method=None, request_param=None,
+ containment=None, attr=None, renderer=None,
+ wrapper=None, xhr=False, header=None,
+ accept=None)
+ obj.__bfg_view_settings__ = [settings]
+ sm = getSiteManager()
+ result = grokker.grok('name', obj, _info='', _registry=sm)
+ self.assertEqual(result, True)
+ wrapped = sm.adapters.lookup((Interface, IRequest), IView,
+ name='foo.html')
+ self.assertEqual(wrapped(None, None), 'OK')
+
+ def test_grok_is_not_bfg_view(self):
+ grokker = self._makeOne()
+ class obj:
+ pass
+ context = DummyContext()
+ result = grokker.grok('name', obj, context=context)
+ self.assertEqual(result, False)
+ actions = context.actions
+ self.assertEqual(len(actions), 0)
+
+class TestDefaultRootFactory(unittest.TestCase):
+ def _getTargetClass(self):
+ from repoze.bfg.registry import DefaultRootFactory
+ return DefaultRootFactory
+
+ def _makeOne(self, environ):
+ return self._getTargetClass()(environ)
+
+ def test_no_matchdict(self):
+ environ = {}
+ root = self._makeOne(environ)
+ self.assertEqual(root.__parent__, None)
+ self.assertEqual(root.__name__, None)
+
+ def test_matchdict(self):
+ class DummyRequest:
+ pass
+ request = DummyRequest()
+ request.matchdict = {'a':1, 'b':2}
+ root = self._makeOne(request)
+ self.assertEqual(root.a, 1)
+ self.assertEqual(root.b, 2)
+
+
+class DummyModule:
+ __path__ = "foo"
+ __name__ = "dummy"
+ __file__ = ''
+
+
+class DummyContext:
+ def __init__(self, resolved=DummyModule):
+ self.actions = []
+ self.info = None
+ self.resolved = resolved
+
+class DummyPackage:
+ def __init__(self, name):
+ self.__name__ = name
+
+class DummyOverrides:
+ def __init__(self, package):
+ self.package = package
+ self.inserted = []
+
+ def insert(self, path, package, prefix):
+ self.inserted.append((path, package, prefix))
diff --git a/repoze/bfg/tests/test_zcml.py b/repoze/bfg/tests/test_zcml.py
index 804fc7b6e..1a7275932 100644
--- a/repoze/bfg/tests/test_zcml.py
+++ b/repoze/bfg/tests/test_zcml.py
@@ -764,9 +764,8 @@ class TestViewDirective(unittest.TestCase):
sm = getSiteManager()
def view(context, request):
""" """
- self.assertRaises(ConfigurationError, self._callFUT,
- context, None, IFoo,
- view=view, header='Host:a\\')
+ self._callFUT(context, None, IFoo, view=view, header='Host:a\\')
+ self.assertRaises(ConfigurationError, context.actions[-1]['callable'])
def test_with_header_noval_match(self):
from zope.component import getSiteManager
@@ -994,9 +993,8 @@ class TestViewDirective(unittest.TestCase):
sm = getSiteManager()
def view(context, request):
""" """
- self.assertRaises(ConfigurationError, self._callFUT,
- context, None, IFoo,
- view=view, path_info='\\')
+ self._callFUT(context, None, IFoo, view=view, path_info='\\')
+ self.assertRaises(ConfigurationError, context.actions[-1]['callable'])
def test_with_path_info_match(self):
from zope.component import getSiteManager
@@ -1535,31 +1533,6 @@ class TestACLAuthorizationPolicyDirective(unittest.TestCase):
policy = getUtility(IAuthorizationPolicy)
self.assertEqual(policy.__class__, ACLAuthorizationPolicy)
-class TestConnectRouteFunction(unittest.TestCase):
- def setUp(self):
- cleanUp()
-
- def tearDown(self):
- cleanUp()
-
- def _callFUT(self, path, name, factory, predicates):
- from repoze.bfg.zcml import connect_route
- return connect_route(path, name, factory, predicates)
-
- def _registerRoutesMapper(self):
- from zope.component import getSiteManager
- sm = getSiteManager()
- mapper = DummyMapper()
- from repoze.bfg.interfaces import IRoutesMapper
- sm.registerUtility(mapper, IRoutesMapper)
- return mapper
-
- def test_defaults(self):
- mapper = self._registerRoutesMapper()
- self._callFUT('path', 'name', 'factory', 'predicates')
- self.assertEqual(mapper.connections, [('path', 'name', 'factory',
- 'predicates')])
-
class TestRouteDirective(unittest.TestCase):
def setUp(self):
cleanUp()
@@ -1571,31 +1544,38 @@ class TestRouteDirective(unittest.TestCase):
from repoze.bfg.zcml import route
return route(*arg, **kw)
+ def _assertRoute(self, name, path, num_predicates=0):
+ from zope.component import getSiteManager
+ from repoze.bfg.interfaces import IRoutesMapper
+ sm = getSiteManager()
+ mapper = sm.getUtility(IRoutesMapper)
+ routes = mapper.get_routes()
+ route = routes[0]
+ self.assertEqual(len(routes), 1)
+ self.assertEqual(route.name, name)
+ self.assertEqual(route.path, path)
+ self.assertEqual(len(routes[0].predicates), num_predicates)
+ return route
+
def test_defaults(self):
- from repoze.bfg.zcml import connect_route
context = DummyContext()
self._callFUT(context, 'name', 'path')
actions = context.actions
self.assertEqual(len(actions), 1)
-
route_action = actions[0]
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
- route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
self.assertEqual(
route_discriminator,
('route', 'name', False, None, None, None, None, None))
- self.assertEqual(route_args[:3], ('path', 'name', None))
- predicates = route_args[3]
- self.assertEqual(len(predicates), 0)
+ route_callable()
+ self._assertRoute('name', 'path')
def test_with_view(self):
from zope.interface import Interface
- from zope.component import getSiteManager
- from repoze.bfg.interfaces import IRouteRequest
- from repoze.bfg.zcml import connect_route
from repoze.bfg.interfaces import IView
+ from repoze.bfg.interfaces import IRouteRequest
+ from zope.component import getSiteManager
context = DummyContext()
def view(context, request):
@@ -1615,7 +1595,6 @@ class TestRouteDirective(unittest.TestCase):
self.assertEqual(view_discriminator, discrim)
register = view_action['callable']
register()
- sm = getSiteManager()
wrapped = sm.adapters.lookup((Interface, request_type), IView, name='')
request = DummyRequest()
self.assertEqual(wrapped(None, request), '123')
@@ -1623,18 +1602,15 @@ class TestRouteDirective(unittest.TestCase):
route_action = actions[1]
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
- route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
self.assertEqual(
route_discriminator,
('route', 'name', False, None, None, None, None, None))
- self.assertEqual(route_args[:3], ('path', 'name', None))
- predicates = route_args[3]
- self.assertEqual(len(predicates), 0)
+
+ route_callable()
+ self._assertRoute('name', 'path')
def test_with_view_and_view_for(self):
from zope.component import getSiteManager
- from repoze.bfg.zcml import connect_route
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
@@ -1661,17 +1637,14 @@ class TestRouteDirective(unittest.TestCase):
route_action = actions[1]
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
- route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
- self.assertEqual(route_discriminator,
- ('route', 'name', False, None, None, None, None,None))
- self.assertEqual(route_args[:3], ('path', 'name', None))
- predicates = route_args[3]
- self.assertEqual(len(predicates), 0)
+ self.assertEqual(
+ route_discriminator,
+ ('route', 'name', False, None, None, None, None, None))
+ route_callable()
+ self._assertRoute('name', 'path')
def test_with_view_and_view_for_alias(self):
from zope.component import getSiteManager
- from repoze.bfg.zcml import connect_route
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
@@ -1698,16 +1671,14 @@ class TestRouteDirective(unittest.TestCase):
route_action = actions[1]
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
- route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
self.assertEqual(route_discriminator,
('route', 'name', False, None, None, None, None,None))
- self.assertEqual(route_args[:3], ('path', 'name', None))
- predicates = route_args[3]
- self.assertEqual(len(predicates), 0)
+ route_callable()
+ self._assertRoute('name', 'path')
def test_without_view(self):
- from repoze.bfg.zcml import connect_route
+ from repoze.bfg.interfaces import IRoutesMapper
+ from zope.component import getSiteManager
context = DummyContext()
self._callFUT(context, 'name', 'path')
actions = context.actions
@@ -1716,15 +1687,19 @@ class TestRouteDirective(unittest.TestCase):
route_action = actions[0]
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
- route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
self.assertEqual(route_discriminator,
('route', 'name', False, None, None, None, None, None))
- self.assertEqual(route_args, ('path', 'name', None, []))
+ route_callable()
+ sm = getSiteManager()
+ mapper = sm.getUtility(IRoutesMapper)
+ routes = mapper.get_routes()
+ self.assertEqual(len(routes), 1)
+ self.assertEqual(routes[0].name, 'name')
+ self.assertEqual(routes[0].path, 'path')
+ self.failIf(routes[0].predicates)
def test_with_view_request_type(self):
from zope.component import getSiteManager
- from repoze.bfg.zcml import connect_route
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
@@ -1751,17 +1726,13 @@ class TestRouteDirective(unittest.TestCase):
route_action = actions[1]
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
- route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
self.assertEqual(route_discriminator,
('route', 'name', False, None, None, None, None,None))
- self.assertEqual(route_args[:3], ('path', 'name', None))
- predicates = route_args[3]
- self.assertEqual(len(predicates), 0)
+ route_callable()
+ self._assertRoute('name', 'path')
def test_with_view_request_type_alias(self):
from zope.component import getSiteManager
- from repoze.bfg.zcml import connect_route
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
@@ -1789,16 +1760,13 @@ class TestRouteDirective(unittest.TestCase):
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
self.assertEqual(route_discriminator,
('route', 'name', False, None, None, None, None,None))
- self.assertEqual(route_args[:3], ('path', 'name', None))
- predicates = route_args[3]
- self.assertEqual(len(predicates), 0)
+ route_callable()
+ self._assertRoute('name', 'path')
def test_with_view_request_method(self):
from zope.component import getSiteManager
- from repoze.bfg.zcml import connect_route
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
@@ -1825,17 +1793,13 @@ class TestRouteDirective(unittest.TestCase):
route_action = actions[1]
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
- route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
self.assertEqual(route_discriminator,
('route', 'name', False, None, None, None, None, None))
- self.assertEqual(route_args[:3], ('path', 'name', None))
- predicates = route_args[3]
- self.assertEqual(len(predicates), 0)
+ route_callable()
+ self._assertRoute('name', 'path')
def test_with_view_containment(self):
from zope.component import getSiteManager
- from repoze.bfg.zcml import connect_route
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
@@ -1861,17 +1825,13 @@ class TestRouteDirective(unittest.TestCase):
route_action = actions[1]
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
- route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
self.assertEqual(route_discriminator,
('route', 'name', False, None, None,None, None, None))
- self.assertEqual(route_args[:3], ('path', 'name', None))
- predicates = route_args[3]
- self.assertEqual(len(predicates), 0)
+ route_callable()
+ self._assertRoute('name', 'path')
def test_with_view_header(self):
from zope.component import getSiteManager
- from repoze.bfg.zcml import connect_route
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
@@ -1897,17 +1857,13 @@ class TestRouteDirective(unittest.TestCase):
route_action = actions[1]
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
- route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
self.assertEqual(route_discriminator,
('route', 'name', False, None, None,None, None, None))
- self.assertEqual(route_args[:3], ('path', 'name', None))
- predicates = route_args[3]
- self.assertEqual(len(predicates), 0)
+ route_callable()
+ self._assertRoute('name', 'path')
def test_with_view_path_info(self):
from zope.component import getSiteManager
- from repoze.bfg.zcml import connect_route
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
@@ -1933,17 +1889,13 @@ class TestRouteDirective(unittest.TestCase):
route_action = actions[1]
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
- route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
self.assertEqual(route_discriminator,
('route', 'name', False, None, None, None, None, None))
- self.assertEqual(route_args[:3], ('path', 'name', None))
- predicates = route_args[3]
- self.assertEqual(len(predicates), 0)
+ route_callable()
+ self._assertRoute('name', 'path')
def test_with_view_xhr(self):
from zope.component import getSiteManager
- from repoze.bfg.zcml import connect_route
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
@@ -1969,17 +1921,13 @@ class TestRouteDirective(unittest.TestCase):
route_action = actions[1]
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
- route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
self.assertEqual(route_discriminator,
('route', 'name', False, None, None, None, None, None))
- self.assertEqual(route_args[:3], ('path', 'name', None))
- predicates = route_args[3]
- self.assertEqual(len(predicates), 0)
+ route_callable()
+ self._assertRoute('name', 'path')
def test_with_view_accept(self):
from zope.component import getSiteManager
- from repoze.bfg.zcml import connect_route
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
@@ -2006,18 +1954,14 @@ class TestRouteDirective(unittest.TestCase):
route_action = actions[1]
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
- route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
self.assertEqual(
route_discriminator,
('route', 'name', False, None, None, None, None, None))
- self.assertEqual(route_args[:3], ('path', 'name', None))
- predicates = route_args[3]
- self.assertEqual(len(predicates), 0)
+ route_callable()
+ self._assertRoute('name', 'path')
def test_with_view_renderer(self):
from zope.component import getSiteManager
- from repoze.bfg.zcml import connect_route
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
from repoze.bfg.interfaces import IRendererFactory
@@ -2050,17 +1994,13 @@ class TestRouteDirective(unittest.TestCase):
route_action = actions[1]
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
- route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
self.assertEqual(route_discriminator,
('route', 'name', False, None, None, None, None, None))
- self.assertEqual(route_args[:3], ('path', 'name', None))
- predicates = route_args[3]
- self.assertEqual(len(predicates), 0)
+ route_callable()
+ self._assertRoute('name', 'path')
def test_with_view_renderer_alias(self):
from zope.component import getSiteManager
- from repoze.bfg.zcml import connect_route
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
from repoze.bfg.interfaces import IRendererFactory
@@ -2092,17 +2032,14 @@ class TestRouteDirective(unittest.TestCase):
route_action = actions[1]
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
- route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
self.assertEqual(route_discriminator,
('route', 'name', False, None, None, None, None, None))
- self.assertEqual(route_args[:3], ('path', 'name', None))
- predicates = route_args[3]
- self.assertEqual(len(predicates), 0)
+ route_callable()
+ self._assertRoute('name', 'path')
+
def test_with_view_permission(self):
from zope.component import getSiteManager
- from repoze.bfg.zcml import connect_route
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
@@ -2131,17 +2068,13 @@ class TestRouteDirective(unittest.TestCase):
route_action = actions[1]
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
- route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
self.assertEqual(route_discriminator,
('route', 'name', False, None, None, None, None, None))
- self.assertEqual(route_args[:3], ('path', 'name', None))
- predicates = route_args[3]
- self.assertEqual(len(predicates), 0)
+ route_callable()
+ self._assertRoute('name', 'path')
def test_with_view_permission_alias(self):
from zope.component import getSiteManager
- from repoze.bfg.zcml import connect_route
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
@@ -2168,17 +2101,13 @@ class TestRouteDirective(unittest.TestCase):
route_action = actions[1]
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
- route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
self.assertEqual(route_discriminator,
('route', 'name', False, None, None, None, None, None))
- self.assertEqual(route_args[:3], ('path', 'name', None))
- predicates = route_args[3]
- self.assertEqual(len(predicates), 0)
+ route_callable()
+ self._assertRoute('name', 'path')
def test_with_view_for(self):
from zope.component import getSiteManager
- from repoze.bfg.zcml import connect_route
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
@@ -2207,17 +2136,13 @@ class TestRouteDirective(unittest.TestCase):
route_action = actions[1]
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
- route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
self.assertEqual(route_discriminator,
('route', 'name', False, None, None, None, None, None))
- self.assertEqual(route_args[:3], ('path', 'name', None))
- predicates = route_args[3]
- self.assertEqual(len(predicates), 0)
+ route_callable()
+ self._assertRoute('name', 'path')
def test_with_view_for_alias(self):
from zope.component import getSiteManager
- from repoze.bfg.zcml import connect_route
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
@@ -2244,17 +2169,13 @@ class TestRouteDirective(unittest.TestCase):
route_action = actions[1]
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
- route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
self.assertEqual(route_discriminator,
('route', 'name', False, None, None, None, None, None))
- self.assertEqual(route_args[:3], ('path', 'name', None))
- predicates = route_args[3]
- self.assertEqual(len(predicates), 0)
+ route_callable()
+ self._assertRoute('name', 'path')
def test_with_request_type_GET(self):
from zope.component import getSiteManager
- from repoze.bfg.zcml import connect_route
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
@@ -2280,19 +2201,15 @@ class TestRouteDirective(unittest.TestCase):
route_action = actions[1]
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
- route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
self.assertEqual(route_discriminator,
('route', 'name', False, None, None, None, None,None))
- self.assertEqual(route_args[:3], ('path', 'name', None))
- predicates = route_args[3]
- self.assertEqual(len(predicates), 0)
+ route_callable()
+ self._assertRoute('name', 'path')
# route predicates
def test_with_xhr(self):
from zope.component import getSiteManager
- from repoze.bfg.zcml import connect_route
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
@@ -2318,20 +2235,16 @@ class TestRouteDirective(unittest.TestCase):
route_action = actions[1]
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
- route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
self.assertEqual(route_discriminator,
('route', 'name', True, None, None, None, None, None))
- self.assertEqual(route_args[:3], ('path', 'name', None))
- predicates = route_args[3]
- self.assertEqual(len(predicates), 1)
+ route_callable()
+ route = self._assertRoute('name', 'path', 1)
request = DummyRequest()
request.is_xhr = True
- self.assertEqual(predicates[0](None, request), True)
+ self.assertEqual(route.predicates[0](None, request), True)
def test_with_request_method(self):
from zope.component import getSiteManager
- from repoze.bfg.zcml import connect_route
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
@@ -2358,20 +2271,16 @@ class TestRouteDirective(unittest.TestCase):
route_action = actions[1]
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
- route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
self.assertEqual(route_discriminator,
('route', 'name', False, 'GET',None, None, None, None))
- self.assertEqual(route_args[:3], ('path', 'name', None))
- predicates = route_args[3]
- self.assertEqual(len(predicates), 1)
+ route_callable()
+ route = self._assertRoute('name', 'path', 1)
request = DummyRequest()
request.method = 'GET'
- self.assertEqual(predicates[0](None, request), True)
+ self.assertEqual(route.predicates[0](None, request), True)
def test_with_path_info(self):
from zope.component import getSiteManager
- from repoze.bfg.zcml import connect_route
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
@@ -2397,20 +2306,16 @@ class TestRouteDirective(unittest.TestCase):
route_action = actions[1]
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
- route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
self.assertEqual(route_discriminator,
('route', 'name', False, None, '/foo',None,None, None))
- self.assertEqual(route_args[:3], ('path', 'name', None))
- predicates = route_args[3]
- self.assertEqual(len(predicates), 1)
+ route_callable()
+ route = self._assertRoute('name', 'path', 1)
request = DummyRequest()
request.path_info = '/foo'
- self.assertEqual(predicates[0](None, request), True)
+ self.assertEqual(route.predicates[0](None, request), True)
def test_with_request_param(self):
from zope.component import getSiteManager
- from repoze.bfg.zcml import connect_route
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
@@ -2437,19 +2342,16 @@ class TestRouteDirective(unittest.TestCase):
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
self.assertEqual(route_discriminator,
('route', 'name', False, None, None,'abc', None, None))
- self.assertEqual(route_args[:3], ('path', 'name', None))
- predicates = route_args[3]
- self.assertEqual(len(predicates), 1)
+ route_callable()
+ route = self._assertRoute('name', 'path', 1)
request = DummyRequest()
request.params = {'abc':'123'}
- self.assertEqual(predicates[0](None, request), True)
+ self.assertEqual(route.predicates[0](None, request), True)
def test_with_header(self):
from zope.component import getSiteManager
- from repoze.bfg.zcml import connect_route
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
@@ -2475,20 +2377,16 @@ class TestRouteDirective(unittest.TestCase):
route_action = actions[1]
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
- route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
self.assertEqual(route_discriminator,
('route', 'name', False, None, None,None,'Host', None))
- self.assertEqual(route_args[:3], ('path', 'name', None))
- predicates = route_args[3]
- self.assertEqual(len(predicates), 1)
+ route_callable()
+ route = self._assertRoute('name', 'path', 1)
request = DummyRequest()
request.headers = {'Host':'example.com'}
- self.assertEqual(predicates[0](None, request), True)
+ self.assertEqual(route.predicates[0](None, request), True)
def test_with_accept(self):
from zope.component import getSiteManager
- from repoze.bfg.zcml import connect_route
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
@@ -2515,17 +2413,14 @@ class TestRouteDirective(unittest.TestCase):
route_action = actions[1]
route_callable = route_action['callable']
route_discriminator = route_action['discriminator']
- route_args = route_action['args']
- self.assertEqual(route_callable, connect_route)
self.assertEqual(
route_discriminator,
('route', 'name', False, None, None, None, None, 'text/xml'))
- self.assertEqual(route_args[:3], ('path', 'name', None))
- predicates = route_args[3]
- self.assertEqual(len(predicates), 1)
+ route_callable()
+ route = self._assertRoute('name', 'path', 1)
request = DummyRequest()
request.accept = ['text/xml']
- self.assertEqual(predicates[0](None, request), True)
+ self.assertEqual(route.predicates[0](None, request), True)
class TestStaticDirective(unittest.TestCase):
def setUp(self):
@@ -2542,10 +2437,10 @@ class TestStaticDirective(unittest.TestCase):
from paste.urlparser import StaticURLParser
from zope.interface import implementedBy
from zope.component import getSiteManager
- from repoze.bfg.zcml import connect_route
from repoze.bfg.static import StaticRootFactory
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
+ from repoze.bfg.interfaces import IRoutesMapper
import os
here = os.path.dirname(__file__)
static_path = os.path.join(here, 'fixtures', 'static')
@@ -2571,20 +2466,23 @@ class TestStaticDirective(unittest.TestCase):
action = actions[1]
callable = action['callable']
discriminator = action['discriminator']
- args = action['args']
- self.assertEqual(callable, connect_route)
self.assertEqual(discriminator,
('route', 'name', False, None, None, None, None, None))
- self.assertEqual(args[0], 'name*subpath')
+ action['callable']()
+ mapper = sm.getUtility(IRoutesMapper)
+ routes = mapper.get_routes()
+ self.assertEqual(len(routes), 1)
+ self.assertEqual(routes[0].path, 'name*subpath')
+ self.assertEqual(routes[0].name, 'name')
def test_package_relative(self):
from repoze.bfg.static import PackageURLParser
from zope.component import getSiteManager
from zope.interface import implementedBy
- from repoze.bfg.zcml import connect_route
from repoze.bfg.static import StaticRootFactory
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
+ from repoze.bfg.interfaces import IRoutesMapper
context = DummyContext()
self._callFUT(context, 'name', 'repoze.bfg.tests:fixtures/static')
actions = context.actions
@@ -2604,22 +2502,24 @@ class TestStaticDirective(unittest.TestCase):
self.assertEqual(view(None, request).__class__, PackageURLParser)
action = actions[1]
- callable = action['callable']
discriminator = action['discriminator']
- args = action['args']
- self.assertEqual(callable, connect_route)
self.assertEqual(discriminator,
('route', 'name', False, None, None, None, None, None))
- self.assertEqual(args[0], 'name*subpath')
+ action['callable']()
+ mapper = sm.getUtility(IRoutesMapper)
+ routes = mapper.get_routes()
+ self.assertEqual(len(routes), 1)
+ self.assertEqual(routes[0].path, 'name*subpath')
+ self.assertEqual(routes[0].name, 'name')
def test_here_relative(self):
from repoze.bfg.static import PackageURLParser
from zope.component import getSiteManager
from zope.interface import implementedBy
- from repoze.bfg.zcml import connect_route
from repoze.bfg.static import StaticRootFactory
from repoze.bfg.interfaces import IView
from repoze.bfg.interfaces import IRouteRequest
+ from repoze.bfg.interfaces import IRoutesMapper
import repoze.bfg.tests
context = DummyContext(repoze.bfg.tests)
self._callFUT(context, 'name', 'fixtures/static')
@@ -2642,11 +2542,14 @@ class TestStaticDirective(unittest.TestCase):
action = actions[1]
callable = action['callable']
discriminator = action['discriminator']
- args = action['args']
- self.assertEqual(callable, connect_route)
self.assertEqual(discriminator,
('route', 'name', False, None, None, None, None, None))
- self.assertEqual(args[0], 'name*subpath')
+ action['callable']()
+ mapper = sm.getUtility(IRoutesMapper)
+ routes = mapper.get_routes()
+ self.assertEqual(len(routes), 1)
+ self.assertEqual(routes[0].path, 'name*subpath')
+ self.assertEqual(routes[0].name, 'name')
class TestResourceDirective(unittest.TestCase):
def setUp(self):
@@ -2677,100 +2580,64 @@ class TestResourceDirective(unittest.TestCase):
'a:foo.pt', 'a:foo/')
def test_no_colons(self):
- from repoze.bfg.zcml import _override
+ from zope.component import getSiteManager
context = DummyContext()
self._callFUT(context, 'a', 'b')
actions = context.actions
self.assertEqual(len(actions), 1)
action = actions[0]
- self.assertEqual(action['callable'], _override)
+ sm = getSiteManager()
+ self.assertEqual(action['callable'], sm.resource)
self.assertEqual(action['discriminator'], None)
- self.assertEqual(action['args'],
- (DummyModule, '', DummyModule, ''))
+ self.assertEqual(action['args'], ('a', 'b', None))
def test_with_colons(self):
- from repoze.bfg.zcml import _override
+ from zope.component import getSiteManager
context = DummyContext()
self._callFUT(context, 'a:foo.pt', 'b:foo.pt')
actions = context.actions
self.assertEqual(len(actions), 1)
action = actions[0]
- self.assertEqual(action['callable'], _override)
+ sm = getSiteManager()
+ self.assertEqual(action['callable'], sm.resource)
self.assertEqual(action['discriminator'], None)
- self.assertEqual(action['args'],
- (DummyModule, 'foo.pt', DummyModule, 'foo.pt'))
+ self.assertEqual(action['args'], ('a:foo.pt', 'b:foo.pt', None))
def test_override_module_with_directory(self):
- from repoze.bfg.zcml import _override
+ from zope.component import getSiteManager
context = DummyContext()
self._callFUT(context, 'a', 'b:foo/')
actions = context.actions
self.assertEqual(len(actions), 1)
action = actions[0]
- self.assertEqual(action['callable'], _override)
+ sm = getSiteManager()
+ self.assertEqual(action['callable'], sm.resource)
self.assertEqual(action['discriminator'], None)
- self.assertEqual(action['args'],
- (DummyModule, '', DummyModule, 'foo/'))
+ self.assertEqual(action['args'], ('a', 'b:foo/', None))
def test_override_directory_with_module(self):
- from repoze.bfg.zcml import _override
+ from zope.component import getSiteManager
context = DummyContext()
self._callFUT(context, 'a:foo/', 'b')
actions = context.actions
self.assertEqual(len(actions), 1)
action = actions[0]
- self.assertEqual(action['callable'], _override)
+ sm = getSiteManager()
+ self.assertEqual(action['callable'], sm.resource)
self.assertEqual(action['discriminator'], None)
- self.assertEqual(action['args'],
- (DummyModule, 'foo/', DummyModule, ''))
+ self.assertEqual(action['args'], ('a:foo/', 'b', None))
def test_override_module_with_module(self):
- from repoze.bfg.zcml import _override
+ from zope.component import getSiteManager
context = DummyContext()
self._callFUT(context, 'a', 'b')
actions = context.actions
self.assertEqual(len(actions), 1)
action = actions[0]
- self.assertEqual(action['callable'], _override)
+ sm = getSiteManager()
+ self.assertEqual(action['callable'], sm.resource)
self.assertEqual(action['discriminator'], None)
- self.assertEqual(action['args'],
- (DummyModule, '', DummyModule, ''))
-
-class Test_OverrideFunction(unittest.TestCase):
- def setUp(self):
- cleanUp()
-
- def tearDown(self):
- cleanUp()
-
- def _callFUT(self, *arg, **kw):
- from repoze.bfg.zcml import _override
- return _override(*arg, **kw)
-
- def _registerOverrides(self, overrides, package_name):
- from repoze.bfg.interfaces import IPackageOverrides
- from zope.component import getSiteManager
- sm = getSiteManager()
- sm.registerUtility(overrides, IPackageOverrides, name=package_name)
-
- def test_overrides_not_yet_registered(self):
- from zope.component import queryUtility
- from repoze.bfg.interfaces import IPackageOverrides
- package = DummyPackage('package')
- opackage = DummyPackage('opackage')
- self._callFUT(package, 'path', opackage, 'oprefix',
- PackageOverrides=DummyOverrides)
- overrides = queryUtility(IPackageOverrides, name='package')
- self.assertEqual(overrides.package, package)
- self.assertEqual(overrides.inserted, [('path', 'opackage', 'oprefix')])
-
- def test_overrides_already_registered(self):
- package = DummyPackage('package')
- opackage = DummyPackage('opackage')
- overrides = DummyOverrides(package)
- self._registerOverrides(overrides, 'package')
- self._callFUT(package, 'path', opackage, 'oprefix')
- self.assertEqual(overrides.inserted, [('path', 'opackage', 'oprefix')])
+ self.assertEqual(action['args'], ('a', 'b', None))
class TestZCMLConfigure(unittest.TestCase):
i = 0
@@ -2818,60 +2685,6 @@ class TestZCMLConfigure(unittest.TestCase):
self.assertRaises(IOError, self._callFUT, 'configure.zcml',
self.module)
-class TestBFGViewGrokker(unittest.TestCase):
- def setUp(self):
- cleanUp()
-
- def tearDown(self):
- cleanUp()
-
- def _getTargetClass(self):
- from repoze.bfg.zcml import BFGViewGrokker
- return BFGViewGrokker
-
- def _makeOne(self, *arg, **kw):
- return self._getTargetClass()(*arg, **kw)
-
- def test_grok_is_bfg_view(self):
- from zope.component import getSiteManager
- from repoze.bfg.interfaces import IRequest
- from repoze.bfg.interfaces import IView
- from zope.interface import Interface
- grokker = self._makeOne()
- class obj:
- def __init__(self, context, request):
- pass
- def __call__(self):
- return 'OK'
- settings = dict(permission='foo', for_=Interface, name='foo.html',
- request_type=IRequest, route_name=None,
- request_method=None, request_param=None,
- containment=None, attr=None, renderer=None,
- wrapper=None, xhr=False, header=None,
- accept=None)
- obj.__bfg_view_settings__ = [settings]
- context = DummyContext()
- result = grokker.grok('name', obj, context=context)
- self.assertEqual(result, True)
- actions = context.actions
- self.assertEqual(len(actions), 1)
- register = actions[0]['callable']
- register()
- sm = getSiteManager()
- wrapped = sm.adapters.lookup((Interface, IRequest), IView,
- name='foo.html')
- self.assertEqual(wrapped(None, None), 'OK')
-
- def test_grok_is_not_bfg_view(self):
- grokker = self._makeOne()
- class obj:
- pass
- context = DummyContext()
- result = grokker.grok('name', obj, context=context)
- self.assertEqual(result, False)
- actions = context.actions
- self.assertEqual(len(actions), 0)
-
class TestZCMLScanDirective(unittest.TestCase):
def setUp(self):
cleanUp()
@@ -2884,32 +2697,18 @@ class TestZCMLScanDirective(unittest.TestCase):
return scan(context, package, martian)
def test_it(self):
- from repoze.bfg.zcml import BFGMultiGrokker
- from repoze.bfg.zcml import exclude
+ from repoze.bfg.registry import BFGMultiGrokker
martian = DummyMartianModule()
module_grokker = DummyModuleGrokker()
dummy_module = DummyModule()
- self._callFUT(None, dummy_module, martian)
+ context = DummyContext()
+ self._callFUT(context, dummy_module, martian)
+ context.actions[-1]['callable']()
self.assertEqual(martian.name, 'dummy')
multi_grokker = martian.module_grokker.multi_grokker
self.assertEqual(multi_grokker.__class__, BFGMultiGrokker)
- self.assertEqual(martian.context, None)
- self.assertEqual(martian.exclude_filter, exclude)
-
-class TestExcludeFunction(unittest.TestCase):
- def setUp(self):
- cleanUp()
-
- def tearDown(self):
- cleanUp()
-
- def _callFUT(self, name):
- from repoze.bfg.zcml import exclude
- return exclude(name)
-
- def test_it(self):
- self.assertEqual(self._callFUT('.foo'), True)
- self.assertEqual(self._callFUT('foo'), False)
+ self.assertEqual(martian.info, context.info)
+ self.failUnless(martian.exclude_filter)
class DummyModule:
__path__ = "foo"
@@ -2921,9 +2720,11 @@ class DummyModuleGrokker:
self.multi_grokker = grokker
class DummyMartianModule:
- def grok_dotted_name(self, name, grokker, context, exclude_filter=None):
+ def grok_dotted_name(self, name, grokker, _info, _registry,
+ exclude_filter=None):
self.name = name
- self.context = context
+ self.info = _info
+ self.registry = _registry
self.exclude_filter = exclude_filter
return True
@@ -2950,13 +2751,6 @@ class DummyContext:
class Dummy:
pass
-class DummyMapper:
- def __init__(self):
- self.connections = []
-
- def connect(self, path, name, factory, predicates=()):
- self.connections.append((path, name, factory, predicates))
-
class DummyRoute:
pass
@@ -2978,15 +2772,3 @@ class DummyRequest:
def copy(self):
return self
-class DummyOverrides:
- def __init__(self, package):
- self.package = package
- self.inserted = []
-
- def insert(self, path, package, prefix):
- self.inserted.append((path, package, prefix))
-
-class DummyPackage:
- def __init__(self, name):
- self.__name__ = name
-
diff --git a/repoze/bfg/zcml.py b/repoze/bfg/zcml.py
index 0213e564c..7bfa1a916 100644
--- a/repoze/bfg/zcml.py
+++ b/repoze/bfg/zcml.py
@@ -1,16 +1,10 @@
-import re
-import sys
-
from zope.component import getSiteManager
-from zope.component import getUtility
from zope.component import queryUtility
from zope.configuration.exceptions import ConfigurationError
from zope.configuration.fields import GlobalObject
from zope.interface import Interface
-from zope.interface import implementedBy
-from zope.interface.interfaces import IInterface
from zope.schema import Bool
from zope.schema import Int
@@ -22,16 +16,11 @@ import martian
from repoze.bfg.interfaces import IAuthenticationPolicy
from repoze.bfg.interfaces import IAuthorizationPolicy
from repoze.bfg.interfaces import IForbiddenView
-from repoze.bfg.interfaces import IMultiView
from repoze.bfg.interfaces import INotFoundView
-from repoze.bfg.interfaces import IPackageOverrides
from repoze.bfg.interfaces import IRendererFactory
from repoze.bfg.interfaces import IRequest
from repoze.bfg.interfaces import IRouteRequest
-from repoze.bfg.interfaces import IRoutesMapper
-from repoze.bfg.interfaces import ISecuredView
from repoze.bfg.interfaces import IView
-from repoze.bfg.interfaces import IViewPermission
from repoze.bfg.authentication import RepozeWho1AuthenticationPolicy
from repoze.bfg.authentication import RemoteUserAuthenticationPolicy
@@ -40,12 +29,8 @@ from repoze.bfg.authorization import ACLAuthorizationPolicy
from repoze.bfg.configuration import zcml_configure
from repoze.bfg.path import package_name
from repoze.bfg.request import route_request_iface
-from repoze.bfg.resource import PackageOverrides
from repoze.bfg.resource import resource_spec
from repoze.bfg.static import StaticRootFactory
-from repoze.bfg.traversal import find_interface
-from repoze.bfg.view import MultiView
-from repoze.bfg.view import derive_view
from repoze.bfg.view import static as static_view
###################### directives ##########################
@@ -148,104 +133,6 @@ class IViewDirective(Interface):
description=(u'Accepts a regular expression.'),
required = False)
-def _make_predicates(xhr=None, request_method=None, path_info=None,
- request_param=None, header=None, accept=None,
- containment=None):
- # Predicates are added to the predicate list in (presumed)
- # computation expense order. All predicates associated with a
- # view must evaluate true for the view to "match" a request.
- # Elsewhere in the code, we evaluate them using a generator
- # expression. The fastest predicate should be evaluated first,
- # then the next fastest, and so on, as if one returns false, the
- # remainder of the predicates won't need to be evaluated.
-
- # Each predicate is associated with a weight value. The weight
- # symbolizes the relative potential "importance" of the predicate
- # to all other predicates. A larger weight indicates greater
- # importance. These weights are subtracted from an aggregate
- # 'weight' variable. The aggregate weight is then divided by the
- # length of the predicate list to compute a "score" for this view.
- # The score represents the ordering in which a "multiview" ( a
- # collection of views that share the same context/request/name
- # triad but differ in other ways via predicates) will attempt to
- # call its set of views. Views with lower scores will be tried
- # first. The intent is to a) ensure that views with more
- # predicates are always evaluated before views with fewer
- # predicates and b) to ensure a stable call ordering of views that
- # share the same number of predicates.
-
- # Views which do not have any predicates get a score of
- # sys.maxint, meaning that they will be tried very last.
-
- predicates = []
- weight = sys.maxint
-
- if xhr:
- def xhr_predicate(context, request):
- return request.is_xhr
- weight = weight - 10
- predicates.append(xhr_predicate)
-
- if request_method is not None:
- def request_method_predicate(context, request):
- return request.method == request_method
- weight = weight - 20
- predicates.append(request_method_predicate)
-
- if path_info is not None:
- try:
- path_info_val = re.compile(path_info)
- except re.error, why:
- raise ConfigurationError(why[0])
- def path_info_predicate(context, request):
- return path_info_val.match(request.path_info) is not None
- weight = weight - 30
- predicates.append(path_info_predicate)
-
- if request_param is not None:
- request_param_val = None
- if '=' in request_param:
- request_param, request_param_val = request_param.split('=', 1)
- def request_param_predicate(context, request):
- if request_param_val is None:
- return request_param in request.params
- return request.params.get(request_param) == request_param_val
- weight = weight - 40
- predicates.append(request_param_predicate)
-
- if header is not None:
- header_name = header
- header_val = None
- if ':' in header:
- header_name, header_val = header.split(':', 1)
- try:
- header_val = re.compile(header_val)
- except re.error, why:
- raise ConfigurationError(why[0])
- def header_predicate(context, request):
- if header_val is None:
- return header_name in request.headers
- val = request.headers.get(header_name)
- return header_val.match(val) is not None
- weight = weight - 50
- predicates.append(header_predicate)
-
- if accept is not None:
- def accept_predicate(context, request):
- return accept in request.accept
- weight = weight - 60
- predicates.append(accept_predicate)
-
- if containment is not None:
- def containment_predicate(context, request):
- return find_interface(context, containment) is not None
- weight = weight - 70
- predicates.append(containment_predicate)
-
- # this will be == sys.maxint if no predicates
- score = weight / (len(predicates) + 1)
- return score, predicates
-
def view(
_context,
permission=None,
@@ -297,66 +184,129 @@ def view(
if renderer and '.' in renderer:
renderer = resource_spec(renderer, package_name(_context.resolve('.')))
- score, predicates = _make_predicates(
- xhr=xhr, request_method=request_method, path_info=path_info,
- request_param=request_param, header=header, accept=accept,
- containment=containment)
-
def register():
- derived_view = derive_view(view, permission, predicates, attr, renderer,
- wrapper, name)
- r_for_ = for_
- r_request_type = request_type
- if r_for_ is None:
- r_for_ = Interface
- if not IInterface.providedBy(r_for_):
- r_for_ = implementedBy(r_for_)
- if not IInterface.providedBy(r_request_type):
- r_request_type = implementedBy(r_request_type)
- old_view = sm.adapters.lookup((r_for_, r_request_type), IView,name=name)
- if old_view is None:
- if hasattr(derived_view, '__call_permissive__'):
- sm.registerAdapter(derived_view, (for_, request_type),
- ISecuredView, name, _context.info)
- if hasattr(derived_view, '__permitted__'):
- # bw compat
- sm.registerAdapter(derived_view.__permitted__,
- (for_, request_type), IViewPermission,
- name, _context.info)
- else:
- sm.registerAdapter(derived_view, (for_, request_type),
- IView, name, _context.info)
- else:
- # XXX we could try to be more efficient here and register
- # a non-secured view for a multiview if none of the
- # multiview's consituent views have a permission
- # associated with them, but this code is getting pretty
- # rough already
- if IMultiView.providedBy(old_view):
- multiview = old_view
- else:
- multiview = MultiView(name)
- multiview.add(old_view, sys.maxint)
- multiview.add(derived_view, score)
- for i in (IView, ISecuredView):
- # unregister any existing views
- sm.adapters.unregister((r_for_, r_request_type), i, name=name)
- sm.registerAdapter(multiview, (for_, request_type), IMultiView,
- name, _context.info)
- # b/w compat
- sm.registerAdapter(multiview.__permitted__,
- (for_, request_type), IViewPermission,
- name, _context.info)
+ sm.view(permission=permission, for_=for_, view=view, name=name,
+ request_type=request_type, route_name=route_name,
+ request_method=request_method, request_param=request_param,
+ containment=containment, attr=attr, renderer=renderer,
+ wrapper=wrapper, xhr=xhr, accept=accept, header=header,
+ path_info=path_info, _info=_context.info)
+
_context.action(
discriminator = ('view', for_, name, request_type, IView, containment,
request_param, request_method, route_name, attr,
xhr, accept, header, path_info),
callable = register,
- args = (),
)
_view = view # for directives that take a view arg
+class IRouteDirective(Interface):
+ """ The interface for the ``route`` ZCML directive
+ """
+ name = TextLine(title=u'name', required=True)
+ path = TextLine(title=u'path', required=True)
+ factory = GlobalObject(title=u'context factory', required=False)
+ view = GlobalObject(title=u'view', required=False)
+
+ view_for = GlobalObject(title=u'view_for', required=False)
+ # alias for view_for
+ for_ = GlobalObject(title=u'for', required=False)
+
+ view_permission = TextLine(title=u'view_permission', required=False)
+ # alias for view_permission
+ permission = TextLine(title=u'permission', required=False)
+
+ view_request_type = TextLine(title=u'view_request_type', required=False)
+ # alias for view_request_type
+ request_type = TextLine(title=u'request_type', required=False)
+
+ view_renderer = TextLine(title=u'view_renderer', required=False)
+ # alias for view_renderer
+ renderer = TextLine(title=u'renderer', required=False)
+
+ view_request_method = TextLine(title=u'view_request_method', required=False)
+ view_containment = GlobalObject(
+ title = u'Dotted name of a containment class or interface',
+ required=False)
+ view_attr = TextLine(title=u'view_attr', required=False)
+ view_header = TextLine(title=u'view_header', required=False)
+ view_accept = TextLine(title=u'view_accept', required=False)
+ view_xhr = Bool(title=u'view_xhr', required=False)
+ view_path_info = TextLine(title=u'view_path_info', required=False)
+
+ request_method = TextLine(title=u'request_method', required=False)
+ request_param = TextLine(title=u'request_param', required=False)
+ header = TextLine(title=u'header', required=False)
+ accept = TextLine(title=u'accept', required=False)
+ xhr = Bool(title=u'xhr', required=False)
+ path_info = TextLine(title=u'path_info', required=False)
+
+def route(_context, name, path, view=None, view_for=None,
+ permission=None, factory=None, request_type=None, for_=None,
+ header=None, xhr=False, accept=None, path_info=None,
+ request_method=None, request_param=None,
+ view_permission=None, view_request_type=None,
+ view_request_method=None, view_request_param=None,
+ view_containment=None, view_attr=None,
+ renderer=None, view_renderer=None, view_header=None,
+ view_accept=None, view_xhr=False,
+ view_path_info=None):
+ """ Handle ``route`` ZCML directives
+ """
+ # the strange ordering of the request kw args above is for b/w
+ # compatibility purposes.
+ # these are route predicates; if they do not match, the next route
+ # in the routelist will be tried
+ sm = getSiteManager()
+
+ if request_type in ('GET', 'HEAD', 'PUT', 'POST', 'DELETE'):
+ # b/w compat for 1.0
+ view_request_method = request_type
+ request_type = None
+
+ request_iface = queryUtility(IRouteRequest, name=name)
+ if request_iface is None:
+ request_iface = route_request_iface(name)
+ sm.registerUtility(request_iface, IRouteRequest, name=name)
+
+ if view:
+ view_for = view_for or for_
+ view_request_type = view_request_type or request_type
+ view_permission = view_permission or permission
+ view_renderer = view_renderer or renderer
+ _view(
+ _context,
+ permission=view_permission,
+ for_=view_for,
+ view=view,
+ name='',
+ request_type=view_request_type,
+ route_name=name,
+ request_method=view_request_method,
+ request_param=view_request_param,
+ containment=view_containment,
+ attr=view_attr,
+ renderer=view_renderer,
+ header=view_header,
+ accept=view_accept,
+ xhr=view_xhr,
+ path_info=view_path_info,
+ )
+
+ def register():
+ sm.route(name, path, factory=factory, header=header,
+ xhr=xhr, accept=accept, path_info=path_info,
+ request_method=request_method, request_param=request_param,
+ _info=_context.info)
+
+ _context.action(
+ discriminator = ('route', name, xhr, request_method, path_info,
+ request_param, header, accept),
+ callable = register,
+ )
+
+
class ISystemViewDirective(Interface):
view = GlobalObject(
title=u"",
@@ -398,10 +348,8 @@ def view_utility(_context, view, attr, renderer, wrapper, iface):
renderer = resource_spec(renderer, package_name(_context.resolve('.')))
def register():
- derived_view = derive_view(view, attr=attr, renderer_name=renderer,
- wrapper_viewname=wrapper)
sm = getSiteManager()
- sm.registerUtility(derived_view, iface, '', _context.info)
+ sm.view_utility(view, attr, renderer, wrapper, iface, _context.info)
_context.action(
discriminator = iface,
@@ -422,19 +370,7 @@ class IResourceDirective(Interface):
description=u"The spec of the resource providing the override.",
required=True)
-def _override(package, path, override_package, override_prefix,
- PackageOverrides=PackageOverrides):
- # PackageOverrides kw arg for tests
- sm = getSiteManager()
- pkg_name = package.__name__
- override_pkg_name = override_package.__name__
- override = queryUtility(IPackageOverrides, name=pkg_name)
- if override is None:
- override = PackageOverrides(package)
- sm.registerUtility(override, IPackageOverrides, name=pkg_name)
- override.insert(path, override_pkg_name, override_prefix)
-
-def resource(context, to_override, override_with):
+def resource(_context, to_override, override_with):
if to_override == override_with:
raise ConfigurationError('You cannot override a resource with itself')
@@ -460,13 +396,12 @@ def resource(context, to_override, override_with):
'A file cannot be overridden with a directory (put a slash '
'at the end of to_override if necessary)')
- package = context.resolve(package)
- override_package = context.resolve(override_package)
+ sm = getSiteManager()
- context.action(
+ _context.action(
discriminator = None,
- callable = _override,
- args = (package, path, override_package, override_prefix),
+ callable = sm.resource,
+ args = (to_override, override_with, _context.info),
)
class IRepozeWho1AuthenticationPolicyDirective(Interface):
@@ -481,7 +416,7 @@ def repozewho1authenticationpolicy(_context, identifier_name='auth_tkt',
# authentication policies must be registered eagerly so they can
# be found by the view registration machinery
sm = getSiteManager()
- sm.registerUtility(policy, IAuthenticationPolicy)
+ sm.authentication_policy(policy, _info=_context.info)
_context.action(discriminator=IAuthenticationPolicy)
class IRemoteUserAuthenticationPolicyDirective(Interface):
@@ -496,7 +431,7 @@ def remoteuserauthenticationpolicy(_context, environ_key='REMOTE_USER',
# authentication policies must be registered eagerly so they can
# be found by the view registration machinery
sm = getSiteManager()
- sm.registerUtility(policy, IAuthenticationPolicy)
+ sm.authentication_policy(policy, _info=_context.info)
_context.action(discriminator=IAuthenticationPolicy)
class IAuthTktAuthenticationPolicyDirective(Interface):
@@ -533,7 +468,7 @@ def authtktauthenticationpolicy(_context,
# authentication policies must be registered eagerly so they can
# be found by the view registration machinery
sm = getSiteManager()
- sm.registerUtility(policy, IAuthenticationPolicy)
+ sm.authentication_policy(policy, _info=_context.info)
_context.action(discriminator=IAuthenticationPolicy)
class IACLAuthorizationPolicyDirective(Interface):
@@ -544,120 +479,9 @@ def aclauthorizationpolicy(_context):
# authorization policies must be registered eagerly so they can be
# found by the view registration machinery
sm = getSiteManager()
- sm.registerUtility(policy, IAuthorizationPolicy)
+ sm.authorization_policy(policy, _info=_context.info)
_context.action(discriminator=IAuthorizationPolicy)
-class IRouteDirective(Interface):
- """ The interface for the ``route`` ZCML directive
- """
- name = TextLine(title=u'name', required=True)
- path = TextLine(title=u'path', required=True)
- factory = GlobalObject(title=u'context factory', required=False)
- view = GlobalObject(title=u'view', required=False)
-
- view_for = GlobalObject(title=u'view_for', required=False)
- # alias for view_for
- for_ = GlobalObject(title=u'for', required=False)
-
- view_permission = TextLine(title=u'view_permission', required=False)
- # alias for view_permission
- permission = TextLine(title=u'permission', required=False)
-
- view_request_type = TextLine(title=u'view_request_type', required=False)
- # alias for view_request_type
- request_type = TextLine(title=u'request_type', required=False)
-
- view_renderer = TextLine(title=u'view_renderer', required=False)
- # alias for view_renderer
- renderer = TextLine(title=u'renderer', required=False)
-
- view_request_method = TextLine(title=u'view_request_method', required=False)
- view_containment = GlobalObject(
- title = u'Dotted name of a containment class or interface',
- required=False)
- view_attr = TextLine(title=u'view_attr', required=False)
- view_header = TextLine(title=u'view_header', required=False)
- view_accept = TextLine(title=u'view_accept', required=False)
- view_xhr = Bool(title=u'view_xhr', required=False)
- view_path_info = TextLine(title=u'view_path_info', required=False)
-
- request_method = TextLine(title=u'request_method', required=False)
- request_param = TextLine(title=u'request_param', required=False)
- header = TextLine(title=u'header', required=False)
- accept = TextLine(title=u'accept', required=False)
- xhr = Bool(title=u'xhr', required=False)
- path_info = TextLine(title=u'path_info', required=False)
-
-def route(_context, name, path, view=None, view_for=None,
- permission=None, factory=None, request_type=None, for_=None,
- header=None, xhr=False, accept=None, path_info=None,
- request_method=None, request_param=None,
- view_permission=None, view_request_type=None,
- view_request_method=None, view_request_param=None,
- view_containment=None, view_attr=None,
- renderer=None, view_renderer=None, view_header=None,
- view_accept=None, view_xhr=False,
- view_path_info=None):
- """ Handle ``route`` ZCML directives
- """
- # the strange ordering of the request kw args above is for b/w
- # compatibility purposes.
- # these are route predicates; if they do not match, the next route
- # in the routelist will be tried
- _, predicates = _make_predicates(xhr=xhr,
- request_method=request_method,
- path_info=path_info,
- request_param=request_param,
- header=header,
- accept=accept)
-
- sm = getSiteManager()
-
- if request_type in ('GET', 'HEAD', 'PUT', 'POST', 'DELETE'):
- # b/w compat for 1.0
- view_request_method = request_type
- request_type = None
-
- request_iface = queryUtility(IRouteRequest, name=name)
- if request_iface is None:
- request_iface = route_request_iface(name)
- sm.registerUtility(request_iface, IRouteRequest, name=name)
-
- if view:
- view_for = view_for or for_
- view_request_type = view_request_type or request_type
- view_permission = view_permission or permission
- view_renderer = view_renderer or renderer
- _view(
- _context,
- permission=view_permission,
- for_=view_for,
- view=view,
- name='',
- request_type=view_request_type,
- route_name=name,
- request_method=view_request_method,
- request_param=view_request_param,
- containment=view_containment,
- attr=view_attr,
- renderer=view_renderer,
- header=view_header,
- accept=view_accept,
- xhr=view_xhr,
- path_info=view_path_info,
- )
-
- _context.action(
- discriminator = ('route', name, xhr, request_method, path_info,
- request_param, header, accept),
- callable = connect_route,
- args = (path, name, factory, predicates),
- )
-
-def connect_route(path, name, factory, predicates):
- mapper = getUtility(IRoutesMapper)
- mapper.connect(path, name, factory, predicates=predicates)
-
class IRendererDirective(Interface):
factory = GlobalObject(
title=u'IRendererFactory implementation',
@@ -671,7 +495,7 @@ def renderer(_context, factory, name=''):
# renderer factories must be registered eagerly so they can be
# found by the view machinery
sm = getSiteManager()
- sm.registerUtility(factory, IRendererFactory, name=name)
+ sm.renderer(factory, name, _info=_context.info)
_context.action(discriminator=(IRendererFactory, name))
class IStaticDirective(Interface):
@@ -709,35 +533,10 @@ class IScanDirective(Interface):
def scan(_context, package, martian=martian):
# martian overrideable only for unit tests
- multi_grokker = BFGMultiGrokker()
- multi_grokker.register(BFGViewGrokker())
- module_grokker = martian.ModuleGrokker(grokker=multi_grokker)
- martian.grok_dotted_name(package.__name__, grokker=module_grokker,
- context=_context, exclude_filter=exclude)
-
-################# utility stuff ####################
-
-class BFGViewMarker(object):
- pass
-
-class BFGMultiGrokker(martian.core.MultiInstanceOrClassGrokkerBase):
- def get_bases(self, obj):
- if hasattr(obj, '__bfg_view_settings__'):
- return [BFGViewMarker]
- return []
-
-class BFGViewGrokker(martian.InstanceGrokker):
- martian.component(BFGViewMarker)
- def grok(self, name, obj, **kw):
- config = getattr(obj, '__bfg_view_settings__', [])
- for settings in config:
- view(kw['context'], view=obj, **settings)
- return bool(config)
-
-def exclude(name):
- if name.startswith('.'):
- return True
- return False
+ def register():
+ sm = getSiteManager()
+ sm.scan(package, _info=_context.info, martian=martian)
+ _context.action(discriminator=None, callable=register)
class Uncacheable(object):
""" Include in discriminators of actions which are not cacheable;