Source

redis-dict / tests.py

Full commit
import unittest

import redis
from redis_dict import RedisDict


class RedisMapTestCase(unittest.TestCase):
    """In order to run this test case you have to start redis server locally"""

    def setUp(self):
        self.connection = redis.Redis()
        self.connection.delete('testmap')
        self.redis_dict = RedisDict('testmap', connection=self.connection)
        self.redis_dict.clear()

    def tearDown(self):
        self.connection.delete('testmap')

    def test_getitem(self):
        self.redis_dict['key'] = 1
        self.assertEqual(self.redis_dict['key'], 1)

    def test_clear(self):
        self.redis_dict['a'] = 1
        self.redis_dict['b'] = 2
        self.redis_dict['c'] = 3
        self.redis_dict.clear()
        self.assertEqual(len(self.redis_dict), 0)

    def test_getitem_raises_exception_for_non_existing_key(self):
        self.assertRaises(KeyError, lambda: self.redis_dict['non existing key'])

    def test_delete_key(self):
        self.redis_dict['key'] = 1
        del self.redis_dict['key']
        self.assertFalse('key' in self.redis_dict)

    def test_values(self):
        self.redis_dict['key'] = 1
        self.assertEqual(self.redis_dict.values(), [1])

    def test_get_method_returns_default_value_for_non_existing_key(self):
        self.assertEqual(self.redis_dict.get('non existing key'), None)
        self.assertEqual(self.redis_dict.get('non existing key', 2), 2)

    def test_execute_update_raises_exception_when_dict_changed(self):
        self.redis_dict['one'] = 1
        self.redis_dict['two'] = 2
        intersect = lambda c: self.redis_dict.__setitem__('two', 'two')
        self.assertRaises(redis.WatchError, lambda: self.redis_dict.execute_update('one', intersect))

    def test_execute_update_retries(self):
        self.redis_dict['one'] = 1
        self.redis_dict['two'] = 2
        class call_counter: count = 0
        def intersect(c):
            self.redis_dict.__setitem__('two', 'two')
            call_counter.count += 1
        self.assertRaises(redis.WatchError, lambda: self.redis_dict.execute_update('one', intersect, retry=3, interval=0))
        self.assertEqual(call_counter.count, 4)

    def test_execute_update(self):
        self.redis_dict['one'] = 1
        self.redis_dict['two'] = 2
        self.redis_dict.execute_update('one', lambda v: v+100)
        self.assertEqual(self.redis_dict['one'], 101)

    def test_items_method(self):
        self.redis_dict['one'] = 1
        self.redis_dict['two'] = 2
        self.assertEqual(set([('one', 1), ('two', 2)]), set(self.redis_dict.items()))