From e4c0570d5c67ddf0ad9502169b59475ba0784d82 Mon Sep 17 00:00:00 2001
From: Michael Merickel <michael@merickel.org>
Date: Thu, 18 Oct 2018 08:27:04 +0200
Subject: [PATCH] move action-related code into pyramid.config.actions
---
tests/test_config/test_testing.py | 2
src/pyramid/config/__init__.py | 526 ---------
src/pyramid/config/actions.py | 525 +++++++++
tests/test_config/test_init.py | 1033 -------------------
tests/test_config/test_actions.py | 1055 +++++++++++++++++++
5 files changed, 1,587 insertions(+), 1,554 deletions(-)
diff --git a/src/pyramid/config/__init__.py b/src/pyramid/config/__init__.py
index f579035..e4dcc6b 100644
--- a/src/pyramid/config/__init__.py
+++ b/src/pyramid/config/__init__.py
@@ -1,9 +1,6 @@
import inspect
-import itertools
import logging
-import operator
import os
-import sys
import threading
import venusian
@@ -23,21 +20,17 @@
from pyramid.authorization import ACLAuthorizationPolicy
-from pyramid.compat import text_, reraise, string_types
+from pyramid.compat import text_, string_types
from pyramid.events import ApplicationCreated
-from pyramid.exceptions import (
- ConfigurationConflictError,
- ConfigurationError,
- ConfigurationExecutionError,
-)
+from pyramid.exceptions import ConfigurationError
from pyramid.httpexceptions import default_exceptionresponse_view
from pyramid.path import caller_package, package_of
-from pyramid.registry import Introspectable, Introspector, Registry, undefer
+from pyramid.registry import Introspectable, Introspector, Registry
from pyramid.router import Router
@@ -47,8 +40,9 @@
from pyramid.util import WeakOrderedSet, object_description
-from pyramid.config.util import ActionInfo, PredicateList, action_method, not_
+from pyramid.config.util import PredicateList, action_method, not_
+from pyramid.config.actions import ActionConfiguratorMixin
from pyramid.config.adapters import AdaptersConfiguratorMixin
from pyramid.config.assets import AssetsConfiguratorMixin
from pyramid.config.factories import FactoriesConfiguratorMixin
@@ -76,6 +70,7 @@
class Configurator(
+ ActionConfiguratorMixin,
TestingConfiguratorMixin,
TweensConfiguratorMixin,
SecurityConfiguratorMixin,
@@ -575,143 +570,6 @@
order=PHASE1_CONFIG,
) # must be registered early
- @property
- def action_info(self):
- info = self.info # usually a ZCML action (ParserInfo) if self.info
- if not info:
- # Try to provide more accurate info for conflict reports
- if self._ainfo:
- info = self._ainfo[0]
- else:
- info = ActionInfo(None, 0, '', '')
- return info
-
- def action(
- self,
- discriminator,
- callable=None,
- args=(),
- kw=None,
- order=0,
- introspectables=(),
- **extra
- ):
- """ Register an action which will be executed when
- :meth:`pyramid.config.Configurator.commit` is called (or executed
- immediately if ``autocommit`` is ``True``).
-
- .. warning:: This method is typically only used by :app:`Pyramid`
- framework extension authors, not by :app:`Pyramid` application
- developers.
-
- The ``discriminator`` uniquely identifies the action. It must be
- given, but it can be ``None``, to indicate that the action never
- conflicts. It must be a hashable value.
-
- The ``callable`` is a callable object which performs the task
- associated with the action when the action is executed. It is
- optional.
-
- ``args`` and ``kw`` are tuple and dict objects respectively, which
- are passed to ``callable`` when this action is executed. Both are
- optional.
-
- ``order`` is a grouping mechanism; an action with a lower order will
- be executed before an action with a higher order (has no effect when
- autocommit is ``True``).
-
- ``introspectables`` is a sequence of :term:`introspectable` objects
- (or the empty sequence if no introspectable objects are associated
- with this action). If this configurator's ``introspection``
- attribute is ``False``, these introspectables will be ignored.
-
- ``extra`` provides a facility for inserting extra keys and values
- into an action dictionary.
- """
- # catch nonhashable discriminators here; most unit tests use
- # autocommit=False, which won't catch unhashable discriminators
- assert hash(discriminator)
-
- if kw is None:
- kw = {}
-
- autocommit = self.autocommit
- action_info = self.action_info
-
- if not self.introspection:
- # if we're not introspecting, ignore any introspectables passed
- # to us
- introspectables = ()
-
- if autocommit:
- # callables can depend on the side effects of resolving a
- # deferred discriminator
- self.begin()
- try:
- undefer(discriminator)
- if callable is not None:
- callable(*args, **kw)
- for introspectable in introspectables:
- introspectable.register(self.introspector, action_info)
- finally:
- self.end()
-
- else:
- action = extra
- action.update(
- dict(
- discriminator=discriminator,
- callable=callable,
- args=args,
- kw=kw,
- order=order,
- info=action_info,
- includepath=self.includepath,
- introspectables=introspectables,
- )
- )
- self.action_state.action(**action)
-
- def _get_action_state(self):
- registry = self.registry
- try:
- state = registry.action_state
- except AttributeError:
- state = ActionState()
- registry.action_state = state
- return state
-
- def _set_action_state(self, state):
- self.registry.action_state = state
-
- action_state = property(_get_action_state, _set_action_state)
-
- _ctx = action_state # bw compat
-
- def commit(self):
- """
- Commit any pending configuration actions. If a configuration
- conflict is detected in the pending configuration actions, this method
- will raise a :exc:`ConfigurationConflictError`; within the traceback
- of this error will be information about the source of the conflict,
- usually including file names and line numbers of the cause of the
- configuration conflicts.
-
- .. warning::
- You should think very carefully before manually invoking
- ``commit()``. Especially not as part of any reusable configuration
- methods. Normally it should only be done by an application author at
- the end of configuration in order to override certain aspects of an
- addon.
-
- """
- self.begin()
- try:
- self.action_state.execute_actions(introspector=self.introspector)
- finally:
- self.end()
- self.action_state = ActionState() # old actions have been processed
-
def include(self, callable, route_prefix=None):
"""Include a configuration callable, to support imperative
application extensibility.
@@ -1082,378 +940,6 @@
self.end()
return app
-
-
-# this class is licensed under the ZPL (stolen from Zope)
-class ActionState(object):
- def __init__(self):
- # NB "actions" is an API, dep'd upon by pyramid_zcml's load_zcml func
- self.actions = []
- self._seen_files = set()
-
- def processSpec(self, spec):
- """Check whether a callable needs to be processed. The ``spec``
- refers to a unique identifier for the callable.
-
- Return True if processing is needed and False otherwise. If
- the callable needs to be processed, it will be marked as
- processed, assuming that the caller will procces the callable if
- it needs to be processed.
- """
- if spec in self._seen_files:
- return False
- self._seen_files.add(spec)
- return True
-
- def action(
- self,
- discriminator,
- callable=None,
- args=(),
- kw=None,
- order=0,
- includepath=(),
- info=None,
- introspectables=(),
- **extra
- ):
- """Add an action with the given discriminator, callable and arguments
- """
- if kw is None:
- kw = {}
- action = extra
- action.update(
- dict(
- discriminator=discriminator,
- callable=callable,
- args=args,
- kw=kw,
- includepath=includepath,
- info=info,
- order=order,
- introspectables=introspectables,
- )
- )
- self.actions.append(action)
-
- def execute_actions(self, clear=True, introspector=None):
- """Execute the configuration actions
-
- This calls the action callables after resolving conflicts
-
- For example:
-
- >>> output = []
- >>> def f(*a, **k):
- ... output.append(('f', a, k))
- >>> context = ActionState()
- >>> context.actions = [
- ... (1, f, (1,)),
- ... (1, f, (11,), {}, ('x', )),
- ... (2, f, (2,)),
- ... ]
- >>> context.execute_actions()
- >>> output
- [('f', (1,), {}), ('f', (2,), {})]
-
- If the action raises an error, we convert it to a
- ConfigurationExecutionError.
-
- >>> output = []
- >>> def bad():
- ... bad.xxx
- >>> context.actions = [
- ... (1, f, (1,)),
- ... (1, f, (11,), {}, ('x', )),
- ... (2, f, (2,)),
- ... (3, bad, (), {}, (), 'oops')
- ... ]
- >>> try:
- ... v = context.execute_actions()
- ... except ConfigurationExecutionError, v:
- ... pass
- >>> print(v)
- exceptions.AttributeError: 'function' object has no attribute 'xxx'
- in:
- oops
-
- Note that actions executed before the error still have an effect:
-
- >>> output
- [('f', (1,), {}), ('f', (2,), {})]
-
- The execution is re-entrant such that actions may be added by other
- actions with the one caveat that the order of any added actions must
- be equal to or larger than the current action.
-
- >>> output = []
- >>> def f(*a, **k):
- ... output.append(('f', a, k))
- ... context.actions.append((3, g, (8,), {}))
- >>> def g(*a, **k):
- ... output.append(('g', a, k))
- >>> context.actions = [
- ... (1, f, (1,)),
- ... ]
- >>> context.execute_actions()
- >>> output
- [('f', (1,), {}), ('g', (8,), {})]
-
- """
- try:
- all_actions = []
- executed_actions = []
- action_iter = iter([])
- conflict_state = ConflictResolverState()
-
- while True:
- # We clear the actions list prior to execution so if there
- # are some new actions then we add them to the mix and resolve
- # conflicts again. This orders the new actions as well as
- # ensures that the previously executed actions have no new
- # conflicts.
- if self.actions:
- all_actions.extend(self.actions)
- action_iter = resolveConflicts(
- self.actions, state=conflict_state
- )
- self.actions = []
-
- action = next(action_iter, None)
- if action is None:
- # we are done!
- break
-
- callable = action['callable']
- args = action['args']
- kw = action['kw']
- info = action['info']
- # we use "get" below in case an action was added via a ZCML
- # directive that did not know about introspectables
- introspectables = action.get('introspectables', ())
-
- try:
- if callable is not None:
- callable(*args, **kw)
- except Exception:
- t, v, tb = sys.exc_info()
- try:
- reraise(
- ConfigurationExecutionError,
- ConfigurationExecutionError(t, v, info),
- tb,
- )
- finally:
- del t, v, tb
-
- if introspector is not None:
- for introspectable in introspectables:
- introspectable.register(introspector, info)
-
- executed_actions.append(action)
-
- self.actions = all_actions
- return executed_actions
-
- finally:
- if clear:
- self.actions = []
-
-
-class ConflictResolverState(object):
- def __init__(self):
- # keep a set of resolved discriminators to test against to ensure
- # that a new action does not conflict with something already executed
- self.resolved_ainfos = {}
-
- # actions left over from a previous iteration
- self.remaining_actions = []
-
- # after executing an action we memoize its order to avoid any new
- # actions sending us backward
- self.min_order = None
-
- # unique tracks the index of the action so we need it to increase
- # monotonically across invocations to resolveConflicts
- self.start = 0
-
-
-# this function is licensed under the ZPL (stolen from Zope)
-def resolveConflicts(actions, state=None):
- """Resolve conflicting actions
-
- Given an actions list, identify and try to resolve conflicting actions.
- Actions conflict if they have the same non-None discriminator.
-
- Conflicting actions can be resolved if the include path of one of
- the actions is a prefix of the includepaths of the other
- conflicting actions and is unequal to the include paths in the
- other conflicting actions.
-
- Actions are resolved on a per-order basis because some discriminators
- cannot be computed until earlier actions have executed. An action in an
- earlier order may execute successfully only to find out later that it was
- overridden by another action with a smaller include path. This will result
- in a conflict as there is no way to revert the original action.
-
- ``state`` may be an instance of ``ConflictResolverState`` that
- can be used to resume execution and resolve the new actions against the
- list of executed actions from a previous call.
-
- """
- if state is None:
- state = ConflictResolverState()
-
- # pick up where we left off last time, but track the new actions as well
- state.remaining_actions.extend(normalize_actions(actions))
- actions = state.remaining_actions
-
- def orderandpos(v):
- n, v = v
- return (v['order'] or 0, n)
-
- def orderonly(v):
- n, v = v
- return v['order'] or 0
-
- sactions = sorted(enumerate(actions, start=state.start), key=orderandpos)
- for order, actiongroup in itertools.groupby(sactions, orderonly):
- # "order" is an integer grouping. Actions in a lower order will be
- # executed before actions in a higher order. All of the actions in
- # one grouping will be executed (its callable, if any will be called)
- # before any of the actions in the next.
- output = []
- unique = {}
-
- # error out if we went backward in order
- if state.min_order is not None and order < state.min_order:
- r = [
- 'Actions were added to order={0} after execution had moved '
- 'on to order={1}. Conflicting actions: '.format(
- order, state.min_order
- )
- ]
- for i, action in actiongroup:
- for line in str(action['info']).rstrip().split('\n'):
- r.append(" " + line)
- raise ConfigurationError('\n'.join(r))
-
- for i, action in actiongroup:
- # Within an order, actions are executed sequentially based on
- # original action ordering ("i").
-
- # "ainfo" is a tuple of (i, action) where "i" is an integer
- # expressing the relative position of this action in the action
- # list being resolved, and "action" is an action dictionary. The
- # purpose of an ainfo is to associate an "i" with a particular
- # action; "i" exists for sorting after conflict resolution.
- ainfo = (i, action)
-
- # wait to defer discriminators until we are on their order because
- # the discriminator may depend on state from a previous order
- discriminator = undefer(action['discriminator'])
- action['discriminator'] = discriminator
-
- if discriminator is None:
- # The discriminator is None, so this action can never conflict.
- # We can add it directly to the result.
- output.append(ainfo)
- continue
-
- L = unique.setdefault(discriminator, [])
- L.append(ainfo)
-
- # Check for conflicts
- conflicts = {}
- for discriminator, ainfos in unique.items():
- # We use (includepath, i) as a sort key because we need to
- # sort the actions by the paths so that the shortest path with a
- # given prefix comes first. The "first" action is the one with the
- # shortest include path. We break sorting ties using "i".
- def bypath(ainfo):
- path, i = ainfo[1]['includepath'], ainfo[0]
- return path, order, i
-
- ainfos.sort(key=bypath)
- ainfo, rest = ainfos[0], ainfos[1:]
- _, action = ainfo
-
- # ensure this new action does not conflict with a previously
- # resolved action from an earlier order / invocation
- prev_ainfo = state.resolved_ainfos.get(discriminator)
- if prev_ainfo is not None:
- _, paction = prev_ainfo
- basepath, baseinfo = paction['includepath'], paction['info']
- includepath = action['includepath']
- # if the new action conflicts with the resolved action then
- # note the conflict, otherwise drop the action as it's
- # effectively overriden by the previous action
- if (
- includepath[: len(basepath)] != basepath
- or includepath == basepath
- ):
- L = conflicts.setdefault(discriminator, [baseinfo])
- L.append(action['info'])
-
- else:
- output.append(ainfo)
-
- basepath, baseinfo = action['includepath'], action['info']
- for _, action in rest:
- includepath = action['includepath']
- # Test whether path is a prefix of opath
- if (
- includepath[: len(basepath)] != basepath
- or includepath == basepath # not a prefix
- ):
- L = conflicts.setdefault(discriminator, [baseinfo])
- L.append(action['info'])
-
- if conflicts:
- raise ConfigurationConflictError(conflicts)
-
- # sort resolved actions by "i" and yield them one by one
- for i, action in sorted(output, key=operator.itemgetter(0)):
- # do not memoize the order until we resolve an action inside it
- state.min_order = action['order']
- state.start = i + 1
- state.remaining_actions.remove(action)
- state.resolved_ainfos[action['discriminator']] = (i, action)
- yield action
-
-
-def normalize_actions(actions):
- """Convert old-style tuple actions to new-style dicts."""
- result = []
- for v in actions:
- if not isinstance(v, dict):
- v = expand_action_tuple(*v)
- result.append(v)
- return result
-
-
-def expand_action_tuple(
- discriminator,
- callable=None,
- args=(),
- kw=None,
- includepath=(),
- info=None,
- order=0,
- introspectables=(),
-):
- if kw is None:
- kw = {}
- return dict(
- discriminator=discriminator,
- callable=callable,
- args=args,
- kw=kw,
- includepath=includepath,
- info=info,
- order=order,
- introspectables=introspectables,
- )
global_registries = WeakOrderedSet()
diff --git a/src/pyramid/config/actions.py b/src/pyramid/config/actions.py
new file mode 100644
index 0000000..353ed5e
--- /dev/null
+++ b/src/pyramid/config/actions.py
@@ -0,0 +1,525 @@
+import itertools
+import operator
+import sys
+
+from pyramid.compat import reraise
+
+from pyramid.config.util import ActionInfo
+
+from pyramid.exceptions import (
+ ConfigurationConflictError,
+ ConfigurationError,
+ ConfigurationExecutionError,
+)
+
+from pyramid.registry import undefer
+
+
+class ActionConfiguratorMixin(object):
+ @property
+ def action_info(self):
+ info = self.info # usually a ZCML action (ParserInfo) if self.info
+ if not info:
+ # Try to provide more accurate info for conflict reports
+ if self._ainfo:
+ info = self._ainfo[0]
+ else:
+ info = ActionInfo(None, 0, '', '')
+ return info
+
+ def action(
+ self,
+ discriminator,
+ callable=None,
+ args=(),
+ kw=None,
+ order=0,
+ introspectables=(),
+ **extra
+ ):
+ """ Register an action which will be executed when
+ :meth:`pyramid.config.Configurator.commit` is called (or executed
+ immediately if ``autocommit`` is ``True``).
+
+ .. warning:: This method is typically only used by :app:`Pyramid`
+ framework extension authors, not by :app:`Pyramid` application
+ developers.
+
+ The ``discriminator`` uniquely identifies the action. It must be
+ given, but it can be ``None``, to indicate that the action never
+ conflicts. It must be a hashable value.
+
+ The ``callable`` is a callable object which performs the task
+ associated with the action when the action is executed. It is
+ optional.
+
+ ``args`` and ``kw`` are tuple and dict objects respectively, which
+ are passed to ``callable`` when this action is executed. Both are
+ optional.
+
+ ``order`` is a grouping mechanism; an action with a lower order will
+ be executed before an action with a higher order (has no effect when
+ autocommit is ``True``).
+
+ ``introspectables`` is a sequence of :term:`introspectable` objects
+ (or the empty sequence if no introspectable objects are associated
+ with this action). If this configurator's ``introspection``
+ attribute is ``False``, these introspectables will be ignored.
+
+ ``extra`` provides a facility for inserting extra keys and values
+ into an action dictionary.
+ """
+ # catch nonhashable discriminators here; most unit tests use
+ # autocommit=False, which won't catch unhashable discriminators
+ assert hash(discriminator)
+
+ if kw is None:
+ kw = {}
+
+ autocommit = self.autocommit
+ action_info = self.action_info
+
+ if not self.introspection:
+ # if we're not introspecting, ignore any introspectables passed
+ # to us
+ introspectables = ()
+
+ if autocommit:
+ # callables can depend on the side effects of resolving a
+ # deferred discriminator
+ self.begin()
+ try:
+ undefer(discriminator)
+ if callable is not None:
+ callable(*args, **kw)
+ for introspectable in introspectables:
+ introspectable.register(self.introspector, action_info)
+ finally:
+ self.end()
+
+ else:
+ action = extra
+ action.update(
+ dict(
+ discriminator=discriminator,
+ callable=callable,
+ args=args,
+ kw=kw,
+ order=order,
+ info=action_info,
+ includepath=self.includepath,
+ introspectables=introspectables,
+ )
+ )
+ self.action_state.action(**action)
+
+ def _get_action_state(self):
+ registry = self.registry
+ try:
+ state = registry.action_state
+ except AttributeError:
+ state = ActionState()
+ registry.action_state = state
+ return state
+
+ def _set_action_state(self, state):
+ self.registry.action_state = state
+
+ action_state = property(_get_action_state, _set_action_state)
+
+ _ctx = action_state # bw compat
+
+ def commit(self):
+ """
+ Commit any pending configuration actions. If a configuration
+ conflict is detected in the pending configuration actions, this method
+ will raise a :exc:`ConfigurationConflictError`; within the traceback
+ of this error will be information about the source of the conflict,
+ usually including file names and line numbers of the cause of the
+ configuration conflicts.
+
+ .. warning::
+ You should think very carefully before manually invoking
+ ``commit()``. Especially not as part of any reusable configuration
+ methods. Normally it should only be done by an application author at
+ the end of configuration in order to override certain aspects of an
+ addon.
+
+ """
+ self.begin()
+ try:
+ self.action_state.execute_actions(introspector=self.introspector)
+ finally:
+ self.end()
+ self.action_state = ActionState() # old actions have been processed
+
+# this class is licensed under the ZPL (stolen from Zope)
+class ActionState(object):
+ def __init__(self):
+ # NB "actions" is an API, dep'd upon by pyramid_zcml's load_zcml func
+ self.actions = []
+ self._seen_files = set()
+
+ def processSpec(self, spec):
+ """Check whether a callable needs to be processed. The ``spec``
+ refers to a unique identifier for the callable.
+
+ Return True if processing is needed and False otherwise. If
+ the callable needs to be processed, it will be marked as
+ processed, assuming that the caller will procces the callable if
+ it needs to be processed.
+ """
+ if spec in self._seen_files:
+ return False
+ self._seen_files.add(spec)
+ return True
+
+ def action(
+ self,
+ discriminator,
+ callable=None,
+ args=(),
+ kw=None,
+ order=0,
+ includepath=(),
+ info=None,
+ introspectables=(),
+ **extra
+ ):
+ """Add an action with the given discriminator, callable and arguments
+ """
+ if kw is None:
+ kw = {}
+ action = extra
+ action.update(
+ dict(
+ discriminator=discriminator,
+ callable=callable,
+ args=args,
+ kw=kw,
+ includepath=includepath,
+ info=info,
+ order=order,
+ introspectables=introspectables,
+ )
+ )
+ self.actions.append(action)
+
+ def execute_actions(self, clear=True, introspector=None):
+ """Execute the configuration actions
+
+ This calls the action callables after resolving conflicts
+
+ For example:
+
+ >>> output = []
+ >>> def f(*a, **k):
+ ... output.append(('f', a, k))
+ >>> context = ActionState()
+ >>> context.actions = [
+ ... (1, f, (1,)),
+ ... (1, f, (11,), {}, ('x', )),
+ ... (2, f, (2,)),
+ ... ]
+ >>> context.execute_actions()
+ >>> output
+ [('f', (1,), {}), ('f', (2,), {})]
+
+ If the action raises an error, we convert it to a
+ ConfigurationExecutionError.
+
+ >>> output = []
+ >>> def bad():
+ ... bad.xxx
+ >>> context.actions = [
+ ... (1, f, (1,)),
+ ... (1, f, (11,), {}, ('x', )),
+ ... (2, f, (2,)),
+ ... (3, bad, (), {}, (), 'oops')
+ ... ]
+ >>> try:
+ ... v = context.execute_actions()
+ ... except ConfigurationExecutionError, v:
+ ... pass
+ >>> print(v)
+ exceptions.AttributeError: 'function' object has no attribute 'xxx'
+ in:
+ oops
+
+ Note that actions executed before the error still have an effect:
+
+ >>> output
+ [('f', (1,), {}), ('f', (2,), {})]
+
+ The execution is re-entrant such that actions may be added by other
+ actions with the one caveat that the order of any added actions must
+ be equal to or larger than the current action.
+
+ >>> output = []
+ >>> def f(*a, **k):
+ ... output.append(('f', a, k))
+ ... context.actions.append((3, g, (8,), {}))
+ >>> def g(*a, **k):
+ ... output.append(('g', a, k))
+ >>> context.actions = [
+ ... (1, f, (1,)),
+ ... ]
+ >>> context.execute_actions()
+ >>> output
+ [('f', (1,), {}), ('g', (8,), {})]
+
+ """
+ try:
+ all_actions = []
+ executed_actions = []
+ action_iter = iter([])
+ conflict_state = ConflictResolverState()
+
+ while True:
+ # We clear the actions list prior to execution so if there
+ # are some new actions then we add them to the mix and resolve
+ # conflicts again. This orders the new actions as well as
+ # ensures that the previously executed actions have no new
+ # conflicts.
+ if self.actions:
+ all_actions.extend(self.actions)
+ action_iter = resolveConflicts(
+ self.actions, state=conflict_state
+ )
+ self.actions = []
+
+ action = next(action_iter, None)
+ if action is None:
+ # we are done!
+ break
+
+ callable = action['callable']
+ args = action['args']
+ kw = action['kw']
+ info = action['info']
+ # we use "get" below in case an action was added via a ZCML
+ # directive that did not know about introspectables
+ introspectables = action.get('introspectables', ())
+
+ try:
+ if callable is not None:
+ callable(*args, **kw)
+ except Exception:
+ t, v, tb = sys.exc_info()
+ try:
+ reraise(
+ ConfigurationExecutionError,
+ ConfigurationExecutionError(t, v, info),
+ tb,
+ )
+ finally:
+ del t, v, tb
+
+ if introspector is not None:
+ for introspectable in introspectables:
+ introspectable.register(introspector, info)
+
+ executed_actions.append(action)
+
+ self.actions = all_actions
+ return executed_actions
+
+ finally:
+ if clear:
+ self.actions = []
+
+
+class ConflictResolverState(object):
+ def __init__(self):
+ # keep a set of resolved discriminators to test against to ensure
+ # that a new action does not conflict with something already executed
+ self.resolved_ainfos = {}
+
+ # actions left over from a previous iteration
+ self.remaining_actions = []
+
+ # after executing an action we memoize its order to avoid any new
+ # actions sending us backward
+ self.min_order = None
+
+ # unique tracks the index of the action so we need it to increase
+ # monotonically across invocations to resolveConflicts
+ self.start = 0
+
+
+# this function is licensed under the ZPL (stolen from Zope)
+def resolveConflicts(actions, state=None):
+ """Resolve conflicting actions
+
+ Given an actions list, identify and try to resolve conflicting actions.
+ Actions conflict if they have the same non-None discriminator.
+
+ Conflicting actions can be resolved if the include path of one of
+ the actions is a prefix of the includepaths of the other
+ conflicting actions and is unequal to the include paths in the
+ other conflicting actions.
+
+ Actions are resolved on a per-order basis because some discriminators
+ cannot be computed until earlier actions have executed. An action in an
+ earlier order may execute successfully only to find out later that it was
+ overridden by another action with a smaller include path. This will result
+ in a conflict as there is no way to revert the original action.
+
+ ``state`` may be an instance of ``ConflictResolverState`` that
+ can be used to resume execution and resolve the new actions against the
+ list of executed actions from a previous call.
+
+ """
+ if state is None:
+ state = ConflictResolverState()
+
+ # pick up where we left off last time, but track the new actions as well
+ state.remaining_actions.extend(normalize_actions(actions))
+ actions = state.remaining_actions
+
+ def orderandpos(v):
+ n, v = v
+ return (v['order'] or 0, n)
+
+ def orderonly(v):
+ n, v = v
+ return v['order'] or 0
+
+ sactions = sorted(enumerate(actions, start=state.start), key=orderandpos)
+ for order, actiongroup in itertools.groupby(sactions, orderonly):
+ # "order" is an integer grouping. Actions in a lower order will be
+ # executed before actions in a higher order. All of the actions in
+ # one grouping will be executed (its callable, if any will be called)
+ # before any of the actions in the next.
+ output = []
+ unique = {}
+
+ # error out if we went backward in order
+ if state.min_order is not None and order < state.min_order:
+ r = [
+ 'Actions were added to order={0} after execution had moved '
+ 'on to order={1}. Conflicting actions: '.format(
+ order, state.min_order
+ )
+ ]
+ for i, action in actiongroup:
+ for line in str(action['info']).rstrip().split('\n'):
+ r.append(" " + line)
+ raise ConfigurationError('\n'.join(r))
+
+ for i, action in actiongroup:
+ # Within an order, actions are executed sequentially based on
+ # original action ordering ("i").
+
+ # "ainfo" is a tuple of (i, action) where "i" is an integer
+ # expressing the relative position of this action in the action
+ # list being resolved, and "action" is an action dictionary. The
+ # purpose of an ainfo is to associate an "i" with a particular
+ # action; "i" exists for sorting after conflict resolution.
+ ainfo = (i, action)
+
+ # wait to defer discriminators until we are on their order because
+ # the discriminator may depend on state from a previous order
+ discriminator = undefer(action['discriminator'])
+ action['discriminator'] = discriminator
+
+ if discriminator is None:
+ # The discriminator is None, so this action can never conflict.
+ # We can add it directly to the result.
+ output.append(ainfo)
+ continue
+
+ L = unique.setdefault(discriminator, [])
+ L.append(ainfo)
+
+ # Check for conflicts
+ conflicts = {}
+ for discriminator, ainfos in unique.items():
+ # We use (includepath, i) as a sort key because we need to
+ # sort the actions by the paths so that the shortest path with a
+ # given prefix comes first. The "first" action is the one with the
+ # shortest include path. We break sorting ties using "i".
+ def bypath(ainfo):
+ path, i = ainfo[1]['includepath'], ainfo[0]
+ return path, order, i
+
+ ainfos.sort(key=bypath)
+ ainfo, rest = ainfos[0], ainfos[1:]
+ _, action = ainfo
+
+ # ensure this new action does not conflict with a previously
+ # resolved action from an earlier order / invocation
+ prev_ainfo = state.resolved_ainfos.get(discriminator)
+ if prev_ainfo is not None:
+ _, paction = prev_ainfo
+ basepath, baseinfo = paction['includepath'], paction['info']
+ includepath = action['includepath']
+ # if the new action conflicts with the resolved action then
+ # note the conflict, otherwise drop the action as it's
+ # effectively overriden by the previous action
+ if (
+ includepath[: len(basepath)] != basepath
+ or includepath == basepath
+ ):
+ L = conflicts.setdefault(discriminator, [baseinfo])
+ L.append(action['info'])
+
+ else:
+ output.append(ainfo)
+
+ basepath, baseinfo = action['includepath'], action['info']
+ for _, action in rest:
+ includepath = action['includepath']
+ # Test whether path is a prefix of opath
+ if (
+ includepath[: len(basepath)] != basepath
+ or includepath == basepath # not a prefix
+ ):
+ L = conflicts.setdefault(discriminator, [baseinfo])
+ L.append(action['info'])
+
+ if conflicts:
+ raise ConfigurationConflictError(conflicts)
+
+ # sort resolved actions by "i" and yield them one by one
+ for i, action in sorted(output, key=operator.itemgetter(0)):
+ # do not memoize the order until we resolve an action inside it
+ state.min_order = action['order']
+ state.start = i + 1
+ state.remaining_actions.remove(action)
+ state.resolved_ainfos[action['discriminator']] = (i, action)
+ yield action
+
+
+def normalize_actions(actions):
+ """Convert old-style tuple actions to new-style dicts."""
+ result = []
+ for v in actions:
+ if not isinstance(v, dict):
+ v = expand_action_tuple(*v)
+ result.append(v)
+ return result
+
+
+def expand_action_tuple(
+ discriminator,
+ callable=None,
+ args=(),
+ kw=None,
+ includepath=(),
+ info=None,
+ order=0,
+ introspectables=(),
+):
+ if kw is None:
+ kw = {}
+ return dict(
+ discriminator=discriminator,
+ callable=callable,
+ args=args,
+ kw=kw,
+ includepath=includepath,
+ info=info,
+ order=order,
+ introspectables=introspectables,
+ )
diff --git a/tests/test_config/test_actions.py b/tests/test_config/test_actions.py
new file mode 100644
index 0000000..ba8b8c1
--- /dev/null
+++ b/tests/test_config/test_actions.py
@@ -0,0 +1,1055 @@
+import unittest
+
+from pyramid.exceptions import ConfigurationConflictError
+from pyramid.exceptions import ConfigurationExecutionError
+
+from pyramid.interfaces import IRequest
+
+
+class ConfiguratorTests(unittest.TestCase):
+ def _makeOne(self, *arg, **kw):
+ from pyramid.config import Configurator
+
+ config = Configurator(*arg, **kw)
+ return config
+
+ def _getViewCallable(
+ self,
+ config,
+ ctx_iface=None,
+ request_iface=None,
+ name='',
+ exception_view=False,
+ ):
+ from zope.interface import Interface
+ from pyramid.interfaces import IView
+ from pyramid.interfaces import IViewClassifier
+ from pyramid.interfaces import IExceptionViewClassifier
+
+ if exception_view: # pragma: no cover
+ classifier = IExceptionViewClassifier
+ else:
+ classifier = IViewClassifier
+ if ctx_iface is None:
+ ctx_iface = Interface
+ if request_iface is None:
+ request_iface = IRequest
+ return config.registry.adapters.lookup(
+ (classifier, request_iface, ctx_iface),
+ IView,
+ name=name,
+ default=None,
+ )
+
+ def test_action_branching_kw_is_None(self):
+ config = self._makeOne(autocommit=True)
+ self.assertEqual(config.action('discrim'), None)
+
+ def test_action_branching_kw_is_not_None(self):
+ config = self._makeOne(autocommit=True)
+ self.assertEqual(config.action('discrim', kw={'a': 1}), None)
+
+ def test_action_autocommit_with_introspectables(self):
+ from pyramid.config.util import ActionInfo
+
+ config = self._makeOne(autocommit=True)
+ intr = DummyIntrospectable()
+ config.action('discrim', introspectables=(intr,))
+ self.assertEqual(len(intr.registered), 1)
+ self.assertEqual(intr.registered[0][0], config.introspector)
+ self.assertEqual(intr.registered[0][1].__class__, ActionInfo)
+
+ def test_action_autocommit_with_introspectables_introspection_off(self):
+ config = self._makeOne(autocommit=True)
+ config.introspection = False
+ intr = DummyIntrospectable()
+ config.action('discrim', introspectables=(intr,))
+ self.assertEqual(len(intr.registered), 0)
+
+ def test_action_branching_nonautocommit_with_config_info(self):
+ config = self._makeOne(autocommit=False)
+ config.info = 'abc'
+ state = DummyActionState()
+ state.autocommit = False
+ config.action_state = state
+ config.action('discrim', kw={'a': 1})
+ self.assertEqual(
+ state.actions,
+ [
+ (
+ (),
+ {
+ 'args': (),
+ 'callable': None,
+ 'discriminator': 'discrim',
+ 'includepath': (),
+ 'info': 'abc',
+ 'introspectables': (),
+ 'kw': {'a': 1},
+ 'order': 0,
+ },
+ )
+ ],
+ )
+
+ def test_action_branching_nonautocommit_without_config_info(self):
+ config = self._makeOne(autocommit=False)
+ config.info = ''
+ config._ainfo = ['z']
+ state = DummyActionState()
+ config.action_state = state
+ state.autocommit = False
+ config.action('discrim', kw={'a': 1})
+ self.assertEqual(
+ state.actions,
+ [
+ (
+ (),
+ {
+ 'args': (),
+ 'callable': None,
+ 'discriminator': 'discrim',
+ 'includepath': (),
+ 'info': 'z',
+ 'introspectables': (),
+ 'kw': {'a': 1},
+ 'order': 0,
+ },
+ )
+ ],
+ )
+
+ def test_action_branching_nonautocommit_with_introspectables(self):
+ config = self._makeOne(autocommit=False)
+ config.info = ''
+ config._ainfo = []
+ state = DummyActionState()
+ config.action_state = state
+ state.autocommit = False
+ intr = DummyIntrospectable()
+ config.action('discrim', introspectables=(intr,))
+ self.assertEqual(state.actions[0][1]['introspectables'], (intr,))
+
+ def test_action_nonautocommit_with_introspectables_introspection_off(self):
+ config = self._makeOne(autocommit=False)
+ config.info = ''
+ config._ainfo = []
+ config.introspection = False
+ state = DummyActionState()
+ config.action_state = state
+ state.autocommit = False
+ intr = DummyIntrospectable()
+ config.action('discrim', introspectables=(intr,))
+ self.assertEqual(state.actions[0][1]['introspectables'], ())
+
+ def test_commit_conflict_simple(self):
+ config = self._makeOne()
+
+ def view1(request): # pragma: no cover
+ pass
+
+ def view2(request): # pragma: no cover
+ pass
+
+ config.add_view(view1)
+ config.add_view(view2)
+ self.assertRaises(ConfigurationConflictError, config.commit)
+
+ def test_commit_conflict_resolved_with_include(self):
+ config = self._makeOne()
+
+ def view1(request): # pragma: no cover
+ pass
+
+ def view2(request): # pragma: no cover
+ pass
+
+ def includeme(config):
+ config.add_view(view2)
+
+ config.add_view(view1)
+ config.include(includeme)
+ config.commit()
+ registeredview = self._getViewCallable(config)
+ self.assertEqual(registeredview.__name__, 'view1')
+
+ def test_commit_conflict_with_two_includes(self):
+ config = self._makeOne()
+
+ def view1(request): # pragma: no cover
+ pass
+
+ def view2(request): # pragma: no cover
+ pass
+
+ def includeme1(config):
+ config.add_view(view1)
+
+ def includeme2(config):
+ config.add_view(view2)
+
+ config.include(includeme1)
+ config.include(includeme2)
+ try:
+ config.commit()
+ except ConfigurationConflictError as why:
+ c1, c2 = _conflictFunctions(why)
+ self.assertEqual(c1, 'includeme1')
+ self.assertEqual(c2, 'includeme2')
+ else: # pragma: no cover
+ raise AssertionError
+
+ def test_commit_conflict_resolved_with_two_includes_and_local(self):
+ config = self._makeOne()
+
+ def view1(request): # pragma: no cover
+ pass
+
+ def view2(request): # pragma: no cover
+ pass
+
+ def view3(request): # pragma: no cover
+ pass
+
+ def includeme1(config):
+ config.add_view(view1)
+
+ def includeme2(config):
+ config.add_view(view2)
+
+ config.include(includeme1)
+ config.include(includeme2)
+ config.add_view(view3)
+ config.commit()
+ registeredview = self._getViewCallable(config)
+ self.assertEqual(registeredview.__name__, 'view3')
+
+ def test_autocommit_no_conflicts(self):
+ from pyramid.renderers import null_renderer
+
+ config = self._makeOne(autocommit=True)
+
+ def view1(request): # pragma: no cover
+ pass
+
+ def view2(request): # pragma: no cover
+ pass
+
+ def view3(request): # pragma: no cover
+ pass
+
+ config.add_view(view1, renderer=null_renderer)
+ config.add_view(view2, renderer=null_renderer)
+ config.add_view(view3, renderer=null_renderer)
+ config.commit()
+ registeredview = self._getViewCallable(config)
+ self.assertEqual(registeredview.__name__, 'view3')
+
+ def test_conflict_set_notfound_view(self):
+ config = self._makeOne()
+
+ def view1(request): # pragma: no cover
+ pass
+
+ def view2(request): # pragma: no cover
+ pass
+
+ config.set_notfound_view(view1)
+ config.set_notfound_view(view2)
+ try:
+ config.commit()
+ except ConfigurationConflictError as why:
+ c1, c2 = _conflictFunctions(why)
+ self.assertEqual(c1, 'test_conflict_set_notfound_view')
+ self.assertEqual(c2, 'test_conflict_set_notfound_view')
+ else: # pragma: no cover
+ raise AssertionError
+
+ def test_conflict_set_forbidden_view(self):
+ config = self._makeOne()
+
+ def view1(request): # pragma: no cover
+ pass
+
+ def view2(request): # pragma: no cover
+ pass
+
+ config.set_forbidden_view(view1)
+ config.set_forbidden_view(view2)
+ try:
+ config.commit()
+ except ConfigurationConflictError as why:
+ c1, c2 = _conflictFunctions(why)
+ self.assertEqual(c1, 'test_conflict_set_forbidden_view')
+ self.assertEqual(c2, 'test_conflict_set_forbidden_view')
+ else: # pragma: no cover
+ raise AssertionError
+
+
+class TestActionState(unittest.TestCase):
+ def _makeOne(self):
+ from pyramid.config.actions import ActionState
+
+ return ActionState()
+
+ def test_it(self):
+ c = self._makeOne()
+ self.assertEqual(c.actions, [])
+
+ def test_action_simple(self):
+ from . import dummyfactory as f
+
+ c = self._makeOne()
+ c.actions = []
+ c.action(1, f, (1,), {'x': 1})
+ self.assertEqual(
+ c.actions,
+ [
+ {
+ 'args': (1,),
+ 'callable': f,
+ 'discriminator': 1,
+ 'includepath': (),
+ 'info': None,
+ 'introspectables': (),
+ 'kw': {'x': 1},
+ 'order': 0,
+ }
+ ],
+ )
+ c.action(None)
+ self.assertEqual(
+ c.actions,
+ [
+ {
+ 'args': (1,),
+ 'callable': f,
+ 'discriminator': 1,
+ 'includepath': (),
+ 'info': None,
+ 'introspectables': (),
+ 'kw': {'x': 1},
+ 'order': 0,
+ },
+ {
+ 'args': (),
+ 'callable': None,
+ 'discriminator': None,
+ 'includepath': (),
+ 'info': None,
+ 'introspectables': (),
+ 'kw': {},
+ 'order': 0,
+ },
+ ],
+ )
+
+ def test_action_with_includepath(self):
+ c = self._makeOne()
+ c.actions = []
+ c.action(None, includepath=('abc',))
+ self.assertEqual(
+ c.actions,
+ [
+ {
+ 'args': (),
+ 'callable': None,
+ 'discriminator': None,
+ 'includepath': ('abc',),
+ 'info': None,
+ 'introspectables': (),
+ 'kw': {},
+ 'order': 0,
+ }
+ ],
+ )
+
+ def test_action_with_info(self):
+ c = self._makeOne()
+ c.action(None, info='abc')
+ self.assertEqual(
+ c.actions,
+ [
+ {
+ 'args': (),
+ 'callable': None,
+ 'discriminator': None,
+ 'includepath': (),
+ 'info': 'abc',
+ 'introspectables': (),
+ 'kw': {},
+ 'order': 0,
+ }
+ ],
+ )
+
+ def test_action_with_includepath_and_info(self):
+ c = self._makeOne()
+ c.action(None, includepath=('spec',), info='bleh')
+ self.assertEqual(
+ c.actions,
+ [
+ {
+ 'args': (),
+ 'callable': None,
+ 'discriminator': None,
+ 'includepath': ('spec',),
+ 'info': 'bleh',
+ 'introspectables': (),
+ 'kw': {},
+ 'order': 0,
+ }
+ ],
+ )
+
+ def test_action_with_order(self):
+ c = self._makeOne()
+ c.actions = []
+ c.action(None, order=99999)
+ self.assertEqual(
+ c.actions,
+ [
+ {
+ 'args': (),
+ 'callable': None,
+ 'discriminator': None,
+ 'includepath': (),
+ 'info': None,
+ 'introspectables': (),
+ 'kw': {},
+ 'order': 99999,
+ }
+ ],
+ )
+
+ def test_action_with_introspectables(self):
+ c = self._makeOne()
+ c.actions = []
+ intr = DummyIntrospectable()
+ c.action(None, introspectables=(intr,))
+ self.assertEqual(
+ c.actions,
+ [
+ {
+ 'args': (),
+ 'callable': None,
+ 'discriminator': None,
+ 'includepath': (),
+ 'info': None,
+ 'introspectables': (intr,),
+ 'kw': {},
+ 'order': 0,
+ }
+ ],
+ )
+
+ def test_processSpec(self):
+ c = self._makeOne()
+ self.assertTrue(c.processSpec('spec'))
+ self.assertFalse(c.processSpec('spec'))
+
+ def test_execute_actions_tuples(self):
+ output = []
+
+ def f(*a, **k):
+ output.append((a, k))
+
+ c = self._makeOne()
+ c.actions = [
+ (1, f, (1,)),
+ (1, f, (11,), {}, ('x',)),
+ (2, f, (2,)),
+ (None, None),
+ ]
+ c.execute_actions()
+ self.assertEqual(output, [((1,), {}), ((2,), {})])
+
+ def test_execute_actions_dicts(self):
+ output = []
+
+ def f(*a, **k):
+ output.append((a, k))
+
+ c = self._makeOne()
+ c.actions = [
+ {
+ 'discriminator': 1,
+ 'callable': f,
+ 'args': (1,),
+ 'kw': {},
+ 'order': 0,
+ 'includepath': (),
+ 'info': None,
+ 'introspectables': (),
+ },
+ {
+ 'discriminator': 1,
+ 'callable': f,
+ 'args': (11,),
+ 'kw': {},
+ 'includepath': ('x',),
+ 'order': 0,
+ 'info': None,
+ 'introspectables': (),
+ },
+ {
+ 'discriminator': 2,
+ 'callable': f,
+ 'args': (2,),
+ 'kw': {},
+ 'order': 0,
+ 'includepath': (),
+ 'info': None,
+ 'introspectables': (),
+ },
+ {
+ 'discriminator': None,
+ 'callable': None,
+ 'args': (),
+ 'kw': {},
+ 'order': 0,
+ 'includepath': (),
+ 'info': None,
+ 'introspectables': (),
+ },
+ ]
+ c.execute_actions()
+ self.assertEqual(output, [((1,), {}), ((2,), {})])
+
+ def test_execute_actions_with_introspectables(self):
+ output = []
+
+ def f(*a, **k):
+ output.append((a, k))
+
+ c = self._makeOne()
+ intr = DummyIntrospectable()
+ c.actions = [
+ {
+ 'discriminator': 1,
+ 'callable': f,
+ 'args': (1,),
+ 'kw': {},
+ 'order': 0,
+ 'includepath': (),
+ 'info': None,
+ 'introspectables': (intr,),
+ }
+ ]
+ introspector = object()
+ c.execute_actions(introspector=introspector)
+ self.assertEqual(output, [((1,), {})])
+ self.assertEqual(intr.registered, [(introspector, None)])
+
+ def test_execute_actions_with_introspectable_no_callable(self):
+ c = self._makeOne()
+ intr = DummyIntrospectable()
+ c.actions = [
+ {
+ 'discriminator': 1,
+ 'callable': None,
+ 'args': (1,),
+ 'kw': {},
+ 'order': 0,
+ 'includepath': (),
+ 'info': None,
+ 'introspectables': (intr,),
+ }
+ ]
+ introspector = object()
+ c.execute_actions(introspector=introspector)
+ self.assertEqual(intr.registered, [(introspector, None)])
+
+ def test_execute_actions_error(self):
+ output = []
+
+ def f(*a, **k):
+ output.append(('f', a, k))
+
+ def bad():
+ raise NotImplementedError
+
+ c = self._makeOne()
+ c.actions = [
+ (1, f, (1,)),
+ (1, f, (11,), {}, ('x',)),
+ (2, f, (2,)),
+ (3, bad, (), {}, (), 'oops'),
+ ]
+ self.assertRaises(ConfigurationExecutionError, c.execute_actions)
+ self.assertEqual(output, [('f', (1,), {}), ('f', (2,), {})])
+
+ def test_reentrant_action(self):
+ output = []
+ c = self._makeOne()
+
+ def f(*a, **k):
+ output.append(('f', a, k))
+ c.actions.append((3, g, (8,), {}))
+
+ def g(*a, **k):
+ output.append(('g', a, k))
+
+ c.actions = [(1, f, (1,))]
+ c.execute_actions()
+ self.assertEqual(output, [('f', (1,), {}), ('g', (8,), {})])
+
+ def test_reentrant_action_with_deferred_discriminator(self):
+ # see https://github.com/Pylons/pyramid/issues/2697
+ from pyramid.registry import Deferred
+
+ output = []
+ c = self._makeOne()
+
+ def f(*a, **k):
+ output.append(('f', a, k))
+ c.actions.append((4, g, (4,), {}, (), None, 2))
+
+ def g(*a, **k):
+ output.append(('g', a, k))
+
+ def h(*a, **k):
+ output.append(('h', a, k))
+
+ def discrim():
+ self.assertEqual(output, [('f', (1,), {}), ('g', (2,), {})])
+ return 3
+
+ d = Deferred(discrim)
+ c.actions = [
+ (d, h, (3,), {}, (), None, 1), # order 1
+ (1, f, (1,)), # order 0
+ (2, g, (2,)), # order 0
+ ]
+ c.execute_actions()
+ self.assertEqual(
+ output,
+ [
+ ('f', (1,), {}),
+ ('g', (2,), {}),
+ ('h', (3,), {}),
+ ('g', (4,), {}),
+ ],
+ )
+
+ def test_reentrant_action_error(self):
+ from pyramid.exceptions import ConfigurationError
+
+ c = self._makeOne()
+
+ def f(*a, **k):
+ c.actions.append((3, g, (8,), {}, (), None, -1))
+
+ def g(*a, **k): # pragma: no cover
+ pass
+
+ c.actions = [(1, f, (1,))]
+ self.assertRaises(ConfigurationError, c.execute_actions)
+
+ def test_reentrant_action_without_clear(self):
+ c = self._makeOne()
+
+ def f(*a, **k):
+ c.actions.append((3, g, (8,)))
+
+ def g(*a, **k):
+ pass
+
+ c.actions = [(1, f, (1,))]
+ c.execute_actions(clear=False)
+ self.assertEqual(c.actions, [(1, f, (1,)), (3, g, (8,))])
+
+ def test_executing_conflicting_action_across_orders(self):
+ from pyramid.exceptions import ConfigurationConflictError
+
+ c = self._makeOne()
+
+ def f(*a, **k):
+ pass
+
+ def g(*a, **k): # pragma: no cover
+ pass
+
+ c.actions = [(1, f, (1,), {}, (), None, -1), (1, g, (2,))]
+ self.assertRaises(ConfigurationConflictError, c.execute_actions)
+
+ def test_executing_conflicting_action_across_reentrant_orders(self):
+ from pyramid.exceptions import ConfigurationConflictError
+
+ c = self._makeOne()
+
+ def f(*a, **k):
+ c.actions.append((1, g, (8,)))
+
+ def g(*a, **k): # pragma: no cover
+ pass
+
+ c.actions = [(1, f, (1,), {}, (), None, -1)]
+ self.assertRaises(ConfigurationConflictError, c.execute_actions)
+
+
+class Test_reentrant_action_functional(unittest.TestCase):
+ def _makeConfigurator(self, *arg, **kw):
+ from pyramid.config import Configurator
+
+ config = Configurator(*arg, **kw)
+ return config
+
+ def test_functional(self):
+ def add_auto_route(config, name, view):
+ def register():
+ config.add_view(route_name=name, view=view)
+ config.add_route(name, '/' + name)
+
+ config.action(('auto route', name), register, order=-30)
+
+ config = self._makeConfigurator()
+ config.add_directive('add_auto_route', add_auto_route)
+
+ def my_view(request): # pragma: no cover
+ return request.response
+
+ config.add_auto_route('foo', my_view)
+ config.commit()
+ from pyramid.interfaces import IRoutesMapper
+
+ mapper = config.registry.getUtility(IRoutesMapper)
+ routes = mapper.get_routes()
+ route = routes[0]
+ self.assertEqual(len(routes), 1)
+ self.assertEqual(route.name, 'foo')
+ self.assertEqual(route.path, '/foo')
+
+ def test_deferred_discriminator(self):
+ # see https://github.com/Pylons/pyramid/issues/2697
+ from pyramid.config import PHASE0_CONFIG
+
+ config = self._makeConfigurator()
+
+ def deriver(view, info):
+ return view
+
+ deriver.options = ('foo',)
+ config.add_view_deriver(deriver, 'foo_view')
+ # add_view uses a deferred discriminator and will fail if executed
+ # prior to add_view_deriver executing its action
+ config.add_view(lambda r: r.response, name='', foo=1)
+
+ def dummy_action():
+ # trigger a re-entrant action
+ config.action(None, lambda: None)
+
+ config.action(None, dummy_action, order=PHASE0_CONFIG)
+ config.commit()
+
+
+class Test_resolveConflicts(unittest.TestCase):
+ def _callFUT(self, actions):
+ from pyramid.config.actions import resolveConflicts
+
+ return resolveConflicts(actions)
+
+ def test_it_success_tuples(self):
+ from . import dummyfactory as f
+
+ result = self._callFUT(
+ [
+ (None, f),
+ (1, f, (1,), {}, (), 'first'),
+ (1, f, (2,), {}, ('x',), 'second'),
+ (1, f, (3,), {}, ('y',), 'third'),
+ (4, f, (4,), {}, ('y',), 'should be last', 99999),
+ (3, f, (3,), {}, ('y',)),
+ (None, f, (5,), {}, ('y',)),
+ ]
+ )
+ result = list(result)
+ self.assertEqual(
+ result,
+ [
+ {
+ 'info': None,
+ 'args': (),
+ 'callable': f,
+ 'introspectables': (),
+ 'kw': {},
+ 'discriminator': None,
+ 'includepath': (),
+ 'order': 0,
+ },
+ {
+ 'info': 'first',
+ 'args': (1,),
+ 'callable': f,
+ 'introspectables': (),
+ 'kw': {},
+ 'discriminator': 1,
+ 'includepath': (),
+ 'order': 0,
+ },
+ {
+ 'info': None,
+ 'args': (3,),
+ 'callable': f,
+ 'introspectables': (),
+ 'kw': {},
+ 'discriminator': 3,
+ 'includepath': ('y',),
+ 'order': 0,
+ },
+ {
+ 'info': None,
+ 'args': (5,),
+ 'callable': f,
+ 'introspectables': (),
+ 'kw': {},
+ 'discriminator': None,
+ 'includepath': ('y',),
+ 'order': 0,
+ },
+ {
+ 'info': 'should be last',
+ 'args': (4,),
+ 'callable': f,
+ 'introspectables': (),
+ 'kw': {},
+ 'discriminator': 4,
+ 'includepath': ('y',),
+ 'order': 99999,
+ },
+ ],
+ )
+
+ def test_it_success_dicts(self):
+ from . import dummyfactory as f
+
+ result = self._callFUT(
+ [
+ (None, f),
+ (1, f, (1,), {}, (), 'first'),
+ (1, f, (2,), {}, ('x',), 'second'),
+ (1, f, (3,), {}, ('y',), 'third'),
+ (4, f, (4,), {}, ('y',), 'should be last', 99999),
+ (3, f, (3,), {}, ('y',)),
+ (None, f, (5,), {}, ('y',)),
+ ]
+ )
+ result = list(result)
+ self.assertEqual(
+ result,
+ [
+ {
+ 'info': None,
+ 'args': (),
+ 'callable': f,
+ 'introspectables': (),
+ 'kw': {},
+ 'discriminator': None,
+ 'includepath': (),
+ 'order': 0,
+ },
+ {
+ 'info': 'first',
+ 'args': (1,),
+ 'callable': f,
+ 'introspectables': (),
+ 'kw': {},
+ 'discriminator': 1,
+ 'includepath': (),
+ 'order': 0,
+ },
+ {
+ 'info': None,
+ 'args': (3,),
+ 'callable': f,
+ 'introspectables': (),
+ 'kw': {},
+ 'discriminator': 3,
+ 'includepath': ('y',),
+ 'order': 0,
+ },
+ {
+ 'info': None,
+ 'args': (5,),
+ 'callable': f,
+ 'introspectables': (),
+ 'kw': {},
+ 'discriminator': None,
+ 'includepath': ('y',),
+ 'order': 0,
+ },
+ {
+ 'info': 'should be last',
+ 'args': (4,),
+ 'callable': f,
+ 'introspectables': (),
+ 'kw': {},
+ 'discriminator': 4,
+ 'includepath': ('y',),
+ 'order': 99999,
+ },
+ ],
+ )
+
+ def test_it_conflict(self):
+ from . import dummyfactory as f
+
+ result = self._callFUT(
+ [
+ (None, f),
+ (1, f, (2,), {}, ('x',), 'eek'), # will conflict
+ (1, f, (3,), {}, ('y',), 'ack'), # will conflict
+ (4, f, (4,), {}, ('y',)),
+ (3, f, (3,), {}, ('y',)),
+ (None, f, (5,), {}, ('y',)),
+ ]
+ )
+ self.assertRaises(ConfigurationConflictError, list, result)
+
+ def test_it_with_actions_grouped_by_order(self):
+ from . import dummyfactory as f
+
+ result = self._callFUT(
+ [
+ (None, f), # X
+ (1, f, (1,), {}, (), 'third', 10), # X
+ (1, f, (2,), {}, ('x',), 'fourth', 10),
+ (1, f, (3,), {}, ('y',), 'fifth', 10),
+ (2, f, (1,), {}, (), 'sixth', 10), # X
+ (3, f, (1,), {}, (), 'seventh', 10), # X
+ (5, f, (4,), {}, ('y',), 'eighth', 99999), # X
+ (4, f, (3,), {}, (), 'first', 5), # X
+ (4, f, (5,), {}, ('y',), 'second', 5),
+ ]
+ )
+ result = list(result)
+ self.assertEqual(len(result), 6)
+ # resolved actions should be grouped by (order, i)
+ self.assertEqual(
+ result,
+ [
+ {
+ 'info': None,
+ 'args': (),
+ 'callable': f,
+ 'introspectables': (),
+ 'kw': {},
+ 'discriminator': None,
+ 'includepath': (),
+ 'order': 0,
+ },
+ {
+ 'info': 'first',
+ 'args': (3,),
+ 'callable': f,
+ 'introspectables': (),
+ 'kw': {},
+ 'discriminator': 4,
+ 'includepath': (),
+ 'order': 5,
+ },
+ {
+ 'info': 'third',
+ 'args': (1,),
+ 'callable': f,
+ 'introspectables': (),
+ 'kw': {},
+ 'discriminator': 1,
+ 'includepath': (),
+ 'order': 10,
+ },
+ {
+ 'info': 'sixth',
+ 'args': (1,),
+ 'callable': f,
+ 'introspectables': (),
+ 'kw': {},
+ 'discriminator': 2,
+ 'includepath': (),
+ 'order': 10,
+ },
+ {
+ 'info': 'seventh',
+ 'args': (1,),
+ 'callable': f,
+ 'introspectables': (),
+ 'kw': {},
+ 'discriminator': 3,
+ 'includepath': (),
+ 'order': 10,
+ },
+ {
+ 'info': 'eighth',
+ 'args': (4,),
+ 'callable': f,
+ 'introspectables': (),
+ 'kw': {},
+ 'discriminator': 5,
+ 'includepath': ('y',),
+ 'order': 99999,
+ },
+ ],
+ )
+
+ def test_override_success_across_orders(self):
+ from . import dummyfactory as f
+
+ result = self._callFUT(
+ [
+ (1, f, (2,), {}, ('x',), 'eek', 0),
+ (1, f, (3,), {}, ('x', 'y'), 'ack', 10),
+ ]
+ )
+ result = list(result)
+ self.assertEqual(
+ result,
+ [
+ {
+ 'info': 'eek',
+ 'args': (2,),
+ 'callable': f,
+ 'introspectables': (),
+ 'kw': {},
+ 'discriminator': 1,
+ 'includepath': ('x',),
+ 'order': 0,
+ }
+ ],
+ )
+
+ def test_conflicts_across_orders(self):
+ from . import dummyfactory as f
+
+ result = self._callFUT(
+ [
+ (1, f, (2,), {}, ('x', 'y'), 'eek', 0),
+ (1, f, (3,), {}, ('x'), 'ack', 10),
+ ]
+ )
+ self.assertRaises(ConfigurationConflictError, list, result)
+
+
+def _conflictFunctions(e):
+ conflicts = e._conflicts.values()
+ for conflict in conflicts:
+ for confinst in conflict:
+ yield confinst.function
+
+
+class DummyActionState(object):
+ autocommit = False
+ info = ''
+
+ def __init__(self):
+ self.actions = []
+
+ def action(self, *arg, **kw):
+ self.actions.append((arg, kw))
+
+
+class DummyIntrospectable(object):
+ def __init__(self):
+ self.registered = []
+
+ def register(self, introspector, action_info):
+ self.registered.append((introspector, action_info))
diff --git a/tests/test_config/test_init.py b/tests/test_config/test_init.py
index 2c92b60..8216b55 100644
--- a/tests/test_config/test_init.py
+++ b/tests/test_config/test_init.py
@@ -1,7 +1,5 @@
import os
import unittest
-from zope.interface import Interface
-from zope.interface import implementer
from pyramid.compat import im_func
from pyramid.testing import skip_on
@@ -10,7 +8,6 @@
from . import dummy_include
from . import dummy_extend
from . import dummy_extend2
-from . import IDummy
from . import DummyContext
from pyramid.exceptions import ConfigurationExecutionError
@@ -35,7 +32,6 @@
exception_view=False,
):
from zope.interface import Interface
- from pyramid.interfaces import IRequest
from pyramid.interfaces import IView
from pyramid.interfaces import IViewClassifier
from pyramid.interfaces import IExceptionViewClassifier
@@ -299,7 +295,6 @@
def test_ctor_httpexception_view_default(self):
from pyramid.interfaces import IExceptionResponse
from pyramid.httpexceptions import default_exceptionresponse_view
- from pyramid.interfaces import IRequest
config = self._makeOne()
view = self._getViewCallable(
@@ -309,7 +304,6 @@
def test_ctor_exceptionresponse_view_None(self):
from pyramid.interfaces import IExceptionResponse
- from pyramid.interfaces import IRequest
config = self._makeOne(exceptionresponse_view=None)
view = self._getViewCallable(
@@ -319,7 +313,6 @@
def test_ctor_exceptionresponse_view_custom(self):
from pyramid.interfaces import IExceptionResponse
- from pyramid.interfaces import IRequest
def exceptionresponse_view(context, request): # pragma: no cover
pass
@@ -548,7 +541,6 @@
def test_setup_registry_explicit_notfound_trumps_iexceptionresponse(self):
from pyramid.renderers import null_renderer
from zope.interface import implementedBy
- from pyramid.interfaces import IRequest
from pyramid.httpexceptions import HTTPNotFound
from pyramid.registry import Registry
@@ -975,110 +967,8 @@
config.include(include)
self.assertTrue(stack[0] is config.registry)
- def test_action_branching_kw_is_None(self):
- config = self._makeOne(autocommit=True)
- self.assertEqual(config.action('discrim'), None)
-
- def test_action_branching_kw_is_not_None(self):
- config = self._makeOne(autocommit=True)
- self.assertEqual(config.action('discrim', kw={'a': 1}), None)
-
- def test_action_autocommit_with_introspectables(self):
- from pyramid.config.util import ActionInfo
-
- config = self._makeOne(autocommit=True)
- intr = DummyIntrospectable()
- config.action('discrim', introspectables=(intr,))
- self.assertEqual(len(intr.registered), 1)
- self.assertEqual(intr.registered[0][0], config.introspector)
- self.assertEqual(intr.registered[0][1].__class__, ActionInfo)
-
- def test_action_autocommit_with_introspectables_introspection_off(self):
- config = self._makeOne(autocommit=True)
- config.introspection = False
- intr = DummyIntrospectable()
- config.action('discrim', introspectables=(intr,))
- self.assertEqual(len(intr.registered), 0)
-
- def test_action_branching_nonautocommit_with_config_info(self):
- config = self._makeOne(autocommit=False)
- config.info = 'abc'
- state = DummyActionState()
- state.autocommit = False
- config.action_state = state
- config.action('discrim', kw={'a': 1})
- self.assertEqual(
- state.actions,
- [
- (
- (),
- {
- 'args': (),
- 'callable': None,
- 'discriminator': 'discrim',
- 'includepath': (),
- 'info': 'abc',
- 'introspectables': (),
- 'kw': {'a': 1},
- 'order': 0,
- },
- )
- ],
- )
-
- def test_action_branching_nonautocommit_without_config_info(self):
- config = self._makeOne(autocommit=False)
- config.info = ''
- config._ainfo = ['z']
- state = DummyActionState()
- config.action_state = state
- state.autocommit = False
- config.action('discrim', kw={'a': 1})
- self.assertEqual(
- state.actions,
- [
- (
- (),
- {
- 'args': (),
- 'callable': None,
- 'discriminator': 'discrim',
- 'includepath': (),
- 'info': 'z',
- 'introspectables': (),
- 'kw': {'a': 1},
- 'order': 0,
- },
- )
- ],
- )
-
- def test_action_branching_nonautocommit_with_introspectables(self):
- config = self._makeOne(autocommit=False)
- config.info = ''
- config._ainfo = []
- state = DummyActionState()
- config.action_state = state
- state.autocommit = False
- intr = DummyIntrospectable()
- config.action('discrim', introspectables=(intr,))
- self.assertEqual(state.actions[0][1]['introspectables'], (intr,))
-
- def test_action_nonautocommit_with_introspectables_introspection_off(self):
- config = self._makeOne(autocommit=False)
- config.info = ''
- config._ainfo = []
- config.introspection = False
- state = DummyActionState()
- config.action_state = state
- state.autocommit = False
- intr = DummyIntrospectable()
- config.action('discrim', introspectables=(intr,))
- self.assertEqual(state.actions[0][1]['introspectables'], ())
-
def test_scan_integration(self):
from zope.interface import alsoProvides
- from pyramid.interfaces import IRequest
from pyramid.view import render_view_to_response
import tests.test_config.pkgs.scannable as package
@@ -1185,7 +1075,6 @@
def test_scan_integration_with_ignore(self):
from zope.interface import alsoProvides
- from pyramid.interfaces import IRequest
from pyramid.view import render_view_to_response
import tests.test_config.pkgs.scannable as package
@@ -1207,7 +1096,6 @@
def test_scan_integration_dottedname_package(self):
from zope.interface import alsoProvides
- from pyramid.interfaces import IRequest
from pyramid.view import render_view_to_response
config = self._makeOne(autocommit=True)
@@ -1305,149 +1193,6 @@
self.assertNotEqual(sm, '123')
finally:
getSiteManager.reset()
-
- def test_commit_conflict_simple(self):
- config = self._makeOne()
-
- def view1(request): # pragma: no cover
- pass
-
- def view2(request): # pragma: no cover
- pass
-
- config.add_view(view1)
- config.add_view(view2)
- self.assertRaises(ConfigurationConflictError, config.commit)
-
- def test_commit_conflict_resolved_with_include(self):
- config = self._makeOne()
-
- def view1(request): # pragma: no cover
- pass
-
- def view2(request): # pragma: no cover
- pass
-
- def includeme(config):
- config.add_view(view2)
-
- config.add_view(view1)
- config.include(includeme)
- config.commit()
- registeredview = self._getViewCallable(config)
- self.assertEqual(registeredview.__name__, 'view1')
-
- def test_commit_conflict_with_two_includes(self):
- config = self._makeOne()
-
- def view1(request): # pragma: no cover
- pass
-
- def view2(request): # pragma: no cover
- pass
-
- def includeme1(config):
- config.add_view(view1)
-
- def includeme2(config):
- config.add_view(view2)
-
- config.include(includeme1)
- config.include(includeme2)
- try:
- config.commit()
- except ConfigurationConflictError as why:
- c1, c2 = _conflictFunctions(why)
- self.assertEqual(c1, 'includeme1')
- self.assertEqual(c2, 'includeme2')
- else: # pragma: no cover
- raise AssertionError
-
- def test_commit_conflict_resolved_with_two_includes_and_local(self):
- config = self._makeOne()
-
- def view1(request): # pragma: no cover
- pass
-
- def view2(request): # pragma: no cover
- pass
-
- def view3(request): # pragma: no cover
- pass
-
- def includeme1(config):
- config.add_view(view1)
-
- def includeme2(config):
- config.add_view(view2)
-
- config.include(includeme1)
- config.include(includeme2)
- config.add_view(view3)
- config.commit()
- registeredview = self._getViewCallable(config)
- self.assertEqual(registeredview.__name__, 'view3')
-
- def test_autocommit_no_conflicts(self):
- from pyramid.renderers import null_renderer
-
- config = self._makeOne(autocommit=True)
-
- def view1(request): # pragma: no cover
- pass
-
- def view2(request): # pragma: no cover
- pass
-
- def view3(request): # pragma: no cover
- pass
-
- config.add_view(view1, renderer=null_renderer)
- config.add_view(view2, renderer=null_renderer)
- config.add_view(view3, renderer=null_renderer)
- config.commit()
- registeredview = self._getViewCallable(config)
- self.assertEqual(registeredview.__name__, 'view3')
-
- def test_conflict_set_notfound_view(self):
- config = self._makeOne()
-
- def view1(request): # pragma: no cover
- pass
-
- def view2(request): # pragma: no cover
- pass
-
- config.set_notfound_view(view1)
- config.set_notfound_view(view2)
- try:
- config.commit()
- except ConfigurationConflictError as why:
- c1, c2 = _conflictFunctions(why)
- self.assertEqual(c1, 'test_conflict_set_notfound_view')
- self.assertEqual(c2, 'test_conflict_set_notfound_view')
- else: # pragma: no cover
- raise AssertionError
-
- def test_conflict_set_forbidden_view(self):
- config = self._makeOne()
-
- def view1(request): # pragma: no cover
- pass
-
- def view2(request): # pragma: no cover
- pass
-
- config.set_forbidden_view(view1)
- config.set_forbidden_view(view2)
- try:
- config.commit()
- except ConfigurationConflictError as why:
- c1, c2 = _conflictFunctions(why)
- self.assertEqual(c1, 'test_conflict_set_forbidden_view')
- self.assertEqual(c2, 'test_conflict_set_forbidden_view')
- else: # pragma: no cover
- raise AssertionError
def test___getattr__missing_when_directives_exist(self):
config = self._makeOne()
@@ -1624,749 +1369,6 @@
)
-class TestActionState(unittest.TestCase):
- def _makeOne(self):
- from pyramid.config import ActionState
-
- return ActionState()
-
- def test_it(self):
- c = self._makeOne()
- self.assertEqual(c.actions, [])
-
- def test_action_simple(self):
- from . import dummyfactory as f
-
- c = self._makeOne()
- c.actions = []
- c.action(1, f, (1,), {'x': 1})
- self.assertEqual(
- c.actions,
- [
- {
- 'args': (1,),
- 'callable': f,
- 'discriminator': 1,
- 'includepath': (),
- 'info': None,
- 'introspectables': (),
- 'kw': {'x': 1},
- 'order': 0,
- }
- ],
- )
- c.action(None)
- self.assertEqual(
- c.actions,
- [
- {
- 'args': (1,),
- 'callable': f,
- 'discriminator': 1,
- 'includepath': (),
- 'info': None,
- 'introspectables': (),
- 'kw': {'x': 1},
- 'order': 0,
- },
- {
- 'args': (),
- 'callable': None,
- 'discriminator': None,
- 'includepath': (),
- 'info': None,
- 'introspectables': (),
- 'kw': {},
- 'order': 0,
- },
- ],
- )
-
- def test_action_with_includepath(self):
- c = self._makeOne()
- c.actions = []
- c.action(None, includepath=('abc',))
- self.assertEqual(
- c.actions,
- [
- {
- 'args': (),
- 'callable': None,
- 'discriminator': None,
- 'includepath': ('abc',),
- 'info': None,
- 'introspectables': (),
- 'kw': {},
- 'order': 0,
- }
- ],
- )
-
- def test_action_with_info(self):
- c = self._makeOne()
- c.action(None, info='abc')
- self.assertEqual(
- c.actions,
- [
- {
- 'args': (),
- 'callable': None,
- 'discriminator': None,
- 'includepath': (),
- 'info': 'abc',
- 'introspectables': (),
- 'kw': {},
- 'order': 0,
- }
- ],
- )
-
- def test_action_with_includepath_and_info(self):
- c = self._makeOne()
- c.action(None, includepath=('spec',), info='bleh')
- self.assertEqual(
- c.actions,
- [
- {
- 'args': (),
- 'callable': None,
- 'discriminator': None,
- 'includepath': ('spec',),
- 'info': 'bleh',
- 'introspectables': (),
- 'kw': {},
- 'order': 0,
- }
- ],
- )
-
- def test_action_with_order(self):
- c = self._makeOne()
- c.actions = []
- c.action(None, order=99999)
- self.assertEqual(
- c.actions,
- [
- {
- 'args': (),
- 'callable': None,
- 'discriminator': None,
- 'includepath': (),
- 'info': None,
- 'introspectables': (),
- 'kw': {},
- 'order': 99999,
- }
- ],
- )
-
- def test_action_with_introspectables(self):
- c = self._makeOne()
- c.actions = []
- intr = DummyIntrospectable()
- c.action(None, introspectables=(intr,))
- self.assertEqual(
- c.actions,
- [
- {
- 'args': (),
- 'callable': None,
- 'discriminator': None,
- 'includepath': (),
- 'info': None,
- 'introspectables': (intr,),
- 'kw': {},
- 'order': 0,
- }
- ],
- )
-
- def test_processSpec(self):
- c = self._makeOne()
- self.assertTrue(c.processSpec('spec'))
- self.assertFalse(c.processSpec('spec'))
-
- def test_execute_actions_tuples(self):
- output = []
-
- def f(*a, **k):
- output.append((a, k))
-
- c = self._makeOne()
- c.actions = [
- (1, f, (1,)),
- (1, f, (11,), {}, ('x',)),
- (2, f, (2,)),
- (None, None),
- ]
- c.execute_actions()
- self.assertEqual(output, [((1,), {}), ((2,), {})])
-
- def test_execute_actions_dicts(self):
- output = []
-
- def f(*a, **k):
- output.append((a, k))
-
- c = self._makeOne()
- c.actions = [
- {
- 'discriminator': 1,
- 'callable': f,
- 'args': (1,),
- 'kw': {},
- 'order': 0,
- 'includepath': (),
- 'info': None,
- 'introspectables': (),
- },
- {
- 'discriminator': 1,
- 'callable': f,
- 'args': (11,),
- 'kw': {},
- 'includepath': ('x',),
- 'order': 0,
- 'info': None,
- 'introspectables': (),
- },
- {
- 'discriminator': 2,
- 'callable': f,
- 'args': (2,),
- 'kw': {},
- 'order': 0,
- 'includepath': (),
- 'info': None,
- 'introspectables': (),
- },
- {
- 'discriminator': None,
- 'callable': None,
- 'args': (),
- 'kw': {},
- 'order': 0,
- 'includepath': (),
- 'info': None,
- 'introspectables': (),
- },
- ]
- c.execute_actions()
- self.assertEqual(output, [((1,), {}), ((2,), {})])
-
- def test_execute_actions_with_introspectables(self):
- output = []
-
- def f(*a, **k):
- output.append((a, k))
-
- c = self._makeOne()
- intr = DummyIntrospectable()
- c.actions = [
- {
- 'discriminator': 1,
- 'callable': f,
- 'args': (1,),
- 'kw': {},
- 'order': 0,
- 'includepath': (),
- 'info': None,
- 'introspectables': (intr,),
- }
- ]
- introspector = object()
- c.execute_actions(introspector=introspector)
- self.assertEqual(output, [((1,), {})])
- self.assertEqual(intr.registered, [(introspector, None)])
-
- def test_execute_actions_with_introspectable_no_callable(self):
- c = self._makeOne()
- intr = DummyIntrospectable()
- c.actions = [
- {
- 'discriminator': 1,
- 'callable': None,
- 'args': (1,),
- 'kw': {},
- 'order': 0,
- 'includepath': (),
- 'info': None,
- 'introspectables': (intr,),
- }
- ]
- introspector = object()
- c.execute_actions(introspector=introspector)
- self.assertEqual(intr.registered, [(introspector, None)])
-
- def test_execute_actions_error(self):
- output = []
-
- def f(*a, **k):
- output.append(('f', a, k))
-
- def bad():
- raise NotImplementedError
-
- c = self._makeOne()
- c.actions = [
- (1, f, (1,)),
- (1, f, (11,), {}, ('x',)),
- (2, f, (2,)),
- (3, bad, (), {}, (), 'oops'),
- ]
- self.assertRaises(ConfigurationExecutionError, c.execute_actions)
- self.assertEqual(output, [('f', (1,), {}), ('f', (2,), {})])
-
- def test_reentrant_action(self):
- output = []
- c = self._makeOne()
-
- def f(*a, **k):
- output.append(('f', a, k))
- c.actions.append((3, g, (8,), {}))
-
- def g(*a, **k):
- output.append(('g', a, k))
-
- c.actions = [(1, f, (1,))]
- c.execute_actions()
- self.assertEqual(output, [('f', (1,), {}), ('g', (8,), {})])
-
- def test_reentrant_action_with_deferred_discriminator(self):
- # see https://github.com/Pylons/pyramid/issues/2697
- from pyramid.registry import Deferred
-
- output = []
- c = self._makeOne()
-
- def f(*a, **k):
- output.append(('f', a, k))
- c.actions.append((4, g, (4,), {}, (), None, 2))
-
- def g(*a, **k):
- output.append(('g', a, k))
-
- def h(*a, **k):
- output.append(('h', a, k))
-
- def discrim():
- self.assertEqual(output, [('f', (1,), {}), ('g', (2,), {})])
- return 3
-
- d = Deferred(discrim)
- c.actions = [
- (d, h, (3,), {}, (), None, 1), # order 1
- (1, f, (1,)), # order 0
- (2, g, (2,)), # order 0
- ]
- c.execute_actions()
- self.assertEqual(
- output,
- [
- ('f', (1,), {}),
- ('g', (2,), {}),
- ('h', (3,), {}),
- ('g', (4,), {}),
- ],
- )
-
- def test_reentrant_action_error(self):
- from pyramid.exceptions import ConfigurationError
-
- c = self._makeOne()
-
- def f(*a, **k):
- c.actions.append((3, g, (8,), {}, (), None, -1))
-
- def g(*a, **k): # pragma: no cover
- pass
-
- c.actions = [(1, f, (1,))]
- self.assertRaises(ConfigurationError, c.execute_actions)
-
- def test_reentrant_action_without_clear(self):
- c = self._makeOne()
-
- def f(*a, **k):
- c.actions.append((3, g, (8,)))
-
- def g(*a, **k):
- pass
-
- c.actions = [(1, f, (1,))]
- c.execute_actions(clear=False)
- self.assertEqual(c.actions, [(1, f, (1,)), (3, g, (8,))])
-
- def test_executing_conflicting_action_across_orders(self):
- from pyramid.exceptions import ConfigurationConflictError
-
- c = self._makeOne()
-
- def f(*a, **k):
- pass
-
- def g(*a, **k): # pragma: no cover
- pass
-
- c.actions = [(1, f, (1,), {}, (), None, -1), (1, g, (2,))]
- self.assertRaises(ConfigurationConflictError, c.execute_actions)
-
- def test_executing_conflicting_action_across_reentrant_orders(self):
- from pyramid.exceptions import ConfigurationConflictError
-
- c = self._makeOne()
-
- def f(*a, **k):
- c.actions.append((1, g, (8,)))
-
- def g(*a, **k): # pragma: no cover
- pass
-
- c.actions = [(1, f, (1,), {}, (), None, -1)]
- self.assertRaises(ConfigurationConflictError, c.execute_actions)
-
-
-class Test_reentrant_action_functional(unittest.TestCase):
- def _makeConfigurator(self, *arg, **kw):
- from pyramid.config import Configurator
-
- config = Configurator(*arg, **kw)
- return config
-
- def test_functional(self):
- def add_auto_route(config, name, view):
- def register():
- config.add_view(route_name=name, view=view)
- config.add_route(name, '/' + name)
-
- config.action(('auto route', name), register, order=-30)
-
- config = self._makeConfigurator()
- config.add_directive('add_auto_route', add_auto_route)
-
- def my_view(request): # pragma: no cover
- return request.response
-
- config.add_auto_route('foo', my_view)
- config.commit()
- from pyramid.interfaces import IRoutesMapper
-
- mapper = config.registry.getUtility(IRoutesMapper)
- routes = mapper.get_routes()
- route = routes[0]
- self.assertEqual(len(routes), 1)
- self.assertEqual(route.name, 'foo')
- self.assertEqual(route.path, '/foo')
-
- def test_deferred_discriminator(self):
- # see https://github.com/Pylons/pyramid/issues/2697
- from pyramid.config import PHASE0_CONFIG
-
- config = self._makeConfigurator()
-
- def deriver(view, info):
- return view
-
- deriver.options = ('foo',)
- config.add_view_deriver(deriver, 'foo_view')
- # add_view uses a deferred discriminator and will fail if executed
- # prior to add_view_deriver executing its action
- config.add_view(lambda r: r.response, name='', foo=1)
-
- def dummy_action():
- # trigger a re-entrant action
- config.action(None, lambda: None)
-
- config.action(None, dummy_action, order=PHASE0_CONFIG)
- config.commit()
-
-
-class Test_resolveConflicts(unittest.TestCase):
- def _callFUT(self, actions):
- from pyramid.config import resolveConflicts
-
- return resolveConflicts(actions)
-
- def test_it_success_tuples(self):
- from . import dummyfactory as f
-
- result = self._callFUT(
- [
- (None, f),
- (1, f, (1,), {}, (), 'first'),
- (1, f, (2,), {}, ('x',), 'second'),
- (1, f, (3,), {}, ('y',), 'third'),
- (4, f, (4,), {}, ('y',), 'should be last', 99999),
- (3, f, (3,), {}, ('y',)),
- (None, f, (5,), {}, ('y',)),
- ]
- )
- result = list(result)
- self.assertEqual(
- result,
- [
- {
- 'info': None,
- 'args': (),
- 'callable': f,
- 'introspectables': (),
- 'kw': {},
- 'discriminator': None,
- 'includepath': (),
- 'order': 0,
- },
- {
- 'info': 'first',
- 'args': (1,),
- 'callable': f,
- 'introspectables': (),
- 'kw': {},
- 'discriminator': 1,
- 'includepath': (),
- 'order': 0,
- },
- {
- 'info': None,
- 'args': (3,),
- 'callable': f,
- 'introspectables': (),
- 'kw': {},
- 'discriminator': 3,
- 'includepath': ('y',),
- 'order': 0,
- },
- {
- 'info': None,
- 'args': (5,),
- 'callable': f,
- 'introspectables': (),
- 'kw': {},
- 'discriminator': None,
- 'includepath': ('y',),
- 'order': 0,
- },
- {
- 'info': 'should be last',
- 'args': (4,),
- 'callable': f,
- 'introspectables': (),
- 'kw': {},
- 'discriminator': 4,
- 'includepath': ('y',),
- 'order': 99999,
- },
- ],
- )
-
- def test_it_success_dicts(self):
- from . import dummyfactory as f
-
- result = self._callFUT(
- [
- (None, f),
- (1, f, (1,), {}, (), 'first'),
- (1, f, (2,), {}, ('x',), 'second'),
- (1, f, (3,), {}, ('y',), 'third'),
- (4, f, (4,), {}, ('y',), 'should be last', 99999),
- (3, f, (3,), {}, ('y',)),
- (None, f, (5,), {}, ('y',)),
- ]
- )
- result = list(result)
- self.assertEqual(
- result,
- [
- {
- 'info': None,
- 'args': (),
- 'callable': f,
- 'introspectables': (),
- 'kw': {},
- 'discriminator': None,
- 'includepath': (),
- 'order': 0,
- },
- {
- 'info': 'first',
- 'args': (1,),
- 'callable': f,
- 'introspectables': (),
- 'kw': {},
- 'discriminator': 1,
- 'includepath': (),
- 'order': 0,
- },
- {
- 'info': None,
- 'args': (3,),
- 'callable': f,
- 'introspectables': (),
- 'kw': {},
- 'discriminator': 3,
- 'includepath': ('y',),
- 'order': 0,
- },
- {
- 'info': None,
- 'args': (5,),
- 'callable': f,
- 'introspectables': (),
- 'kw': {},
- 'discriminator': None,
- 'includepath': ('y',),
- 'order': 0,
- },
- {
- 'info': 'should be last',
- 'args': (4,),
- 'callable': f,
- 'introspectables': (),
- 'kw': {},
- 'discriminator': 4,
- 'includepath': ('y',),
- 'order': 99999,
- },
- ],
- )
-
- def test_it_conflict(self):
- from . import dummyfactory as f
-
- result = self._callFUT(
- [
- (None, f),
- (1, f, (2,), {}, ('x',), 'eek'), # will conflict
- (1, f, (3,), {}, ('y',), 'ack'), # will conflict
- (4, f, (4,), {}, ('y',)),
- (3, f, (3,), {}, ('y',)),
- (None, f, (5,), {}, ('y',)),
- ]
- )
- self.assertRaises(ConfigurationConflictError, list, result)
-
- def test_it_with_actions_grouped_by_order(self):
- from . import dummyfactory as f
-
- result = self._callFUT(
- [
- (None, f), # X
- (1, f, (1,), {}, (), 'third', 10), # X
- (1, f, (2,), {}, ('x',), 'fourth', 10),
- (1, f, (3,), {}, ('y',), 'fifth', 10),
- (2, f, (1,), {}, (), 'sixth', 10), # X
- (3, f, (1,), {}, (), 'seventh', 10), # X
- (5, f, (4,), {}, ('y',), 'eighth', 99999), # X
- (4, f, (3,), {}, (), 'first', 5), # X
- (4, f, (5,), {}, ('y',), 'second', 5),
- ]
- )
- result = list(result)
- self.assertEqual(len(result), 6)
- # resolved actions should be grouped by (order, i)
- self.assertEqual(
- result,
- [
- {
- 'info': None,
- 'args': (),
- 'callable': f,
- 'introspectables': (),
- 'kw': {},
- 'discriminator': None,
- 'includepath': (),
- 'order': 0,
- },
- {
- 'info': 'first',
- 'args': (3,),
- 'callable': f,
- 'introspectables': (),
- 'kw': {},
- 'discriminator': 4,
- 'includepath': (),
- 'order': 5,
- },
- {
- 'info': 'third',
- 'args': (1,),
- 'callable': f,
- 'introspectables': (),
- 'kw': {},
- 'discriminator': 1,
- 'includepath': (),
- 'order': 10,
- },
- {
- 'info': 'sixth',
- 'args': (1,),
- 'callable': f,
- 'introspectables': (),
- 'kw': {},
- 'discriminator': 2,
- 'includepath': (),
- 'order': 10,
- },
- {
- 'info': 'seventh',
- 'args': (1,),
- 'callable': f,
- 'introspectables': (),
- 'kw': {},
- 'discriminator': 3,
- 'includepath': (),
- 'order': 10,
- },
- {
- 'info': 'eighth',
- 'args': (4,),
- 'callable': f,
- 'introspectables': (),
- 'kw': {},
- 'discriminator': 5,
- 'includepath': ('y',),
- 'order': 99999,
- },
- ],
- )
-
- def test_override_success_across_orders(self):
- from . import dummyfactory as f
-
- result = self._callFUT(
- [
- (1, f, (2,), {}, ('x',), 'eek', 0),
- (1, f, (3,), {}, ('x', 'y'), 'ack', 10),
- ]
- )
- result = list(result)
- self.assertEqual(
- result,
- [
- {
- 'info': 'eek',
- 'args': (2,),
- 'callable': f,
- 'introspectables': (),
- 'kw': {},
- 'discriminator': 1,
- 'includepath': ('x',),
- 'order': 0,
- }
- ],
- )
-
- def test_conflicts_across_orders(self):
- from . import dummyfactory as f
-
- result = self._callFUT(
- [
- (1, f, (2,), {}, ('x', 'y'), 'eek', 0),
- (1, f, (3,), {}, ('x'), 'ack', 10),
- ]
- )
- self.assertRaises(ConfigurationConflictError, list, result)
-
-
class TestGlobalRegistriesIntegration(unittest.TestCase):
def setUp(self):
from pyramid.config import global_registries
@@ -2430,11 +1432,6 @@
self.popped = True
-@implementer(IDummy)
-class DummyEvent:
- pass
-
-
class DummyRegistry(object):
def __init__(self, adaptation=None, util=None):
self.utilities = []
@@ -2457,36 +1454,6 @@
def queryUtility(self, *arg, **kw):
return self.util
-
-
-class IOther(Interface):
- pass
-
-
-def _conflictFunctions(e):
- conflicts = e._conflicts.values()
- for conflict in conflicts:
- for confinst in conflict:
- yield confinst.function
-
-
-class DummyActionState(object):
- autocommit = False
- info = ''
-
- def __init__(self):
- self.actions = []
-
- def action(self, *arg, **kw):
- self.actions.append((arg, kw))
-
-
-class DummyIntrospectable(object):
- def __init__(self):
- self.registered = []
-
- def register(self, introspector, action_info):
- self.registered.append((introspector, action_info))
class DummyPredicate(object):
diff --git a/tests/test_config/test_testing.py b/tests/test_config/test_testing.py
index ce66596..ede31e1 100644
--- a/tests/test_config/test_testing.py
+++ b/tests/test_config/test_testing.py
@@ -104,7 +104,7 @@
def test_testing_add_subscriber_dottedname(self):
config = self._makeOne(autocommit=True)
- L = config.testing_add_subscriber('tests.test_config.test_init.IDummy')
+ L = config.testing_add_subscriber('tests.test_config.IDummy')
event = DummyEvent()
config.registry.notify(event)
self.assertEqual(len(L), 1)
--
Gitblit v1.9.3