openstack/swift
Code Issues Proposed changes

tests: fix FakeSwift HEAD with query param

Browse Source 
The existing FakeSwift implementation already supports using registered
GET responses for GET requests with a query param.  It also supports
using registered GET responses for HEAD requests (if they either both
had the exact SAME matching query params, or both did not have ANY query
params).  But it did not support using registered GET responses w/o
query params for HEAD requests with a query param, even though a GET
with the same query param would work.

This change makes it a little more consistent: any client or test that
makes a GET request, should be able to make a similar HEAD request and
expect consistent response headers and status.

This test infra improvement is needed as we're going to be extending
test_slo with a bunch of tests that assert consistent response headers
for both GET and HEAD requests w/ new query params.

Change-Id: Idb4020fdeee87a9164312dc9647ab0820b098ff8
This commit is contained in:
indianwhocodes 2023-07-24 10:45:21 -07:00 committed by Alistair Coles
parent 9e065e2d23
commit dab7192e1e
5 changed files with 149 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
test/unit/common/middleware/helpers.py
View File

@ -115,6 +115,40 @@ class FakeSwift(object):
(method, path),))
return resp
def _select_response(self, env, method, path):
# in some cases we can borrow different registered response
# ... the order is brittle and significant
preferences = [(method, path)]
if env.get('QUERY_STRING'):
# we can always reuse response w/o query string
preferences.append((method, env['PATH_INFO']))
if method == 'HEAD':
# any path suitable for GET always works for HEAD
# N.B. list(preferences) to avoid iter+modify/sigkill
preferences.extend(('GET', p) for _, p in list(preferences))
for m, p in preferences:
try:
resp_class, headers, body = self._find_response(m, p)
except KeyError:
pass
else:
break
else:
# special case for re-reading an uploaded file
# ... uploaded is only objects and always raw path
if method in ('GET', 'HEAD') and path in self.uploaded:
resp_class = swob.HTTPOk
headers, body = self.uploaded[path]
else:
raise KeyError("Didn't find %r in allowed responses" % (
(method, path),))
if method == 'HEAD':
# HEAD resp never has body
body = None
return resp_class, HeaderKeyDict(headers), body
def __call__(self, env, start_response):
if self.can_ignore_range:
# we might pop off the Range header
@ -139,26 +173,7 @@ class FakeSwift(object):
self.swift_sources.append(env.get('swift.source'))
self.txn_ids.append(env.get('swift.trans_id'))
try:
resp_class, raw_headers, body = self._find_response(method, path)
headers = HeaderKeyDict(raw_headers)
except KeyError:
if (env.get('QUERY_STRING')
and (method, env['PATH_INFO']) in self._responses):
resp_class, raw_headers, body = self._find_response(
method, env['PATH_INFO'])
headers = HeaderKeyDict(raw_headers)
elif method == 'HEAD' and ('GET', path) in self._responses:
resp_class, raw_headers, body = self._find_response(
'GET', path)
body = None
headers = HeaderKeyDict(raw_headers)
elif method == 'GET' and obj and path in self.uploaded:
resp_class = swob.HTTPOk
headers, body = self.uploaded[path]
else:
raise KeyError("Didn't find %r in allowed responses" % (
(method, path),))
resp_class, headers, body = self._select_response(env, method, path)
ignore_range_meta = req.headers.get(
'x-backend-ignore-range-if-metadata-present')
@ -182,9 +197,10 @@ class FakeSwift(object):
headers.setdefault('Content-Length', len(req_body))
# keep it for subsequent GET requests later
self.uploaded[path] = (dict(req.headers), req_body)
resp_headers = dict(req.headers)
if "CONTENT_TYPE" in env:
self.uploaded[path][0]['Content-Type'] = env["CONTENT_TYPE"]
resp_headers['Content-Type'] = env["CONTENT_TYPE"]
self.uploaded[path] = (resp_headers, req_body)
# simulate object POST
elif method == 'POST' and obj:

24
test/unit/common/middleware/test_dlo.py
View File

