From 64107009a869a6ddb6cea129e65735d9740e697b Mon Sep 17 00:00:00 2001
From: Victor Shyba <victor1984@riseup.net>
Date: Sat, 8 Jul 2017 07:04:49 -0300
Subject: [feature] add get/set flags

IncomingBox spec has a flags feature for the processing flow of
messages. This commit adds it using a .flags file.

-- Resolves: #8869
---
 testing/tests/server/test_blobs_server.py | 46 +++++++++++++++++++++++++++++++
 1 file changed, 46 insertions(+)

(limited to 'testing/tests/server')

diff --git a/testing/tests/server/test_blobs_server.py b/testing/tests/server/test_blobs_server.py
index f4c9119b..0db64256 100644
--- a/testing/tests/server/test_blobs_server.py
+++ b/testing/tests/server/test_blobs_server.py
@@ -26,9 +26,11 @@ from twisted.internet import reactor
 from twisted.internet import defer
 from treq._utils import set_global_pool
 
+from leap.soledad.common.blobs import Flags
 from leap.soledad.server import _blobs as server_blobs
 from leap.soledad.client._db.blobs import BlobManager
 from leap.soledad.client._db.blobs import BlobAlreadyExistsError
+from leap.soledad.client._db.blobs import InvalidFlagsError
 
 
 class BlobServerTestCase(unittest.TestCase):
@@ -55,6 +57,50 @@ class BlobServerTestCase(unittest.TestCase):
         blob, size = yield manager._download_and_decrypt('blob_id')
         self.assertEquals(blob.getvalue(), "save me")
 
+    @defer.inlineCallbacks
+    @pytest.mark.usefixtures("method_tmpdir")
+    def test_set_get_flags(self):
+        manager = BlobManager('', self.uri, self.secret,
+                              self.secret, 'user')
+        fd = BytesIO("flag me")
+        yield manager._encrypt_and_upload('blob_id', fd)
+        yield manager.set_flags('blob_id', [Flags.PROCESSING])
+        flags = yield manager.get_flags('blob_id')
+        self.assertEquals([Flags.PROCESSING], flags)
+
+    @defer.inlineCallbacks
+    @pytest.mark.usefixtures("method_tmpdir")
+    def test_cant_set_invalid_flags(self):
+        manager = BlobManager('', self.uri, self.secret,
+                              self.secret, 'user')
+        fd = BytesIO("flag me")
+        yield manager._encrypt_and_upload('blob_id', fd)
+        with pytest.raises(InvalidFlagsError):
+            yield manager.set_flags('blob_id', ['invalid'])
+        flags = yield manager.get_flags('blob_id')
+        self.assertEquals([], flags)
+
+    @defer.inlineCallbacks
+    @pytest.mark.usefixtures("method_tmpdir")
+    def test_get_empty_flags(self):
+        manager = BlobManager('', self.uri, self.secret,
+                              self.secret, 'user')
+        fd = BytesIO("flag me")
+        yield manager._encrypt_and_upload('blob_id', fd)
+        flags = yield manager.get_flags('blob_id')
+        self.assertEquals([], flags)
+
+    @defer.inlineCallbacks
+    @pytest.mark.usefixtures("method_tmpdir")
+    def test_flags_ignored_by_listing(self):
+        manager = BlobManager('', self.uri, self.secret,
+                              self.secret, 'user')
+        fd = BytesIO("flag me")
+        yield manager._encrypt_and_upload('blob_id', fd)
+        yield manager.set_flags('blob_id', [Flags.PROCESSING])
+        blobs_list = yield manager.remote_list()
+        self.assertEquals(['blob_id'], blobs_list)
+
     @defer.inlineCallbacks
     @pytest.mark.usefixtures("method_tmpdir")
     def test_upload_changes_remote_list(self):
-- 
cgit v1.2.3