Commits

Andy Mikhailenko committed 3947fbb

Added base classes for most unit tests; no idea why it stayed uncommitted for such a long time (since around 2011-02).

  • Participants
  • Parent commits f402271

Comments (0)

Files changed (1)

File tests/base_storage.py

+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+Base for backend-specific storage tests.
+
+Not intended to be run itself.
+"""
+from datetime import date
+import os
+import unittest2 as unittest
+
+from doqu import get_db
+
+
+__test__ = False  # tell nose to skip this module
+
+
+class BaseStorageTestCase(unittest.TestCase):
+    FIXTURE = {
+        'john-id': {u'name': u'John', u'last_name': u'Connor', u'age': 30,
+                    u'birth_date': date(1985,2,28)},
+        'mary-id': {u'name': u'Mary', u'last_name': u'Connor', u'age': 25}
+    }
+    TMP_FILENAME_EXTENSION = 'tmp'  # matters for Tokyo Cabinet
+
+    #-- These should be overloaded:
+
+    def get_connection(self):
+        raise NotImplementedError('Backend test case must implement '
+                                  'the method get_connection().')
+
+    #-- All below till EOF should not be touched or overloaded
+
+    @property
+    def _tmp_filename(self):
+        return os.path.abspath('doqu_tests_{modname}.{clsname}.{ext}'.format(
+            modname = __name__,
+            clsname = self.__class__.__name__,
+            ext = self.TMP_FILENAME_EXTENSION
+        ))
+
+    def setUp(self):
+        self.db = self.get_connection()
+        self.db.clear()
+        for pk, data in self.FIXTURE.items():
+            self.db.save(pk, data)
+
+    def tearDown(self):
+        if self.db:
+            self.db.clear()
+            self.db.disconnect()
+        if os.path.exists(self._tmp_filename):
+            os.unlink(self._tmp_filename)
+
+    def _unified(self, data):
+        """Returns given dictionary with values in a unified form so that
+        backend-specific data conversion stuff won't affect comparison.
+        """
+        #return dict((k, unicode(v)) for k,v in d.items())
+        return dict((k, unicode(self.db.value_to_db(v)))
+                    for k,v in data.items())
+
+    def test_get(self):
+        # it's up to the backend how exactly the values are stored; some
+        # backends (such as TC) coerce everything to strings. Metadata can be
+        # used to restore the type but that's a higher level of abstraction; on
+        # this level the backend may return strings instead of integers.
+        #unified = lambda d: dict((k, unicode(v)) for k,v in d.items())
+        key, data = self.db.get('mary-id')
+        data = self._unified(data)
+        expected = self._unified(self.FIXTURE['mary-id'])
+        self.assertEquals(data, expected)
+
+        with self.assertRaises(KeyError):
+            self.db.get('void')
+
+    def test_get_many(self):
+        pairs = self.db.get_many(['mary-id', 'john-id'])
+        obtained = [(pk, self._unified(v)) for pk, v in sorted(pairs)]
+        expected = [(pk, self._unified(v)) for pk, v in sorted(self.FIXTURE.items())]
+        self.assertEquals(obtained, expected)
+
+        with self.assertRaises(KeyError):
+            self.db.get_many(['mary-id', 'foo', 'bar'])
+
+    def test_save(self):
+        data = {'foo': 'bar'}
+        key = self.db.save(None, data)
+        saved_key, saved_data = self.db.get(key)
+        self.assertEquals(key, saved_key)  # this is redundant but present
+        self.assertEquals(saved_data, data)
+
+    def test_save_new_key(self):
+        data = {'foo': 'bar'}
+        key = self.db.save('quux', data)
+        saved_key, saved_data = self.db.get(key)
+        self.assertEquals(key, saved_key)
+        self.assertEquals(saved_data, data)
+
+    def test_len(self):
+        self.assertEquals(2, len(self.db))
+
+    def test_iter(self):
+        self.assertEquals(sorted(self.FIXTURE), sorted(list(self.db)))
+
+    def test_nonzero(self):
+        self.assertEquals(bool(self.db), True)
+        self.db.disconnect()
+        self.assertEquals(bool(self.db), False)
+        self.db.connect()
+        self.assertEquals(bool(self.db), True)
+
+    def test_clear(self):
+        self.assertEquals(len(self.db), 2)
+        self.db.clear()
+        self.assertEquals(len(self.db), 0)
+
+    def test_connect(self):
+        self.db.disconnect()
+        self.db.connect()
+        self.assertEquals(2, len(self.db))
+
+    def test_delete(self):
+        self.assertEquals(2, len(self.db))
+        self.db.delete('john-id')
+        self.assertEquals(1, len(self.db))
+        self.assertEquals(['mary-id'], list(self.db))
+
+    def test_disconnect(self):
+        self.db.disconnect()
+        self.assertEquals(bool(self.db), False)
+
+    def test_reconnect(self):
+        self.db.reconnect()
+        self.assertEquals(2, len(self.db))
+
+if __name__ == '__main__':
+    unittest.main()