openstack/swift
Code Issues Proposed changes

Merge "s3api: fix test_service with pre-existing buckets"

Browse Source
This commit is contained in:
Zuul
2025-10-08 15:56:18 +00:00
committed by Gerrit Code Review
parent 97e00e208f b5e6964a22
commit e963d13979
4 changed files with 86 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
roles/dsvm-additional-middlewares/tasks/main.yaml
View File

@@ -21,12 +21,12 @@
value: example.com
become: true
- name: Turn on s3_acl_tests_disabled in test.conf (for Keystone tests)
- name: Turn off s3_acl_tests_enabled in test.conf (for Keystone tests)
ini_file:
path: /etc/swift/test.conf
section: s3api_test
option: s3_acl_tests_disabled
value: true
option: s3_acl_tests_enabled
value: false
become: true
- name: Set storage_domain in test/sample.conf (for tempauth tests)
@@ -37,12 +37,12 @@
value: example.com
become: true
- name: Turn on s3_acl_tests_disabled in test/sample.conf (for tempauth tests)
- name: Turn off s3_acl_tests_enabled in test/sample.conf (for tempauth tests)
ini_file:
path: "{{ ansible_env.HOME }}/{{ zuul.project.src_dir }}/../swift/test/sample.conf"
section: s3api_test
option: s3_acl_tests_disabled
value: true
option: s3_acl_tests_enabled
value: false
become: true
- name: Enable object versioning

13
test/s3api/__init__.py
View File

