06a6329793
There is an infinite loop if multiple quota limits are set and exceeded,
eventually resulting in a 500 response due to a RecursionError ("maximum
recursion depth exceeded").
The issue is the delayed rejection, required to support container_acls.
If any quota is exceeded the middleware needs to return directly,
without proceeding to check other quota settings.
The fix is basically to add a "return self.app". However, there is quite
some redundant code, thus moving this into its own method.
Another test with multiple exceeded quotas has been added, which is
failing without the bugfix.
Closes-Bug: #2118758
Change-Id: I49ec4c5f6c83f36ce1d38f2f1687081c71488286
Signed-off-by: Christian Schwede <cschwede@redhat.com>
1108 lines
50 KiB
Python
1108 lines
50 KiB
Python
# 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 unittest
|
|
|
|
from swift.common.swob import Request, wsgify, HTTPForbidden, HTTPOk, \
|
|
HTTPServiceUnavailable, HTTPNotFound
|
|
|
|
from swift.common.middleware import account_quotas, copy
|
|
|
|
from test.unit import patch_policies
|
|
from test.unit.common.middleware.helpers import FakeSwift
|
|
|
|
|
|
class FakeCache(object):
|
|
def __init__(self, val):
|
|
self.val = val
|
|
|
|
def get(self, *args):
|
|
return self.val
|
|
|
|
def set(self, *args, **kwargs):
|
|
pass
|
|
|
|
|
|
class FakeAuthFilter(object):
|
|
|
|
def __init__(self, app):
|
|
self.app = app
|
|
|
|
@wsgify
|
|
def __call__(self, req):
|
|
def authorize(req):
|
|
if req.headers['x-auth-token'] == 'secret':
|
|
return
|
|
return HTTPForbidden(request=req)
|
|
req.environ['swift.authorize'] = authorize
|
|
return req.get_response(self.app)
|
|
|
|
|
|
class TestAccountQuota(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.app = FakeSwift()
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-bytes-used': '1000'})
|
|
self.app.register('HEAD', '/v1/a/c', HTTPOk, {
|
|
'x-backend-storage-policy-index': '1'})
|
|
self.app.register('POST', '/v1/a', HTTPOk, {})
|
|
self.app.register('PUT', '/v1/a/c/o', HTTPOk, {})
|
|
|
|
def test_unauthorized(self):
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o',
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
# Response code of 200 because authentication itself is not done here
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_no_quotas(self):
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o',
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_obj_request_ignores_attempt_to_set_quotas(self):
|
|
# If you try to set X-Account-Meta-* on an object, it's ignored, so
|
|
# the quota middleware shouldn't complain about it even if we're not a
|
|
# reseller admin.
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o',
|
|
headers={'X-Account-Meta-Quota-Bytes': '99999'},
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_container_request_ignores_attempt_to_set_quotas(self):
|
|
# As with an object, if you try to set X-Account-Meta-* on a
|
|
# container, it's ignored.
|
|
self.app.register('PUT', '/v1/a/c', HTTPOk, {})
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c',
|
|
headers={'X-Account-Meta-Quota-Bytes': '99999'},
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_bogus_quota_is_ignored(self):
|
|
# This can happen if the metadata was set by a user prior to the
|
|
# activation of the account-quota middleware
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-bytes-used': '1000',
|
|
'x-account-sysmeta-quota-bytes': 'pasty-plastogene'})
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o',
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_exceed_bytes_quota_legacy(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-bytes-used': '1000',
|
|
'x-account-meta-quota-bytes': '0'})
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o',
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 413)
|
|
self.assertEqual(res.body, b'Upload exceeds quota.')
|
|
|
|
def test_exceed_bytes_quota(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-bytes-used': '1000',
|
|
'x-account-sysmeta-quota-bytes': '0'})
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o',
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 413)
|
|
self.assertEqual(res.body, b'Upload exceeds quota.')
|
|
|
|
@patch_policies
|
|
def test_exceed_per_policy_quota(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-bytes-used': '100',
|
|
'x-account-storage-policy-unu-bytes-used': '100',
|
|
'x-account-sysmeta-quota-bytes-policy-1': '10',
|
|
'x-account-sysmeta-quota-bytes': '1000'})
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o',
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 413)
|
|
self.assertEqual(res.body, b'Upload exceeds policy quota.')
|
|
|
|
@patch_policies
|
|
def test_policy_quota_translation_legacy_loses(self):
|
|
# if we have both meta (legacy) and sysmeta, ensure sysmeta wins
|
|
def do_test(method):
|
|
self.app.register(method, '/v1/a', HTTPOk, {
|
|
'x-account-bytes-used': '100',
|
|
'x-account-storage-policy-unu-bytes-used': '100',
|
|
'x-account-sysmeta-quota-bytes-policy-1': '10',
|
|
'x-account-sysmeta-quota-bytes': '1000',
|
|
'x-account-meta-quota-bytes': '2000'})
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a', method=method, environ={
|
|
'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
self.assertEqual(res.headers.get(
|
|
'X-Account-Quota-Bytes'), '1000')
|
|
self.assertEqual(res.headers.get(
|
|
'X-Account-Meta-Quota-Bytes'), '2000')
|
|
self.assertEqual(res.headers.get(
|
|
'X-Account-Sysmeta-Quota-Bytes-Policy-1'), '10')
|
|
self.assertEqual(res.headers.get(
|
|
'X-Account-Quota-Bytes-Policy-Unu'), '10')
|
|
self.assertEqual(res.headers.get(
|
|
'X-Account-Storage-Policy-Unu-Bytes-Used'), '100')
|
|
|
|
do_test('GET')
|
|
do_test('HEAD')
|
|
|
|
@patch_policies
|
|
def test_policy_quota_translation(self):
|
|
def do_test(method):
|
|
self.app.register(method, '/v1/a', HTTPOk, {
|
|
'x-account-bytes-used': '100',
|
|
'x-account-storage-policy-unu-bytes-used': '100',
|
|
'x-account-sysmeta-quota-bytes-policy-1': '10',
|
|
'x-account-sysmeta-quota-bytes': '1000'})
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a', method=method, environ={
|
|
'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
self.assertEqual(res.headers.get(
|
|
'X-Account-Sysmeta-Quota-Bytes'), '1000')
|
|
self.assertEqual(res.headers.get(
|
|
'X-Account-Quota-Bytes'), '1000')
|
|
self.assertEqual(res.headers.get(
|
|
'X-Account-Sysmeta-Quota-Bytes-Policy-1'), '10')
|
|
self.assertEqual(res.headers.get(
|
|
'X-Account-Quota-Bytes-Policy-Unu'), '10')
|
|
self.assertEqual(res.headers.get(
|
|
'X-Account-Storage-Policy-Unu-Bytes-Used'), '100')
|
|
|
|
do_test('GET')
|
|
do_test('HEAD')
|
|
|
|
def test_exceed_quota_not_authorized(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-bytes-used': '1000',
|
|
'x-account-sysmeta-quota-bytes': '0'})
|
|
app = FakeAuthFilter(account_quotas.AccountQuotaMiddleware(self.app))
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o', method='PUT',
|
|
headers={'x-auth-token': 'bad-secret'},
|
|
environ={'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 403)
|
|
|
|
def test_exceed_quota_authorized_legacy(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-bytes-used': '1000',
|
|
'x-account-meta-quota-bytes': '0'})
|
|
app = FakeAuthFilter(account_quotas.AccountQuotaMiddleware(self.app))
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o', method='PUT',
|
|
headers={'x-auth-token': 'secret'},
|
|
environ={'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 413)
|
|
|
|
def test_exceed_quota_authorized(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-bytes-used': '1000',
|
|
'x-account-sysmeta-quota-bytes': '0'})
|
|
app = FakeAuthFilter(account_quotas.AccountQuotaMiddleware(self.app))
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o', method='PUT',
|
|
headers={'x-auth-token': 'secret'},
|
|
environ={'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 413)
|
|
|
|
def test_under_quota_not_authorized(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-bytes-used': '0',
|
|
'x-account-sysmeta-quota-bytes': '1000'})
|
|
app = FakeAuthFilter(account_quotas.AccountQuotaMiddleware(self.app))
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o', method='PUT',
|
|
headers={'x-auth-token': 'bad-secret'},
|
|
environ={'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 403)
|
|
|
|
def test_under_quota_authorized_legacy(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-bytes-used': '0',
|
|
'x-account-meta-quota-bytes': '1000'})
|
|
app = FakeAuthFilter(account_quotas.AccountQuotaMiddleware(self.app))
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o', method='PUT',
|
|
headers={'x-auth-token': 'secret'},
|
|
environ={'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_under_quota_authorized(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-bytes-used': '0',
|
|
'x-account-sysmeta-quota-bytes': '1000'})
|
|
app = FakeAuthFilter(account_quotas.AccountQuotaMiddleware(self.app))
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o', method='PUT',
|
|
headers={'x-auth-token': 'secret'},
|
|
environ={'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_exceed_quota_bytes_on_empty_account_authorized(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-bytes-used': '0',
|
|
'x-account-sysmeta-quota-bytes': '10'})
|
|
app = FakeAuthFilter(account_quotas.AccountQuotaMiddleware(self.app))
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o', method='PUT',
|
|
headers={'x-auth-token': 'secret',
|
|
'content-length': '100'},
|
|
environ={'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 413)
|
|
self.assertEqual(res.body, b'Upload exceeds quota.')
|
|
|
|
def test_exceed_quota_bytes_authorized(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-bytes-used': '100',
|
|
'x-account-sysmeta-quota-bytes': '1000'})
|
|
app = FakeAuthFilter(account_quotas.AccountQuotaMiddleware(self.app))
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o', method='PUT',
|
|
headers={'x-auth-token': 'secret',
|
|
'content-length': '901'},
|
|
environ={'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 413)
|
|
self.assertEqual(res.body, b'Upload exceeds quota.')
|
|
|
|
def test_exceed_multiple_quota_authorized(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-bytes-used': '100',
|
|
'x-account-object-count': '10',
|
|
'x-account-sysmeta-quota-count': '10',
|
|
'x-account-sysmeta-quota-bytes': '10'})
|
|
app = FakeAuthFilter(account_quotas.AccountQuotaMiddleware(self.app))
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o', method='PUT',
|
|
headers={'x-auth-token': 'secret',
|
|
'content-length': '901'},
|
|
environ={'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 413)
|
|
self.assertEqual(res.body, b'Upload exceeds quota.')
|
|
|
|
def test_over_quota_container_create_still_works(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-bytes-used': '1001',
|
|
'x-account-sysmeta-quota-bytes': '1000'})
|
|
self.app.register('PUT', '/v1/a/new_container', HTTPOk, {})
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/new_container',
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'HTTP_X_CONTAINER_META_BERT': 'ernie',
|
|
'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_over_quota_container_post_still_works(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-bytes-used': '1001',
|
|
'x-account-sysmeta-quota-bytes': '1000'})
|
|
self.app.register('POST', '/v1/a/new_container', HTTPOk, {})
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/new_container',
|
|
environ={'REQUEST_METHOD': 'POST',
|
|
'HTTP_X_CONTAINER_META_BERT': 'ernie',
|
|
'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_over_quota_obj_post_still_works(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-bytes-used': '1001',
|
|
'x-account-sysmeta-quota-bytes': '1000'})
|
|
self.app.register('POST', '/v1/a/c/o', HTTPOk, {})
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o',
|
|
environ={'REQUEST_METHOD': 'POST',
|
|
'HTTP_X_OBJECT_META_BERT': 'ernie',
|
|
'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_exceed_bytes_quota_reseller(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-bytes-used': '1000',
|
|
'x-account-sysmeta-quota-bytes': '0'})
|
|
self.app.register('PUT', '/v1/a', HTTPOk, {})
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a',
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache,
|
|
'reseller_request': True})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_exceed_bytes_quota_reseller_copy_from(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-bytes-used': '500',
|
|
'x-account-sysmeta-quota-bytes': '1000'})
|
|
self.app.register('GET', '/v1/a/c2/o2', HTTPOk, {
|
|
'content-length': '1000'}, b'a' * 1000)
|
|
app = copy.filter_factory({})(
|
|
account_quotas.AccountQuotaMiddleware(self.app))
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o',
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache,
|
|
'reseller_request': True},
|
|
headers={'x-copy-from': 'c2/o2'})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_exceed_bytes_quota_reseller_copy_verb(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-bytes-used': '500',
|
|
'x-account-sysmeta-quota-bytes': '1000'})
|
|
self.app.register('GET', '/v1/a/c2/o2', HTTPOk, {
|
|
'content-length': '1000'}, b'a' * 1000)
|
|
app = copy.filter_factory({})(
|
|
account_quotas.AccountQuotaMiddleware(self.app))
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c2/o2',
|
|
environ={'REQUEST_METHOD': 'COPY',
|
|
'swift.cache': cache,
|
|
'reseller_request': True},
|
|
headers={'Destination': 'c/o'})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_bad_application_quota(self):
|
|
self.app.register('PUT', '/v1/a/c/o', HTTPNotFound, {})
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o',
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 404)
|
|
|
|
def test_no_info_quota(self):
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o',
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_not_exceed_bytes_quota(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-bytes-used': '1000',
|
|
'x-account-sysmeta-quota-bytes': '2000'})
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o',
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_invalid_quotas(self):
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a',
|
|
environ={'REQUEST_METHOD': 'POST',
|
|
'swift.cache': cache,
|
|
'HTTP_X_ACCOUNT_QUOTA_BYTES': 'abc',
|
|
'reseller_request': True})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 400)
|
|
self.assertEqual(self.app.calls, [])
|
|
|
|
@patch_policies
|
|
def test_invalid_policy_quota(self):
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a', environ={
|
|
'REQUEST_METHOD': 'POST',
|
|
'swift.cache': cache,
|
|
'HTTP_X_ACCOUNT_QUOTA_BYTES_POLICY_NULO': 'abc',
|
|
'reseller_request': True})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 400)
|
|
self.assertEqual(self.app.calls, [])
|
|
|
|
def test_valid_quotas_non_admin_fails(self):
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a',
|
|
environ={'REQUEST_METHOD': 'POST',
|
|
'swift.cache': cache,
|
|
'HTTP_X_ACCOUNT_QUOTA_BYTES': '100'})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 403)
|
|
self.assertEqual(self.app.calls, [])
|
|
|
|
@patch_policies
|
|
def test_valid_policy_quota_admin(self):
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a', environ={
|
|
'REQUEST_METHOD': 'POST',
|
|
'swift.cache': cache,
|
|
'HTTP_X_ACCOUNT_QUOTA_BYTES_POLICY_UNU': '100'})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 403)
|
|
self.assertEqual(self.app.calls, [])
|
|
|
|
def test_valid_quotas_reseller_legacy(self):
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a',
|
|
environ={'REQUEST_METHOD': 'POST',
|
|
'swift.cache': cache,
|
|
'HTTP_X_ACCOUNT_META_QUOTA_BYTES': '100',
|
|
'reseller_request': True})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
self.assertEqual(self.app.calls_with_headers, [
|
|
('POST', '/v1/a', {'Host': 'localhost:80',
|
|
'X-Account-Quota-Bytes': '100',
|
|
'X-Account-Meta-Quota-Bytes': '100',
|
|
'X-Account-Sysmeta-Quota-Bytes': '100'})])
|
|
|
|
def test_valid_quotas_reseller_legacy_loses(self):
|
|
# if we have both meta (legacy) and sysmeta, ensure sysmeta wins
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a',
|
|
environ={'REQUEST_METHOD': 'POST',
|
|
'swift.cache': cache,
|
|
'HTTP_X_ACCOUNT_QUOTA_BYTES': '100',
|
|
'HTTP_X_ACCOUNT_META_QUOTA_BYTES': '200',
|
|
'reseller_request': True})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
self.assertEqual(self.app.calls_with_headers, [
|
|
('POST', '/v1/a', {'Host': 'localhost:80',
|
|
'X-Account-Quota-Bytes': '100',
|
|
'X-Account-Meta-Quota-Bytes': '200',
|
|
'X-Account-Sysmeta-Quota-Bytes': '100'})])
|
|
|
|
def test_valid_quotas_reseller(self):
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a',
|
|
environ={'REQUEST_METHOD': 'POST',
|
|
'swift.cache': cache,
|
|
'HTTP_X_ACCOUNT_QUOTA_BYTES': '100',
|
|
'reseller_request': True})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
self.assertEqual(self.app.calls_with_headers, [
|
|
('POST', '/v1/a', {'Host': 'localhost:80',
|
|
'X-Account-Quota-Bytes': '100',
|
|
'X-Account-Sysmeta-Quota-Bytes': '100'})])
|
|
|
|
@patch_policies
|
|
def test_valid_policy_quota_reseller(self):
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a', environ={
|
|
'REQUEST_METHOD': 'POST',
|
|
'swift.cache': cache,
|
|
'HTTP_X_ACCOUNT_QUOTA_BYTES_POLICY_NULO': '100',
|
|
'reseller_request': True})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
self.assertEqual(self.app.calls_with_headers, [
|
|
('POST', '/v1/a', {
|
|
'Host': 'localhost:80',
|
|
'X-Account-Sysmeta-Quota-Bytes-Policy-0': '100'})])
|
|
|
|
def test_delete_quotas(self):
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a',
|
|
environ={'REQUEST_METHOD': 'POST',
|
|
'swift.cache': cache,
|
|
'HTTP_X_ACCOUNT_QUOTA_BYTES': ''})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 403)
|
|
|
|
def test_delete_quotas_with_remove_header(self):
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a', environ={
|
|
'REQUEST_METHOD': 'POST',
|
|
'swift.cache': cache,
|
|
'HTTP_X_REMOVE_ACCOUNT_QUOTA_BYTES': 'True'})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 403)
|
|
|
|
def test_delete_quotas_reseller(self):
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
req = Request.blank('/v1/a',
|
|
environ={'REQUEST_METHOD': 'POST',
|
|
'HTTP_X_ACCOUNT_QUOTA_BYTES': '',
|
|
'reseller_request': True})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_delete_quotas_with_remove_header_reseller(self):
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a', environ={
|
|
'REQUEST_METHOD': 'POST',
|
|
'swift.cache': cache,
|
|
'HTTP_X_REMOVE_ACCOUNT_QUOTA_BYTES': 'True',
|
|
'reseller_request': True})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_invalid_request_exception(self):
|
|
self.app.register('PUT', '/v1', HTTPServiceUnavailable, {})
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1',
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 503)
|
|
|
|
def test_obj_request_ignores_attempt_to_set_count_quotas(self):
|
|
# If you try to set X-Account-Meta-* on an object, it's ignored, so
|
|
# the quota middleware shouldn't complain about it even if we're not a
|
|
# reseller admin.
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o',
|
|
headers={'X-Account-Meta-Quota-Count': '99999'},
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_container_request_ignores_attempt_to_set_count_quotas(self):
|
|
# As with an object, if you try to set X-Account-Meta-* on a
|
|
# container, it's ignored.
|
|
self.app.register('PUT', '/v1/a/c', HTTPOk, {})
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c',
|
|
headers={'X-Account-Meta-Quota-Count': '99999'},
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_bogus_count_quota_is_ignored(self):
|
|
# This can happen if the metadata was set by a user prior to the
|
|
# activation of the account-quota middleware
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-bytes-used': '1000',
|
|
'x-account-meta-quota-count': 'pasty-plastogene'})
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o',
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_exceed_count_quota(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-object-count': '10',
|
|
'x-account-sysmeta-quota-count': '10'})
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o',
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 413)
|
|
self.assertEqual(res.body, b'Upload exceeds quota.')
|
|
|
|
def test_exceed_quota_count_not_authorized(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-sysmeta-quota-count': '0'})
|
|
app = FakeAuthFilter(account_quotas.AccountQuotaMiddleware(self.app))
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o', method='PUT',
|
|
headers={'x-auth-token': 'bad-secret'},
|
|
environ={'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 403)
|
|
|
|
def test_exceed_count_quota_authorized(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-sysmeta-quota-count': '0'})
|
|
app = FakeAuthFilter(account_quotas.AccountQuotaMiddleware(self.app))
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o', method='PUT',
|
|
headers={'x-auth-token': 'secret'},
|
|
environ={'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 413)
|
|
|
|
def test_under_quota_count_not_authorized(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-object-count': '0',
|
|
'x-account-sysmeta-quota-count': '5'})
|
|
app = FakeAuthFilter(account_quotas.AccountQuotaMiddleware(self.app))
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o', method='PUT',
|
|
headers={'x-auth-token': 'bad-secret'},
|
|
environ={'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 403)
|
|
|
|
def test_under_quota_count_authorized(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-object-count': '0',
|
|
'x-account-sysmeta-quota-count': '5'})
|
|
app = FakeAuthFilter(account_quotas.AccountQuotaMiddleware(self.app))
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o', method='PUT',
|
|
headers={'x-auth-token': 'secret'},
|
|
environ={'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_exceed_quota_count_on_empty_account_not_authorized(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-object-count': '0',
|
|
'x-account-sysmeta-quota-count': '0'})
|
|
app = FakeAuthFilter(account_quotas.AccountQuotaMiddleware(self.app))
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o', method='PUT',
|
|
headers={'x-auth-token': 'bad-secret'},
|
|
environ={'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 403)
|
|
|
|
def test_exceed_quota_count_authorized(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-object-count': '5',
|
|
'x-account-sysmeta-quota-count': '5'})
|
|
app = FakeAuthFilter(account_quotas.AccountQuotaMiddleware(self.app))
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o', method='PUT',
|
|
headers={'x-auth-token': 'secret'},
|
|
environ={'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 413)
|
|
self.assertEqual(res.body, b'Upload exceeds quota.')
|
|
|
|
def test_over_quota_count_container_create_still_works(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-object-count': '6',
|
|
'x-account-sysmeta-quota-count': '5'})
|
|
self.app.register('PUT', '/v1/a/new_container', HTTPOk, {})
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/new_container',
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'HTTP_X_CONTAINER_META_BERT': 'ernie',
|
|
'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_over_quota_count_container_post_still_works(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-quota-count': '6',
|
|
'x-account-sysmeta-quota-count': '5'})
|
|
self.app.register('POST', '/v1/a/new_container', HTTPOk, {})
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/new_container',
|
|
environ={'REQUEST_METHOD': 'POST',
|
|
'HTTP_X_CONTAINER_META_BERT': 'ernie',
|
|
'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_over_count_quota_obj_post_still_works(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-object-count': '101',
|
|
'x-account-sysmeta-quota-count': '100'})
|
|
self.app.register('POST', '/v1/a/c/o', HTTPOk, {})
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o',
|
|
environ={'REQUEST_METHOD': 'POST',
|
|
'HTTP_X_OBJECT_META_BERT': 'ernie',
|
|
'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_exceed_count_quota_reseller(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-object-count': '1000',
|
|
'x-account-sysmeta-quota-count': '0'})
|
|
self.app.register('PUT', '/v1/a', HTTPOk, {})
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a',
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache,
|
|
'reseller_request': True})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_exceed_count_quota_reseller_copy_from(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-object-count': '10',
|
|
'x-account-sysmeta-quota-count': '10'})
|
|
self.app.register('GET', '/v1/a/c2/o2', HTTPOk, {
|
|
'content-length': '1000'}, b'a' * 1000)
|
|
app = copy.filter_factory({})(
|
|
account_quotas.AccountQuotaMiddleware(self.app))
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o',
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache,
|
|
'reseller_request': True},
|
|
headers={'x-copy-from': 'c2/o2'})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_exceed_count_quota_reseller_copy_verb(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-object-count': '99',
|
|
'x-account-sysmeta-quota-count': '100'})
|
|
self.app.register('GET', '/v1/a/c2/o2', HTTPOk, {
|
|
'content-length': '1000'}, b'a' * 1000)
|
|
app = copy.filter_factory({})(
|
|
account_quotas.AccountQuotaMiddleware(self.app))
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c2/o2',
|
|
environ={'REQUEST_METHOD': 'COPY',
|
|
'swift.cache': cache,
|
|
'reseller_request': True},
|
|
headers={'Destination': 'c/o'})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_not_exceed_count_quota(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk, {
|
|
'x-account-object-count': '10',
|
|
'x-account-sysmeta-quota-count': '20'})
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o',
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_invalid_count_quotas_on_object(self):
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a',
|
|
environ={'REQUEST_METHOD': 'POST',
|
|
'swift.cache': cache,
|
|
'HTTP_X_ACCOUNT_QUOTA_COUNT': 'abc',
|
|
'reseller_request': True})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 400)
|
|
self.assertEqual(self.app.calls, [])
|
|
|
|
def test_valid_count_quotas_admin(self):
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a',
|
|
environ={'REQUEST_METHOD': 'POST',
|
|
'swift.cache': cache,
|
|
'HTTP_X_ACCOUNT_QUOTA_COUNT': '100'})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 403)
|
|
self.assertEqual(self.app.calls, [])
|
|
|
|
@patch_policies
|
|
def test_valid_policy_count_quota_admin(self):
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a', environ={
|
|
'REQUEST_METHOD': 'POST',
|
|
'swift.cache': cache,
|
|
'HTTP_X_ACCOUNT_QUOTA_COUNT_POLICY_UNU': '100'})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 403)
|
|
self.assertEqual(self.app.calls, [])
|
|
|
|
def test_valid_count_quota_reseller(self):
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a',
|
|
environ={'REQUEST_METHOD': 'POST',
|
|
'swift.cache': cache,
|
|
'HTTP_X_ACCOUNT_QUOTA_COUNT': '100',
|
|
'reseller_request': True})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
self.assertEqual(self.app.calls_with_headers, [
|
|
('POST', '/v1/a', {'Host': 'localhost:80',
|
|
'X-Account-Quota-Count': '100',
|
|
'X-Account-Sysmeta-Quota-Count': '100'})])
|
|
|
|
@patch_policies
|
|
def test_valid_policy_count_quota_reseller(self):
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a', environ={
|
|
'REQUEST_METHOD': 'POST',
|
|
'swift.cache': cache,
|
|
'HTTP_X_ACCOUNT_QUOTA_COUNT_POLICY_NULO': '100',
|
|
'reseller_request': True})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
self.assertEqual(self.app.calls_with_headers, [
|
|
('POST', '/v1/a', {
|
|
'Host': 'localhost:80',
|
|
'X-Account-Sysmeta-Quota-Count-Policy-0': '100'})])
|
|
|
|
def test_delete_count_quotas(self):
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a',
|
|
environ={'REQUEST_METHOD': 'POST',
|
|
'swift.cache': cache,
|
|
'HTTP_X_ACCOUNT_QUOTA_COUNT': ''})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 403)
|
|
|
|
def test_delete_count_quotas_with_remove_header(self):
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a', environ={
|
|
'REQUEST_METHOD': 'POST',
|
|
'swift.cache': cache,
|
|
'HTTP_X_REMOVE_ACCOUNT_QUOTA_COUNT': 'True'})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 403)
|
|
|
|
def test_delete_count_quotas_reseller(self):
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
req = Request.blank('/v1/a',
|
|
environ={'REQUEST_METHOD': 'POST',
|
|
'HTTP_X_ACCOUNT_QUOTA_COUNT': '',
|
|
'reseller_request': True})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_delete_count_quotas_with_remove_header_reseller(self):
|
|
app = account_quotas.AccountQuotaMiddleware(self.app)
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a', environ={
|
|
'REQUEST_METHOD': 'POST',
|
|
'swift.cache': cache,
|
|
'HTTP_X_REMOVE_ACCOUNT_QUOTA_COUNT': 'True',
|
|
'reseller_request': True})
|
|
res = req.get_response(app)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
|
|
class AccountQuotaCopyingTestCases(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.app = FakeSwift()
|
|
self.app.register('HEAD', '/v1/a/c', HTTPOk, {
|
|
'x-backend-storage-policy-index': '1'})
|
|
self.app.register('GET', '/v1/a/c2/o2', HTTPOk, {
|
|
'content-length': '1000'})
|
|
self.aq_filter = account_quotas.filter_factory({})(self.app)
|
|
self.copy_filter = copy.filter_factory({})(self.aq_filter)
|
|
|
|
def test_exceed_bytes_quota_copy_from(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk,
|
|
[('x-account-bytes-used', '500'),
|
|
('x-account-sysmeta-quota-bytes', '1000')])
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o',
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache},
|
|
headers={'x-copy-from': '/c2/o2'})
|
|
res = req.get_response(self.copy_filter)
|
|
self.assertEqual(res.status_int, 413)
|
|
self.assertEqual(res.body, b'Upload exceeds quota.')
|
|
|
|
def test_exceed_bytes_quota_copy_verb(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk,
|
|
[('x-account-bytes-used', '500'),
|
|
('x-account-sysmeta-quota-bytes', '1000')])
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c2/o2',
|
|
environ={'REQUEST_METHOD': 'COPY',
|
|
'swift.cache': cache},
|
|
headers={'Destination': '/c/o'})
|
|
res = req.get_response(self.copy_filter)
|
|
self.assertEqual(res.status_int, 413)
|
|
self.assertEqual(res.body, b'Upload exceeds quota.')
|
|
|
|
def test_not_exceed_bytes_quota_copy_from(self):
|
|
self.app.register('PUT', '/v1/a/c/o', HTTPOk, {})
|
|
self.app.register('HEAD', '/v1/a', HTTPOk,
|
|
[('x-account-bytes-used', '0'),
|
|
('x-account-sysmeta-quota-bytes', '1000')])
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o',
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache},
|
|
headers={'x-copy-from': '/c2/o2'})
|
|
res = req.get_response(self.copy_filter)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_not_exceed_bytes_quota_copy_verb(self):
|
|
self.app.register('PUT', '/v1/a/c/o', HTTPOk, {})
|
|
self.app.register('HEAD', '/v1/a', HTTPOk,
|
|
[('x-account-bytes-used', '0'),
|
|
('x-account-sysmeta-quota-bytes', '1000')])
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c2/o2',
|
|
environ={'REQUEST_METHOD': 'COPY',
|
|
'swift.cache': cache},
|
|
headers={'Destination': '/c/o'})
|
|
res = req.get_response(self.copy_filter)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_quota_copy_from_bad_src(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk,
|
|
[('x-account-bytes-used', '0'),
|
|
('x-account-sysmeta-quota-bytes', '1000')])
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o',
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache},
|
|
headers={'x-copy-from': 'bad_path'})
|
|
res = req.get_response(self.copy_filter)
|
|
self.assertEqual(res.status_int, 412)
|
|
|
|
self.app.register('HEAD', '/v1/a', HTTPOk,
|
|
[('x-account-bytes-used', '1000'),
|
|
('x-account-sysmeta-quota-bytes', '0')])
|
|
res = req.get_response(self.copy_filter)
|
|
self.assertEqual(res.status_int, 412)
|
|
|
|
def test_exceed_bytes_count_quota_copy_from(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk,
|
|
[('x-account-object-count', '5'),
|
|
('x-account-sysmeta-quota-count', '5')])
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o',
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache},
|
|
headers={'x-copy-from': '/c2/o2'})
|
|
res = req.get_response(self.copy_filter)
|
|
self.assertEqual(res.status_int, 413)
|
|
self.assertEqual(res.body, b'Upload exceeds quota.')
|
|
|
|
def test_exceed_bytes_count_quota_copy_verb(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk,
|
|
[('x-account-object-count', '5'),
|
|
('x-account-sysmeta-quota-count', '5')])
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c2/o2',
|
|
environ={'REQUEST_METHOD': 'COPY',
|
|
'swift.cache': cache},
|
|
headers={'Destination': '/c/o'})
|
|
res = req.get_response(self.copy_filter)
|
|
self.assertEqual(res.status_int, 413)
|
|
self.assertEqual(res.body, b'Upload exceeds quota.')
|
|
|
|
def test_not_exceed_bytes_count_quota_copy_from(self):
|
|
self.app.register('PUT', '/v1/a/c/o', HTTPOk, {})
|
|
self.app.register('HEAD', '/v1/a', HTTPOk,
|
|
[('x-account-object-count', '5'),
|
|
('x-account-sysmeta-quota-count', '6')])
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o',
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache},
|
|
headers={'x-copy-from': '/c2/o2'})
|
|
res = req.get_response(self.copy_filter)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_not_exceed_bytes_count_quota_copy_verb(self):
|
|
self.app.register('PUT', '/v1/a/c/o', HTTPOk, {})
|
|
self.app.register('HEAD', '/v1/a', HTTPOk,
|
|
[('x-account-object-count', '5'),
|
|
('x-account-sysmeta-quota-count', '6')])
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c2/o2',
|
|
environ={'REQUEST_METHOD': 'COPY',
|
|
'swift.cache': cache},
|
|
headers={'Destination': '/c/o'})
|
|
res = req.get_response(self.copy_filter)
|
|
self.assertEqual(res.status_int, 200)
|
|
|
|
def test_count_quota_copy_from_bad_src(self):
|
|
self.app.register('HEAD', '/v1/a', HTTPOk,
|
|
[('x-account-object-count', '0'),
|
|
('x-account-sysmeta-quota-count', '1')])
|
|
cache = FakeCache(None)
|
|
req = Request.blank('/v1/a/c/o',
|
|
environ={'REQUEST_METHOD': 'PUT',
|
|
'swift.cache': cache},
|
|
headers={'x-copy-from': 'bad_path'})
|
|
res = req.get_response(self.copy_filter)
|
|
self.assertEqual(res.status_int, 412)
|
|
|
|
self.app.register('HEAD', '/v1/a', HTTPOk,
|
|
[('x-account-object-count', '1'),
|
|
('x-account-sysmeta-quota-count', '0')])
|
|
res = req.get_response(self.copy_filter)
|
|
self.assertEqual(res.status_int, 412)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|