vbatts/quay
Archived
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Add missing copy_to method to Swift

Browse source 
Along with tests

Fixes https://coreosdev.atlassian.net/browse/QS-22
This commit is contained in:
Joseph Schorr 2017-10-04 15:25:39 -04:00
parent 3bef21253d
commit ab886226f3
2 changed files with 69 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

48
storage/test/test_swift.py
View file

@ -1,6 +1,7 @@
import io
import pytest
import hashlib
import copy
from collections import defaultdict
from mock import MagicMock
@ -26,9 +27,9 @@ class MockSwiftStorage(SwiftStorage):
return self._connection
class FakeSwiftStorage(SwiftStorage):
def __init__(self, fail_checksum=False, *args, **kwargs):
def __init__(self, fail_checksum=False, connection=None, *args, **kwargs):
super(FakeSwiftStorage, self).__init__(*args, **kwargs)
self._connection = FakeSwift(fail_checksum=fail_checksum)
self._connection = connection or FakeSwift(fail_checksum=fail_checksum)
def _get_connection(self):
return self._connection
@ -42,6 +43,11 @@ class FakeSwift(object):
def head_object(self, container, path):
return self.containers[container].get(path)
def copy_object(self, container, path, target):
pieces = target.split('/', 2)
_, content = self.get_object(container, path)
self.put_object(pieces[1], pieces[2], content)
def put_object(self, container, path, content, chunk_size=None, content_type=None, headers=None):
if not isinstance(content, str):
if hasattr(content, 'read'):
@ -53,7 +59,7 @@ class FakeSwift(object):
'content': content,
'chunk_size': chunk_size,
'content_type': content_type,
'headers': headers,
'headers': headers or {},
}
digest = hashlib.md5()
@ -103,7 +109,6 @@ def test_fixed_path_concat():
swift.exists('object/path')
swift._get_connection().head_object.assert_called_with('container-name', 'basepath/object/path')
def test_simple_path_concat():
simple_concat_args = dict(base_args)
simple_concat_args['simple_path_concat'] = True
@ -150,6 +155,41 @@ def test_remove():
swift.remove('somepath')
assert not swift.exists('somepath')
def test_copy_to():
swift = FakeSwiftStorage(**base_args)
modified_args = copy.deepcopy(base_args)
modified_args['swift_container'] = 'another_container'
another_swift = FakeSwiftStorage(connection=swift._connection, **modified_args)
swift.put_content('somepath', 'some content here')
swift.copy_to(another_swift, 'somepath')
assert swift.exists('somepath')
assert another_swift.exists('somepath')
assert swift.get_content('somepath') == 'some content here'
assert another_swift.get_content('somepath') == 'some content here'
def test_copy_to_different():
swift = FakeSwiftStorage(**base_args)
modified_args = copy.deepcopy(base_args)
modified_args['swift_user'] = 'foobarbaz'
modified_args['swift_container'] = 'another_container'
another_swift = FakeSwiftStorage(**modified_args)
swift.put_content('somepath', 'some content here')
swift.copy_to(another_swift, 'somepath')
assert swift.exists('somepath')
assert another_swift.exists('somepath')
assert swift.get_content('somepath') == 'some content here'
assert another_swift.get_content('somepath') == 'some content here'
def test_checksum():
swift = FakeSwiftStorage(**base_args)
swift.put_content('somepath', 'hello world!')