doqu / 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


__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):
            list(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()
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.