[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

1 files changed, 46 insertions, 0 deletions

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):
@@ -57,6 +59,50 @@ class BlobServerTestCase(unittest.TestCase):
 
     @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):
         manager = BlobManager('', self.uri, self.secret,
                               self.secret, 'user')