@ -333,6 +333,16 @@ class TestDloGetManifest(DloTestCase):
self.assertEqual(body, b'manifest-contents')
self.assertFalse(self.app.unread_requests)
# HEAD query param worked, since GET with query param is registered
req = swob.Request.blank(
'/v1/AUTH_test/mancon/manifest',
environ={'REQUEST_METHOD': 'HEAD',
'QUERY_STRING': 'multipart-manifest=get'})
status, headers, body = self.call_dlo(req)
headers = HeaderKeyDict(headers)
self.assertEqual(headers["Etag"], "manifest-etag")
self.assertEqual(body, b'')
def test_error_passthrough(self):
self.app.register(
'GET', '/v1/AUTH_test/gone/404ed',
@ -342,6 +352,20 @@ class TestDloGetManifest(DloTestCase):
status, headers, body = self.call_dlo(req)
self.assertEqual(status, '404 Not Found')
# ... and multipart-manifest=get also returns registered 404 response
req = swob.Request.blank('/v1/AUTH_test/gone/404ed',
method='GET',
params={'multipart-manifest': 'get'})
status, headers, body = self.call_dlo(req)
self.assertEqual(status, '404 Not Found')
# HEAD with same params find same registered GET
req = swob.Request.blank('/v1/AUTH_test/gone/404ed',
method='HEAD',
params={'multipart-manifest': 'get'})
status, headers, body = self.call_dlo(req)
self.assertEqual(status, '404 Not Found')
def test_get_range(self):
req = swob.Request.blank('/v1/AUTH_test/mancon/manifest',
environ={'REQUEST_METHOD': 'GET'},

47
test/unit/common/middleware/test_object_versioning.py
View File

@ -114,6 +114,7 @@ class ObjectVersioningBaseTestCase(unittest.TestCase):
self.expected_unread_requests)
def call_ov(self, req):
# authorized gets reset everytime
self.authorized = []
def authorize(req):
@ -1903,7 +1904,7 @@ class ObjectVersioningTestVersionAPI(ObjectVersioningBaseTestCase):
status, headers, body = self.call_ov(req)
self.assertEqual(status, '400 Bad Request')
def test_GET(self):
def test_GET_and_HEAD(self):
self.app.register(
'GET',
self.build_versions_path(obj='o', version='9999999939.99999'),
@ -1917,6 +1918,16 @@ class ObjectVersioningTestVersionAPI(ObjectVersioningBaseTestCase):
self.assertIn(('X-Object-Version-Id', '0000000060.00000'),
headers)
self.assertEqual(b'foobar', body)
# HEAD with same params find same registered GET
req = Request.blank(
'/v1/a/c/o', method='HEAD',
environ={'swift.cache': self.cache_version_on},
params={'version-id': '0000000060.00000'})
status, headers, body = self.call_ov(req)
self.assertEqual(status, '200 OK')
self.assertIn(('X-Object-Version-Id', '0000000060.00000'),
headers)
self.assertEqual(b'', body)
def test_GET_404(self):
self.app.register(
@ -1957,12 +1968,25 @@ class ObjectVersioningTestVersionAPI(ObjectVersioningBaseTestCase):
'/v1/a/c/o', method='GET',
environ={'swift.cache': self.cache_version_on},
params={'version-id': 'null'})
# N.B. GET w/ query param found registered raw_path GET
status, headers, body = self.call_ov(req)
self.assertEqual(status, '200 OK')
self.assertEqual(1, len(self.authorized))
self.assertRequestEqual(req, self.authorized[0])
self.assertEqual(1, len(self.app.calls))
self.assertIn(('X-Object-Version-Id', 'null'), headers)
self.assertEqual(b'foobar', body)
# and HEAD w/ same params finds same registered GET
req = Request.blank(
'/v1/a/c/o?version-id=null', method='HEAD',
environ={'swift.cache': self.cache_version_on})
status, headers, body = self.call_ov(req)
self.assertEqual(status, '200 OK')
self.assertEqual(len(self.authorized), 1)
self.assertRequestEqual(req, self.authorized[0])
self.assertEqual(2, len(self.app.calls))
self.assertIn(('X-Object-Version-Id', 'null'), headers)
self.assertEqual(b'', body)
def test_GET_null_id_versioned_obj(self):
self.app.register(
@ -1993,8 +2017,22 @@ class ObjectVersioningTestVersionAPI(ObjectVersioningBaseTestCase):
status, headers, body = self.call_ov(req)
self.assertEqual(status, '404 Not Found')
self.assertEqual(1, len(self.authorized))
self.assertRequestEqual(req, self.authorized[0])
self.assertEqual(1, len(self.app.calls))
self.assertNotIn(('X-Object-Version-Id', 'null'), headers)
# and HEAD w/ same params finds same registered GET
# we have test_HEAD_null_id, the following test is meant to illustrate
# that FakeSwift works for HEADs even if only GETs are registered.
req = Request.blank(
'/v1/a/c/o', method='HEAD',
environ={'swift.cache': self.cache_version_on},
params={'version-id': 'null'})
status, headers, body = self.call_ov(req)
self.assertEqual(status, '404 Not Found')
self.assertEqual(1, len(self.authorized))
self.assertRequestEqual(req, self.authorized[0])
self.assertEqual(2, len(self.app.calls))
self.assertNotIn(('X-Object-Version-Id', 'null'), headers)
def test_HEAD_null_id(self):
self.app.register(
@ -2009,6 +2047,13 @@ class ObjectVersioningTestVersionAPI(ObjectVersioningBaseTestCase):
self.assertEqual(1, len(self.app.calls))
self.assertIn(('X-Object-Version-Id', 'null'), headers)
self.assertIn(('X-Object-Meta-Foo', 'bar'), headers)
# N.B. GET on explicitly registered HEAD raised KeyError
req = Request.blank(
'/v1/a/c/o', method='GET',
environ={'swift.cache': self.cache_version_on},
params={'version-id': 'null'})
with self.assertRaises(KeyError):
status, headers, body = self.call_ov(req)
def test_HEAD_delete_marker(self):
self.app.register(

29
test/unit/common/middleware/test_slo.py
View File

@ -1865,6 +1865,12 @@ class TestSloDeleteManifest(SloTestCase):
class TestSloHeadOldManifest(SloTestCase):
"""
Exercise legacy manifests written before we added etag/size SLO Sysmeta
N.B. We used to GET the whole manifest to calculate etag/size, just to
respond to HEAD requests.
"""
slo_etag = md5hex("seg01-hashseg02-hash")
def setUp(self):
@ -1887,6 +1893,7 @@ class TestSloHeadOldManifest(SloTestCase):
'X-Static-Large-Object': 'true',
'X-Object-Sysmeta-Artisanal-Etag': 'bespoke',
'Etag': self.manifest_json_etag}
# see TestSloHeadManifest for tests w/ manifest_has_sysmeta = True
manifest_headers.update(getattr(self, 'extra_manifest_headers', {}))
self.manifest_has_sysmeta = all(h in manifest_headers for h in (
'X-Object-Sysmeta-Slo-Etag', 'X-Object-Sysmeta-Slo-Size'))
@ -1912,6 +1919,24 @@ class TestSloHeadOldManifest(SloTestCase):
expected_app_calls.append(('GET', '/v1/AUTH_test/headtest/man'))
self.assertEqual(self.app.calls, expected_app_calls)
def test_get_manifest_passthrough(self):
req = Request.blank(
'/v1/AUTH_test/headtest/man?multipart-manifest=get',
environ={'REQUEST_METHOD': 'HEAD'})
status, headers, body = self.call_slo(req)
self.assertEqual(status, '200 OK')
self.assertIn(('Etag', self.manifest_json_etag), headers)
self.assertIn(('Content-Type', 'application/json; charset=utf-8'),
headers)
self.assertIn(('X-Static-Large-Object', 'true'), headers)
self.assertIn(('X-Object-Sysmeta-Artisanal-Etag', 'bespoke'), headers)
self.assertEqual(body, b'') # it's a HEAD request, after all
expected_app_calls = [(
'HEAD', '/v1/AUTH_test/headtest/man?multipart-manifest=get')]
self.assertEqual(self.app.calls, expected_app_calls)
def test_if_none_match_etag_matching(self):
req = Request.blank(
'/v1/AUTH_test/headtest/man',
@ -1986,6 +2011,10 @@ class TestSloHeadOldManifest(SloTestCase):
class TestSloHeadManifest(TestSloHeadOldManifest):
"""
Exercise manifests written after we added etag/size SLO Sysmeta
"""
def setUp(self):
self.extra_manifest_headers = {
'X-Object-Sysmeta-Slo-Etag': self.slo_etag,

13
test/unit/common/middleware/test_symlink.py
View File

@ -400,13 +400,24 @@ class TestSymlinkMiddleware(TestSymlinkMiddlewareBase):
def test_get_symlink(self):
self.app.register('GET', '/v1/a/c/symlink', swob.HTTPOk,
{'X-Object-Sysmeta-Symlink-Target': 'c1/o'})
{'X-Object-Sysmeta-Symlink-Target': 'c1/o',
'X-Object-Meta-Color': 'Red'})
req = Request.blank('/v1/a/c/symlink?symlink=get', method='GET')
status, headers, body = self.call_sym(req)
self.assertEqual(status, '200 OK')
self.assertIsInstance(headers, list)
self.assertIn(('X-Symlink-Target', 'c1/o'), headers)
self.assertNotIn('X-Symlink-Target-Account', dict(headers))
self.assertIn(('X-Object-Meta-Color', 'Red'), headers)
self.assertEqual(body, b'')
# HEAD with same params find same registered GET
req = Request.blank('/v1/a/c/symlink?symlink=get', method='HEAD')
head_status, head_headers, head_body = self.call_sym(req)
self.assertEqual(head_status, '200 OK')
self.assertIn(('X-Symlink-Target', 'c1/o'), head_headers)
self.assertNotIn('X-Symlink-Target-Account', dict(head_headers))
self.assertIn(('X-Object-Meta-Color', 'Red'), head_headers)
self.assertEqual(head_body, b'')
def test_get_symlink_with_account(self):
self.app.register('GET', '/v1/a/c/symlink', swob.HTTPOk,