Andriy Kornatskyy avatar Andriy Kornatskyy committed b57bd18

Introduced NullTPCSession.

Comments (0)

Files changed (2)

src/wheezy/core/db.py

 
     @property
     def connection(self):
-        """ Return the session connection.
+        """ Return the session connection. Not intended to be used
+            directly, use `cursor` method instead.
         """
         if self.__connection:
             return self.__connection
         self.status = SESSION_STATUS_ENTERED
         return self
 
+    @property
+    def connection(self):
+        raise AssertionError('Not intended to be used directly. '
+            'Use cursor() method instead.')
+
+    def cursor(self, *args, **kwargs):
+        """ Ensure session is entered.
+        """
+        assert self.status == SESSION_STATUS_ENTERED
+
     def commit(self):
         """ Simulates commit. Asserts the session is used in scope.
         """
     def __exit__(self, exc_type, exc_value, traceback):
         assert self.status == SESSION_STATUS_ENTERED
         self.status = SESSION_STATUS_IDLE
+
+
+class NullTPCSession(object):
+    """ Null TPC session is supposed to be used in mock scenarios.
+    """
+
+    def __init__(self):
+        self.status = SESSION_STATUS_IDLE
+
+    def __enter__(self):
+        assert self.status == SESSION_STATUS_IDLE
+        self.status = SESSION_STATUS_ENTERED
+        return self
+
+    def enlist(self, session):
+        """ Ensure session is entered.
+        """
+        assert session
+        assert self.status == SESSION_STATUS_ENTERED
+        self.status = SESSION_STATUS_ACTIVE
+
+    def commit(self):
+        """ Simulates commit. Asserts the session is used in scope.
+        """
+        assert self.status != SESSION_STATUS_IDLE
+        self.status = SESSION_STATUS_ENTERED
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        assert self.status != SESSION_STATUS_IDLE
+        self.status = SESSION_STATUS_IDLE

src/wheezy/core/tests/test_db.py

 
 class NullSessionTestCase(unittest.TestCase):
 
+    def setUp(self):
+        from wheezy.core.db import NullSession
+        self.session = NullSession()
+
     def test_enter(self):
         """ Enter returns session instance.
         """
-        from wheezy.core.db import NullSession
-        session = NullSession()
-        assert session == session.__enter__()
+        assert self.session == self.session.__enter__()
+
+    def test_connection_raise_error(self):
+        """ Not intended to be used directly.
+        """
+        self.assertRaises(AssertionError, lambda: self.session.connection)
+
+    def test_cursor_raise_error(self):
+        """ If session is not entered raise error.
+        """
+        self.assertRaises(AssertionError, lambda: self.session.cursor())
+
+    def test_cursor(self):
+        """ Noop if session is entered.
+        """
+        self.session.__enter__()
+        self.session.cursor()
+
+    def test_commit_raise_error(self):
+        """ If session is not entered raise error.
+        """
+        self.assertRaises(AssertionError, lambda: self.session.commit())
 
     def test_commit(self):
+        """ Noop if session is entered.
+        """
+        self.session.__enter__()
+        self.session.commit()
+
+    def test_exit_raise_error(self):
         """ If session is not entered raise error.
         """
-        from wheezy.core.db import NullSession
-        session = NullSession()
-        session.__enter__()
-        session.commit()
-
-        session = NullSession()
-        self.assertRaises(AssertionError, lambda: session.commit())
+        self.assertRaises(AssertionError,
+                          lambda: self.session.__exit__(None, None, None))
 
     def test_exit(self):
+        """ Noop if session is entered.
+        """
+        self.session.__enter__()
+        self.session.__exit__(None, None, None)
+
+
+class NullTPCSessionTestCase(unittest.TestCase):
+
+    def setUp(self):
+        from wheezy.core.db import NullTPCSession
+        self.session = NullTPCSession()
+
+    def test_enter(self):
+        """ Enter returns session instance.
+        """
+        assert self.session == self.session.__enter__()
+
+    def test_enlist_raise_error(self):
         """ If session is not entered raise error.
         """
-        from wheezy.core.db import NullSession
-        session = NullSession()
-        session.__enter__()
-        session.__exit__(None, None, None)
+        self.assertRaises(AssertionError, lambda: self.session.enlist('x'))
 
-        session = NullSession()
+    def test_enlist(self):
+        """ Noop if session is entered.
+        """
+        self.session.__enter__()
+        self.session.enlist('x')
+
+    def test_commit_raise_error(self):
+        """ If session is not entered raise error.
+        """
+        self.assertRaises(AssertionError, lambda: self.session.commit())
+
+    def test_commit(self):
+        """ Noop if session is entered.
+        """
+        self.session.__enter__()
+        self.session.commit()
+
+    def test_exit_raise_error(self):
+        """ If session is not entered raise error.
+        """
         self.assertRaises(AssertionError,
-                          lambda: session.__exit__(None, None, None))
+                          lambda: self.session.__exit__(None, None, None))
+    def test_exit(self):
+        """ Noop if session is entered.
+        """
+        self.session.__enter__()
+        self.session.__exit__(None, None, None)
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.