Merge "do not allow redis job reclaim by same owner"

taskflow/jobs/backends/impl_redis.py

@@ -410,12 +410,10 @@ if redis.call("hexists", listings_key, job_key) == 1 then
             -- Owner is the same, leave it alone...
             redis.call("set", last_modified_key, last_modified_blob)
             apply_ttl(owner_key, ms_expiry)
-            result["status"] = "${ok}"
-        else
+        end
         result["status"] = "${error}"
         result["reason"] = "${already_claimed}"
         result["owner"] = owner
-        end
     else
         redis.call("set", owner_key, expected_owner)
         redis.call("set", last_modified_key, last_modified_blob)

taskflow/tests/unit/jobs/test_redis_job.py

@@ -20,6 +20,7 @@ from oslo_utils import uuidutils
 import six
 import testtools
 
+from taskflow import exceptions as excp
 from taskflow.jobs.backends import impl_redis
 from taskflow import states
 from taskflow import test
@@ -76,6 +77,28 @@ class RedisJobboardTest(test.TestCase, base.BoardTestMixin):
             possible_jobs = list(self.board.iterjobs(only_unclaimed=True))
             self.assertEqual(1, len(possible_jobs))
 
+    def test_posting_claim_same_owner(self):
+        with base.connect_close(self.board):
+            with self.flush(self.client):
+                self.board.post('test', p_utils.temporary_log_book())
+
+            self.assertEqual(1, self.board.job_count)
+            possible_jobs = list(self.board.iterjobs(only_unclaimed=True))
+            self.assertEqual(1, len(possible_jobs))
+            j = possible_jobs[0]
+            self.assertEqual(states.UNCLAIMED, j.state)
+
+            with self.flush(self.client):
+                self.board.claim(j, self.board.name)
+
+            possible_jobs = list(self.board.iterjobs())
+            self.assertEqual(1, len(possible_jobs))
+            with self.flush(self.client):
+                self.assertRaises(excp.UnclaimableJob, self.board.claim,
+                                  possible_jobs[0], self.board.name)
+            possible_jobs = list(self.board.iterjobs(only_unclaimed=True))
+            self.assertEqual(0, len(possible_jobs))
+
     def setUp(self):
         super(RedisJobboardTest, self).setUp()
         self.client, self.board = self.create_board()