fix: claimed message require claim_id to delete

But bulk deletion can still delete both unclaimed and claimed
messages.  However, the major purpose to support bulk deletion
is to make it easy for a user to issue a DELETE on the Location
header returned from a POST; if a user intend to use claim, then
this URI does not help.

Change-Id: Ieed92fc0a07d2e873b729242fb2fb14536462780
Closes-Bug: #1215484

marconi/storage/exceptions.py

@@ -84,6 +84,13 @@ class MessageDoesNotExist(DoesNotExist):
         super(MessageDoesNotExist, self).__init__(msg)
 
 
+class MessageIsClaimed(NotPermitted):
+
+    def __init__(self, mid):
+        msg = (u'Message %(mid)s is claimed' % dict(mid=mid))
+        super(MessageIsClaimed, self).__init__(msg)
+
+
 class ClaimDoesNotExist(DoesNotExist):
 
     def __init__(self, cid, queue, project):
@@ -93,9 +100,9 @@ class ClaimDoesNotExist(DoesNotExist):
         super(ClaimDoesNotExist, self).__init__(msg)
 
 
-class ClaimNotPermitted(NotPermitted):
+class MessageIsClaimedBy(NotPermitted):
 
     def __init__(self, mid, cid):
         msg = (u'Message %(mid)s is not claimed by %(cid)s' %
                dict(cid=cid, mid=mid))
-        super(ClaimNotPermitted, self).__init__(msg)
+        super(MessageIsClaimedBy, self).__init__(msg)

marconi/storage/mongodb/messages.py

@@ -609,7 +609,6 @@ class MessageController(storage.MessageBase):
             '_id': mid
         }
 
-        if claim:
         # NOTE(cpp-cabrera): return early - the user gaves us an
         # invalid claim id and that renders the rest of this
         # request moot
@@ -622,16 +621,20 @@ class MessageController(storage.MessageBase):
         message = self._col.find_one(query)
 
         if message is None:
-                return None
+            return
 
-            if not ('c' in message and
+        is_claimed = (message['c']['id'] is not None and
-                    message['c']['id'] == cid and
+                      message['c']['e'] > now)
-                    message['c']['e'] > now):
+
-                raise exceptions.ClaimNotPermitted(message_id, claim)
+        if claim is None:
+            if is_claimed:
+                raise exceptions.MessageIsClaimed(message_id)
+
+        else:
+            if message['c']['id'] != cid:
+                raise exceptions.MessageIsClaimedBy(message_id, claim)
 
         self._col.remove(query['_id'], w=0)
-        else:
-            self._col.remove(query, w=0)
 
     @utils.raises_conn_error
     def bulk_delete(self, queue_name, message_ids, project=None):

marconi/storage/sqlite/messages.py

@@ -223,15 +223,6 @@ class MessageController(base.MessageBase):
         if id is None:
             return
 
-        if not claim:
-            self.driver.run('''
-                delete from Messages
-                 where id = ?
-                   and qid = (select id from Queues
-                               where project = ? and name = ?)
-            ''', id, project, queue)
-            return
-
         with self.driver('immediate'):
             message_exists, = self.driver.get('''
                 select count(M.id)
@@ -244,8 +235,24 @@ class MessageController(base.MessageBase):
             if not message_exists:
                 return
 
+            if claim is None:
+                self.__delete_unclaimed(id)
+            else:
                 self.__delete_claimed(id, claim)
 
+    def __delete_unclaimed(self, id):
+        self.driver.run('''
+            delete from Messages
+             where id = ?
+               and not exists (select *
+                                 from Claims join Locked
+                                   on id = cid
+                                where ttl > julianday() * 86400.0 - created)
+        ''', id)
+
+        if not self.driver.affected:
+            raise exceptions.MessageIsClaimed(id)
+
     def __delete_claimed(self, id, claim):
         # Precondition: id exists in a specific queue
         cid = utils.cid_decode(claim)
@@ -263,7 +270,7 @@ class MessageController(base.MessageBase):
         ''', id, cid)
 
         if not self.driver.affected:
-            raise exceptions.ClaimNotPermitted(id, claim)
+            raise exceptions.MessageIsClaimedBy(id, claim)
 
     def bulk_delete(self, queue, message_ids, project):
         if project is None:

marconi/tests/transport/wsgi/test_claims.py

@@ -145,6 +145,10 @@ class ClaimsBaseTest(base.TestBase):
                           claim_href)
         self.assertEquals(claim['ttl'], 100)
 
+        # Try to delete the message without submitting a claim_id
+        self.simulate_delete(message_href, self.project_id)
+        self.assertEquals(self.srmock.status, falcon.HTTP_403)
+
         # Delete the message and its associated claim
         self.simulate_delete(message_href, self.project_id,
                              query_string=params)

marconi/transport/wsgi/messages.py

@@ -290,10 +290,11 @@ class ItemResource(object):
 
         except storage_exceptions.NotPermitted as ex:
             LOG.exception(ex)
-            title = _(u'Invalid claim')
+            title = _(u'Unable to delete')
-            description = _(u'The specified claim either does not '
+            description = _(u'This message is claimed; it cannot be '
-                            u'exist or has expired.')
+                            u'deleted without a valid claim_id.')
             raise falcon.HTTPForbidden(title, description)
+
         except Exception as ex:
             LOG.exception(ex)
             description = _(u'Message could not be deleted.')