summaryrefslogtreecommitdiff
path: root/testing
diff options
context:
space:
mode:
Diffstat (limited to 'testing')
-rw-r--r--testing/tests/blobs/test_blob_manager.py (renamed from testing/tests/blobs/test_local_backend.py)17
-rw-r--r--testing/tests/blobs/test_decrypter_buffer.py (renamed from testing/tests/blobs/test_blobs.py)7
-rw-r--r--testing/tests/blobs/test_fs_backend.py15
-rw-r--r--testing/tests/blobs/test_sqlcipher_client_backend.py63
4 files changed, 96 insertions, 6 deletions
diff --git a/testing/tests/blobs/test_local_backend.py b/testing/tests/blobs/test_blob_manager.py
index 202d0611..69a272c8 100644
--- a/testing/tests/blobs/test_local_backend.py
+++ b/testing/tests/blobs/test_blob_manager.py
@@ -15,11 +15,12 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""
-Tests for sqlcipher backend on blobs client.
+Tests for BlobManager.
"""
from twisted.trial import unittest
from twisted.internet import defer
from leap.soledad.client._database.blobs import BlobManager, BlobDoc, FIXED_REV
+from leap.soledad.client._database.blobs import BlobAlreadyExistsError
from io import BytesIO
from mock import Mock
import pytest
@@ -115,3 +116,17 @@ class BlobManagerTestCase(unittest.TestCase):
call_blob_id, call_fd = call_list[0][0]
self.assertEquals('missing_id', call_blob_id)
self.assertEquals('test', call_fd.getvalue())
+
+ @defer.inlineCallbacks
+ @pytest.mark.usefixtures("method_tmpdir")
+ def test_duplicated_blob_error_on_put(self):
+ self.manager._encrypt_and_upload = Mock(return_value=None)
+ content = "Blob content"
+ doc1 = BlobDoc(BytesIO(content), 'existing_id')
+ yield self.manager.put(doc1, len(content))
+ doc2 = BlobDoc(BytesIO(content), 'existing_id')
+ # reset mock, so we can check that upload wasnt called
+ self.manager._encrypt_and_upload = Mock(return_value=None)
+ with pytest.raises(BlobAlreadyExistsError):
+ yield self.manager.put(doc2, len(content))
+ self.assertFalse(self.manager._encrypt_and_upload.called)
diff --git a/testing/tests/blobs/test_blobs.py b/testing/tests/blobs/test_decrypter_buffer.py
index 9a0feb61..edaa66e2 100644
--- a/testing/tests/blobs/test_blobs.py
+++ b/testing/tests/blobs/test_decrypter_buffer.py
@@ -15,7 +15,8 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""
-Tests for blobs handling.
+Tests for blobs decrypter buffer. A component which is used as a decryption
+sink during blob stream download.
"""
from io import BytesIO
from mock import Mock
@@ -29,7 +30,7 @@ from leap.soledad.client._database.blobs import FIXED_REV
from leap.soledad.client import _crypto
-class BlobTestCase(unittest.TestCase):
+class DecrypterBufferCase(unittest.TestCase):
class doc_info:
doc_id = 'D-BLOB-ID'
@@ -53,7 +54,7 @@ class BlobTestCase(unittest.TestCase):
self.assertEquals(fd.getvalue(), 'rosa de foc')
@defer.inlineCallbacks
- def test_blob_manager_encrypted_upload(self):
+ def test_decrypt_uploading_encrypted_blob(self):
@defer.inlineCallbacks
def _check_result(uri, data, *args, **kwargs):
diff --git a/testing/tests/blobs/test_fs_backend.py b/testing/tests/blobs/test_fs_backend.py
index 6da22621..0d7e9789 100644
--- a/testing/tests/blobs/test_fs_backend.py
+++ b/testing/tests/blobs/test_fs_backend.py
@@ -18,6 +18,7 @@
Tests for blobs backend on server side.
"""
from twisted.trial import unittest
+from twisted.internet import defer
from twisted.web.test.test_web import DummyRequest
from leap.soledad.server import _blobs
from io import BytesIO
@@ -57,17 +58,19 @@ class FilesystemBackendTestCase(unittest.TestCase):
render_mock.render_GET.assert_called_once_with(request)
@mock.patch.object(os.path, 'isfile')
+ @defer.inlineCallbacks
def test_cannot_overwrite(self, isfile):
isfile.return_value = True
backend = _blobs.FilesystemBlobsBackend()
backend._get_path = Mock(return_value='path')
request = DummyRequest([''])
- result = yield backend.write_blob('user', 'blob_id', request)
- self.assertEquals(result, "Blob already exists: blob_id")
+ yield backend.write_blob('user', 'blob_id', request)
+ self.assertEquals(request.written[0], "Blob already exists: blob_id")
self.assertEquals(request.responseCode, 409)
@pytest.mark.usefixtures("method_tmpdir")
@mock.patch.object(os.path, 'isfile')
+ @defer.inlineCallbacks
def test_write_cannot_exceed_quota(self, isfile):
isfile.return_value = False
backend = _blobs.FilesystemBlobsBackend()
@@ -94,3 +97,11 @@ class FilesystemBackendTestCase(unittest.TestCase):
walk_mock.return_value = [(_, _, ['blob_0']), (_, _, ['blob_1'])]
result = json.loads(backend.list_blobs('user', DummyRequest([''])))
self.assertEquals(result, ['blob_0', 'blob_1'])
+
+ @pytest.mark.usefixtures("method_tmpdir")
+ def test_path_validation_for_subdirectories(self):
+ blobs_path = self.tempdir
+ backend = _blobs.FilesystemBlobsBackend(blobs_path)
+ self.assertFalse(backend._valid_subdir('/'))
+ self.assertFalse(backend._valid_subdir(blobs_path + '../../../../../'))
+ self.assertTrue(backend._valid_subdir(os.path.join(blobs_path, 'x')))
diff --git a/testing/tests/blobs/test_sqlcipher_client_backend.py b/testing/tests/blobs/test_sqlcipher_client_backend.py
new file mode 100644
index 00000000..865a64e1
--- /dev/null
+++ b/testing/tests/blobs/test_sqlcipher_client_backend.py
@@ -0,0 +1,63 @@
+# -*- coding: utf-8 -*-
+# test_sqlcipher_client_backend.py
+# Copyright (C) 2017 LEAP
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+"""
+Tests for sqlcipher backend on blobs client.
+"""
+from twisted.trial import unittest
+from twisted.internet import defer
+from leap.soledad.client._database.blobs import SQLiteBlobBackend
+from io import BytesIO
+import pytest
+
+
+class SQLBackendTestCase(unittest.TestCase):
+
+ def setUp(self):
+ self.key = "A" * 96
+ self.local = SQLiteBlobBackend(self.tempdir, self.key)
+ self.addCleanup(self.local.close)
+
+ @defer.inlineCallbacks
+ @pytest.mark.usefixtures("method_tmpdir")
+ def test_get_inexistent(self):
+ bad_blob_id = 'inexsistent_id'
+ self.assertFalse((yield self.local.exists(bad_blob_id)))
+ result = yield self.local.get(bad_blob_id)
+ self.assertIsNone(result)
+
+ @defer.inlineCallbacks
+ @pytest.mark.usefixtures("method_tmpdir")
+ def test_get_existent(self):
+ blob_id = 'blob_id'
+ content = "x"
+ yield self.local.put(blob_id, BytesIO(content), len(content))
+ result = yield self.local.get(blob_id)
+ self.assertTrue((yield self.local.exists(blob_id)))
+ self.assertEquals(result.getvalue(), content)
+
+ @defer.inlineCallbacks
+ @pytest.mark.usefixtures("method_tmpdir")
+ def test_list(self):
+ blob_ids = [('blob_id%s' % i) for i in range(10)]
+ content = "x"
+ deferreds = []
+ for blob_id in blob_ids:
+ deferreds.append(self.local.put(blob_id, BytesIO(content),
+ len(content)))
+ yield defer.gatherResults(deferreds)
+ result = yield self.local.list()
+ self.assertEquals(blob_ids, result)