diff options
Diffstat (limited to 'test/functionalnosetests/test_object.py')
-rwxr-xr-x | test/functionalnosetests/test_object.py | 211 |
1 files changed, 101 insertions, 110 deletions
diff --git a/test/functionalnosetests/test_object.py b/test/functionalnosetests/test_object.py index 3972aaf..97cd8d0 100755 --- a/test/functionalnosetests/test_object.py +++ b/test/functionalnosetests/test_object.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. @@ -55,8 +40,9 @@ class TestObject(unittest.TestCase): self.obj = uuid4().hex def put(url, token, parsed, conn): - conn.request('PUT', '%s/%s/%s' % (parsed.path, self.container, - self.obj), 'test', {'X-Auth-Token': token}) + conn.request('PUT', '%s/%s/%s' % ( + parsed.path, self.container, self.obj), 'test', + {'X-Auth-Token': token}) return check_response(conn) resp = retry(put) resp.read() @@ -182,7 +168,7 @@ class TestObject(unittest.TestCase): 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 ')) def post(url, token, parsed, conn): @@ -207,7 +193,7 @@ class TestObject(unittest.TestCase): 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 ')) def test_private_object(self): @@ -216,9 +202,9 @@ class TestObject(unittest.TestCase): # Ensure we can't access the object with the third account def get(url, token, parsed, conn): - conn.request('GET', '%s/%s/%s' % (parsed.path, self.container, - self.obj), '', - {'X-Auth-Token': token}) + conn.request('GET', '%s/%s/%s' % ( + parsed.path, self.container, self.obj), '', + {'X-Auth-Token': token}) return check_response(conn) resp = retry(get, use_account=3) resp.read() @@ -228,11 +214,11 @@ class TestObject(unittest.TestCase): shared_container = uuid4().hex def put(url, token, parsed, conn): - conn.request('PUT', '%s/%s' % (parsed.path, - shared_container), '', - {'X-Auth-Token': token, - 'X-Container-Read': swift_test_perm[2], - 'X-Container-Write': swift_test_perm[2]}) + conn.request('PUT', '%s/%s' % ( + parsed.path, shared_container), '', + {'X-Auth-Token': token, + 'X-Container-Read': swift_test_perm[2], + 'X-Container-Write': swift_test_perm[2]}) return check_response(conn) resp = retry(put) resp.read() @@ -240,13 +226,11 @@ class TestObject(unittest.TestCase): # verify third account can not copy from private container def copy(url, token, parsed, conn): - conn.request('PUT', '%s/%s/%s' % (parsed.path, - shared_container, - 'private_object'), - '', {'X-Auth-Token': token, - 'Content-Length': '0', - 'X-Copy-From': '%s/%s' % (self.container, - self.obj)}) + conn.request('PUT', '%s/%s/%s' % ( + parsed.path, shared_container, 'private_object'), '', + {'X-Auth-Token': token, + 'Content-Length': '0', + 'X-Copy-From': '%s/%s' % (self.container, self.obj)}) return check_response(conn) resp = retry(copy, use_account=3) resp.read() @@ -254,8 +238,9 @@ class TestObject(unittest.TestCase): # verify third account can write "obj1" to shared container def put(url, token, parsed, conn): - conn.request('PUT', '%s/%s/%s' % (parsed.path, shared_container, - 'obj1'), 'test', {'X-Auth-Token': token}) + conn.request('PUT', '%s/%s/%s' % ( + parsed.path, shared_container, 'obj1'), 'test', + {'X-Auth-Token': token}) return check_response(conn) resp = retry(put, use_account=3) resp.read() @@ -263,12 +248,10 @@ class TestObject(unittest.TestCase): # verify third account can copy "obj1" to shared container def copy2(url, token, parsed, conn): - conn.request('COPY', '%s/%s/%s' % (parsed.path, - shared_container, - 'obj1'), - '', {'X-Auth-Token': token, - 'Destination': '%s/%s' % (shared_container, - 'obj1')}) + conn.request('COPY', '%s/%s/%s' % ( + parsed.path, shared_container, 'obj1'), '', + {'X-Auth-Token': token, + 'Destination': '%s/%s' % (shared_container, 'obj1')}) return check_response(conn) resp = retry(copy2, use_account=3) resp.read() @@ -276,12 +259,11 @@ class TestObject(unittest.TestCase): # verify third account STILL can not copy from private container def copy3(url, token, parsed, conn): - conn.request('COPY', '%s/%s/%s' % (parsed.path, - self.container, - self.obj), - '', {'X-Auth-Token': token, - 'Destination': '%s/%s' % (shared_container, - 'private_object')}) + conn.request('COPY', '%s/%s/%s' % ( + parsed.path, self.container, self.obj), '', + {'X-Auth-Token': token, + 'Destination': '%s/%s' % (shared_container, + 'private_object')}) return check_response(conn) resp = retry(copy3, use_account=3) resp.read() @@ -289,8 +271,9 @@ class TestObject(unittest.TestCase): # clean up "obj1" def delete(url, token, parsed, conn): - conn.request('DELETE', '%s/%s/%s' % (parsed.path, shared_container, - 'obj1'), '', {'X-Auth-Token': token}) + conn.request('DELETE', '%s/%s/%s' % ( + parsed.path, shared_container, 'obj1'), '', + {'X-Auth-Token': token}) return check_response(conn) resp = retry(delete) resp.read() @@ -316,8 +299,8 @@ class TestObject(unittest.TestCase): # Upload the first set of segments def put(url, token, parsed, conn, objnum): - conn.request('PUT', '%s/%s/segments1/%s' % (parsed.path, - self.container, str(objnum)), segments1[objnum], + conn.request('PUT', '%s/%s/segments1/%s' % ( + parsed.path, self.container, str(objnum)), segments1[objnum], {'X-Auth-Token': token}) return check_response(conn) for objnum in xrange(len(segments1)): @@ -327,10 +310,11 @@ class TestObject(unittest.TestCase): # Upload the manifest def put(url, token, parsed, conn): - conn.request('PUT', '%s/%s/manifest' % (parsed.path, - self.container), '', {'X-Auth-Token': token, - 'X-Object-Manifest': '%s/segments1/' % self.container, - 'Content-Type': 'text/jibberish', 'Content-Length': '0'}) + conn.request('PUT', '%s/%s/manifest' % ( + parsed.path, self.container), '', { + 'X-Auth-Token': token, + 'X-Object-Manifest': '%s/segments1/' % self.container, + 'Content-Type': 'text/jibberish', 'Content-Length': '0'}) return check_response(conn) resp = retry(put) resp.read() @@ -338,8 +322,8 @@ class TestObject(unittest.TestCase): # Get the manifest (should get all the segments as the body) def get(url, token, parsed, conn): - conn.request('GET', '%s/%s/manifest' % (parsed.path, - self.container), '', {'X-Auth-Token': token}) + conn.request('GET', '%s/%s/manifest' % ( + parsed.path, self.container), '', {'X-Auth-Token': token}) return check_response(conn) resp = retry(get) self.assertEquals(resp.read(), ''.join(segments1)) @@ -348,9 +332,9 @@ class TestObject(unittest.TestCase): # Get with a range at the start of the second segment def get(url, token, parsed, conn): - conn.request('GET', '%s/%s/manifest' % (parsed.path, - self.container), '', {'X-Auth-Token': token, 'Range': - 'bytes=3-'}) + conn.request('GET', '%s/%s/manifest' % ( + parsed.path, self.container), '', { + 'X-Auth-Token': token, 'Range': 'bytes=3-'}) return check_response(conn) resp = retry(get) self.assertEquals(resp.read(), ''.join(segments1[1:])) @@ -358,9 +342,9 @@ class TestObject(unittest.TestCase): # Get with a range in the middle of the second segment def get(url, token, parsed, conn): - conn.request('GET', '%s/%s/manifest' % (parsed.path, - self.container), '', {'X-Auth-Token': token, 'Range': - 'bytes=5-'}) + conn.request('GET', '%s/%s/manifest' % ( + parsed.path, self.container), '', { + 'X-Auth-Token': token, 'Range': 'bytes=5-'}) return check_response(conn) resp = retry(get) self.assertEquals(resp.read(), ''.join(segments1)[5:]) @@ -368,9 +352,9 @@ class TestObject(unittest.TestCase): # Get with a full start and stop range def get(url, token, parsed, conn): - conn.request('GET', '%s/%s/manifest' % (parsed.path, - self.container), '', {'X-Auth-Token': token, 'Range': - 'bytes=5-10'}) + conn.request('GET', '%s/%s/manifest' % ( + parsed.path, self.container), '', { + 'X-Auth-Token': token, 'Range': 'bytes=5-10'}) return check_response(conn) resp = retry(get) self.assertEquals(resp.read(), ''.join(segments1)[5:11]) @@ -378,8 +362,8 @@ class TestObject(unittest.TestCase): # Upload the second set of segments def put(url, token, parsed, conn, objnum): - conn.request('PUT', '%s/%s/segments2/%s' % (parsed.path, - self.container, str(objnum)), segments2[objnum], + conn.request('PUT', '%s/%s/segments2/%s' % ( + parsed.path, self.container, str(objnum)), segments2[objnum], {'X-Auth-Token': token}) return check_response(conn) for objnum in xrange(len(segments2)): @@ -389,8 +373,8 @@ class TestObject(unittest.TestCase): # Get the manifest (should still be the first segments of course) def get(url, token, parsed, conn): - conn.request('GET', '%s/%s/manifest' % (parsed.path, - self.container), '', {'X-Auth-Token': token}) + conn.request('GET', '%s/%s/manifest' % ( + parsed.path, self.container), '', {'X-Auth-Token': token}) return check_response(conn) resp = retry(get) self.assertEquals(resp.read(), ''.join(segments1)) @@ -398,10 +382,11 @@ class TestObject(unittest.TestCase): # Update the manifest def put(url, token, parsed, conn): - conn.request('PUT', '%s/%s/manifest' % (parsed.path, - self.container), '', {'X-Auth-Token': token, - 'X-Object-Manifest': '%s/segments2/' % self.container, - 'Content-Length': '0'}) + conn.request('PUT', '%s/%s/manifest' % ( + parsed.path, self.container), '', { + 'X-Auth-Token': token, + 'X-Object-Manifest': '%s/segments2/' % self.container, + 'Content-Length': '0'}) return check_response(conn) resp = retry(put) resp.read() @@ -409,8 +394,8 @@ class TestObject(unittest.TestCase): # Get the manifest (should be the second set of segments now) def get(url, token, parsed, conn): - conn.request('GET', '%s/%s/manifest' % (parsed.path, - self.container), '', {'X-Auth-Token': token}) + conn.request('GET', '%s/%s/manifest' % ( + parsed.path, self.container), '', {'X-Auth-Token': token}) return check_response(conn) resp = retry(get) self.assertEquals(resp.read(), ''.join(segments2)) @@ -420,8 +405,8 @@ class TestObject(unittest.TestCase): # Ensure we can't access the manifest with the third account def get(url, token, parsed, conn): - conn.request('GET', '%s/%s/manifest' % (parsed.path, - self.container), '', {'X-Auth-Token': token}) + conn.request('GET', '%s/%s/manifest' % ( + parsed.path, self.container), '', {'X-Auth-Token': token}) return check_response(conn) resp = retry(get, use_account=3) resp.read() @@ -430,8 +415,8 @@ class TestObject(unittest.TestCase): # Grant access to the third account def post(url, token, parsed, conn): conn.request('POST', '%s/%s' % (parsed.path, self.container), - '', {'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() @@ -439,8 +424,8 @@ class TestObject(unittest.TestCase): # The third account should be able to get the manifest now def get(url, token, parsed, conn): - conn.request('GET', '%s/%s/manifest' % (parsed.path, - self.container), '', {'X-Auth-Token': token}) + conn.request('GET', '%s/%s/manifest' % ( + parsed.path, self.container), '', {'X-Auth-Token': token}) return check_response(conn) resp = retry(get, use_account=3) self.assertEquals(resp.read(), ''.join(segments2)) @@ -459,8 +444,8 @@ class TestObject(unittest.TestCase): # Upload the third set of segments in the other container def put(url, token, parsed, conn, objnum): - conn.request('PUT', '%s/%s/segments3/%s' % (parsed.path, - acontainer, str(objnum)), segments3[objnum], + conn.request('PUT', '%s/%s/segments3/%s' % ( + parsed.path, acontainer, str(objnum)), segments3[objnum], {'X-Auth-Token': token}) return check_response(conn) for objnum in xrange(len(segments3)): @@ -470,10 +455,11 @@ class TestObject(unittest.TestCase): # Update the manifest def put(url, token, parsed, conn): - conn.request('PUT', '%s/%s/manifest' % (parsed.path, - self.container), '', {'X-Auth-Token': token, - 'X-Object-Manifest': '%s/segments3/' % acontainer, - 'Content-Length': '0'}) + conn.request('PUT', '%s/%s/manifest' % ( + parsed.path, self.container), '', + {'X-Auth-Token': token, + 'X-Object-Manifest': '%s/segments3/' % acontainer, + 'Content-Length': '0'}) return check_response(conn) resp = retry(put) resp.read() @@ -481,8 +467,8 @@ class TestObject(unittest.TestCase): # Get the manifest to ensure it's the third set of segments def get(url, token, parsed, conn): - conn.request('GET', '%s/%s/manifest' % (parsed.path, - self.container), '', {'X-Auth-Token': token}) + conn.request('GET', '%s/%s/manifest' % ( + parsed.path, self.container), '', {'X-Auth-Token': token}) return check_response(conn) resp = retry(get) self.assertEquals(resp.read(), ''.join(segments3)) @@ -495,8 +481,8 @@ class TestObject(unittest.TestCase): # manifest itself is not). def get(url, token, parsed, conn): - conn.request('GET', '%s/%s/manifest' % (parsed.path, - self.container), '', {'X-Auth-Token': token}) + conn.request('GET', '%s/%s/manifest' % ( + parsed.path, self.container), '', {'X-Auth-Token': token}) return check_response(conn) resp = retry(get, use_account=3) resp.read() @@ -505,8 +491,8 @@ class TestObject(unittest.TestCase): # Grant access to the third account def post(url, token, parsed, conn): conn.request('POST', '%s/%s' % (parsed.path, acontainer), - '', {'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() @@ -514,8 +500,8 @@ class TestObject(unittest.TestCase): # The third account should be able to get the manifest now def get(url, token, parsed, conn): - conn.request('GET', '%s/%s/manifest' % (parsed.path, - self.container), '', {'X-Auth-Token': token}) + conn.request('GET', '%s/%s/manifest' % ( + parsed.path, self.container), '', {'X-Auth-Token': token}) return check_response(conn) resp = retry(get, use_account=3) self.assertEquals(resp.read(), ''.join(segments3)) @@ -523,7 +509,8 @@ class TestObject(unittest.TestCase): # Delete the manifest def delete(url, token, parsed, conn, objnum): - conn.request('DELETE', '%s/%s/manifest' % (parsed.path, + conn.request('DELETE', '%s/%s/manifest' % ( + parsed.path, self.container), '', {'X-Auth-Token': token}) return check_response(conn) resp = retry(delete, objnum) @@ -532,8 +519,9 @@ class TestObject(unittest.TestCase): # Delete the third set of segments def delete(url, token, parsed, conn, objnum): - conn.request('DELETE', '%s/%s/segments3/%s' % (parsed.path, - acontainer, str(objnum)), '', {'X-Auth-Token': token}) + conn.request('DELETE', '%s/%s/segments3/%s' % ( + parsed.path, acontainer, str(objnum)), '', + {'X-Auth-Token': token}) return check_response(conn) for objnum in xrange(len(segments3)): resp = retry(delete, objnum) @@ -542,8 +530,9 @@ class TestObject(unittest.TestCase): # Delete the second set of segments def delete(url, token, parsed, conn, objnum): - conn.request('DELETE', '%s/%s/segments2/%s' % (parsed.path, - self.container, str(objnum)), '', {'X-Auth-Token': token}) + conn.request('DELETE', '%s/%s/segments2/%s' % ( + parsed.path, self.container, str(objnum)), '', + {'X-Auth-Token': token}) return check_response(conn) for objnum in xrange(len(segments2)): resp = retry(delete, objnum) @@ -552,8 +541,9 @@ class TestObject(unittest.TestCase): # Delete the first set of segments def delete(url, token, parsed, conn, objnum): - conn.request('DELETE', '%s/%s/segments1/%s' % (parsed.path, - self.container, str(objnum)), '', {'X-Auth-Token': token}) + conn.request('DELETE', '%s/%s/segments1/%s' % ( + parsed.path, self.container, str(objnum)), '', + {'X-Auth-Token': token}) return check_response(conn) for objnum in xrange(len(segments1)): resp = retry(delete, objnum) @@ -563,7 +553,7 @@ class TestObject(unittest.TestCase): # Delete the extra container def delete(url, token, parsed, conn): conn.request('DELETE', '%s/%s' % (parsed.path, acontainer), '', - {'X-Auth-Token': token}) + {'X-Auth-Token': token}) return check_response(conn) resp = retry(delete) resp.read() @@ -574,8 +564,8 @@ class TestObject(unittest.TestCase): raise SkipTest def put(url, token, parsed, conn): - conn.request('PUT', '%s/%s/hi' % (parsed.path, - self.container), 'there', {'X-Auth-Token': token}) + conn.request('PUT', '%s/%s/hi' % (parsed.path, self.container), + 'there', {'X-Auth-Token': token}) return check_response(conn) resp = retry(put) resp.read() @@ -583,7 +573,7 @@ class TestObject(unittest.TestCase): def delete(url, token, parsed, conn): conn.request('DELETE', '%s/%s/hi' % (parsed.path, self.container), - '', {'X-Auth-Token': token}) + '', {'X-Auth-Token': token}) return check_response(conn) resp = retry(delete) resp.read() @@ -596,7 +586,8 @@ class TestObject(unittest.TestCase): raise SkipTest def put(url, token, parsed, conn): - conn.request('PUT', '%s/%s/abc%%00def' % (parsed.path, + conn.request('PUT', '%s/%s/abc%%00def' % ( + parsed.path, self.container), 'test', {'X-Auth-Token': token}) return check_response(conn) resp = retry(put) |