Merge branch 'redis_upgrade'

Author: Matt George

pyres/__init__.py

@@ -3,8 +3,6 @@
 from redis import Redis
 import pyres.json_parser as json
 
-import types
-
 def my_import(name):
     mod = __import__(name)    
     components = name.split('.')    
@@ -60,18 +58,18 @@ class ResQ(object):
     def __init__(self, server="localhost:6379", password=None, 
                  timeout=None, retry_connection=True):
         self.timeout = timeout
-        self.retry_connection = retry_connection
+        #self.retry_connection = retry_connection
         self.redis = server
         if password:
             self.redis.auth(password)
         self._watched_queues = set()
 
     def push(self, queue, item):
         self.watch_queue(queue)
-        self.redis.push("resque:queue:%s" % queue, ResQ.encode(item))
+        self.redis.rpush("resque:queue:%s" % queue, ResQ.encode(item))
 
     def pop(self, queue):
-        ret = self.redis.pop("resque:queue:%s" % queue)
+        ret = self.redis.lpop("resque:queue:%s" % queue)
         if ret:
             return ResQ.decode(ret)
         return ret
@@ -90,7 +88,7 @@ def peek(self, queue, start=0, count=1):
         return self.list_range('resque:queue:%s' % queue, start, count)
 
     def list_range(self, key, start, count):
-        items = self.redis.lrange(key, start,start+count-1)
+        items = self.redis.lrange(key, start,start+count-1) or []
         ret_list = []
         for i in items:
             ret_list.append(ResQ.decode(i))
@@ -103,9 +101,7 @@ def _set_redis(self, server):
         if isinstance(server, basestring):
             self.dsn = server
             host, port = server.split(':')
-            self._redis = Redis(host=host, port=int(port), 
-                                retry_connection=self.retry_connection,
-                                timeout=self.timeout)
+            self._redis = Redis(host=host, port=int(port))
         elif isinstance(server, Redis):
             self.dsn = '%s:%s' % (server.host,server.port)
             self._redis = server
@@ -129,7 +125,7 @@ def enqueue_from_string(self, klass_as_string, queue, *args):
         self.push(queue, {'class':klass_as_string,'args':args})
 
     def queues(self):
