diff options
Diffstat (limited to 'test/functionalnosetests/test_container.py')
| -rwxr-xr-x | test/functionalnosetests/test_container.py | 149 |
1 files changed, 103 insertions, 46 deletions
diff --git a/test/functionalnosetests/test_container.py b/test/functionalnosetests/test_container.py index af78a7a..15f7fc1 100755 --- a/test/functionalnosetests/test_container.py +++ b/test/functionalnosetests/test_container.py @@ -1,21 +1,6 @@ #!/usr/bin/python -# Copyright (c) 2010-2013 OpenStack, LLC. -# -# 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. - -# Copyright (c) 2013 Red Hat, Inc. +# Copyright (c) 2010-2012 OpenStack Foundation # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -39,7 +24,7 @@ from swift.common.constraints import MAX_META_COUNT, MAX_META_NAME_LENGTH, \ MAX_META_OVERALL_SIZE, MAX_META_VALUE_LENGTH from swift_testing import check_response, retry, skip, skip2, skip3, \ - swift_test_perm, web_front_end + swift_test_perm, web_front_end class TestContainer(unittest.TestCase): @@ -48,10 +33,12 @@ class TestContainer(unittest.TestCase): if skip: raise SkipTest self.name = uuid4().hex + def put(url, token, parsed, conn): conn.request('PUT', parsed.path + '/' + self.name, '', {'X-Auth-Token': token}) return check_response(conn) + resp = retry(put) resp.read() self.assertEquals(resp.status, 201) @@ -59,15 +46,18 @@ class TestContainer(unittest.TestCase): def tearDown(self): if skip: raise SkipTest + def get(url, token, parsed, conn): conn.request('GET', parsed.path + '/' + self.name + '?format=json', '', {'X-Auth-Token': token}) return check_response(conn) + def delete(url, token, parsed, conn, obj): conn.request('DELETE', '/'.join([parsed.path, self.name, obj['name']]), '', {'X-Auth-Token': token}) return check_response(conn) + while True: resp = retry(get) body = resp.read() @@ -79,10 +69,12 @@ class TestContainer(unittest.TestCase): resp = retry(delete, obj) resp.read() self.assertEquals(resp.status, 204) + def delete(url, token, parsed, conn): conn.request('DELETE', parsed.path + '/' + self.name, '', {'X-Auth-Token': token}) return check_response(conn) + resp = retry(delete) resp.read() self.assertEquals(resp.status, 204) @@ -90,14 +82,17 @@ class TestContainer(unittest.TestCase): def test_multi_metadata(self): if skip: raise SkipTest + def post(url, token, parsed, conn, name, value): conn.request('POST', parsed.path + '/' + self.name, '', - {'X-Auth-Token': token, name: value}) + {'X-Auth-Token': token, name: value}) return check_response(conn) + def head(url, token, parsed, conn): conn.request('HEAD', parsed.path + '/' + self.name, '', {'X-Auth-Token': token}) return check_response(conn) + resp = retry(post, 'X-Container-Meta-One', '1') resp.read() self.assertEquals(resp.status, 204) @@ -154,27 +149,33 @@ class TestContainer(unittest.TestCase): resp.read() self.assert_(resp.status in (200, 204), resp.status) self.assertEquals(resp.getheader(uni_key.encode('utf-8')), - uni_value.encode('utf-8')) + uni_value.encode('utf-8')) def test_PUT_metadata(self): if skip: raise SkipTest + def put(url, token, parsed, conn, name, value): conn.request('PUT', parsed.path + '/' + name, '', - {'X-Auth-Token': token, 'X-Container-Meta-Test': value}) + {'X-Auth-Token': token, + 'X-Container-Meta-Test': value}) return check_response(conn) + def head(url, token, parsed, conn, name): conn.request('HEAD', parsed.path + '/' + name, '', {'X-Auth-Token': token}) return check_response(conn) + def get(url, token, parsed, conn, name): conn.request('GET', parsed.path + '/' + name, '', {'X-Auth-Token': token}) return check_response(conn) + def delete(url, token, parsed, conn, name): conn.request('DELETE', parsed.path + '/' + name, '', {'X-Auth-Token': token}) return check_response(conn) + name = uuid4().hex resp = retry(put, name, 'Value') resp.read() @@ -210,18 +211,23 @@ class TestContainer(unittest.TestCase): def test_POST_metadata(self): if skip: raise SkipTest + def post(url, token, parsed, conn, value): conn.request('POST', parsed.path + '/' + self.name, '', - {'X-Auth-Token': token, 'X-Container-Meta-Test': value}) + {'X-Auth-Token': token, + 'X-Container-Meta-Test': value}) return check_response(conn) + def head(url, token, parsed, conn): conn.request('HEAD', parsed.path + '/' + self.name, '', {'X-Auth-Token': token}) return check_response(conn) + def get(url, token, parsed, conn): conn.request('GET', parsed.path + '/' + self.name, '', {'X-Auth-Token': token}) return check_response(conn) + resp = retry(head) resp.read() self.assert_(resp.status in (200, 204), resp.status) @@ -245,26 +251,31 @@ class TestContainer(unittest.TestCase): def test_PUT_bad_metadata(self): if skip: raise SkipTest + def put(url, token, parsed, conn, name, extra_headers): headers = {'X-Auth-Token': token} headers.update(extra_headers) conn.request('PUT', parsed.path + '/' + name, '', headers) return check_response(conn) + def delete(url, token, parsed, conn, name): conn.request('DELETE', parsed.path + '/' + name, '', {'X-Auth-Token': token}) return check_response(conn) + name = uuid4().hex - resp = retry(put, name, - {'X-Container-Meta-' + ('k' * MAX_META_NAME_LENGTH): 'v'}) + resp = retry( + put, name, + {'X-Container-Meta-' + ('k' * MAX_META_NAME_LENGTH): 'v'}) resp.read() self.assertEquals(resp.status, 201) resp = retry(delete, name) resp.read() self.assertEquals(resp.status, 204) name = uuid4().hex - resp = retry(put, name, - {'X-Container-Meta-' + ('k' * (MAX_META_NAME_LENGTH + 1)): 'v'}) + resp = retry( + put, name, + {'X-Container-Meta-' + ('k' * (MAX_META_NAME_LENGTH + 1)): 'v'}) resp.read() self.assertEquals(resp.status, 400) resp = retry(delete, name) @@ -272,16 +283,18 @@ class TestContainer(unittest.TestCase): self.assertEquals(resp.status, 404) name = uuid4().hex - resp = retry(put, name, - {'X-Container-Meta-Too-Long': 'k' * MAX_META_VALUE_LENGTH}) + resp = retry( + put, name, + {'X-Container-Meta-Too-Long': 'k' * MAX_META_VALUE_LENGTH}) resp.read() self.assertEquals(resp.status, 201) resp = retry(delete, name) resp.read() self.assertEquals(resp.status, 204) name = uuid4().hex - resp = retry(put, name, - {'X-Container-Meta-Too-Long': 'k' * (MAX_META_VALUE_LENGTH + 1)}) + resp = retry( + put, name, + {'X-Container-Meta-Too-Long': 'k' * (MAX_META_VALUE_LENGTH + 1)}) resp.read() self.assertEquals(resp.status, 400) resp = retry(delete, name) @@ -340,26 +353,32 @@ class TestContainer(unittest.TestCase): def test_POST_bad_metadata(self): if skip: raise SkipTest + def post(url, token, parsed, conn, extra_headers): headers = {'X-Auth-Token': token} headers.update(extra_headers) conn.request('POST', parsed.path + '/' + self.name, '', headers) return check_response(conn) - resp = retry(post, - {'X-Container-Meta-' + ('k' * MAX_META_NAME_LENGTH): 'v'}) + + resp = retry( + post, + {'X-Container-Meta-' + ('k' * MAX_META_NAME_LENGTH): 'v'}) resp.read() self.assertEquals(resp.status, 204) - resp = retry(post, - {'X-Container-Meta-' + ('k' * (MAX_META_NAME_LENGTH + 1)): 'v'}) + resp = retry( + post, + {'X-Container-Meta-' + ('k' * (MAX_META_NAME_LENGTH + 1)): 'v'}) resp.read() self.assertEquals(resp.status, 400) - resp = retry(post, - {'X-Container-Meta-Too-Long': 'k' * MAX_META_VALUE_LENGTH}) + resp = retry( + post, + {'X-Container-Meta-Too-Long': 'k' * MAX_META_VALUE_LENGTH}) resp.read() self.assertEquals(resp.status, 204) - resp = retry(post, - {'X-Container-Meta-Too-Long': 'k' * (MAX_META_VALUE_LENGTH + 1)}) + resp = retry( + post, + {'X-Container-Meta-Too-Long': 'k' * (MAX_META_VALUE_LENGTH + 1)}) resp.read() self.assertEquals(resp.status, 400) @@ -399,36 +418,42 @@ class TestContainer(unittest.TestCase): def test_public_container(self): if skip: raise SkipTest + def get(url, token, parsed, conn): conn.request('GET', parsed.path + '/' + self.name) return check_response(conn) + try: resp = retry(get) raise Exception('Should not have been able to GET') - except Exception, err: + except Exception as err: self.assert_(str(err).startswith('No result after '), err) + def post(url, token, parsed, conn): conn.request('POST', parsed.path + '/' + self.name, '', {'X-Auth-Token': token, 'X-Container-Read': '.r:*,.rlistings'}) return check_response(conn) + resp = retry(post) resp.read() self.assertEquals(resp.status, 204) resp = retry(get) resp.read() self.assertEquals(resp.status, 204) + def post(url, token, parsed, conn): conn.request('POST', parsed.path + '/' + self.name, '', {'X-Auth-Token': token, 'X-Container-Read': ''}) return check_response(conn) + resp = retry(post) resp.read() self.assertEquals(resp.status, 204) try: resp = retry(get) raise Exception('Should not have been able to GET') - except Exception, err: + except Exception as err: self.assert_(str(err).startswith('No result after '), err) def test_cross_account_container(self): @@ -436,27 +461,34 @@ class TestContainer(unittest.TestCase): raise SkipTest # Obtain the first account's string first_account = ['unknown'] + def get1(url, token, parsed, conn): first_account[0] = parsed.path conn.request('HEAD', parsed.path + '/' + self.name, '', {'X-Auth-Token': token}) return check_response(conn) + resp = retry(get1) resp.read() + # Ensure we can't access the container with the second account def get2(url, token, parsed, conn): conn.request('GET', first_account[0] + '/' + self.name, '', {'X-Auth-Token': token}) return check_response(conn) + resp = retry(get2, use_account=2) resp.read() self.assertEquals(resp.status, 403) + # Make the container accessible by the second account def post(url, token, parsed, conn): conn.request('POST', parsed.path + '/' + self.name, '', - {'X-Auth-Token': token, 'X-Container-Read': swift_test_perm[1], - 'X-Container-Write': swift_test_perm[1]}) + {'X-Auth-Token': token, + 'X-Container-Read': swift_test_perm[1], + 'X-Container-Write': swift_test_perm[1]}) return check_response(conn) + resp = retry(post) resp.read() self.assertEquals(resp.status, 204) @@ -464,12 +496,14 @@ class TestContainer(unittest.TestCase): resp = retry(get2, use_account=2) resp.read() self.assertEquals(resp.status, 204) + # Make the container private again def post(url, token, parsed, conn): conn.request('POST', parsed.path + '/' + self.name, '', {'X-Auth-Token': token, 'X-Container-Read': '', 'X-Container-Write': ''}) return check_response(conn) + resp = retry(post) resp.read() self.assertEquals(resp.status, 204) @@ -483,27 +517,33 @@ class TestContainer(unittest.TestCase): raise SkipTest # Obtain the first account's string first_account = ['unknown'] + def get1(url, token, parsed, conn): first_account[0] = parsed.path conn.request('HEAD', parsed.path + '/' + self.name, '', {'X-Auth-Token': token}) return check_response(conn) + resp = retry(get1) resp.read() + # Ensure we can't access the container with the second account def get2(url, token, parsed, conn): conn.request('GET', first_account[0] + '/' + self.name, '', {'X-Auth-Token': token}) return check_response(conn) + resp = retry(get2, use_account=2) resp.read() self.assertEquals(resp.status, 403) + # Make the container completely public def post(url, token, parsed, conn): conn.request('POST', parsed.path + '/' + self.name, '', {'X-Auth-Token': token, 'X-Container-Read': '.r:*,.rlistings'}) return check_response(conn) + resp = retry(post) resp.read() self.assertEquals(resp.status, 204) @@ -511,20 +551,24 @@ class TestContainer(unittest.TestCase): resp = retry(get2, use_account=2) resp.read() self.assertEquals(resp.status, 204) + # But we shouldn't be able to write with the second account def put2(url, token, parsed, conn): conn.request('PUT', first_account[0] + '/' + self.name + '/object', 'test object', {'X-Auth-Token': token}) return check_response(conn) + resp = retry(put2, use_account=2) resp.read() self.assertEquals(resp.status, 403) + # Now make the container also writeable by the second account def post(url, token, parsed, conn): conn.request('POST', parsed.path + '/' + self.name, '', - {'X-Auth-Token': token, - 'X-Container-Write': swift_test_perm[1]}) + {'X-Auth-Token': token, + 'X-Container-Write': swift_test_perm[1]}) return check_response(conn) + resp = retry(post) resp.read() self.assertEquals(resp.status, 204) @@ -542,26 +586,33 @@ class TestContainer(unittest.TestCase): raise SkipTest # Obtain the first account's string first_account = ['unknown'] + def get1(url, token, parsed, conn): first_account[0] = parsed.path conn.request('HEAD', parsed.path + '/' + self.name, '', {'X-Auth-Token': token}) return check_response(conn) + resp = retry(get1) resp.read() + # Ensure we can't access the container with the third account def get3(url, token, parsed, conn): conn.request('GET', first_account[0] + '/' + self.name, '', {'X-Auth-Token': token}) return check_response(conn) + resp = retry(get3, use_account=3) resp.read() self.assertEquals(resp.status, 403) + # Make the container accessible by the third account def post(url, token, parsed, conn): conn.request('POST', parsed.path + '/' + self.name, '', - {'X-Auth-Token': token, 'X-Container-Read': swift_test_perm[2]}) + {'X-Auth-Token': token, + 'X-Container-Read': swift_test_perm[2]}) return check_response(conn) + resp = retry(post) resp.read() self.assertEquals(resp.status, 204) @@ -569,20 +620,24 @@ class TestContainer(unittest.TestCase): resp = retry(get3, use_account=3) resp.read() self.assertEquals(resp.status, 204) + # But we shouldn't be able to write with the third account def put3(url, token, parsed, conn): conn.request('PUT', first_account[0] + '/' + self.name + '/object', 'test object', {'X-Auth-Token': token}) return check_response(conn) + resp = retry(put3, use_account=3) resp.read() self.assertEquals(resp.status, 403) + # Now make the container also writeable by the third account def post(url, token, parsed, conn): conn.request('POST', parsed.path + '/' + self.name, '', {'X-Auth-Token': token, 'X-Container-Write': swift_test_perm[2]}) return check_response(conn) + resp = retry(post) resp.read() self.assertEquals(resp.status, 204) @@ -601,9 +656,10 @@ class TestContainer(unittest.TestCase): def put(url, token, parsed, conn): container_name = 'X' * 2048 - conn.request('PUT', '%s/%s' % (parsed.path, - container_name), 'there', {'X-Auth-Token': token}) + conn.request('PUT', '%s/%s' % (parsed.path, container_name), + 'there', {'X-Auth-Token': token}) return check_response(conn) + resp = retry(put) resp.read() self.assertEquals(resp.status, 400) @@ -618,6 +674,7 @@ class TestContainer(unittest.TestCase): conn.request('PUT', '%s/abc%%00def' % parsed.path, '', {'X-Auth-Token': token}) return check_response(conn) + resp = retry(put) if (web_front_end == 'apache2'): self.assertEquals(resp.status, 404) |