Permalink
Browse files

Back to 100% unit test coverage of pickle.py

  • Loading branch information...
1 parent 10ea1f9 commit db3bef9e4bb4f7ca84591a272844acd04d097439 Andrew McFague committed Aug 25, 2011
Showing with 145 additions and 8 deletions.
  1. +4 −4 linesman/backends/pickle.py
  2. +141 −4 linesman/tests/backends/test_backend_pickle.py
View
8 linesman/backends/pickle.py
@@ -40,7 +40,7 @@ class PickleBackend(Backend):
def __init__(self, filename="sessions.dat"):
self.filename = filename
- def __flush(self):
+ def _flush(self):
"""
Writes the session history to disk, in pickled form.
"""
@@ -72,15 +72,15 @@ def add(self, session):
Adds a session to this dictionary and flushes it to disk.
"""
self._session_history[session.uuid] = session
- self.__flush()
+ self._flush()
def delete(self, session_uuid):
"""
Remove a session from the dictionary and flush it to disk.
"""
if self.get(session_uuid):
del self._session_history[session_uuid]
- self.__flush()
+ self._flush()
return 1
return 0
@@ -91,7 +91,7 @@ def delete_all(self):
"""
deleted_rows = len(self._session_history)
self._session_history.clear()
- self.__flush()
+ self._flush()
return deleted_rows
View
145 linesman/tests/backends/test_backend_pickle.py
@@ -1,16 +1,153 @@
import os
+from cPickle import HIGHEST_PROTOCOL
+from nose.tools import raises
+from mock import MagicMock, Mock, patch
+from tempfile import TemporaryFile
-from linesman.backends.pickle import PickleBackend
+import linesman.backends.pickle
from linesman.tests import get_temporary_filename
from linesman.tests.backends import TestBackend
+MOCK_SESSION_UUID = "abcd1234"
class TestBackendPickle(TestBackend):
def setUp(self):
self.filename = get_temporary_filename()
- self.backend = PickleBackend(self.filename)
- self.backend.setup()
+ self.backend = linesman.backends.pickle.PickleBackend(self.filename)
def tearDown(self):
- os.remove(self.db_filename)
+ os.remove(self.filename)
+
+ @patch("cPickle.dump")
+ def test_flush(self, mock_dump):
+ """ Test that the file is opened by cPickle. """
+ test_fd = TemporaryFile()
+ with patch("__builtin__.open", Mock(side_effect=IOError())):
+ self.backend.setup()
+
+ with patch("__builtin__.open") as mock_open:
+ mock_open.return_value = test_fd
+ self.backend._flush()
+ mock_open.assert_called_once_with(
+ self.backend.filename, "w+b")
+
+ mock_dump.assert_called_once_with(
+ self.backend._session_history,
+ test_fd,
+ HIGHEST_PROTOCOL)
+
+ @patch("__builtin__.open")
+ @patch("cPickle.load")
+ def test_setup(self, mock_load, mock_open):
+ """ Test that setup will load pickled data. """
+ mock_open.return_value = MagicMock(spec=file)
+ self.backend.setup()
+ mock_open.assert_called_once_with(self.filename, "rb")
+ mock_load.assert_called_once()
+
+ @patch("__builtin__.open")
+ @patch("linesman.backends.pickle.OrderedDict")
+ def test_setup_ioerror(self, mock_ordered_dict, mock_open):
+ """ Test that setup will create a new ordered dict if no file exists """
+ mock_open.side_effect = IOError()
+ self.backend.setup()
+ mock_ordered_dict.assert_called_once_with()
+
+ @raises(ValueError)
+ @patch("__builtin__.open")
+ def test_setup_value_error(self, mock_open):
+ """ Test that bad pickled data raises a ValueError. """
+ mock_open.side_effect = ValueError("Could not unpickle!")
+ self.backend.setup()
+
+ @patch("__builtin__.open", Mock(side_effect=IOError()))
+ @patch("linesman.backends.pickle.PickleBackend._flush")
+ def test_add(self, mock_flush):
+ """ Test that add creates a new entry in the session. """
+ mock_session = MagicMock()
+ mock_session.uuid = MOCK_SESSION_UUID
+
+ self.backend.setup()
+ self.backend.add(mock_session)
+
+ self.assertTrue(mock_session.uuid in self.backend._session_history)
+ mock_flush.assert_called_once()
+
+ @patch("__builtin__.open", Mock(side_effect=IOError()))
+ @patch("linesman.backends.pickle.PickleBackend._flush")
+ def test_delete(self, mock_flush):
+ """ Test that deleting an existing UUID returns 0. """
+ mock_session = MagicMock()
+ mock_session.uuid = MOCK_SESSION_UUID
+
+ self.backend.setup()
+ self.backend.add(mock_session)
+ self.assertEquals(self.backend.delete(mock_session.uuid), 1)
+ mock_flush.assert_called_once()
+
+ @patch("__builtin__.open", Mock(side_effect=IOError()))
+ @patch("linesman.backends.pickle.PickleBackend._flush")
+ def test_delete_non_existent_uuid(self, mock_flush):
+ """ Test that deleting a non-existing UUID returns 0. """
+ mock_session = MagicMock()
+ mock_session.uuid = MOCK_SESSION_UUID
+
+ self.backend.setup()
+ self.backend.add(mock_session)
+ self.assertEquals(self.backend.delete("basb3144"), 0)
+ mock_flush.assert_called_once()
+
+ @patch("__builtin__.open", Mock(side_effect=IOError()))
+ @patch("linesman.backends.pickle.PickleBackend._flush")
+ def test_delete_all(self, mock_flush):
+ """ Test that deleting a non-existing UUID returns 0. """
+ mock_session1 = MagicMock()
+ mock_session1.uuid = MOCK_SESSION_UUID
+
+ mock_session2 = MagicMock()
+ mock_session2.uuid = "something else"
+
+ self.backend.setup()
+ self.backend.add(mock_session1)
+ self.backend.add(mock_session2)
+ self.assertEquals(self.backend.delete_all(), 2)
+ self.assertEquals(len(self.backend._session_history), 0)
+ mock_flush.assert_called_once()
+
+ @patch("__builtin__.open", Mock(side_effect=IOError()))
+ @patch("linesman.backends.pickle.PickleBackend._flush")
+ def test_delete_all_empty(self, mock_flush):
+ """ Test that callign delete all on an empty dict returns 0. """
+ self.backend.setup()
+ self.assertEquals(self.backend.delete_all(), 0)
+ mock_flush.assert_called_once()
+
+ @patch("__builtin__.open", Mock(side_effect=IOError()))
+ @patch("linesman.backends.pickle.PickleBackend._flush", Mock())
+ def test_get(self):
+ """ Test that retrieving an existing UUID succeeds. """
+ mock_session = MagicMock()
+ mock_session.uuid = MOCK_SESSION_UUID
+
+ self.backend.setup()
+ self.backend.add(mock_session)
+ self.assertEquals(self.backend.get(MOCK_SESSION_UUID), mock_session)
+
+ @patch("__builtin__.open", Mock(side_effect=IOError()))
+ def test_get_non_existent_uuid(self):
+ """ Test that retrieving an non-existent UUID returns None. """
+ self.backend.setup()
+ self.assertEquals(self.backend.get(MOCK_SESSION_UUID), None)
+
+ @patch("__builtin__.open", Mock(side_effect=IOError()))
+ @patch("linesman.backends.pickle.PickleBackend._flush", Mock())
+ def test_get_all(self):
+ """ Test that getting all results returns a copy. """
+ mock_session = MagicMock()
+ mock_session.uuid = MOCK_SESSION_UUID
+
+ self.backend.setup()
+ self.backend.add(mock_session)
+ session_history_copy = self.backend.get_all()
+ assert self.backend._session_history is not session_history_copy

0 comments on commit db3bef9

Please sign in to comment.