From cd288b183d1d5df6d8c504fa2303e83f49097f75 Mon Sep 17 00:00:00 2001 From: Tim Burke Date: Fri, 6 Sep 2024 12:26:39 -0700 Subject: [PATCH] tests: Functionally test account quotas Change-Id: Ied0ff6bea7e054fad3fe9579c85d9ae5c9c0b255 --- test/functional/__init__.py | 33 +++++---- test/functional/test_account.py | 127 ++++++++++++++++++++++++++++++++ 2 files changed, 147 insertions(+), 13 deletions(-) diff --git a/test/functional/__init__.py b/test/functional/__init__.py index 71b5e97b15..fdd48a5978 100644 --- a/test/functional/__init__.py +++ b/test/functional/__init__.py @@ -607,9 +607,9 @@ def in_process_setup(the_object_server=object_server): 'reseller_prefix': 'AUTH, SERVICE', 'SERVICE_require_group': 'service', # Reseller admin user (needs reseller_admin_role) - 'account6': 'test6', - 'username6': 'tester6', - 'password6': 'testing6' + 'account6': 'admin', + 'username6': 'admin', + 'password6': 'admin' }) acc1lis = listen_zero() @@ -912,6 +912,13 @@ def setup_package(): swift_test_tenant[4] = config['account5'] except KeyError: pass # no service token tests can be run + try: + swift_test_user[5] = '%s%s' % ( + '%s:' % config['account6'], config['username6']) + swift_test_key[5] = config['password6'] + swift_test_tenant[5] = config['account6'] + except KeyError: + pass # no reseller admin tests can be run for _ in range(3): swift_test_perm[_] = swift_test_user[_] @@ -1067,20 +1074,20 @@ class InternalServerError(Exception): pass -url = [None, None, None, None, None] -token = [None, None, None, None, None] -service_token = [None, None, None, None, None] -parsed = [None, None, None, None, None] -conn = [None, None, None, None, None] +url = [None, None, None, None, None, None] +token = [None, None, None, None, None, None] +service_token = [None, None, None, None, None, None] +parsed = [None, None, None, None, None, None] +conn = [None, None, None, None, None, None] def reset_globals(): global url, token, service_token, parsed, conn, config - url = [None, None, None, None, None] - token = [None, None, None, None, None] - service_token = [None, None, None, None, None] - parsed = [None, None, None, None, None] - conn = [None, None, None, None, None] + url = [None, None, None, None, None, None] + token = [None, None, None, None, None, None] + service_token = [None, None, None, None, None, None] + parsed = [None, None, None, None, None, None] + conn = [None, None, None, None, None, None] if config: config = {} diff --git a/test/functional/test_account.py b/test/functional/test_account.py index 6cf30968d5..20515842e6 100644 --- a/test/functional/test_account.py +++ b/test/functional/test_account.py @@ -924,5 +924,132 @@ class TestAccountInNonDefaultDomain(unittest.TestCase): self.assertIn('X-Account-Project-Domain-Id', resp.headers) +class TestAccountQuotas(unittest.TestCase): + def setUp(self): + if 'account_quotas' not in tf.cluster_info: + raise SkipTest('Account quotas are not enabled') + + self.policies = tf.FunctionalStoragePolicyCollection.from_info() + + def _check_user_cannot_post(self, headers): + def post(url, token, parsed, conn): + conn.request('POST', parsed.path, '', + dict({'X-Auth-Token': token}, **headers)) + return check_response(conn) + + resp = retry(post) + resp.read() + self.assertEqual(resp.status, 403) + + def test_user_cannot_set_own_quota(self): + self._check_user_cannot_post({'X-Account-Meta-Quota-Bytes': '0'}) + + def test_user_cannot_set_own_policy_quota(self): + policy = self.policies.select()['name'] + self._check_user_cannot_post( + {'X-Account-Quota-Bytes-Policy-' + policy: '0'}) + + def test_user_cannot_remove_own_quota(self): + self._check_user_cannot_post( + {'X-Remove-Account-Meta-Quota-Bytes': 't'}) + + def test_user_cannot_remove_own_policy_quota(self): + policy = self.policies.select()['name'] + self._check_user_cannot_post( + {'X-Remove-Account-Quota-Bytes-Policy-' + policy: 't'}) + + def _check_admin_can_post(self, headers): + def post(url, token, parsed, conn): + conn.request('POST', parsed.path, '', + dict({'X-Auth-Token': token}, **headers)) + return check_response(conn) + + resp = retry(post, use_account=6, url_account=1) + resp.read() + self.assertEqual(resp.status, 204) + + def test_admin_can_set_and_remove_user_quota(self): + if tf.skip_if_no_reseller_admin: + raise SkipTest('No admin user configured') + quota_header = 'X-Account-Meta-Quota-Bytes' + + def get_current_quota(): + def head(url, token, parsed, conn): + conn.request('HEAD', parsed.path, '', + {'X-Auth-Token': token}) + return check_response(conn) + + # Use user, not admin, to ensure globals in test.functional + # are properly populated before issuing POSTs + resp = retry(head) + resp.read() + self.assertEqual(resp.status, 204) + return resp.headers.get(quota_header) + + original_quota = get_current_quota() + + try: + self._check_admin_can_post({quota_header: '123'}) + self.assertEqual('123', get_current_quota()) + + self._check_admin_can_post( + {quota_header.replace('X-', 'X-Remove-'): 't'}) + self.assertIsNone(get_current_quota()) + + self._check_admin_can_post({quota_header: '111'}) + self.assertEqual('111', get_current_quota()) + + # Can also remove with an explicit empty string + self._check_admin_can_post({quota_header: ''}) + self.assertIsNone(get_current_quota()) + + self._check_admin_can_post({quota_header: '0'}) + self.assertEqual('0', get_current_quota()) + finally: + self._check_admin_can_post({quota_header: original_quota or ''}) + + def test_admin_can_set_and_remove_user_policy_quota(self): + if tf.skip_if_no_reseller_admin: + raise SkipTest('No admin user configured') + policy = self.policies.select()['name'] + quota_header = 'X-Account-Quota-Bytes-Policy-' + policy + + def get_current_quota(): + def head(url, token, parsed, conn): + conn.request('HEAD', parsed.path, '', + {'X-Auth-Token': token}) + return check_response(conn) + + # Use user, not admin, to ensure globals in test.functional + # are properly populated before issuing POSTs + resp = retry(head) + resp.read() + self.assertEqual(resp.status, 204) + return resp.headers.get(quota_header) + + original_quota = get_current_quota() + + try: + self._check_admin_can_post({quota_header: '123'}) + self.assertEqual('123', get_current_quota()) + + self._check_admin_can_post( + {quota_header.replace('X-', 'X-Remove-'): 't'}) + # TODO: well that seems like the opposite of what was intended... + self.assertEqual('0', get_current_quota()) + + self._check_admin_can_post({quota_header: '111'}) + self.assertEqual('111', get_current_quota()) + + # Can actually remove with an explicit empty string + self._check_admin_can_post({quota_header: ''}) + self.assertIsNone(get_current_quota()) + + self._check_admin_can_post({quota_header: '0'}) + self.assertEqual('0', get_current_quota()) + finally: + self._check_admin_can_post({quota_header: original_quota or ''}) + + if __name__ == '__main__': unittest.main()