From 7690dc61d1186b4c1b350566f0e6e1600165bf33 Mon Sep 17 00:00:00 2001 From: Victor Shyba Date: Thu, 13 Jul 2017 03:33:48 -0300 Subject: [feature] LIST blobs - filter by flags filter_flag parameter can now be used to filter listing blobs by a specific flags. Eg: I can ask for blobs on incoming namespace flagged as pending. -- Resolves: #8913 --- testing/tests/server/test_blobs_server.py | 32 +++++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) (limited to 'testing/tests') diff --git a/testing/tests/server/test_blobs_server.py b/testing/tests/server/test_blobs_server.py index 249d76bc..869ef9a0 100644 --- a/testing/tests/server/test_blobs_server.py +++ b/testing/tests/server/test_blobs_server.py @@ -68,6 +68,38 @@ class BlobServerTestCase(unittest.TestCase): flags = yield manager.get_flags('blob_id') self.assertEquals([Flags.PROCESSING], flags) + @defer.inlineCallbacks + @pytest.mark.usefixtures("method_tmpdir") + def test_list_filter_flag(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(filter_flag=Flags.PENDING) + self.assertEquals([], blobs_list) + blobs_list = yield manager.remote_list(filter_flag=Flags.PROCESSING) + self.assertEquals(['blob_id'], blobs_list) + + @defer.inlineCallbacks + @pytest.mark.usefixtures("method_tmpdir") + def test_list_filter_flag_order_by_date(self): + manager = BlobManager('', self.uri, self.secret, + self.secret, 'user') + yield manager._encrypt_and_upload('blob_id1', BytesIO("x")) + yield manager._encrypt_and_upload('blob_id2', BytesIO("x")) + yield manager._encrypt_and_upload('blob_id3', BytesIO("x")) + yield manager.set_flags('blob_id1', [Flags.PROCESSING]) + yield manager.set_flags('blob_id2', [Flags.PROCESSING]) + yield manager.set_flags('blob_id3', [Flags.PROCESSING]) + blobs_list = yield manager.remote_list(filter_flag=Flags.PROCESSING, + order_by='+date') + expected_list = ['blob_id1', 'blob_id2', 'blob_id3'] + self.assertEquals(expected_list, blobs_list) + blobs_list = yield manager.remote_list(filter_flag=Flags.PROCESSING, + order_by='-date') + self.assertEquals(list(reversed(expected_list)), blobs_list) + @defer.inlineCallbacks @pytest.mark.usefixtures("method_tmpdir") def test_cant_set_invalid_flags(self): -- cgit v1.2.3