[refactor] use producer/consumer on write/read_blob respectivelly

3 files changed, 47 insertions, 31 deletions

diff --git a/tests/benchmarks/test_blobs_fs_backend.py b/tests/benchmarks/test_blobs_fs_backend.py
index d3a663d2..bb5b9b88 100644
--- a/tests/benchmarks/test_blobs_fs_backend.py
+++ b/tests/benchmarks/test_blobs_fs_backend.py
@@ -2,7 +2,7 @@ import pytest
 from io import BytesIO
 from leap.soledad.server._blobs import FilesystemBlobsBackend
 from twisted.internet import defer
-from twisted.web.test.test_web import DummyRequest
+from twisted.web.client import FileBodyProducer
 
 
 def create_write_test(amount, size):
@@ -12,12 +12,11 @@ def create_write_test(amount, size):
     def test(txbenchmark, payload, tmpdir):
         backend = FilesystemBlobsBackend(blobs_path=tmpdir.strpath)
         data = payload(size)
+        semaphore = defer.DeferredSemaphore(100)
         deferreds = []
         for i in xrange(amount):
-            fd = BytesIO(data)
-            request = DummyRequest([''])
-            request.content = fd
-            d = backend.write_blob('user', str(i), request)
+            producer = FileBodyProducer(BytesIO(data))
+            d = semaphore.run(backend.write_blob, 'user', str(i), producer)
             deferreds.append(d)
         yield txbenchmark(defer.gatherResults, deferreds)
 
@@ -30,6 +29,12 @@ test_blobs_fs_backend_write_100_100k = create_write_test(100, 100 * 1000)
 test_blobs_fs_backend_write_1000_10k = create_write_test(1000, 10 * 1000)
 
 
+class DevNull(object):
+
+    def write(self, data):
+        pass
+
+
 def create_read_test(amount, size):
 
     @pytest.inlineCallbacks
@@ -39,22 +44,20 @@ def create_read_test(amount, size):
         data = payload(size)
 
         # first write blobs to the backend...
+        semaphore = defer.DeferredSemaphore(100)
         deferreds = []
         for i in xrange(amount):
-            fd = BytesIO(data)
-            request = DummyRequest([''])
-            request.content = fd
-            d = backend.write_blob('user', str(i), request)
+            producer = FileBodyProducer(BytesIO(data))
+            d = semaphore.run(backend.write_blob, 'user', str(i), producer)
             deferreds.append(d)
         yield defer.gatherResults(deferreds)
 
         # ... then measure the read operation
         deferreds = []
         for i in xrange(amount):
-            request = DummyRequest([''])
-            d = request.notifyFinish()
+            consumer = DevNull()
+            d = semaphore.run(backend.read_blob, 'user', str(i), consumer)
             deferreds.append(d)
-            backend.read_blob('user', str(i), request)
         yield txbenchmark(defer.gatherResults, deferreds)
 
     return test
diff --git a/tests/blobs/test_fs_backend.py b/tests/blobs/test_fs_backend.py
index 5b3ff30a..fc3d649a 100644
--- a/tests/blobs/test_fs_backend.py
+++ b/tests/blobs/test_fs_backend.py
@@ -28,7 +28,6 @@ import mock
 import os
 import base64
 import io
-import json
 import pytest
 
 
@@ -64,14 +63,16 @@ class FilesystemBackendTestCase(unittest.TestCase):
 
     @pytest.mark.usefixtures("method_tmpdir")
     @mock.patch('leap.soledad.server._blobs.open')
-    @mock.patch.object(_blobs.FilesystemBlobsBackend, '_get_path',
-                       Mock(return_value='path'))
+    @mock.patch('leap.soledad.server._blobs.FilesystemBlobsBackend._get_path')
     @defer.inlineCallbacks
-    def test_read_blob(self, open):
+    def test_read_blob(self, get_path, open):
+        get_path.return_value = 'path'
+        open.return_value = io.BytesIO('content')
         backend = _blobs.FilesystemBlobsBackend(blobs_path=self.tempdir)
-        yield backend.read_blob('user', 'blob_id')
-        open.assert_called_once_with('path')
-        backend._get_path.assert_called_once_with('user', 'blob_id', '')
+        consumer = Mock()
+        yield backend.read_blob('user', 'blob_id', consumer)
+        consumer.write.assert_called_with('content')
+        get_path.assert_called_once_with('user', 'blob_id', '')
 
     @pytest.mark.usefixtures("method_tmpdir")
     @mock.patch.object(os.path, 'isfile')
@@ -215,22 +216,32 @@ class FilesystemBackendTestCase(unittest.TestCase):
                                  namespace='custom')
         default = yield backend.list_blobs('user')
         custom = yield backend.list_blobs('user', namespace='custom')
-        self.assertEquals([], json.loads(default))
-        self.assertEquals(['blob_id'], json.loads(custom))
+        self.assertEquals([], default)
+        self.assertEquals(['blob_id'], custom)
 
     @pytest.mark.usefixtures("method_tmpdir")
     @defer.inlineCallbacks
     def test_count(self):
         backend = _blobs.FilesystemBlobsBackend(blobs_path=self.tempdir)
         content = 'blah'
-        yield backend.write_blob('user', 'blob_id_1', io.BytesIO(content))
-        yield backend.write_blob('user', 'blob_id_2', io.BytesIO(content))
-        yield backend.write_blob('user', 'blob_id_3', io.BytesIO(content))
+
+        ids = range(5)
+
+        def _write(namespace=''):
+            producer = FileBodyProducer(io.BytesIO(content))
+            d = backend.write_blob('user', str(ids.pop()), producer,
+                                   namespace=namespace)
+            return d
+
+        yield _write()
+        yield _write()
+        yield _write()
+
         count = yield backend.count('user')
         self.assertEqual(3, count)
-        yield backend.write_blob('user', 'blob_id_1', io.BytesIO(content),
-                                 namespace='xfiles')
-        yield backend.write_blob('user', 'blob_id_2', io.BytesIO(content),
-                                 namespace='xfiles')
+
+        yield _write(namespace='xfiles')
+        yield _write(namespace='xfiles')
+
         count = yield backend.count('user', namespace='xfiles')
         self.assertEqual(2, count)
diff --git a/tests/server/test_incoming_server.py b/tests/server/test_incoming_server.py
index f278be74..23c0aa90 100644
--- a/tests/server/test_incoming_server.py
+++ b/tests/server/test_incoming_server.py
@@ -18,12 +18,13 @@
 Integration tests for incoming API
 """
 import pytest
+import mock
+import treq
 from io import BytesIO
 from uuid import uuid4
 from twisted.web.server import Site
 from twisted.internet import reactor
 from twisted.internet import defer
-import treq
 
 from leap.soledad.server._incoming import IncomingResource
 from leap.soledad.server._blobs import BlobsServerState
@@ -82,9 +83,10 @@ class IncomingOnCouchServerTestCase(CouchDBTestCase):
         yield treq.put(incoming_endpoint, BytesIO(content), persistent=False)
 
         db = self.state.open_database(user_id)
-        fd = yield db.read_blob(user_id, doc_id, namespace='MX')
+        consumer = mock.Mock()
+        yield db.read_blob(user_id, doc_id, consumer, namespace='MX')
         flags = yield db.get_flags(user_id, doc_id, namespace='MX')
-        data = fd.read()
+        data = consumer.write.call_args[0][0]
         expected_preamble = formatter.preamble(content, doc_id)
         expected_preamble = decode_preamble(expected_preamble, True)
         written_preamble, written_content = data.split()