Finish implementation of pubsub

mockredis/client.py

@@ -17,6 +17,7 @@
 from mockredis.pipeline import MockRedisPipeline
 from mockredis.script import Script
 from mockredis.sortedset import SortedSet
+from mockredis.pubsub import Pubsub
 
 if sys.version_info >= (3, 0):
     long = int
@@ -55,10 +56,9 @@ def __init__(self,
         self.redis = defaultdict(dict)
         self.redis_config = defaultdict(dict)
         self.timeouts = defaultdict(dict)
-        # The 'PubSub' store
-        self.pubsub = defaultdict(list)
         # Dictionary from script to sha ''Script''
         self.shas = dict()
+        self._pubsub = None
 
     @classmethod
     def from_url(cls, url, db=None, **kwargs):
@@ -270,7 +270,7 @@ def do_expire(self):
 
     def flushdb(self):
         self.redis.clear()
-        self.pubsub.clear()
+        self.pubsub().clear()
         self.timeouts.clear()
 
     def rename(self, old_key, new_key):
@@ -1427,8 +1427,14 @@ def config_get(self, pattern='*'):
 
     # PubSub commands #
 
+    def pubsub(self, **kwargs):
+        """ Return a mocked 'PubSub' object """
+        if not self._pubsub:
+            self._pubsub = Pubsub(self, **kwargs)
+        return self._pubsub
+
     def publish(self, channel, message):
-        self.pubsub[channel].append(message)
+        self.pubsub().publish(channel, message)
 
     # Internal #
 

mockredis/pubsub.py

@@ -0,0 +1,86 @@
+
+class Pubsub(dict):
+
+    def __init__(self, connection_pool, shard_hint=None,
+                 ignore_subscribe_messages=False):
+
+        # there is no connection pool, but we want a reference to the parent
+        self.redis = connection_pool
+        self.shard_hint = shard_hint
+        self.ignore_subscribe_messages=ignore_subscribe_messages
+
+        self.channels = []
+        self.patterns = []
+
+    def publish(self, channel, message):
+        """ emulate publish """
+        if not channel in self:
+            self.channels.append(channel)
+            self[channel] = []
+
+        self[channel].append(message)
+
+    def reset(self):
+        """ emulate reset """
+        self.clear()
+        self.channels = []
+
+
+    def close(self):
+        """ emulate close """
+        self.reset()
+
+    def encode(self, value):
+        """ emulate encode by calling the parent's """
+        return self.redis._encode(value)
+
+    def on_connect(self, connection):
+        """ do nothing while mocking """
+        pass
+
+    @property
+    def subscribed(self):
+        """ emulate subscribed """
+        return bool(self.channels or self.patterns)
+
+
+    def execute_command(self, *args, **kwargs):
+        """ do nothing while mocking """
+        return
+
+    def parse_response(self, block=True, timeout=0):
+        """ do nothing while mocking """
+        return
+
+    def psubscribe(self, *args, **kwargs):
+        """ call no callbacks while mocking """
+        return
+
+    def punsubscribe(self, *args, **kwargs):
+        """ call no callbacks while mocking """
+        return
+
+    def subscribe(self, *args, **kwargs):
+        """ call no callbacks while mocking """
+        return
+
+    def unsubscribe(self, *args, **kwargs):
+        """ call no callbacks while mocking """
+        return
+
+    def listen(self):
+        """ do nothing while mocking """
+        return
+
+    def get_message(self, ignore_subscribe_messages=False, timeout=0):
+        """ do nothing while mocking """
+        return
+
+    def handle_message(self, response, ignore_subscribe_messages=False):
+        """ do nothing while mocking """
+        return
+
+    def run_in_thread(self, sleep_time=0):
+        """ do nothing while mocking """
+        return
+

mockredis/tests/test_pubsub.py

@@ -16,4 +16,4 @@ def test_publish(self):
         channel = 'ch#1'
         msg = 'test message'
         self.redis.publish(channel, msg)
-        eq_(self.redis.pubsub[channel], [msg])
+        eq_(self.redis.pubsub()[channel], [msg])