Rooba
/
kombu
mirror of https://github.com/celery/kombu.git
1
1
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fixes tests for #607

This commit is contained in:
Ask Solem 2016-07-08 14:42:15 -07:00
parent 278a7e4b0c
commit 1dadd0b957
1 changed files with 5 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
kombu/tests/transport/test_redis.py
View File

@ -82,6 +82,9 @@ class Client(object):
def smembers(self, key): def smembers(self, key):
return self.sets.get(key, set()) return self.sets.get(key, set())
def ping(self, *args, **kwargs):
return True
def srem(self, key, *args): def srem(self, key, *args):
self.sets.pop(key, None) self.sets.pop(key, None)
zrem = srem zrem = srem
@ -273,7 +276,7 @@ class test_Channel(Case):
self.assertFalse(chan.ack_emulation) self.assertFalse(chan.ack_emulation)
self.assertEqual(chan.QoS, virtual.QoS) self.assertEqual(chan.QoS, virtual.QoS)
def test_redis_info_raises(self): def test_redis_ping_raises(self):
pool = Mock(name='pool') pool = Mock(name='pool')
pool_at_init = [pool] pool_at_init = [pool]
client = Mock(name='client') client = Mock(name='client')
@ -291,7 +294,7 @@ class test_Channel(Case):
Channel = XChannel Channel = XChannel
conn = Connection(transport=XTransport) conn = Connection(transport=XTransport)
client.info.side_effect = RuntimeError() client.ping.side_effect = RuntimeError()
with self.assertRaises(RuntimeError): with self.assertRaises(RuntimeError):
conn.channel() conn.channel()
pool.disconnect.assert_called_with() pool.disconnect.assert_called_with()