Skip to content

Instantly share code, notes, and snippets.

@chianingwang
Created November 18, 2017 01:45
Show Gist options
  • Save chianingwang/aa98a9e2db2daee37e05118c29048225 to your computer and use it in GitHub Desktop.
Save chianingwang/aa98a9e2db2daee37e05118c29048225 to your computer and use it in GitHub Desktop.
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