-        return self.redis.smembers("resque:queues")
+        return self.redis.smembers("resque:queues") or []
 
     def info(self):
         """

pyres/failure/redis.py

@@ -15,7 +15,7 @@ def save(self, resq=None):
         if self._worker:
             data['worker'] = self._worker
         data = ResQ.encode(data)
-        resq.redis.push('resque:failed', data)
+        resq.redis.rpush('resque:failed', data)
 
     @classmethod
     def count(cls, resq):

pyres/worker.py

@@ -212,7 +212,8 @@ def all(cls, host="localhost:6379"):
             resq = ResQ(host)
         elif isinstance(host, ResQ):
             resq = host
-        return [Worker.find(w,resq) for w in resq.redis.smembers('resque:workers')]
+        
+        return [Worker.find(w,resq) for w in resq.redis.smembers('resque:workers') or []]
 
     @classmethod
     def working(cls, host):

requirements.txt

@@ -1,4 +1,4 @@
 simplejson==2.0.9
 itty==0.6.2
-redis==0.6.0
+redis==1.34.1
 pystache==0.1.0

resweb/views.py

@@ -313,12 +313,14 @@ def key_type(self):
     def items(self):
         items = []
         if self.key_type() == 'list':
-            for k in self.resq.redis.lrange('resque:'+self.stat_id,0,20):
+            lst = self.resq.redis.lrange('resque:'+self.stat_id,0,20) or []
+            for k in lst:
                 items.append({
                     'row':str(k)
                 })
         elif self.key_type() == 'set':
-            for k in self.resq.redis.smembers('resque:'+self.stat_id):
+            st = self.resq.redis.smembers('resque:'+self.stat_id) or set([])
+            for k in st:
                 items.append({
                     'row':str(k)
                 })

tests/__init__.py

@@ -43,17 +43,26 @@ def perform(sleep_time):
         import time
         time.sleep(sleep_time)
         print 'Done Sleeping'
+
 def test_str_to_class():
     ret = str_to_class('tests.Basic')
     assert ret
+    assert ret == Basic
+    assert str_to_class('hello.World') == None
+
+def test_safe_str_to_class():
+    from pyres import safe_str_to_class
+    assert safe_str_to_class('tests.Basic') == Basic
+    assert safe_str_to_class('test.Mine') == None
+    assert safe_str_to_class('hello.World') == None
 
 class PyResTests(unittest.TestCase):
     def setUp(self):
         self.resq = ResQ()
         self.redis = self.resq.redis
-        self.redis.flush(True)
+        self.redis.flushall()
 
     def tearDown(self):
-        self.redis.flush(True)
+        self.redis.flushall()
         del self.redis
         del self.resq

tests/test_resq.py

@@ -42,7 +42,7 @@ def test_redis_property(self):
         from redis import Redis
         rq = ResQ(server="localhost:6379")
         red = Redis()
-        rq2 = ResQ(server=red)
+        #rq2 = ResQ(server=red)
         self.assertRaises(Exception, rq.redis,[Basic])
 
     def test_info(self):
@@ -89,3 +89,14 @@ def test_keys(self):
         self.resq.enqueue_from_string('tests.Basic','basic','test1')
         assert 'queue:basic' in self.resq.keys()
         assert 'queues' in self.resq.keys()
+    
+    def test_queues(self):
+        assert self.resq.queues() == []
+        self.resq.enqueue_from_string('tests.Basic','basic','test1')
+        assert len(self.resq.queues()) == 1
+        self.resq.enqueue_from_string('tests.Basic','basic','test1')
+        assert len(self.resq.queues()) == 1
+        self.resq.enqueue_from_string('tests.Basic','basic2','test1')
+        assert len(self.resq.queues()) == 2
+        assert 'test' not in self.resq.queues()
+        assert 'basic' in self.resq.queues()

tests/test_stats.py

@@ -4,22 +4,22 @@ class StatTests(PyResTests):
     def test_incr(self):
         stat_obj = Stat('test_stat', self.resq)
         stat_obj.incr()
-        assert self.redis.get('resque:stat:test_stat') == 1
+        assert self.redis.get('resque:stat:test_stat') == str(1)
         stat_obj.incr()
-        assert self.redis.get('resque:stat:test_stat') == 2
+        assert self.redis.get('resque:stat:test_stat') == str(2)
         stat_obj.incr(2)
-        assert self.redis.get('resque:stat:test_stat') == 4
+        assert self.redis.get('resque:stat:test_stat') == str(4)
 
     def test_decr(self):
         stat_obj = Stat('test_stat', self.resq)
         stat_obj.incr()
         stat_obj.incr()
-        assert self.redis.get('resque:stat:test_stat') == 2
+        assert self.redis.get('resque:stat:test_stat') == str(2)
         stat_obj.decr()
-        assert self.redis.get('resque:stat:test_stat') == 1
+        assert self.redis.get('resque:stat:test_stat') == str(1)
         stat_obj.incr()
         stat_obj.decr(2)
-        assert self.redis.get('resque:stat:test_stat') == 0
+        assert self.redis.get('resque:stat:test_stat') == str(0)
 
     def test_get(self):
         stat_obj = Stat('test_stat', self.resq)

tests/test_worker.py

@@ -48,23 +48,27 @@ def test_processed(self):
         worker.processed()
         assert self.redis.exists("resque:stat:processed")
         assert self.redis.exists("resque:stat:processed:%s" % name)
-        assert self.redis.get("resque:stat:processed") == 1
-        assert self.redis.get("resque:stat:processed:%s" % name) == 1
+        assert self.redis.get("resque:stat:processed") == str(1)
+        assert self.redis.get("resque:stat:processed:%s" % name) == str(1)
+        assert worker.get_processed() == 1
         worker.processed()
-        assert self.redis.get("resque:stat:processed") == 2
-        assert self.redis.get("resque:stat:processed:%s" % name) == 2
+        assert self.redis.get("resque:stat:processed") == str(2)
+        assert self.redis.get("resque:stat:processed:%s" % name) == str(2)
+        assert worker.get_processed() == 2
 
     def test_failed(self):
         name = "%s:%s:%s" % (os.uname()[1],os.getpid(),'basic')
         worker = Worker(['basic'])
         worker.failed()
         assert self.redis.exists("resque:stat:failed")
         assert self.redis.exists("resque:stat:failed:%s" % name)
-        assert self.redis.get("resque:stat:failed") == 1
-        assert self.redis.get("resque:stat:failed:%s" % name) == 1
+        assert self.redis.get("resque:stat:failed") == str(1)
+        assert self.redis.get("resque:stat:failed:%s" % name) == str(1)
+        assert worker.get_failed() == 1
         worker.failed()
-        assert self.redis.get("resque:stat:failed") == 2
-        assert self.redis.get("resque:stat:failed:%s" % name) == 2
+        assert self.redis.get("resque:stat:failed") == str(2)
+        assert self.redis.get("resque:stat:failed:%s" % name) == str(2)
+        assert worker.get_failed() == 2
 
     def test_process(self):
         name = "%s:%s:%s" % (os.uname()[1],os.getpid(),'basic')
@@ -102,8 +106,8 @@ def test_job_failure(self):
         worker.process()
         name = "%s:%s:%s" % (os.uname()[1],os.getpid(),'basic')
         assert not self.redis.get('resque:worker:%s' % worker)
-        assert self.redis.get("resque:stat:failed") == 1
-        assert self.redis.get("resque:stat:failed:%s" % name) == 1
+        assert self.redis.get("resque:stat:failed") == str(1)
+        assert self.redis.get("resque:stat:failed:%s" % name) == str(1)
 
     def test_get_job(self):
         worker = Worker(['basic'])
@@ -112,6 +116,7 @@ def test_get_job(self):
         worker.working_on(job)
         name = "%s:%s:%s" % (os.uname()[1],os.getpid(),'basic')
         assert worker.job() == ResQ.decode(self.redis.get('resque:worker:%s' % name))
+        assert worker.processing() == ResQ.decode(self.redis.get('resque:worker:%s' % name))
         worker.done_working()
         w2 = Worker(['basic'])
         print w2.job()
@@ -138,4 +143,16 @@ def test_started(self):
         assert self.redis.get('resque:worker:%s:started' % name) == dt.strftime('%Y-%m-%d %H:%M:%S')
         assert worker.started == datetime.datetime.strptime(dt.strftime('%Y-%m-%d %H:%M:%S'),'%Y-%m-%d %H:%M:%S')
         worker.started = None
-        assert not self.redis.exists('resque:worker:%s:started' % name)
+        assert not self.redis.exists('resque:worker:%s:started' % name)
+    
+    def test_state(self):
+        worker = Worker(['basic'])
+        assert worker.state() == 'idle'
+        self.resq.enqueue_from_string('tests.Basic','basic','test1')
+        worker.register_worker()
+        job = Job.reserve('basic', self.resq)
+        worker.working_on(job)
+        assert worker.state() == 'working'
+        worker.done_working()
+        assert worker.state() == 'idle'
+