Merge "Add config option for whether to skip s3_acl-requiring tests"

openstack/swift

Code Issues Proposed changes

Browse Source

This commit is contained in:

Zuul

2025年05月28日 18:05:43 +00:00

committed by Gerrit Code Review

parent f64269b981 7a3b7373bc

commit 1c244b3cd5

3 changed files with 62 additions and 34 deletions

Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL

Download Patch File Download Diff File Expand all files Collapse all files

11
test/s3api/init.py

View File

@@ -12,7 +12,7 @@

# implied.

# See the License for the specific language governing permissions and

# limitations under the License.

import functools

import logging

import os

import unittest

@@ -144,6 +144,15 @@ def get_s3_client(user=1, signature_version='s3v4', addressing_style='path'):

)

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')

return func(*args, **kwargs)

return wrapper

def etag_from_resp(response):

return response['ETag']

74
test/s3api/test_service.py

View File

@@ -15,38 +15,43 @@

import unittest

from test.s3api import BaseS3TestCase, ConfigError

from test.s3api import BaseS3TestCase, ConfigError, \

skip_if_s3_acl_tests_disabled

class TestGetServiceSigV4(BaseS3TestCase):

def _do_test_empty_service(self, client):

access_key = client._request_signer._credentials.access_key

resp = client.list_buckets()

self.assertEqual(200, resp['ResponseMetadata']['HTTPStatusCode'])

self.assertEqual([], resp['Buckets'])

self.assertIn('x-amz-request-id',

resp['ResponseMetadata']['HTTPHeaders'])

self.assertIn('DisplayName', resp['Owner'])

self.assertEqual(access_key, resp['Owner']['DisplayName'])

self.assertIn('ID', resp['Owner'])

def test_empty_service(self):

def do_test(client):

access_key = client._request_signer._credentials.access_key

resp = client.list_buckets()

self.assertEqual(200, resp['ResponseMetadata']['HTTPStatusCode'])

self.assertEqual([], resp['Buckets'])

self.assertIn('x-amz-request-id',

resp['ResponseMetadata']['HTTPHeaders'])

self.assertIn('DisplayName', resp['Owner'])

self.assertEqual(access_key, resp['Owner']['DisplayName'])

self.assertIn('ID', resp['Owner'])

client1 = self.get_s3_client(1)

self._do_test_empty_service(client1)

client = self.get_s3_client(1)

do_test(client)

@skip_if_s3_acl_tests_disabled

def test_empty_service_client3(self):

try:

client = self.get_s3_client(3)

except ConfigError:

pass

client3 = self.get_s3_client(3)

except ConfigError as err:

raise unittest.SkipTest(str(err))

else:

do_test(client)

self._do_test_empty_service(client3)

def test_service_with_buckets(self):

c = self.get_s3_client(1)

def _create_buckets(self, client):

buckets = [self.create_name('bucket%s' % i) for i in range(5)]

for bucket in buckets:

c.create_bucket(Bucket=bucket)

client.create_bucket(Bucket=bucket)

return buckets

resp = c.list_buckets()

def _do_test_service_with_buckets(self, client, buckets):

resp = client.list_buckets()

self.assertEqual(200, resp['ResponseMetadata']['HTTPStatusCode'])

self.assertEqual(sorted(buckets), [

bucket['Name'] for bucket in resp['Buckets']])

@@ -55,27 +60,38 @@ class TestGetServiceSigV4(BaseS3TestCase):

self.assertIn('x-amz-request-id',

resp['ResponseMetadata']['HTTPHeaders'])

self.assertIn('DisplayName', resp['Owner'])

access_key = c._request_signer._credentials.access_key

access_key = client._request_signer._credentials.access_key

self.assertEqual(access_key, resp['Owner']['DisplayName'])

self.assertIn('ID', 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)

@skip_if_s3_acl_tests_disabled

def test_service_with_buckets_client2(self):

# Second user can only see its own buckets

try:

c2 = self.get_s3_client(2)

client2 = self.get_s3_client(2)

except ConfigError as err:

raise unittest.SkipTest(str(err))

buckets2 = [self.create_name('bucket%s' % i) for i in range(2)]

for bucket in buckets2:

c2.create_bucket(Bucket=bucket)

client1 = self.get_s3_client(1)

self._create_buckets(client1)

buckets2 = self._create_buckets(client2)

self.assertEqual(sorted(buckets2), [

bucket['Name'] for bucket in c2.list_buckets()['Buckets']])

bucket['Name'] for bucket in client2.list_buckets()['Buckets']])

@skip_if_s3_acl_tests_disabled

def test_service_with_buckets_client3(self):

# Unprivileged user can't see anything

try:

c3 = self.get_s3_client(3)

client3 = self.get_s3_client(3)

except ConfigError as err:

raise unittest.SkipTest(str(err))

self.assertEqual([], c3.list_buckets()['Buckets'])

client1 = self.get_s3_client(1)

self._create_buckets(client1)

self.assertEqual([], client3.list_buckets()['Buckets'])

class TestGetServiceSigV2(TestGetServiceSigV4):