diff options
author | Tim Burke <tim.burke@gmail.com> | 2019-10-09 21:05:47 -0700 |
---|---|---|
committer | Tim Burke <tim.burke@gmail.com> | 2019-11-06 03:48:02 +0000 |
commit | 1eda8f9f3eac55953b3d54e32754e2ec312ab348 (patch) | |
tree | a60d70e87ff3bf280464e4c6808648ac35d43160 /tests/functional/test_swiftclient.py | |
parent | 9527d0497f19487114e158de6387fb5483cf3182 (diff) | |
download | python-swiftclient-1eda8f9f3eac55953b3d54e32754e2ec312ab348.tar.gz |
Rename "tests" directory to be "test" like in the swift repo
In addition to being less confusing for devs, this lets us actually run
tempauth tests in swiftclient dsvm jobs.
The job definition (over in the swift repo) specifies test/sample.conf,
which does not exist in this repo. As a result, those tests would skip
with
SKIPPING FUNCTIONAL TESTS DUE TO NO CONFIG
Change-Id: I558dbf9a657d442e6e19468e543bbec855129eeb
Diffstat (limited to 'tests/functional/test_swiftclient.py')
-rw-r--r-- | tests/functional/test_swiftclient.py | 557 |
1 files changed, 0 insertions, 557 deletions
diff --git a/tests/functional/test_swiftclient.py b/tests/functional/test_swiftclient.py deleted file mode 100644 index 9a74c63..0000000 --- a/tests/functional/test_swiftclient.py +++ /dev/null @@ -1,557 +0,0 @@ -# Copyright (c) 2014 Christian Schwede <christian.schwede@enovance.com> -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or -# implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -import os -import unittest -import time -from io import BytesIO - -import six -from six.moves import configparser - -import swiftclient - - -class TestFunctional(unittest.TestCase): - - def __init__(self, *args, **kwargs): - super(TestFunctional, self).__init__(*args, **kwargs) - self.skip_tests = False - self._get_config() - - self.test_data = b'42' * 10 - self.etag = '2704306ec982238d85d4b235c925d58e' - - self.containername = "functional-tests-container-%s" % int(time.time()) - self.containername_2 = self.containername + '_second' - self.containername_3 = self.containername + '_third' - self.objectname = "functional-tests-object-%s" % int(time.time()) - self.objectname_2 = self.objectname + '_second' - - def _get_config(self): - config_file = os.environ.get('SWIFT_TEST_CONFIG_FILE', - '/etc/swift/test.conf') - config = configparser.ConfigParser({'auth_version': '1'}) - config.read(config_file) - self.config = config - if config.has_section('func_test'): - if config.has_option('func_test', 'auth_uri'): - self.auth_url = config.get('func_test', 'auth_uri') - try: - self.auth_version = config.get('func_test', 'auth_version') - except configparser.NoOptionError: - last_piece = self.auth_url.rstrip('/').rsplit('/', 1)[1] - if last_piece.endswith('.0'): - last_piece = last_piece[:-2] - if last_piece in ('1', '2', '3'): - self.auth_version = last_piece - else: - raise - else: - auth_host = config.get('func_test', 'auth_host') - auth_port = config.getint('func_test', 'auth_port') - auth_ssl = config.getboolean('func_test', 'auth_ssl') - auth_prefix = config.get('func_test', 'auth_prefix') - self.auth_version = config.get('func_test', 'auth_version') - self.auth_url = "" - if auth_ssl: - self.auth_url += "https://" - else: - self.auth_url += "http://" - self.auth_url += "%s:%s%s" % ( - auth_host, auth_port, auth_prefix) - if self.auth_version == "1": - self.auth_url += 'v1.0' - - try: - self.account_username = config.get('func_test', - 'account_username') - except configparser.NoOptionError: - account = config.get('func_test', 'account') - username = config.get('func_test', 'username') - self.account_username = "%s:%s" % (account, username) - self.password = config.get('func_test', 'password') - else: - self.skip_tests = True - - def _get_connection(self): - """ - Subclasses may override to use different connection setup - """ - return swiftclient.Connection( - self.auth_url, self.account_username, self.password, - auth_version=self.auth_version) - - def setUp(self): - super(TestFunctional, self).setUp() - if self.skip_tests: - self.skipTest('SKIPPING FUNCTIONAL TESTS DUE TO NO CONFIG') - - self.conn = self._get_connection() - self.conn.put_container(self.containername) - self.conn.put_container(self.containername_2) - self.conn.put_object( - self.containername, self.objectname, self.test_data) - self.conn.put_object( - self.containername, self.objectname_2, self.test_data) - - def tearDown(self): - super(TestFunctional, self).tearDown() - for obj in [self.objectname, self.objectname_2]: - try: - self.conn.delete_object(self.containername, obj) - except swiftclient.ClientException: - pass - - for container in [self.containername, - self.containername_2, - self.containername_3, - self.containername + '_segments']: - try: - self.conn.delete_container(container) - except swiftclient.ClientException: - pass - self.conn.close() - - def _check_account_headers(self, headers): - headers_to_check = [ - 'content-length', - 'x-account-object-count', - 'x-timestamp', - 'x-trans-id', - 'date', - 'x-account-bytes-used', - 'x-account-container-count', - 'content-type', - 'accept-ranges', - ] - for h in headers_to_check: - self.assertIn(h, headers) - self.assertTrue(headers[h]) - - def test_stat_account(self): - headers = self.conn.head_account() - self._check_account_headers(headers) - - def test_list_account(self): - headers, containers = self.conn.get_account() - self._check_account_headers(headers) - - self.assertTrue(len(containers)) - test_container = [c - for c in containers - if c.get('name') == self.containername][0] - self.assertTrue(test_container.get('bytes') >= 0) - self.assertTrue(test_container.get('count') >= 0) - - # Check if list limit is working - headers, containers = self.conn.get_account(limit=1) - self.assertEqual(1, len(containers)) - - # Check full listing - headers, containers = self.conn.get_account(limit=1, full_listing=True) - self.assertTrue(len(containers) >= 2) # there might be more containers - - # Test marker - headers, containers = self.conn.get_account(marker=self.containername) - self.assertTrue(len(containers) >= 1) - self.assertEqual(self.containername_2, containers[0].get('name')) - - # Test prefix - _, containers = self.conn.get_account(prefix='dne') - self.assertEqual(0, len(containers)) - - # Test delimiter - _, containers = self.conn.get_account( - prefix=self.containername, delimiter='_') - self.assertEqual(2, len(containers)) - self.assertEqual(self.containername, containers[0].get('name')) - self.assertTrue( - self.containername_2.startswith(containers[1].get('subdir'))) - - def _check_container_headers(self, headers): - self.assertTrue(headers.get('content-length')) - self.assertTrue(headers.get('x-container-object-count')) - self.assertTrue(headers.get('x-timestamp')) - self.assertTrue(headers.get('x-trans-id')) - self.assertTrue(headers.get('date')) - self.assertTrue(headers.get('x-container-bytes-used')) - self.assertTrue(headers.get('x-container-object-count')) - self.assertTrue(headers.get('content-type')) - self.assertTrue(headers.get('accept-ranges')) - - def test_stat_container(self): - headers = self.conn.head_container(self.containername) - self._check_container_headers(headers) - - def test_list_container(self): - headers, objects = self.conn.get_container(self.containername) - self._check_container_headers(headers) - self.assertTrue(len(objects)) - test_object = [o - for o in objects - if o.get('name') == self.objectname][0] - self.assertEqual(len(self.test_data), test_object.get('bytes')) - self.assertEqual(self.etag, test_object.get('hash')) - self.assertEqual('application/octet-stream', - test_object.get('content_type')) - - # Check if list limit is working - headers, objects = self.conn.get_container(self.containername, limit=1) - self.assertEqual(1, len(objects)) - - # Check full listing - headers, objects = self.conn.get_container( - self.containername, limit=1, full_listing=True) - self.assertEqual(2, len(objects)) - - # Test marker - headers, objects = self.conn.get_container( - self.containername, marker=self.objectname) - self.assertEqual(1, len(objects)) - self.assertEqual(self.objectname_2, objects[0].get('name')) - - def test_create_container(self): - self.conn.put_container(self.containername_3) - self.assertTrue(self.conn.head_container(self.containername_3)) - - def test_delete(self): - self.conn.delete_object(self.containername, self.objectname) - self.conn.delete_object(self.containername, self.objectname_2) - self.conn.delete_container(self.containername) - - # Container HEAD will raise an exception if container doesn't exist - # which is only possible if previous requests succeeded - self.assertRaises( - swiftclient.ClientException, - self.conn.head_container, - self.containername) - - def test_upload_object(self): - # Object with content from string - self.conn.put_object( - self.containername, self.objectname, contents=self.test_data) - hdrs = self.conn.head_object(self.containername, self.objectname) - self.assertEqual(str(len(self.test_data)), - hdrs.get('content-length')) - self.assertEqual(self.etag, hdrs.get('etag')) - self.assertEqual('application/octet-stream', - hdrs.get('content-type')) - - # Same but with content_type - self.conn.put_object( - self.containername, self.objectname, - content_type='text/plain', contents=self.test_data) - hdrs = self.conn.head_object(self.containername, self.objectname) - self.assertEqual(str(len(self.test_data)), - hdrs.get('content-length')) - self.assertEqual(self.etag, hdrs.get('etag')) - self.assertEqual('text/plain', - hdrs.get('content-type')) - - # Same but with content-type in headers - self.conn.put_object( - self.containername, self.objectname, - headers={'Content-Type': 'text/plain'}, contents=self.test_data) - hdrs = self.conn.head_object(self.containername, self.objectname) - self.assertEqual(str(len(self.test_data)), - hdrs.get('content-length')) - self.assertEqual(self.etag, hdrs.get('etag')) - self.assertEqual('text/plain', - hdrs.get('content-type')) - - # content_type rewrites content-type in headers - self.conn.put_object( - self.containername, self.objectname, - content_type='image/jpeg', - headers={'Content-Type': 'text/plain'}, contents=self.test_data) - hdrs = self.conn.head_object(self.containername, self.objectname) - self.assertEqual(str(len(self.test_data)), - hdrs.get('content-length')) - self.assertEqual(self.etag, hdrs.get('etag')) - self.assertEqual('image/jpeg', - hdrs.get('content-type')) - - # Same but with content-length - self.conn.put_object( - self.containername, self.objectname, - contents=self.test_data, content_length=len(self.test_data)) - hdrs = self.conn.head_object(self.containername, self.objectname) - self.assertEqual(str(len(self.test_data)), - hdrs.get('content-length')) - self.assertEqual(self.etag, hdrs.get('etag')) - self.assertEqual('application/octet-stream', hdrs.get('content-type')) - - # Content from File-like object - fileobj = BytesIO(self.test_data) - self.conn.put_object( - self.containername, self.objectname, contents=fileobj) - hdrs = self.conn.head_object(self.containername, self.objectname) - self.assertEqual(str(len(self.test_data)), - hdrs.get('content-length')) - self.assertEqual(self.etag, hdrs.get('etag')) - self.assertEqual('application/octet-stream', hdrs.get('content-type')) - - # Content from File-like object, but read in chunks - fileobj = BytesIO(self.test_data) - self.conn.put_object( - self.containername, self.objectname, - contents=fileobj, content_length=len(self.test_data), - chunk_size=10) - hdrs = self.conn.head_object(self.containername, self.objectname) - self.assertEqual(str(len(self.test_data)), - hdrs.get('content-length')) - self.assertEqual(self.etag, hdrs.get('etag')) - self.assertEqual('application/octet-stream', hdrs.get('content-type')) - - # Wrong etag arg, should raise an exception - self.assertRaises( - swiftclient.ClientException, - self.conn.put_object, - self.containername, self.objectname, - contents=self.test_data, etag='invalid') - - def test_download_object(self): - # Download whole object - hdrs, body = self.conn.get_object(self.containername, self.objectname) - self.assertEqual(self.test_data, body) - - # Download in chunks, should return a generator - hdrs, body = self.conn.get_object( - self.containername, self.objectname, - resp_chunk_size=10) - downloaded_contents = b'' - while True: - try: - chunk = next(body) - except StopIteration: - break - downloaded_contents += chunk - self.assertEqual(self.test_data, downloaded_contents) - - # Download in chunks, should also work with read - hdrs, body = self.conn.get_object( - self.containername, self.objectname, - resp_chunk_size=10) - num_bytes = 5 - downloaded_contents = body.read(num_bytes) - self.assertEqual(num_bytes, len(downloaded_contents)) - downloaded_contents += body.read() - self.assertEqual(self.test_data, downloaded_contents) - - def test_put_object_using_generator(self): - # verify that put using a generator yielding empty strings does not - # cause connection to be closed - def data(): - yield b"should" - yield b"" - yield b" tolerate" - yield b"" - yield b" empty chunks" - - self.conn.put_object( - self.containername, self.objectname, data()) - hdrs, body = self.conn.get_object(self.containername, self.objectname) - self.assertEqual(b"should tolerate empty chunks", body) - - def test_download_object_retry_chunked(self): - resp_chunk_size = 2 - hdrs, body = self.conn.get_object(self.containername, - self.objectname, - resp_chunk_size=resp_chunk_size) - data = next(body) - self.assertEqual(self.test_data[:resp_chunk_size], data) - self.assertTrue(1, self.conn.attempts) - for chunk in body.resp: - # Flush remaining data from underlying response - # (simulate a dropped connection) - pass - # Trigger the retry - for chunk in body: - data += chunk - self.assertEqual(self.test_data, data) - self.assertEqual(2, self.conn.attempts) - - def test_download_object_retry_chunked_auth_failure(self): - resp_chunk_size = 2 - self.conn.token = 'invalid' - hdrs, body = self.conn.get_object(self.containername, - self.objectname, - resp_chunk_size=resp_chunk_size) - self.assertEqual(2, self.conn.attempts) - for chunk in body.resp: - # Flush remaining data from underlying response - # (simulate a dropped connection) - pass - - self.conn.token = 'invalid' - data = next(body) - self.assertEqual(4, self.conn.attempts) - - for chunk in body: - data += chunk - - self.assertEqual(self.test_data, data) - self.assertEqual(4, self.conn.attempts) - - def test_download_object_non_chunked(self): - hdrs, body = self.conn.get_object(self.containername, self.objectname) - data = body - self.assertEqual(self.test_data, data) - self.assertTrue(1, self.conn.attempts) - - hdrs, body = self.conn.get_object(self.containername, self.objectname, - resp_chunk_size=0) - data = body - self.assertEqual(self.test_data, data) - self.assertTrue(1, self.conn.attempts) - - def test_post_account(self): - self.conn.post_account({'x-account-meta-data': 'Something'}) - headers = self.conn.head_account() - self.assertEqual('Something', headers.get('x-account-meta-data')) - - def test_post_container(self): - self.conn.post_container( - self.containername, {'x-container-meta-color': 'Something'}) - - headers = self.conn.head_container(self.containername) - self.assertEqual('Something', headers.get('x-container-meta-color')) - - def test_post_object(self): - self.conn.post_object(self.containername, - self.objectname, - {'x-object-meta-color': 'Something', - 'x-object-meta-uni': b'\xd8\xaa'.decode('utf8'), - 'x-object-meta-int': 123, - 'x-object-meta-float': 45.67, - 'x-object-meta-bool': False}) - - headers = self.conn.head_object(self.containername, self.objectname) - self.assertEqual('Something', headers.get('x-object-meta-color')) - self.assertEqual(b'\xd8\xaa'.decode('utf-8'), - headers.get('x-object-meta-uni')) - self.assertEqual('123', headers.get('x-object-meta-int')) - self.assertEqual('45.67', headers.get('x-object-meta-float')) - self.assertEqual('False', headers.get('x-object-meta-bool')) - - def test_post_object_unicode_header_name(self): - self.conn.post_object(self.containername, - self.objectname, - {u'x-object-meta-\U0001f44d': u'\U0001f44d'}) - - # Note that we can't actually read this header back on py3; see - # https://bugs.python.org/issue37093 - # We'll have to settle for just testing that the POST doesn't blow up - # with a UnicodeDecodeError - if six.PY2: - headers = self.conn.head_object( - self.containername, self.objectname) - self.assertIn(u'x-object-meta-\U0001f44d', headers) - self.assertEqual(u'\U0001f44d', - headers.get(u'x-object-meta-\U0001f44d')) - - def test_copy_object(self): - self.conn.put_object( - self.containername, self.objectname, self.test_data) - self.conn.copy_object(self.containername, - self.objectname, - headers={'x-object-meta-color': 'Something'}) - - headers = self.conn.head_object(self.containername, self.objectname) - self.assertEqual('Something', headers.get('x-object-meta-color')) - - self.conn.copy_object(self.containername, - self.objectname, - headers={'x-object-meta-taste': 'Second'}) - - headers = self.conn.head_object(self.containername, self.objectname) - self.assertEqual('Something', headers.get('x-object-meta-color')) - self.assertEqual('Second', headers.get('x-object-meta-taste')) - - destination = "/%s/%s" % (self.containername, self.objectname_2) - self.conn.copy_object(self.containername, - self.objectname, - destination, - headers={'x-object-meta-taste': 'Second'}) - headers, data = self.conn.get_object(self.containername, - self.objectname_2) - self.assertEqual(self.test_data, data) - self.assertEqual('Something', headers.get('x-object-meta-color')) - self.assertEqual('Second', headers.get('x-object-meta-taste')) - - destination = "/%s/%s" % (self.containername, self.objectname_2) - self.conn.copy_object(self.containername, - self.objectname, - destination, - headers={'x-object-meta-color': 'Else'}, - fresh_metadata=True) - - headers = self.conn.head_object(self.containername, self.objectname_2) - self.assertEqual('Else', headers.get('x-object-meta-color')) - self.assertIsNone(headers.get('x-object-meta-taste')) - - def test_get_capabilities(self): - resp = self.conn.get_capabilities() - self.assertTrue(resp.get('swift')) - - -class TestUsingKeystone(TestFunctional): - """ - Repeat tests using os_options parameter to Connection. - """ - - def _get_connection(self): - account = username = password = None - if self.auth_version not in ('2', '3'): - self.skipTest('SKIPPING KEYSTONE-SPECIFIC FUNCTIONAL TESTS') - try: - account = self.config.get('func_test', 'account') - username = self.config.get('func_test', 'username') - password = self.config.get('func_test', 'password') - except Exception: - self.skipTest('SKIPPING KEYSTONE-SPECIFIC FUNCTIONAL TESTS' + - ' - NO CONFIG') - os_options = {'tenant_name': account} - return swiftclient.Connection( - self.auth_url, username, password, auth_version=self.auth_version, - os_options=os_options) - - -class TestUsingKeystoneV3(TestFunctional): - """ - Repeat tests using a keystone user with domain specified. - """ - - def _get_connection(self): - account = username = password = project_domain = user_domain = None - if self.auth_version != '3': - self.skipTest('SKIPPING KEYSTONE-V3-SPECIFIC FUNCTIONAL TESTS') - try: - account = self.config.get('func_test', 'account4') - username = self.config.get('func_test', 'username4') - user_domain = self.config.get('func_test', 'domain4') - project_domain = self.config.get('func_test', 'domain4') - password = self.config.get('func_test', 'password4') - except Exception: - self.skipTest('SKIPPING KEYSTONE-V3-SPECIFIC FUNCTIONAL TESTS' + - ' - NO CONFIG') - - os_options = {'project_name': account, - 'project_domain_name': project_domain, - 'user_domain_name': user_domain} - return swiftclient.Connection(self.auth_url, username, password, - auth_version=self.auth_version, - os_options=os_options) |