Got nearly 100% test coverage for AnkiServer/collection.py

.gitignore

@@ -1,5 +1,6 @@
 *~
 *.pyc
+.coverage
 AnkiServer.egg-info
 development.ini
 server.log

AnkiServer/collection.py

@@ -22,6 +22,10 @@ class CollectionWrapper(object):
         self.setup_new_collection = setup_new_collection
         self.__col = None
 
+    def __del__(self):
+        """Close the collection if the user forgot to do so."""
+        self.close()
+
     def execute(self, func, args=[], kw={}, waitForReturn=True):
         """ Executes the given function with the underlying anki.storage.Collection
         object as the first argument and any additional arguments specified by *args
@@ -55,11 +59,13 @@ class CollectionWrapper(object):
             else:
                 raise
 
-        self.__col = ank.storage.Collection(self.path)
+        col = anki.storage.Collection(self.path)
 
         # Do any special setup
         if self.setup_new_collection is not None:
-            self.setup_new_collection(self.__col)
+            self.setup_new_collection(col)
+
+        return col
 
     def open(self):
         """Open the collection, or create it if it doesn't exist."""

tests/test_collection.py

@@ -0,0 +1,140 @@
+
+import os
+import shutil
+import tempfile
+import unittest
+
+import mock
+from mock import MagicMock, sentinel
+
+import AnkiServer
+from AnkiServer.collection import CollectionWrapper, CollectionManager
+
+class CollectionWrapperTest(unittest.TestCase):
+    def setUp(self):
+        self.temp_dir = tempfile.mkdtemp()
+        self.collection_path = os.path.join(self.temp_dir, 'collection.anki2');
+
+    def tearDown(self):
+        shutil.rmtree(self.temp_dir)
+
+    def test_lifecycle_real(self):
+        """Testing common life-cycle with existing and non-existant collections. This
+        test uses the real Anki objects and actually creates a new collection on disk."""
+
+        wrapper = CollectionWrapper(self.collection_path)
+        self.assertFalse(os.path.exists(self.collection_path))
+        self.assertFalse(wrapper.opened())
+
+        wrapper.open()
+        self.assertTrue(os.path.exists(self.collection_path))
+        self.assertTrue(wrapper.opened())
+
+        # calling open twice shouldn't break anything
+        wrapper.open()
+
+        wrapper.close()
+        self.assertTrue(os.path.exists(self.collection_path))
+        self.assertFalse(wrapper.opened())
+
+        # open the same collection again (not a creation)
+        wrapper = CollectionWrapper(self.collection_path)
+        self.assertFalse(wrapper.opened())
+        wrapper.open()
+        self.assertTrue(wrapper.opened())
+        wrapper.close()
+        self.assertFalse(wrapper.opened())
+        self.assertTrue(os.path.exists(self.collection_path))
+
+    def test_del(self):
+        with mock.patch('anki.storage.Collection') as anki_storage_Collection:
+            col = anki_storage_Collection.return_value
+            wrapper = CollectionWrapper(self.collection_path)
+            wrapper.open()
+            wrapper = None
+            col.close.assert_called_with()
+
+    def test_setup_func(self):
+        # Run it when the collection doesn't exist
+        with mock.patch('anki.storage.Collection') as anki_storage_Collection:
+            col = anki_storage_Collection.return_value
+            setup_new_collection = MagicMock()
+            self.assertFalse(os.path.exists(self.collection_path))
+            wrapper = CollectionWrapper(self.collection_path, setup_new_collection)
+            wrapper.open()
+            anki_storage_Collection.assert_called_with(self.collection_path)
+            setup_new_collection.assert_called_with(col)
+            wrapper = None
+
+        # Make sure that no collection was actually created
+        self.assertFalse(os.path.exists(self.collection_path))
+        
+        # Create a faux collection file
+        with file(self.collection_path, 'wt') as fd:
+            fd.write('Collection!')
+
+        # Run it when the collection does exist
+        with mock.patch('anki.storage.Collection'):
+            setup_new_collection = lambda col: self.fail("Setup function called when collection already exists!")
+            self.assertTrue(os.path.exists(self.collection_path))
+            wrapper = CollectionWrapper(self.collection_path, setup_new_collection)
+            wrapper.open()
+            anki_storage_Collection.assert_called_with(self.collection_path)
+            wrapper = None
+
+    def test_execute(self):
+        with mock.patch('anki.storage.Collection') as anki_storage_Collection:
+            col = anki_storage_Collection.return_value
+            func = MagicMock()
+            func.return_value = sentinel.some_object
+
+            # check that execute works and auto-creates the collection
+            wrapper = CollectionWrapper(self.collection_path)
+            ret = wrapper.execute(func, [1, 2, 3], {'key': 'aoeu'})
+            self.assertEqual(ret, sentinel.some_object)
+            anki_storage_Collection.assert_called_with(self.collection_path)
+            func.assert_called_with(col, 1, 2, 3, key='aoeu')
+
+            # check that execute always returns False if waitForReturn=False
+            func.reset_mock()
+            ret = wrapper.execute(func, [1, 2, 3], {'key': 'aoeu'}, waitForReturn=False)
+            self.assertEqual(ret, None)
+            func.assert_called_with(col, 1, 2, 3, key='aoeu')
+
+class CollectionManagerTest(unittest.TestCase):
+    def test_lifecycle(self):
+        with mock.patch('AnkiServer.collection.CollectionManager.collection_wrapper') as CollectionWrapper:
+            wrapper = MagicMock()
+            CollectionWrapper.return_value = wrapper
+
+            manager = CollectionManager()
+
+            # check getting a new collection
+            ret = manager.get_collection('path1')
+            CollectionWrapper.assert_called_with(os.path.realpath('path1'), None)
+            self.assertEqual(ret, wrapper)
+
+            # change the return value, so that it would return a new object
+            new_wrapper = MagicMock()
+            CollectionWrapper.return_value = new_wrapper
+            CollectionWrapper.reset_mock()
+
+            # get the new wrapper
+            ret = manager.get_collection('path2')
+            CollectionWrapper.assert_called_with(os.path.realpath('path2'), None)
+            self.assertEqual(ret, new_wrapper)
+
+            # make sure the wrapper and new_wrapper are different
+            self.assertNotEqual(wrapper, new_wrapper)
+
+            # assert that calling with the first path again, returns the first wrapper
+            ret = manager.get_collection('path1')
+            self.assertEqual(ret, wrapper)
+
+            manager.shutdown()
+            wrapper.close.assert_called_with()
+            new_wrapper.close.assert_called_with()
+
+if __name__ == '__main__':
+    unittest.main()
+