updated with some tests

Author: Matt George

pyres/horde.py

@@ -154,7 +154,6 @@ def __init__(self, pool_size=5, queues=[], server='localhost:6379', password=Non
         self._workers = OrderedDict()
         self.server = server
         self.password = password
-        
         #self._workers = list()
 
     def setup_resq(self):
@@ -196,15 +195,12 @@ def add_child(self, signum, frame):
     def register_khan(self):
         if not hasattr(self, 'resq'):
             self.setup_resq()
-        
         self.resq.redis.sadd('resque:khans',str(self))
         self.started = datetime.datetime.now()
 
     def _check_commands(self):
         if not self._shutdown:
             logging.debug('Checking commands')
-            command_key = 'resque:khan:%s' % self
-            
             command = self.resq.redis.lpop('resque:khan:%s' % str(self))
             logging.debug('COMMAND FOUND: %s ' % command)
             if command:
@@ -223,15 +219,17 @@ def process_command(self, command):
 
     def add_minion(self):
         m = self._add_minion()
-        m.start()
+        self.resq.redis.srem('resque:khans',str(self))
+        self.pool_size += 1
+        self.resq.redis.sadd('resque:khans',str(self))
 
     def _add_minion(self):
         logging.info('Adding minion')
-        #parent_conn, child_conn = multiprocessing.Pipe()
         m = Minion(self.queues, self.server, self.password)
-        #m.start()
+        m.start()
+        self._workers[m.pid] = m
+        logging.info('minion added at: %s' % m.pid)
         return m
-        #self._workers.append(m)
 
     def _shutdown_minions(self):
         """
@@ -246,20 +244,23 @@ def _remove_minion(self, pid=None):
         #    m = self._workers.pop(pid)
         pid, m = self._workers.popitem(False)
         m.terminate()
+        self.resq.redis.srem('resque:khans',str(self))
+        self.pool_size -= 1
+        self.resq.redis.sadd('resque:khans',str(self))
         return m
 
     def unregister_khan(self):
         logging.debug('unregistering khan')
         self.resq.redis.srem('resque:khans',str(self))
         self.started = None
 
+    def setup_minions(self):
+        for i in range(self.pool_size):
+            self._add_minion()
+
     def work(self, interval=2):
         self.startup()
-        for i in range(self.pool_size):
-            m = self._add_minion()
-            m.start()
-            self._workers[m.pid] = m
-            logging.info('minion added at %s' % m.pid)
+        self.setup_minions()
         self.setup_resq()
         self.register_khan()
         while True:
@@ -275,7 +276,7 @@ def work(self, interval=2):
 
     def __str__(self):
         hostname = os.uname()[1]
-        return '%s:%s' % (hostname, self.pid)
+        return '%s:%s:%s' % (hostname, self.pid, self.pool_size)
 
     @classmethod
     def run(cls, pool_size=5, queues=[], server='localhost:6379'):

tests/test_horde.py

@@ -11,5 +11,49 @@ def test_khan_init(self):
     def test_register_khan(self):
         khan = horde.Khan(pool_size=1, queues=['basic'])
         khan.register_khan()
-        name = "%s:%s" % (os.uname()[1],os.getpid())
-        assert self.redis.sismember('resque:khans',name)
+        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