pyres/tests/test_horde.py at master · ggtracker/pyres · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
from tests import PyResTests, Basic, TestProcess
from pyres import horde
import os

class KhanTests(PyResTests):
    def test_khan_init(self):
        from pyres.exceptions import NoQueueError
        self.assertRaises(NoQueueError, horde.Khan, 2, [])
        self.assertRaises(ValueError, horde.Khan, 'test', ['test'])

    def test_register_khan(self):
        khan = horde.Khan(pool_size=1, queues=['basic'])
        khan.register_khan()
        name = "%s:%s:1" % (os.uname()[1],os.getpid())
        assert self.redis.sismember('resque:khans',name)

    def test_unregister_khan(self):
        khan = horde.Khan(pool_size=1, queues=['basic'])
        khan.register_khan()
        name = "%s:%s:1" % (os.uname()[1],os.getpid())
        assert self.redis.sismember('resque:khans',name)
        assert self.redis.scard('resque:khans') == 1
        khan.unregister_khan()
        assert not self.redis.sismember('resque:khans', name)
        assert self.redis.scard('resque:khans') == 0

    def test_setup_minions(self):
        khan = horde.Khan(pool_size=1, queues=['basic'])
        khan.setup_minions()
        assert len(khan._workers) == 1
        khan._shutdown_minions()

    def test_setup_resq(self):
        khan = horde.Khan(pool_size=1, queues=['basic'])
        assert not hasattr(khan, 'resq')
        khan.setup_resq()
        assert hasattr(khan, 'resq')

    def test_add_minion(self):
        khan = horde.Khan(pool_size=1, queues=['basic'])
        khan.setup_minions()
        khan.register_khan()
        name = "%s:%s:1" % (os.uname()[1],os.getpid())
        assert self.redis.sismember('resque:khans',name)
        khan.add_minion()
        assert len(khan._workers) == 2
        assert not self.redis.sismember('resque:khans',name)
        name = '%s:%s:2' % (os.uname()[1], os.getpid())
        assert khan.pool_size == 2
        assert self.redis.sismember('resque:khans',name)
        khan._shutdown_minions()

    def test_remove_minion(self):
        khan = horde.Khan(pool_size=1, queues=['basic'])
        khan.setup_minions()
        khan.register_khan()
        assert khan.pool_size == 1
        khan._remove_minion()
        assert khan.pool_size == 0