shep

Multi-state key stores using bit masks for python3
git clone git://git.defalsify.org/shep.git
Info | Log | Files | Refs | LICENSE

test_redis.py (3171B)

      1 # standard imports
      2 import unittest
      3 import os
      4 import logging
      5 import sys
      6 import importlib
      7 
      8 # local imports
      9 from shep.persist import PersistedState
     10 from shep.error import (
     11         StateExists,
     12         StateInvalid,
     13         StateItemExists,
     14         StateItemNotFound,
     15         )
     16 
     17 logging.basicConfig(level=logging.DEBUG)
     18 logg = logging.getLogger()
     19 
     20 
     21 class TestRedisStore(unittest.TestCase):
     22         
     23     def setUp(self):
     24         from shep.store.redis import RedisStoreFactory
     25         self.factory = RedisStoreFactory()
     26         self.factory.redis.flushall()
     27         self.states = PersistedState(self.factory.add, 3)
     28         self.states.add('foo') 
     29         self.states.add('bar') 
     30         self.states.add('baz') 
     31 
     32 
     33     def test_add(self):
     34         self.states.put('abcd', state=self.states.FOO, contents='baz')
     35         v = self.states.get('abcd')
     36         self.assertEqual(v, 'baz')
     37         v = self.states.state('abcd')
     38         self.assertEqual(v, self.states.FOO)
     39 
     40 
     41     def test_next(self):
     42         self.states.put('abcd')
     43 
     44         self.states.next('abcd')
     45         self.assertEqual(self.states.state('abcd'), self.states.FOO)
     46         
     47         self.states.next('abcd')
     48         self.assertEqual(self.states.state('abcd'), self.states.BAR)
     49 
     50         self.states.next('abcd')
     51         self.assertEqual(self.states.state('abcd'), self.states.BAZ)
     52 
     53         with self.assertRaises(StateInvalid):
     54             self.states.next('abcd')
     55 
     56         v = self.states.state('abcd')
     57         self.assertEqual(v, self.states.BAZ)
     58 
     59 
     60     def test_replace(self):
     61         with self.assertRaises(StateItemNotFound):
     62             self.states.replace('abcd', contents='foo')
     63 
     64         self.states.put('abcd', state=self.states.FOO, contents='baz')
     65         self.states.replace('abcd', contents='bar')
     66         v = self.states.get('abcd')
     67         self.assertEqual(v, 'bar')
     68 
     69 
     70     def test_factory_ls(self):
     71         r = self.factory.ls()
     72         self.assertEqual(len(r), 0)
     73 
     74         self.states.put('abcd')
     75         self.states.put('xxxx', state=self.states.BAZ)
     76         r = self.factory.ls()
     77         self.assertEqual(len(r), 2)
     78 
     79         self.states.put('yyyy', state=self.states.BAZ)
     80         r = self.factory.ls()
     81         self.assertEqual(len(r), 2)
     82 
     83         self.states.put('zzzz', state=self.states.BAR)
     84         r = self.factory.ls()
     85         self.assertEqual(len(r), 3)
     86 
     87 
     88 if __name__ == '__main__':
     89     noredis = False
     90     redis = None
     91     try:
     92         redis = importlib.import_module('redis')
     93     except ModuleNotFoundError:
     94         logg.critical('redis module not available, skipping tests.')
     95         sys.exit(0)
     96 
     97     host = os.environ.get('REDIS_HOST', 'localhost')
     98     port = os.environ.get('REDIS_PORT', 6379)
     99     port = int(port)
    100     db = os.environ.get('REDIS_DB', 0)
    101     db = int(db)
    102     r = redis.Redis(host=host, port=port, db=db)
    103     try:
    104         r.get('foo')
    105     except redis.exceptions.ConnectionError:
    106         logg.critical('could not connect to redis, skipping tests.')
    107         sys.exit(0)
    108     except redis.exceptions.InvalidResponse as e:
    109         logg.critical('is that really redis running on {}:{}? Got unexpected response: {}'.format(host, port, e))
    110         sys.exit(0)
    111 
    112     unittest.main()