More coverage for celery.events.state

celery/events/state.py

@@ -62,7 +62,7 @@ class _lamportinfo(tuple):
         return tuple.__new__(cls, (clock, timestamp, id, obj))
 
     def __repr__(self):
-        return '_lamport(clock={0}, timestamp={1}, id={2} {3!r}'.format(*self)
+        return '_lamport(clock={0}, timestamp={1}, id={2} {3!r})'.format(*self)
 
     def __getnewargs__(self):
         return tuple(self)
@@ -74,7 +74,7 @@ class _lamportinfo(tuple):
             # uses logical clock value first
             if A and B:  # use logical clock if available
                 if A == B:  # equal clocks use lower process id
-                    return self[3] < other[3]
+                    return self[2] < other[2]
                 return A < B
             return self[1] < other[1]  # ... or use timestamp
         except IndexError:

celery/tests/events/test_state.py

@@ -1,12 +1,20 @@
 from __future__ import absolute_import
 
-from time import time
+import pickle
 
+from time import time
 from itertools import count
+from mock import Mock
 
 from celery import states
 from celery.events import Event
-from celery.events.state import State, Worker, Task, HEARTBEAT_EXPIRE_WINDOW
+from celery.events.state import (
+    State,
+    Worker,
+    Task,
+    HEARTBEAT_EXPIRE_WINDOW,
+    _lamportinfo
+)
 from celery.utils import uuid
 from celery.tests.case import Case
 
@@ -170,6 +178,9 @@ class test_State(Case):
     def test_repr(self):
         self.assertTrue(repr(State()))
 
+    def test_pickleable(self):
+        self.assertTrue(pickle.loads(pickle.dumps(State())))
+
     def test_worker_online_offline(self):
         r = ev_worker_online_offline(State())
         next(r)
@@ -350,3 +361,48 @@ class test_State(Case):
         s = State(callback=callback)
         s.event({'type': 'worker-online'})
         self.assertTrue(scratch.get('recv'))
+
+
+class test_lamportinfo(Case):
+
+    def test_repr(self):
+        x = _lamportinfo(133, time(), 'id', Mock())
+        self.assertTrue(repr(x))
+
+    def test_pickleable(self):
+        x = _lamportinfo(133, time(), 'id', 'obj')
+        self.assertEqual(pickle.loads(pickle.dumps(x)), tuple(x))
+
+    def test_order(self):
+        t1 = time()
+        a = _lamportinfo(133, t1, 'A', 'obj')
+        b = _lamportinfo(140, t1, 'A', 'obj')
+        self.assertTrue(a.__getnewargs__())
+        self.assertEqual(a.clock, 133)
+        self.assertEqual(a.timestamp, t1)
+        self.assertEqual(a.id, 'A')
+        self.assertEqual(a.obj, 'obj')
+        self.assertTrue(
+            a <= b,
+        )
+        self.assertTrue(
+            b >= a,
+        )
+
+        self.assertEqual(
+            _lamportinfo(134, time(), 'A', 'obj').__lt__(tuple()),
+            NotImplemented,
+        )
+        self.assertGreater(
+            _lamportinfo(134, time(), 'A', 'obj'),
+            _lamportinfo(133, time(), 'A', 'obj'),
+        )
+        self.assertGreater(
+            _lamportinfo(134, t1, 'B', 'obj'),
+            _lamportinfo(134, t1, 'A', 'obj'),
+        )
+
+        self.assertGreater(
+            _lamportinfo(None, time(), 'B', 'obj'),
+            _lamportinfo(None, t1, 'A', 'obj'),
+        )