Merge "Refactor CORS unit tests"
This commit is contained in:
Jenkins
@@ -6083,21 +6083,18 @@ class TestObjectController(unittest.TestCase):
|
|||||||
7)
|
7)
|
||||||
self.assertEqual('999', resp.headers['access-control-max-age'])
|
self.assertEqual('999', resp.headers['access-control-max-age'])
|
||||||
|
|
||||||
def test_CORS_valid(self):
|
def _get_CORS_response(self, container_cors, strict_mode, object_get=None):
|
||||||
with save_globals():
|
with save_globals():
|
||||||
controller = ReplicatedObjectController(
|
controller = ReplicatedObjectController(
|
||||||
self.app, 'a', 'c', 'o')
|
self.app, 'a', 'c', 'o')
|
||||||
|
|
||||||
def stubContainerInfo(*args):
|
def stubContainerInfo(*args):
|
||||||
return {
|
return {
|
||||||
'cors': {
|
'cors': container_cors
|
||||||
'allow_origin': 'http://not.foo.bar',
|
|
||||||
'expose_headers': 'X-Object-Meta-Color '
|
|
||||||
'X-Object-Meta-Color-Ex'
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
controller.container_info = stubContainerInfo
|
controller.container_info = stubContainerInfo
|
||||||
controller.app.strict_cors_mode = False
|
controller.app.strict_cors_mode = strict_mode
|
||||||
|
|
||||||
def objectGET(controller, req):
|
def objectGET(controller, req):
|
||||||
return Response(headers={
|
return Response(headers={
|
||||||
@@ -6105,116 +6102,119 @@ class TestObjectController(unittest.TestCase):
|
|||||||
'X-Super-Secret': 'hush',
|
'X-Super-Secret': 'hush',
|
||||||
})
|
})
|
||||||
|
|
||||||
req = Request.blank(
|
mock_object_get = object_get or objectGET
|
||||||
'/v1/a/c/o.jpg',
|
|
||||||
{'REQUEST_METHOD': 'GET'},
|
|
||||||
headers={'Origin': 'http://foo.bar'})
|
|
||||||
|
|
||||||
resp = cors_validation(objectGET)(controller, req)
|
|
||||||
|
|
||||||
self.assertEqual(200, resp.status_int)
|
|
||||||
self.assertEqual('http://foo.bar',
|
|
||||||
resp.headers['access-control-allow-origin'])
|
|
||||||
self.assertEqual('red', resp.headers['x-object-meta-color'])
|
|
||||||
# X-Super-Secret is in the response, but not "exposed"
|
|
||||||
self.assertEqual('hush', resp.headers['x-super-secret'])
|
|
||||||
self.assertIn('access-control-expose-headers', resp.headers)
|
|
||||||
exposed = set(
|
|
||||||
h.strip() for h in
|
|
||||||
resp.headers['access-control-expose-headers'].split(','))
|
|
||||||
expected_exposed = set(['cache-control', 'content-language',
|
|
||||||
'content-type', 'expires', 'last-modified',
|
|
||||||
'pragma', 'etag', 'x-timestamp',
|
|
||||||
'x-trans-id', 'x-object-meta-color',
|
|
||||||
'x-object-meta-color-ex'])
|
|
||||||
self.assertEqual(expected_exposed, exposed)
|
|
||||||
|
|
||||||
controller.app.strict_cors_mode = True
|
|
||||||
req = Request.blank(
|
|
||||||
'/v1/a/c/o.jpg',
|
|
||||||
{'REQUEST_METHOD': 'GET'},
|
|
||||||
headers={'Origin': 'http://foo.bar'})
|
|
||||||
|
|
||||||
resp = cors_validation(objectGET)(controller, req)
|
|
||||||
|
|
||||||
self.assertEqual(200, resp.status_int)
|
|
||||||
self.assertNotIn('access-control-expose-headers', resp.headers)
|
|
||||||
self.assertNotIn('access-control-allow-origin', resp.headers)
|
|
||||||
|
|
||||||
controller.app.strict_cors_mode = False
|
|
||||||
|
|
||||||
def stubContainerInfoWithAsteriskAllowOrigin(*args):
|
|
||||||
return {
|
|
||||||
'cors': {
|
|
||||||
'allow_origin': '*'
|
|
||||||
}
|
|
||||||
}
|
|
||||||
controller.container_info = \
|
|
||||||
stubContainerInfoWithAsteriskAllowOrigin
|
|
||||||
|
|
||||||
req = Request.blank(
|
req = Request.blank(
|
||||||
'/v1/a/c/o.jpg',
|
'/v1/a/c/o.jpg',
|
||||||
{'REQUEST_METHOD': 'GET'},
|
{'REQUEST_METHOD': 'GET'},
|
||||||
headers={'Origin': 'http://foo.bar'})
|
headers={'Origin': 'http://foo.bar'})
|
||||||
|
|
||||||
resp = cors_validation(objectGET)(controller, req)
|
resp = cors_validation(mock_object_get)(controller, req)
|
||||||
|
|
||||||
self.assertEqual(200, resp.status_int)
|
return resp
|
||||||
self.assertEqual('*',
|
|
||||||
resp.headers['access-control-allow-origin'])
|
|
||||||
|
|
||||||
def stubContainerInfoWithEmptyAllowOrigin(*args):
|
def test_CORS_valid_non_strict(self):
|
||||||
return {
|
# test expose_headers to non-allowed origins
|
||||||
'cors': {
|
container_cors = {'allow_origin': 'http://not.foo.bar',
|
||||||
'allow_origin': ''
|
'expose_headers': 'X-Object-Meta-Color '
|
||||||
}
|
'X-Object-Meta-Color-Ex'}
|
||||||
}
|
resp = self._get_CORS_response(
|
||||||
controller.container_info = stubContainerInfoWithEmptyAllowOrigin
|
container_cors=container_cors, strict_mode=False)
|
||||||
|
|
||||||
req = Request.blank(
|
self.assertEqual(200, resp.status_int)
|
||||||
'/v1/a/c/o.jpg',
|
self.assertEqual('http://foo.bar',
|
||||||
{'REQUEST_METHOD': 'GET'},
|
resp.headers['access-control-allow-origin'])
|
||||||
headers={'Origin': 'http://foo.bar'})
|
self.assertEqual('red', resp.headers['x-object-meta-color'])
|
||||||
|
# X-Super-Secret is in the response, but not "exposed"
|
||||||
|
self.assertEqual('hush', resp.headers['x-super-secret'])
|
||||||
|
self.assertIn('access-control-expose-headers', resp.headers)
|
||||||
|
exposed = set(
|
||||||
|
h.strip() for h in
|
||||||
|
resp.headers['access-control-expose-headers'].split(','))
|
||||||
|
expected_exposed = set(['cache-control', 'content-language',
|
||||||
|
'content-type', 'expires', 'last-modified',
|
||||||
|
'pragma', 'etag', 'x-timestamp',
|
||||||
|
'x-trans-id', 'x-object-meta-color',
|
||||||
|
'x-object-meta-color-ex'])
|
||||||
|
self.assertEqual(expected_exposed, exposed)
|
||||||
|
|
||||||
resp = cors_validation(objectGET)(controller, req)
|
# test allow_origin *
|
||||||
|
container_cors = {'allow_origin': '*'}
|
||||||
|
|
||||||
self.assertEqual(200, resp.status_int)
|
resp = self._get_CORS_response(
|
||||||
self.assertEqual('http://foo.bar',
|
container_cors=container_cors, strict_mode=False)
|
||||||
resp.headers['access-control-allow-origin'])
|
self.assertEqual(200, resp.status_int)
|
||||||
|
self.assertEqual('*',
|
||||||
|
resp.headers['access-control-allow-origin'])
|
||||||
|
|
||||||
|
# test allow_origin empty
|
||||||
|
container_cors = {'allow_origin': ''}
|
||||||
|
resp = self._get_CORS_response(
|
||||||
|
container_cors=container_cors, strict_mode=False)
|
||||||
|
self.assertEqual(200, resp.status_int)
|
||||||
|
self.assertEqual('http://foo.bar',
|
||||||
|
resp.headers['access-control-allow-origin'])
|
||||||
|
|
||||||
|
def test_CORS_valid_strict(self):
|
||||||
|
# test expose_headers to non-allowed origins
|
||||||
|
container_cors = {'allow_origin': 'http://not.foo.bar',
|
||||||
|
'expose_headers': 'X-Object-Meta-Color '
|
||||||
|
'X-Object-Meta-Color-Ex'}
|
||||||
|
resp = self._get_CORS_response(
|
||||||
|
container_cors=container_cors, strict_mode=True)
|
||||||
|
|
||||||
|
self.assertEqual(200, resp.status_int)
|
||||||
|
self.assertNotIn('access-control-expose-headers', resp.headers)
|
||||||
|
self.assertNotIn('access-control-allow-origin', resp.headers)
|
||||||
|
|
||||||
|
# test allow_origin *
|
||||||
|
container_cors = {'allow_origin': '*'}
|
||||||
|
|
||||||
|
resp = self._get_CORS_response(
|
||||||
|
container_cors=container_cors, strict_mode=True)
|
||||||
|
self.assertEqual(200, resp.status_int)
|
||||||
|
self.assertEqual('*',
|
||||||
|
resp.headers['access-control-allow-origin'])
|
||||||
|
self.assertEqual('red', resp.headers['x-object-meta-color'])
|
||||||
|
# X-Super-Secret is in the response, but not "exposed"
|
||||||
|
self.assertEqual('hush', resp.headers['x-super-secret'])
|
||||||
|
self.assertIn('access-control-expose-headers', resp.headers)
|
||||||
|
exposed = set(
|
||||||
|
h.strip() for h in
|
||||||
|
resp.headers['access-control-expose-headers'].split(','))
|
||||||
|
expected_exposed = set(['cache-control', 'content-language',
|
||||||
|
'content-type', 'expires', 'last-modified',
|
||||||
|
'pragma', 'etag', 'x-timestamp',
|
||||||
|
'x-trans-id', 'x-object-meta-color'])
|
||||||
|
self.assertEqual(expected_exposed, exposed)
|
||||||
|
|
||||||
|
# test allow_origin empty
|
||||||
|
container_cors = {'allow_origin': ''}
|
||||||
|
resp = self._get_CORS_response(
|
||||||
|
container_cors=container_cors, strict_mode=True)
|
||||||
|
self.assertNotIn('access-control-expose-headers', resp.headers)
|
||||||
|
self.assertNotIn('access-control-allow-origin', resp.headers)
|
||||||
|
|
||||||
def test_CORS_valid_with_obj_headers(self):
|
def test_CORS_valid_with_obj_headers(self):
|
||||||
with save_globals():
|
container_cors = {'allow_origin': 'http://foo.bar'}
|
||||||
controller = ReplicatedObjectController(
|
|
||||||
self.app, 'a', 'c', 'o')
|
|
||||||
|
|
||||||
def stubContainerInfo(*args):
|
def objectGET(controller, req):
|
||||||
return {
|
return Response(headers={
|
||||||
'cors': {
|
'X-Object-Meta-Color': 'red',
|
||||||
'allow_origin': 'http://foo.bar'
|
'X-Super-Secret': 'hush',
|
||||||
}
|
'Access-Control-Allow-Origin': 'http://obj.origin',
|
||||||
}
|
'Access-Control-Expose-Headers': 'x-trans-id'
|
||||||
controller.container_info = stubContainerInfo
|
})
|
||||||
|
|
||||||
def objectGET(controller, req):
|
resp = self._get_CORS_response(
|
||||||
return Response(headers={
|
container_cors=container_cors, strict_mode=True,
|
||||||
'X-Object-Meta-Color': 'red',
|
object_get=objectGET)
|
||||||
'X-Super-Secret': 'hush',
|
|
||||||
'Access-Control-Allow-Origin': 'http://obj.origin',
|
|
||||||
'Access-Control-Expose-Headers': 'x-trans-id'
|
|
||||||
})
|
|
||||||
|
|
||||||
req = Request.blank(
|
self.assertEqual(200, resp.status_int)
|
||||||
'/v1/a/c/o.jpg',
|
self.assertEqual('http://obj.origin',
|
||||||
{'REQUEST_METHOD': 'GET'},
|
resp.headers['access-control-allow-origin'])
|
||||||
headers={'Origin': 'http://foo.bar'})
|
self.assertEqual('x-trans-id',
|
||||||
|
resp.headers['access-control-expose-headers'])
|
||||||
resp = cors_validation(objectGET)(controller, req)
|
|
||||||
|
|
||||||
self.assertEqual(200, resp.status_int)
|
|
||||||
self.assertEqual('http://obj.origin',
|
|
||||||
resp.headers['access-control-allow-origin'])
|
|
||||||
self.assertEqual('x-trans-id',
|
|
||||||
resp.headers['access-control-expose-headers'])
|
|
||||||
|
|
||||||
def _gather_x_container_headers(self, controller_call, req, *connect_args,
|
def _gather_x_container_headers(self, controller_call, req, *connect_args,
|
||||||
**kwargs):
|
**kwargs):
|
||||||
|
|||||||
Reference in New Issue
Block a user