@@ -144,11 +144,20 @@ def get_s3_client(user=1, signature_version='s3v4', addressing_style='path'):
)
def is_s3_acl_tests_enabled():
explicit_opt = get_opt('s3_acl_tests_enabled', None)
if explicit_opt is not None:
return config_true_value(explicit_opt)
else:
legacy_opt = get_opt('s3_acl_tests_disabled', 'false')
return not config_true_value(legacy_opt)
def skip_if_s3_acl_tests_disabled(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
if config_true_value(get_opt('s3_acl_tests_disabled', 'false')):
raise unittest.SkipTest('s3_acl_tests_disabled is true')
if not is_s3_acl_tests_enabled():
raise unittest.SkipTest('s3_acl_tests_enabled is false')
return func(*args, **kwargs)
return wrapper

91
test/s3api/test_service.py
View File

@@ -14,44 +14,84 @@
# limitations under the License.
import unittest
from collections import defaultdict
import botocore.exceptions
from test.s3api import BaseS3TestCase, ConfigError, \
skip_if_s3_acl_tests_disabled
skip_if_s3_acl_tests_disabled, is_s3_acl_tests_enabled
class TestGetServiceSigV4(BaseS3TestCase):
def _do_test_empty_service(self, client):
def setUp(self):
super().setUp()
# Capture existing buckets before running tests
self.existing_buckets = defaultdict(list)
self.existing_buckets[1] = self._get_buckets(1)
if is_s3_acl_tests_enabled():
# client2 seems to be always able to list buckets
try:
self.get_s3_client(2)
except ConfigError:
pass
else:
self.existing_buckets[2] = self._get_buckets(2)
# client3 gets AccessDenied unless s3_acl = True
try:
self.get_s3_client(3)
except ConfigError:
pass
else:
try:
self.existing_buckets[3] = self._get_buckets(3)
except botocore.exceptions.ClientError as e:
# but the lack of the existing_buckets doesn't really
# matter to most tests
if e.response['Error']['Code'] == 'AccessDenied':
pass
else:
raise
def _get_buckets(self, client_num):
client = self.get_s3_client(client_num)
resp = client.list_buckets()
return [bucket['Name'] for bucket in resp['Buckets']]
def _do_test_existing_service(self, client_num):
client = self.get_s3_client(client_num)
resp = client.list_buckets()
self.assertEqual(200, resp['ResponseMetadata']['HTTPStatusCode'])
self.assertEqual([], resp['Buckets'])
found_buckets = [bucket['Name'] for bucket in resp['Buckets']]
self.assertEqual(self.existing_buckets[client_num], found_buckets)
self.assertIn('x-amz-request-id',
resp['ResponseMetadata']['HTTPHeaders'])
self.check_owner(resp['Owner'])
self.assertIn('ID', resp['Owner'])
def test_empty_service(self):
client1 = self.get_s3_client(1)
self._do_test_empty_service(client1)
def test_existing_service(self):
self._do_test_existing_service(1)
@skip_if_s3_acl_tests_disabled
def test_empty_service_client3(self):
def test_existing_service_client3(self):
try:
client3 = self.get_s3_client(3)
self.get_s3_client(3)
except ConfigError as err:
raise unittest.SkipTest(str(err))
else:
self._do_test_empty_service(client3)
self._do_test_existing_service(3)
def _create_buckets(self, client):
def _create_buckets(self, client_num):
client = self.get_s3_client(client_num)
buckets = [self.create_name('bucket%s' % i) for i in range(5)]
for bucket in buckets:
client.create_bucket(Bucket=bucket)
return buckets
def _do_test_service_with_buckets(self, client, buckets):
def _do_test_service_with_buckets(self, client_num, buckets):
client = self.get_s3_client(client_num)
resp = client.list_buckets()
self.assertEqual(200, resp['ResponseMetadata']['HTTPStatusCode'])
self.assertEqual(sorted(buckets), [
expected_buckets = buckets + self.existing_buckets[client_num]
self.assertEqual(sorted(expected_buckets), [
bucket['Name'] for bucket in resp['Buckets']])
self.assertTrue(all('CreationDate' in bucket
for bucket in resp['Buckets']))
@@ -60,9 +100,9 @@ class TestGetServiceSigV4(BaseS3TestCase):
self.check_owner(resp['Owner'])
def test_service_with_buckets(self):
client = self.get_s3_client(1)
buckets = self._create_buckets(client)
self._do_test_service_with_buckets(client, buckets)
client_num = 1
buckets = self._create_buckets(client_num)
self._do_test_service_with_buckets(client_num, buckets)
@skip_if_s3_acl_tests_disabled
def test_service_with_buckets_client2(self):
@@ -71,22 +111,25 @@ class TestGetServiceSigV4(BaseS3TestCase):
client2 = self.get_s3_client(2)
except ConfigError as err:
raise unittest.SkipTest(str(err))
client1 = self.get_s3_client(1)
self._create_buckets(client1)
buckets2 = self._create_buckets(client2)
self.assertEqual(sorted(buckets2), [
bucket['Name'] for bucket in client2.list_buckets()['Buckets']])
self._create_buckets(1)
buckets2 = self._create_buckets(2)
expected_buckets = buckets2 + self.existing_buckets[2]
resp = client2.list_buckets()
found_buckets = [bucket['Name'] for bucket in resp['Buckets']]
self.assertEqual(sorted(expected_buckets), found_buckets)
@skip_if_s3_acl_tests_disabled
def test_service_with_buckets_client3(self):
# Unprivileged user can't see anything
# Unprivileged user can only see its own buckets
# (which should be empty)
try:
client3 = self.get_s3_client(3)
except ConfigError as err:
raise unittest.SkipTest(str(err))
client1 = self.get_s3_client(1)
self._create_buckets(client1)
self.assertEqual([], client3.list_buckets()['Buckets'])
self._create_buckets(1)
resp = client3.list_buckets()
found_buckets = [bucket['Name'] for bucket in resp['Buckets']]
self.assertEqual(self.existing_buckets[3], found_buckets)
class TestGetServiceSigV2(TestGetServiceSigV4):

4
test/sample.conf
View File

@@ -15,8 +15,8 @@ secret_key3 = testing3
# following non-default options to the s3api section of your proxy-server.conf
# s3_acl = True
# check_bucket_owner = True
# Alternatively, skip those tests by setting this option to True
s3_acl_tests_disabled = False
# Alternatively, skip those tests by setting this option to False
s3_acl_tests_enabled = True
[func_test]
# Sample config for Swift with tempauth