- Instances of ``pyramid.testing.DummyRequest`` now have a ``session``

  object, which is mostly a dictionary, but also implements the other session
  API methods for flash and CSRF.

3 files changed, 119 insertions, 0 deletions

diff --git a/CHANGES.txt b/CHANGES.txt
index 520a73847..e66675e64 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -66,6 +66,10 @@ Features
   should be the name of a previously registered renderer.  Useful to provide
   "omnipresent" RPC using existing rendered views.
 
+- Instances of ``pyramid.testing.DummyRequest`` now have a ``session``
+  object, which is mostly a dictionary, but also implements the other session
+  API methods for flash and CSRF.
+
 Backwards Incompatibilities
 ---------------------------
 
diff --git a/pyramid/testing.py b/pyramid/testing.py
index 25aba33ce..96388b709 100644
--- a/pyramid/testing.py
+++ b/pyramid/testing.py
@@ -11,6 +11,7 @@ from pyramid.interfaces import IRequest
 from pyramid.interfaces import ISecuredView
 from pyramid.interfaces import IView
 from pyramid.interfaces import IViewClassifier
+from pyramid.interfaces import ISession
 
 from pyramid.config import Configurator
 from pyramid.exceptions import Forbidden
@@ -511,6 +512,37 @@ class DummyResource:
 
 DummyModel = DummyResource # b/w compat (forever)
 
+class DummySession(dict):
+    implements(ISession)
+    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 = 'csrft'
+        self['_csrft_'] = token
+        return token
+
+    def get_csrf_token(self):
+        return self.get('_csrft_', None)
+        
 class DummyRequest(object):
     """ A dummy request object (imitates a :term:`request` object).
 
@@ -572,6 +604,7 @@ class DummyRequest(object):
         self.virtual_root = None
         self.marshalled = params # repoze.monty
         self.registry = get_current_registry()
+        self.session = DummySession()
         self.__dict__.update(kw)
 
     def add_response_callback(self, callback):
diff --git a/pyramid/tests/test_testing.py b/pyramid/tests/test_testing.py
index d2ed957f2..eaaad6aef 100644
--- a/pyramid/tests/test_testing.py
+++ b/pyramid/tests/test_testing.py
@@ -413,6 +413,7 @@ class TestDummyRequest(unittest.TestCase):
 
     def test_defaults(self):
         from pyramid.threadlocal import get_current_registry
+        from pyramid.testing import DummySession
         request = self._makeOne()
         self.assertEqual(request.method, 'GET')
         self.assertEqual(request.application_url, 'http://example.com')
@@ -438,6 +439,7 @@ class TestDummyRequest(unittest.TestCase):
         self.assertEqual(request.virtual_root, None)
         self.assertEqual(request.virtual_root_path, ())
         self.assertEqual(request.registry, get_current_registry())
+        self.assertEqual(request.session.__class__, DummySession)
 
     def test_params_explicit(self):
         request = self._makeOne(params = {'foo':'bar'})
@@ -723,6 +725,86 @@ class Test_skip_on(unittest.TestCase):
         decorated = self._callFUT('ok')(foo)
         self.assertEqual(decorated(), True)
 
+class TestDummySession(unittest.TestCase):
+    def _makeOne(self):
+        from pyramid.testing import DummySession
+        return DummySession()
+
+    def test_instance_conforms(self):
+        from zope.interface.verify import verifyObject
+        from pyramid.interfaces import ISession
+        session = self._makeOne()
+        verifyObject(ISession, session)
+
+    def test_changed(self):
+        session = self._makeOne()
+        self.assertEqual(session.changed(), None)
+
+    def test_invalidate(self):
+        session = self._makeOne()
+        session['a'] = 1
+        self.assertEqual(session.invalidate(), None)
+        self.failIf('a' in session)
+
+    def test_flash_default(self):
+        session = self._makeOne()
+        session.flash('msg1')
+        session.flash('msg2')
+        self.assertEqual(session['_f_'], ['msg1', 'msg2'])
+        
+    def test_flash_mixed(self):
+        session = self._makeOne()
+        session.flash('warn1', 'warn')
+        session.flash('warn2', 'warn')
+        session.flash('err1', 'error')
+        session.flash('err2', 'error')
+        self.assertEqual(session['_f_warn'], ['warn1', 'warn2'])
+
+    def test_pop_flash_default_queue(self):
+        session = self._makeOne()
+        queue = ['one', 'two']
+        session['_f_'] = queue
+        result = session.pop_flash()
+        self.assertEqual(result, queue)
+        self.assertEqual(session.get('_f_'), None)
+
+    def test_pop_flash_nodefault_queue(self):
+        session = self._makeOne()
+        queue = ['one', 'two']
+        session['_f_error'] = queue
+        result = session.pop_flash('error')
+        self.assertEqual(result, queue)
+        self.assertEqual(session.get('_f_error'), None)
+
+    def test_peek_flash_default_queue(self):
+        session = self._makeOne()
+        queue = ['one', 'two']
+        session['_f_'] = queue
+        result = session.peek_flash()
+        self.assertEqual(result, queue)
+        self.assertEqual(session.get('_f_'), queue)
+
+    def test_peek_flash_nodefault_queue(self):
+        session = self._makeOne()
+        queue = ['one', 'two']
+        session['_f_error'] = queue
+        result = session.peek_flash('error')
+        self.assertEqual(result, queue)
+        self.assertEqual(session.get('_f_error'), queue)
+
+    def test_new_csrf_token(self):
+        session = self._makeOne()
+        token = session.new_csrf_token()
+        self.assertEqual(token, session['_csrft_'])
+
+    def test_get_csrf_token(self):
+        session = self._makeOne()
+        session['_csrft_'] = 'token'
+        token = session.get_csrf_token()
+        self.assertEqual(token, 'token')
+        self.failUnless('_csrft_' in session)
+
+
 from zope.interface import Interface
 from zope.interface import implements