diff options
author | drebs <drebs@leap.se> | 2017-12-19 18:00:23 -0200 |
---|---|---|
committer | drebs <drebs@leap.se> | 2017-12-26 09:04:50 -0200 |
commit | 5a3d6cd05e8f12aba2527b2d711cd0aa7a27f3e8 (patch) | |
tree | b2f0b7b55feda0f3cd7ddd1e97a0975f2c1af794 /tests/server/test_blobs_server.py | |
parent | 0cf535b26c8989a60df646f64dfcb14b59de1a86 (diff) |
[feature] add ranges to blobs backend
Diffstat (limited to 'tests/server/test_blobs_server.py')
-rw-r--r-- | tests/server/test_blobs_server.py | 52 |
1 files changed, 52 insertions, 0 deletions
diff --git a/tests/server/test_blobs_server.py b/tests/server/test_blobs_server.py index bf929386..6fed6d65 100644 --- a/tests/server/test_blobs_server.py +++ b/tests/server/test_blobs_server.py @@ -19,6 +19,8 @@ Integration tests for blobs server """ import os import pytest +import re +import treq from urlparse import urljoin from uuid import uuid4 from io import BytesIO @@ -49,6 +51,11 @@ def sleep(x): return d +def _get(*args, **kwargs): + kwargs.update({'persistent': False}) + return treq.get(*args, **kwargs) + + class BlobServerTestCase(unittest.TestCase): def setUp(self): @@ -455,3 +462,48 @@ class BlobServerTestCase(unittest.TestCase): self.addCleanup(manager.close) with pytest.raises(SoledadError): yield manager.delete('missing_id') + + @defer.inlineCallbacks + @pytest.mark.usefixtures("method_tmpdir") + def test_get_range(self): + user_id = uuid4().hex + manager = BlobManager(self.tempdir, self.uri, self.secret, + self.secret, user_id) + self.addCleanup(manager.close) + blob_id, content = 'blob_id', '0123456789' + doc = BlobDoc(BytesIO(content), blob_id) + yield manager.put(doc, len(content)) + uri = urljoin(self.uri, '%s/%s' % (user_id, blob_id)) + res = yield _get(uri, headers={'Range': 'bytes=10-20'}) + text = yield res.text() + self.assertTrue(res.headers.hasHeader('content-range')) + content_range = res.headers.getRawHeaders('content-range').pop() + self.assertIsNotNone(re.match('^bytes 10-20/[0-9]+$', content_range)) + self.assertEqual(10, len(text)) + + @defer.inlineCallbacks + @pytest.mark.usefixtures("method_tmpdir") + def test_get_range_not_satisfiable(self): + # put a blob in place + user_id = uuid4().hex + manager = BlobManager(self.tempdir, self.uri, self.secret, + self.secret, user_id) + self.addCleanup(manager.close) + blob_id, content = uuid4().hex, 'content' + doc = BlobDoc(BytesIO(content), blob_id) + yield manager.put(doc, len(content)) + # and check possible parsing errors + uri = urljoin(self.uri, '%s/%s' % (user_id, blob_id)) + ranges = [ + 'bytes', + 'bytes=', + 'bytes=1', + 'bytes=blah-100', + 'potatoes=10-100' + 'blah' + ] + for range in ranges: + res = yield _get(uri, headers={'Range': range}) + self.assertEqual(416, res.code) + content_range = res.headers.getRawHeaders('content-range').pop() + self.assertIsNotNone(re.match('^bytes \*/[0-9]+$', content_range)) |