Created
November 18, 2017 01:45
-
-
Save chianingwang/aa98a9e2db2daee37e05118c29048225 to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
sync_swift.py | |
============== | |
... | |
def _get_client_factory(self): | |
# TODO: support LDAP auth | |
# TODO: support v2 auth | |
username = self.settings['aws_identity'] | |
key = self.settings['aws_secret'] | |
# Endpoint must be defined for the Swift clusters and should be the | |
# auth URL | |
endpoint = self.settings['aws_endpoint'] | |
common_connection_kwargs = {'authurl': endpoint, | |
'user': endpoint, | |
'key': key, | |
'retries': 3} | |
# new setting - auth_type which mapp with UI list includes value | |
# swift, keystone_v1, keystone_v2 and keystone_v3 | |
authtype = self.settings['auth_type'] | |
# TODO:test v3 options w/o os_options | |
def swift_client_factory(): | |
if not authtype or authtype == 'swift': | |
return swiftclient.client.Connection( | |
**common_connection_kwargs) | |
elif authtype == 'keystone_v1': | |
return swiftclient.client.Connection(auth_version='1', | |
**common_connection_kwargs) | |
elif authtype == 'keystone_v2': | |
_tenant_name = self.settings['tenant_name'] | |
if _tenant_name: | |
return swiftclient.client.Connection(auth_version='2', | |
tenant_name=_tenant_name, **common_connection_kwargs) | |
else: | |
raise NotImplementedError() | |
elif authtype == 'keystone_v3': | |
_os_options = { | |
'user_domain_name': self.settings['user_domain_name'], | |
'project_domain_name': self.settings['project_domain_name'], | |
'project_name': self.settings['project_name'] | |
} | |
return swiftclient.client.Connection(auth_version='3', | |
os_options=_os_options, **common_connection_kwargs) | |
return swift_client_factory | |
================== | |
test_sync_swift.py | |
================== | |
... | |
def setUp(self): | |
self.aws_bucket = 'bucket' | |
self.scratch_space = 'scratch' | |
self.sync_swift = SyncSwift( | |
{'aws_bucket': self.aws_bucket, | |
'aws_identity': 'identity', | |
'aws_secret': 'credential', | |
'account': 'account', | |
'container': 'container', | |
'aws_endpoint': 'http://swift.url/auth/v1.0', | |
'auth_type':'swift'}) | |
... | |
@mock.patch('s3_sync.sync_swift.swiftclient.client.Connection') | |
def test_per_account_bucket(self, mock_swift): | |
mock_swift.return_value = mock.Mock() | |
# in this case, the "bucket" is actually the prefix | |
aws_bucket = 'sync_' | |
sync_swift = SyncSwift( | |
{'aws_bucket': aws_bucket, | |
'aws_identity': 'identity', | |
'aws_secret': 'credential', | |
'account': 'account', | |
'container': 'container', | |
'auth_type': 'swift', | |
'aws_endpoint': 'http://swift.url/auth/v1.0'}, | |
per_account=True) | |
self.assertEqual('sync_container', sync_swift.remote_container) | |
... | |
@mock.patch('s3_sync.sync_swift.swiftclient.client.Connection') | |
def test_keystone_v1_auth(self, mock_swift): | |
mock_swift.return_value = mock.Mock() | |
# in this case, the "bucket" is actually the prefix | |
aws_bucket = 'sync_' | |
sync_swift = SyncSwift( | |
{'aws_bucket': aws_bucket, | |
'aws_identity': 'identity', | |
'aws_secret': 'credential', | |
'account': 'account', | |
'container': 'container', | |
'auth_type': 'keystone_v1', | |
'aws_endpoint': 'http://swift.url/auth/v1.0'}, | |
per_account=True) | |
self.assertEqual('sync_container', sync_swift.remote_container) | |
@mock.patch('s3_sync.sync_swift.swiftclient.client.Connection') | |
def test_keystone_v2_auth(self, mock_swift): | |
mock_swift.return_value = mock.Mock() | |
# in this case, the "bucket" is actually the prefix | |
aws_bucket = 'sync_' | |
sync_swift = SyncSwift( | |
{'aws_bucket': aws_bucket, | |
'aws_identity': 'identity', | |
'aws_secret': 'credential', | |
'account': 'account', | |
'container': 'container', | |
'auth_type': 'keystone_v2', | |
'tenant_name': 'tenantname', | |
'aws_endpoint': 'http://swift.url/auth/v1.0'}, | |
per_account=True) | |
self.assertEqual('sync_container', sync_swift.remote_container) | |
@mock.patch('s3_sync.sync_swift.swiftclient.client.Connection') | |
def test_keystone_v3_auth(self, mock_swift): | |
mock_swift.return_value = mock.Mock() | |
# in this case, the "bucket" is actually the prefix | |
aws_bucket = 'sync_' | |
sync_swift = SyncSwift( | |
{'aws_bucket': aws_bucket, | |
'aws_identity': 'identity', | |
'aws_secret': 'credential', | |
'account': 'account', | |
'container': 'container', | |
'auth_type': 'keystone_v3', | |
'user_domain_name':'userdomainname', | |
'project_domain_name':'projectdomainname', | |
'project_name':'projectname', | |
'aws_endpoint': 'http://swift.url/auth/v1.0'}, | |
per_account=True) | |
self.assertEqual('sync_container', sync_swift.remote_container) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment