/usr/lib/python2.7/dist-packages/pyramid/testing.py is in python-pyramid 1.6+dfsg-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 | import copy
import os
from contextlib import contextmanager
from zope.interface import (
implementer,
alsoProvides,
)
from pyramid.interfaces import (
IRequest,
ISession,
)
from pyramid.compat import (
PY3,
PYPY,
class_types,
)
from pyramid.config import Configurator
from pyramid.decorator import reify
from pyramid.path import caller_package
from pyramid.response import _get_response_factory
from pyramid.registry import Registry
from pyramid.security import (
Authenticated,
Everyone,
AuthenticationAPIMixin,
AuthorizationAPIMixin,
)
from pyramid.threadlocal import (
get_current_registry,
manager,
)
from pyramid.i18n import LocalizerRequestMixin
from pyramid.request import CallbackMethodsMixin
from pyramid.url import URLMethodsMixin
from pyramid.util import InstancePropertyMixin
_marker = object()
class DummyRootFactory(object):
__parent__ = None
__name__ = None
def __init__(self, request):
if 'bfg.routes.matchdict' in request:
self.__dict__.update(request['bfg.routes.matchdict'])
class DummySecurityPolicy(object):
""" A standin for both an IAuthentication and IAuthorization policy """
def __init__(self, userid=None, groupids=(), permissive=True,
remember_result=None, forget_result=None):
self.userid = userid
self.groupids = groupids
self.permissive = permissive
if remember_result is None:
remember_result = []
if forget_result is None:
forget_result = []
self.remember_result = remember_result
self.forget_result = forget_result
def authenticated_userid(self, request):
return self.userid
def unauthenticated_userid(self, request):
return self.userid
def effective_principals(self, request):
effective_principals = [Everyone]
if self.userid:
effective_principals.append(Authenticated)
effective_principals.append(self.userid)
effective_principals.extend(self.groupids)
return effective_principals
def remember(self, request, userid, **kw):
self.remembered = userid
return self.remember_result
def forget(self, request):
self.forgotten = True
return self.forget_result
def permits(self, context, principals, permission):
return self.permissive
def principals_allowed_by_permission(self, context, permission):
return self.effective_principals(None)
class DummyTemplateRenderer(object):
"""
An instance of this class is returned from
:meth:`pyramid.config.Configurator.testing_add_renderer`. It has a
helper function (``assert_``) that makes it possible to make an
assertion which compares data passed to the renderer by the view
function against expected key/value pairs.
"""
def __init__(self, string_response=''):
self._received = {}
self._string_response = string_response
self._implementation = MockTemplate(string_response)
# For in-the-wild test code that doesn't create its own renderer,
# but mutates our internals instead. When all you read is the
# source code, *everything* is an API!
def _get_string_response(self):
return self._string_response
def _set_string_response(self, response):
self._string_response = response
self._implementation.response = response
string_response = property(_get_string_response, _set_string_response)
def implementation(self):
return self._implementation
def __call__(self, kw, system=None):
if system:
self._received.update(system)
self._received.update(kw)
return self.string_response
def __getattr__(self, k):
""" Backwards compatibility """
val = self._received.get(k, _marker)
if val is _marker:
val = self._implementation._received.get(k, _marker)
if val is _marker:
raise AttributeError(k)
return val
def assert_(self, **kw):
""" Accept an arbitrary set of assertion key/value pairs. For
each assertion key/value pair assert that the renderer
(eg. :func:`pyramid.renderers.render_to_response`)
received the key with a value that equals the asserted
value. If the renderer did not receive the key at all, or the
value received by the renderer doesn't match the assertion
value, raise an :exc:`AssertionError`."""
for k, v in kw.items():
myval = self._received.get(k, _marker)
if myval is _marker:
myval = self._implementation._received.get(k, _marker)
if myval is _marker:
raise AssertionError(
'A value for key "%s" was not passed to the renderer'
% k)
if myval != v:
raise AssertionError(
'\nasserted value for %s: %r\nactual value: %r' % (
k, v, myval))
return True
class DummyResource:
""" A dummy :app:`Pyramid` :term:`resource` object."""
def __init__(self, __name__=None, __parent__=None, __provides__=None,
**kw):
""" The resource's ``__name__`` attribute will be set to the
value of the ``__name__`` argument, and the resource's
``__parent__`` attribute will be set to the value of the
``__parent__`` argument. If ``__provides__`` is specified, it
should be an interface object or tuple of interface objects
that will be attached to the resulting resource via
:func:`zope.interface.alsoProvides`. Any extra keywords passed
in the ``kw`` argumnent will be set as direct attributes of
the resource object.
.. note:: For backwards compatibility purposes, this class can also
be imported as :class:`pyramid.testing.DummyModel`.
"""
self.__name__ = __name__
self.__parent__ = __parent__
if __provides__ is not None:
alsoProvides(self, __provides__)
self.kw = kw
self.__dict__.update(**kw)
self.subs = {}
def __setitem__(self, name, val):
""" When the ``__setitem__`` method is called, the object
passed in as ``val`` will be decorated with a ``__parent__``
attribute pointing at the dummy resource and a ``__name__``
attribute that is the value of ``name``. The value will then
be returned when dummy resource's ``__getitem__`` is called with
the name ``name```."""
val.__name__ = name
val.__parent__ = self
self.subs[name] = val
def __getitem__(self, name):
""" Return a named subobject (see ``__setitem__``)"""
ob = self.subs[name]
return ob
def __delitem__(self, name):
del self.subs[name]
def get(self, name, default=None):
return self.subs.get(name, default)
def values(self):
""" Return the values set by __setitem__ """
return self.subs.values()
def items(self):
""" Return the items set by __setitem__ """
return self.subs.items()
def keys(self):
""" Return the keys set by __setitem__ """
return self.subs.keys()
__iter__ = keys
def __nonzero__(self):
return True
__bool__ = __nonzero__
def __len__(self):
return len(self.subs)
def __contains__(self, name):
return name in self.subs
def clone(self, __name__=_marker, __parent__=_marker, **kw):
""" Create a clone of the resource object. If ``__name__`` or
``__parent__`` arguments are passed, use these values to
override the existing ``__name__`` or ``__parent__`` of the
resource. If any extra keyword args are passed in via the ``kw``
argument, use these keywords to add to or override existing
resource keywords (attributes)."""
oldkw = self.kw.copy()
oldkw.update(kw)
inst = self.__class__(self.__name__, self.__parent__, **oldkw)
inst.subs = copy.deepcopy(self.subs)
if __name__ is not _marker:
inst.__name__ = __name__
if __parent__ is not _marker:
inst.__parent__ = __parent__
return inst
DummyModel = DummyResource # b/w compat (forever)
@implementer(ISession)
class DummySession(dict):
created = None
new = True
def changed(self):
pass
def invalidate(self):
self.clear()
def flash(self, msg, queue='', allow_duplicate=True):
storage = self.setdefault('_f_' + queue, [])
if allow_duplicate or (msg not in storage):
storage.append(msg)
def pop_flash(self, queue=''):
storage = self.pop('_f_' + queue, [])
return storage
def peek_flash(self, queue=''):
storage = self.get('_f_' + queue, [])
return storage
def new_csrf_token(self):
token = '0123456789012345678901234567890123456789'
self['_csrft_'] = token
return token
def get_csrf_token(self):
token = self.get('_csrft_', None)
if token is None:
token = self.new_csrf_token()
return token
@implementer(IRequest)
class DummyRequest(
URLMethodsMixin,
CallbackMethodsMixin,
InstancePropertyMixin,
LocalizerRequestMixin,
AuthenticationAPIMixin,
AuthorizationAPIMixin,
):
""" A DummyRequest object (incompletely) imitates a :term:`request` object.
The ``params``, ``environ``, ``headers``, ``path``, and
``cookies`` arguments correspond to their :term:`WebOb`
equivalents.
The ``post`` argument, if passed, populates the request's
``POST`` attribute, but *not* ``params``, in order to allow testing
that the app accepts data for a given view only from POST requests.
This argument also sets ``self.method`` to "POST".
Extra keyword arguments are assigned as attributes of the request
itself.
Note that DummyRequest does not have complete fidelity with a "real"
request. For example, by default, the DummyRequest ``GET`` and ``POST``
attributes are of type ``dict``, unlike a normal Request's GET and POST,
which are of type ``MultiDict``. If your code uses the features of
MultiDict, you should either use a real :class:`pyramid.request.Request`
or adapt your DummyRequest by replacing the attributes with ``MultiDict``
instances.
Other similar incompatibilities exist. If you need all the features of
a Request, use the :class:`pyramid.request.Request` class itself rather
than this class while writing tests.
"""
method = 'GET'
application_url = 'http://example.com'
host = 'example.com:80'
domain = 'example.com'
content_length = 0
query_string = ''
charset = 'UTF-8'
script_name = ''
_registry = None
request_iface = IRequest
def __init__(self, params=None, environ=None, headers=None, path='/',
cookies=None, post=None, **kw):
if environ is None:
environ = {}
if params is None:
params = {}
if headers is None:
headers = {}
if cookies is None:
cookies = {}
self.environ = environ
self.headers = headers
self.params = params
self.cookies = cookies
self.matchdict = {}
self.GET = params
if post is not None:
self.method = 'POST'
self.POST = post
else:
self.POST = params
self.host_url = self.application_url
self.path_url = self.application_url
self.url = self.application_url
self.path = path
self.path_info = path
self.script_name = ''
self.path_qs = ''
self.body = ''
self.view_name = ''
self.subpath = ()
self.traversed = ()
self.virtual_root_path = ()
self.context = None
self.root = None
self.virtual_root = None
self.marshalled = params # repoze.monty
self.session = DummySession()
self.__dict__.update(kw)
def _get_registry(self):
if self._registry is None:
return get_current_registry()
return self._registry
def _set_registry(self, registry):
self._registry = registry
def _del_registry(self):
self._registry = None
registry = property(_get_registry, _set_registry, _del_registry)
@reify
def response(self):
f = _get_response_factory(self.registry)
return f(self)
have_zca = True
def setUp(registry=None, request=None, hook_zca=True, autocommit=True,
settings=None, package=None):
"""
Set :app:`Pyramid` registry and request thread locals for the
duration of a single unit test.
Use this function in the ``setUp`` method of a unittest test case
which directly or indirectly uses:
- any method of the :class:`pyramid.config.Configurator`
object returned by this function.
- the :func:`pyramid.threadlocal.get_current_registry` or
:func:`pyramid.threadlocal.get_current_request` functions.
If you use the ``get_current_*`` functions (or call :app:`Pyramid` code
that uses these functions) without calling ``setUp``,
:func:`pyramid.threadlocal.get_current_registry` will return a *global*
:term:`application registry`, which may cause unit tests to not be
isolated with respect to registrations they perform.
If the ``registry`` argument is ``None``, a new empty
:term:`application registry` will be created (an instance of the
:class:`pyramid.registry.Registry` class). If the ``registry``
argument is not ``None``, the value passed in should be an
instance of the :class:`pyramid.registry.Registry` class or a
suitable testing analogue.
After ``setUp`` is finished, the registry returned by the
:func:`pyramid.threadlocal.get_current_registry` function will
be the passed (or constructed) registry until
:func:`pyramid.testing.tearDown` is called (or
:func:`pyramid.testing.setUp` is called again) .
If the ``hook_zca`` argument is ``True``, ``setUp`` will attempt
to perform the operation ``zope.component.getSiteManager.sethook(
pyramid.threadlocal.get_current_registry)``, which will cause
the :term:`Zope Component Architecture` global API
(e.g. :func:`zope.component.getSiteManager`,
:func:`zope.component.getAdapter`, and so on) to use the registry
constructed by ``setUp`` as the value it returns from
:func:`zope.component.getSiteManager`. If the
:mod:`zope.component` package cannot be imported, or if
``hook_zca`` is ``False``, the hook will not be set.
If ``settings`` is not ``None``, it must be a dictionary representing the
values passed to a Configurator as its ``settings=`` argument.
If ``package`` is ``None`` it will be set to the caller's package. The
``package`` setting in the :class:`pyramid.config.Configurator` will
affect any relative imports made via
:meth:`pyramid.config.Configurator.include` or
:meth:`pyramid.config.Configurator.maybe_dotted`.
This function returns an instance of the
:class:`pyramid.config.Configurator` class, which can be
used for further configuration to set up an environment suitable
for a unit or integration test. The ``registry`` attribute
attached to the Configurator instance represents the 'current'
:term:`application registry`; the same registry will be returned
by :func:`pyramid.threadlocal.get_current_registry` during the
execution of the test.
"""
manager.clear()
if registry is None:
registry = Registry('testing')
if package is None:
package = caller_package()
config = Configurator(registry=registry, autocommit=autocommit,
package=package)
if settings is None:
settings = {}
if getattr(registry, 'settings', None) is None:
config._set_settings(settings)
if hasattr(registry, 'registerUtility'):
# Sometimes nose calls us with a non-registry object because
# it thinks this function is module test setup. Likewise,
# someone may be passing us an esoteric "dummy" registry, and
# the below won't succeed if it doesn't have a registerUtility
# method.
config.add_default_renderers()
config.add_default_view_predicates()
config.add_default_route_predicates()
config.commit()
global have_zca
try:
have_zca and hook_zca and config.hook_zca()
except ImportError: # pragma: no cover
# (dont choke on not being able to import z.component)
have_zca = False
config.begin(request=request)
return config
def tearDown(unhook_zca=True):
"""Undo the effects of :func:`pyramid.testing.setUp`. Use this
function in the ``tearDown`` method of a unit test that uses
:func:`pyramid.testing.setUp` in its ``setUp`` method.
If the ``unhook_zca`` argument is ``True`` (the default), call
:func:`zope.component.getSiteManager.reset`. This undoes the
action of :func:`pyramid.testing.setUp` when called with the
argument ``hook_zca=True``. If :mod:`zope.component` cannot be
imported, ``unhook_zca`` is set to ``False``.
"""
global have_zca
if unhook_zca and have_zca:
try:
from zope.component import getSiteManager
getSiteManager.reset()
except ImportError: # pragma: no cover
have_zca = False
info = manager.pop()
manager.clear()
if info is not None:
registry = info['registry']
if hasattr(registry, '__init__') and hasattr(registry, '__name__'):
try:
registry.__init__(registry.__name__)
except TypeError:
# calling __init__ is largely for the benefit of
# people who want to use the global ZCA registry;
# however maybe somebody's using a registry we don't
# understand, let's not blow up
pass
def cleanUp(*arg, **kw):
""" An alias for :func:`pyramid.testing.setUp`. """
package = kw.get('package', None)
if package is None:
package = caller_package()
kw['package'] = package
return setUp(*arg, **kw)
class DummyRendererFactory(object):
""" Registered by
:meth:`pyramid.config.Configurator.testing_add_renderer` as
a dummy renderer factory. The indecision about what to use as a
key (a spec vs. a relative name) is caused by test suites in the
wild believing they can register either. The ``factory`` argument
passed to this constructor is usually the *real* template renderer
factory, found when ``testing_add_renderer`` is called."""
def __init__(self, name, factory):
self.name = name
self.factory = factory # the "real" renderer factory reg'd previously
self.renderers = {}
def add(self, spec, renderer):
self.renderers[spec] = renderer
if ':' in spec:
package, relative = spec.split(':', 1)
self.renderers[relative] = renderer
def __call__(self, info):
spec = info.name
renderer = self.renderers.get(spec)
if renderer is None:
if ':' in spec:
package, relative = spec.split(':', 1)
renderer = self.renderers.get(relative)
if renderer is None:
if self.factory:
renderer = self.factory(info)
else:
raise KeyError('No testing renderer registered for %r' %
spec)
return renderer
class MockTemplate(object):
def __init__(self, response):
self._received = {}
self.response = response
def __getattr__(self, attrname):
return self
def __getitem__(self, attrname):
return self
def __call__(self, *arg, **kw):
self._received.update(kw)
return self.response
def skip_on(*platforms): # pragma: no cover
skip = False
for platform in platforms:
if skip_on.os_name.startswith(platform):
skip = True
if platform == 'pypy' and PYPY:
skip = True
if platform == 'py3' and PY3:
skip = True
def decorator(func):
if isinstance(func, class_types):
if skip: return None
else: return func
else:
def wrapper(*args, **kw):
if skip:
return
return func(*args, **kw)
wrapper.__name__ = func.__name__
wrapper.__doc__ = func.__doc__
return wrapper
return decorator
skip_on.os_name = os.name # for testing
@contextmanager
def testConfig(registry=None,
request=None,
hook_zca=True,
autocommit=True,
settings=None):
"""Returns a context manager for test set up.
This context manager calls :func:`pyramid.testing.setUp` when
entering and :func:`pyramid.testing.tearDown` when exiting.
All arguments are passed directly to :func:`pyramid.testing.setUp`.
If the ZCA is hooked, it will always be un-hooked in tearDown.
This context manager allows you to write test code like this:
.. code-block:: python
:linenos:
with testConfig() as config:
config.add_route('bar', '/bar/{id}')
req = DummyRequest()
resp = myview(req),
"""
config = setUp(registry=registry,
request=request,
hook_zca=hook_zca,
autocommit=autocommit,
settings=settings)
try:
yield config
finally:
tearDown(unhook_zca=hook_zca)
|