summaryrefslogtreecommitdiff
path: root/tests/integration/test_main.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/integration/test_main.py')
-rw-r--r--tests/integration/test_main.py1577
1 files changed, 824 insertions, 753 deletions
diff --git a/tests/integration/test_main.py b/tests/integration/test_main.py
index 661094e..098c3d3 100644
--- a/tests/integration/test_main.py
+++ b/tests/integration/test_main.py
@@ -15,18 +15,14 @@
"""Test generic usage
"""
-import asyncio
import pathlib
-from yarl import URL
+import pytest
-from tests.aio_test_base import AioTestBase, create_server, asynctest
-
-import aiohttp
from aiohttp import web
from aiohttp import hdrs
-from aiohttp_cors import setup, ResourceOptions
+from aiohttp_cors import setup as _setup, ResourceOptions, CorsViewMixin
TEST_BODY = "Hello, world"
@@ -34,9 +30,8 @@ SERVER_CUSTOM_HEADER_NAME = "X-Server-Custom-Header"
SERVER_CUSTOM_HEADER_VALUE = "some value"
-@asyncio.coroutine
# pylint: disable=unused-argument
-def handler(request: web.Request) -> web.StreamResponse:
+async def handler(request: web.Request) -> web.StreamResponse:
"""Dummy request handler, returning `TEST_BODY`."""
response = web.Response(text=TEST_BODY)
@@ -45,800 +40,876 @@ def handler(request: web.Request) -> web.StreamResponse:
return response
-class AioAiohttpAppTestBase(AioTestBase):
- """Base class for tests that create single aiohttp server.
+class WebViewHandler(web.View, CorsViewMixin):
- Class manages server creation using create_server() method and proper
- server shutdown.
- """
+ async def get(self) -> web.StreamResponse:
+ """Dummy request handler, returning `TEST_BODY`."""
+ response = web.Response(text=TEST_BODY)
- def setUp(self):
- super().setUp()
+ response.headers[SERVER_CUSTOM_HEADER_NAME] = \
+ SERVER_CUSTOM_HEADER_VALUE
- self.handler = None
- self.app = None
- self.url = None
+ return response
- self.server = None
- self.session = aiohttp.ClientSession(loop=self.loop)
+@pytest.fixture(params=['resource', 'view', 'route'])
+def make_app(request):
+ def inner(defaults, route_config):
+ app = web.Application()
+ cors = _setup(app, defaults=defaults)
- def tearDown(self):
- self.session.close()
+ if request.param == 'resource':
+ resource = cors.add(app.router.add_resource("/resource"))
+ cors.add(resource.add_route("GET", handler), route_config)
+ elif request.param == 'view':
+ WebViewHandler.cors_config = route_config
+ cors.add(
+ app.router.add_route("*", "/resource", WebViewHandler))
+ elif request.param == 'route':
+ cors.add(
+ app.router.add_route("GET", "/resource", handler),
+ route_config)
+ else:
+ raise RuntimeError('unknown parameter {}'.format(request.param))
- if self.server is not None:
- self.loop.run_until_complete(self.shutdown_server())
+ return app
- super().tearDown()
+ return inner
- @asyncio.coroutine
- def create_server(self, app: web.Application):
- """Create server listening on random port."""
- assert self.app is None
- self.app = app
+async def test_message_roundtrip(aiohttp_client):
+ """Test that aiohttp server is correctly setup in the base class."""
- assert self.handler is None
- self.handler = app.make_handler()
+ app = web.Application()
+ app.router.add_route("GET", "/", handler)
- self.server = (yield from create_server(self.handler, self.loop))
+ client = await aiohttp_client(app)
- return self.server
+ resp = await client.get('/')
+ assert resp.status == 200
+ data = await resp.text()
- @property
- def server_url(self):
- """Server navigatable URL."""
- assert self.server is not None
- hostaddr, port = self.server.sockets[0].getsockname()
- return "http://{host}:{port}/".format(host=hostaddr, port=port)
+ assert data == TEST_BODY
- @asyncio.coroutine
- def shutdown_server(self):
- """Shutdown server."""
- assert self.server is not None
- self.server.close()
- yield from self.handler.finish_connections()
- yield from self.server.wait_closed()
- yield from self.app.cleanup()
+async def test_dummy_setup(aiohttp_server):
+ """Test a dummy configuration."""
+ app = web.Application()
+ _setup(app)
- self.server = None
- self.app = None
- self.handler = None
+ await aiohttp_server(app)
-class TestMain(AioAiohttpAppTestBase):
- """Tests CORS server by issuing CORS requests."""
+async def test_dummy_setup_roundtrip(aiohttp_client):
+ """Test a dummy configuration with a message round-trip."""
+ app = web.Application()
+ _setup(app)
- @asynctest
- @asyncio.coroutine
- def test_message_roundtrip(self):
- """Test that aiohttp server is correctly setup in the base class."""
+ app.router.add_route("GET", "/", handler)
- app = web.Application()
+ client = await aiohttp_client(app)
- app.router.add_route("GET", "/", handler)
+ resp = await client.get('/')
+ assert resp.status == 200
+ data = await resp.text()
- yield from self.create_server(app)
+ assert data == TEST_BODY
- response = yield from self.session.request("GET", self.server_url)
- self.assertEqual(response.status, 200)
- data = yield from response.text()
- self.assertEqual(data, TEST_BODY)
+async def test_dummy_setup_roundtrip_resource(aiohttp_client):
+ """Test a dummy configuration with a message round-trip."""
+ app = web.Application()
+ _setup(app)
- @asynctest
- @asyncio.coroutine
- def test_dummy_setup(self):
- """Test a dummy configuration."""
- app = web.Application()
- setup(app)
+ app.router.add_resource("/").add_route("GET", handler)
- yield from self.create_server(app)
+ client = await aiohttp_client(app)
- @asynctest
- @asyncio.coroutine
- def test_dummy_setup_roundtrip(self):
- """Test a dummy configuration with a message round-trip."""
- app = web.Application()
- setup(app)
+ resp = await client.get('/')
+ assert resp.status == 200
+ data = await resp.text()
- app.router.add_route("GET", "/", handler)
+ assert data == TEST_BODY
- yield from self.create_server(app)
- response = yield from self.session.request("GET", self.server_url)
- self.assertEqual(response.status, 200)
- data = yield from response.text()
+async def test_simple_no_origin(aiohttp_client, make_app):
+ app = make_app(None, {"http://client1.example.org":
+ ResourceOptions()})
- self.assertEqual(data, TEST_BODY)
+ client = await aiohttp_client(app)
- @asynctest
- @asyncio.coroutine
- def test_dummy_setup_roundtrip_resource(self):
- """Test a dummy configuration with a message round-trip."""
- app = web.Application()
- setup(app)
-
- app.router.add_resource("/").add_route("GET", handler)
-
- yield from self.create_server(app)
-
- response = yield from self.session.request("GET", self.server_url)
- self.assertEqual(response.status, 200)
- data = yield from response.text()
-
- self.assertEqual(data, TEST_BODY)
-
- @asyncio.coroutine
- def _run_simple_requests_tests(self,
- tests_descriptions,
- use_resources):
- """Runs CORS simple requests (without a preflight request) based
- on the passed tests descriptions.
- """
-
- @asyncio.coroutine
- def run_test(test):
- """Run single test"""
-
- response = yield from self.session.get(
- self.server_url + "resource",
- headers=test.get("request_headers", {}))
- self.assertEqual(response.status, 200)
- self.assertEqual((yield from response.text()), TEST_BODY)
-
- for header_name, header_value in test.get(
- "in_response_headers", {}).items():
- with self.subTest(header_name=header_name):
- self.assertEqual(
- response.headers.get(header_name),
- header_value)
-
- for header_name in test.get("not_in_request_headers", {}).items():
- self.assertNotIn(header_name, response.headers)
-
- for test_descr in tests_descriptions:
- with self.subTest(group_name=test_descr["name"]):
- app = web.Application()
- cors = setup(app, defaults=test_descr["defaults"])
-
- if use_resources:
- resource = cors.add(app.router.add_resource("/resource"))
- cors.add(resource.add_route("GET", handler),
- test_descr["route_config"])
-
- else:
- cors.add(
- app.router.add_route("GET", "/resource", handler),
- test_descr["route_config"])
-
- yield from self.create_server(app)
-
- try:
- for test_data in test_descr["tests"]:
- with self.subTest(name=test_data["name"]):
- yield from run_test(test_data)
- finally:
- yield from self.shutdown_server()
-
- @asynctest
- @asyncio.coroutine
- def test_simple_default(self):
- """Test CORS simple requests with a route with the default
- configuration.
-
- The default configuration means that:
- * no credentials are allowed,
- * no headers are exposed,
- * no client headers are allowed.
- """
-
- client1 = "http://client1.example.org"
- client2 = "http://client2.example.org"
- client1_80 = "http://client1.example.org:80"
- client1_https = "https://client2.example.org"
-
- tests_descriptions = [
- {
- "name": "default",
- "defaults": None,
- "route_config":
- {
- client1: ResourceOptions(),
- },
- "tests": [
- {
- "name": "no origin header",
- "not_in_response_headers": {
- hdrs.ACCESS_CONTROL_ALLOW_ORIGIN,
- hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
- hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
- }
- },
- {
- "name": "allowed origin",
- "request_headers": {
- hdrs.ORIGIN: client1,
- },
- "in_response_headers": {
- hdrs.ACCESS_CONTROL_ALLOW_ORIGIN: client1,
- },
- "not_in_response_headers": {
- hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
- hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
- }
- },
- {
- "name": "not allowed origin",
- "request_headers": {
- hdrs.ORIGIN: client2,
- },
- "not_in_response_headers": {
- hdrs.ACCESS_CONTROL_ALLOW_ORIGIN,
- hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
- hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
- }
- },
- {
- "name": "explicitly specified default port",
- # CORS specification says, that origins may compared
- # as strings, so "example.org:80" is not the same as
- # "example.org".
- "request_headers": {
- hdrs.ORIGIN: client1_80,
- },
- "not_in_response_headers": {
- hdrs.ACCESS_CONTROL_ALLOW_ORIGIN,
- hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
- hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
- }
- },
- {
- "name": "different scheme",
- "request_headers": {
- hdrs.ORIGIN: client1_https,
- },
- "not_in_response_headers": {
- hdrs.ACCESS_CONTROL_ALLOW_ORIGIN,
- hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
- hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
- }
- },
- ],
- },
- ]
-
- yield from self._run_simple_requests_tests(tests_descriptions, False)
- yield from self._run_simple_requests_tests(tests_descriptions, True)
-
- @asynctest
- @asyncio.coroutine
- def test_simple_with_credentials(self):
- """Test CORS simple requests with a route with enabled authorization.
-
- Route with enabled authorization must return
- Origin: <origin as requested, NOT "*">
- Access-Control-Allow-Credentials: true
- """
-
- client1 = "http://client1.example.org"
- client2 = "http://client2.example.org"
-
- credential_tests = [
- {
- "name": "no origin header",
- "not_in_response_headers": {
- hdrs.ACCESS_CONTROL_ALLOW_ORIGIN,
- hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
- hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
- }
- },
- {
- "name": "allowed origin",
- "request_headers": {
- hdrs.ORIGIN: client1,
- },
- "in_response_headers": {
- hdrs.ACCESS_CONTROL_ALLOW_ORIGIN: client1,
- hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS: "true",
- },
- "not_in_response_headers": {
- hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
- }
- },
- {
- "name": "disallowed origin",
- "request_headers": {
- hdrs.ORIGIN: client2,
- },
- "not_in_response_headers": {
- hdrs.ACCESS_CONTROL_ALLOW_ORIGIN,
- hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
- hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
- }
- },
- ]
-
- tests_descriptions = [
- {
- "name": "route settings",
- "defaults": None,
- "route_config":
- {
- client1: ResourceOptions(allow_credentials=True),
- },
- "tests": credential_tests,
- },
- {
- "name": "cors default settings",
- "defaults":
- {
- client1: ResourceOptions(allow_credentials=True),
- },
- "route_config": None,
- "tests": credential_tests,
- },
- ]
-
- yield from self._run_simple_requests_tests(tests_descriptions, False)
- yield from self._run_simple_requests_tests(tests_descriptions, True)
-
- @asynctest
- @asyncio.coroutine
- def test_simple_expose_headers(self):
- """Test CORS simple requests with a route that exposes header."""
-
- client1 = "http://client1.example.org"
- client2 = "http://client2.example.org"
-
- tests_descriptions = [
- {
- "name": "default",
- "defaults": None,
- "route_config":
- {
- client1: ResourceOptions(
- expose_headers=(SERVER_CUSTOM_HEADER_NAME,)),
- },
- "tests": [
- {
- "name": "no origin header",
- "not_in_response_headers": {
- hdrs.ACCESS_CONTROL_ALLOW_ORIGIN,
- hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
- hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
- }
- },
- {
- "name": "allowed origin",
- "request_headers": {
- hdrs.ORIGIN: client1,
- },
- "in_response_headers": {
- hdrs.ACCESS_CONTROL_ALLOW_ORIGIN: client1,
- hdrs.ACCESS_CONTROL_EXPOSE_HEADERS:
- SERVER_CUSTOM_HEADER_NAME,
- },
- "not_in_response_headers": {
- hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
- }
- },
- {
- "name": "not allowed origin",
- "request_headers": {
- hdrs.ORIGIN: client2,
- },
- "not_in_response_headers": {
- hdrs.ACCESS_CONTROL_ALLOW_ORIGIN,
- hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
- hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
- }
- },
- ],
- },
- ]
-
- yield from self._run_simple_requests_tests(tests_descriptions, False)
- yield from self._run_simple_requests_tests(tests_descriptions, True)
- yield from self._run_simple_requests_tests(tests_descriptions, True)
-
- @asyncio.coroutine
- def _run_preflight_requests_tests(self, tests_descriptions, use_resources):
- """Runs CORS preflight requests based on the passed tests descriptions.
- """
-
- @asyncio.coroutine
- def run_test(test):
- """Run single test"""
-
- response = yield from self.session.options(
- self.server_url + "resource",
- headers=test.get("request_headers", {}))
- self.assertEqual(response.status, test.get("response_status", 200))
- response_text = yield from response.text()
- in_response = test.get("in_response")
- if in_response is not None:
- self.assertIn(in_response, response_text)
- else:
- self.assertEqual(response_text, "")
-
- for header_name, header_value in test.get(
- "in_response_headers", {}).items():
- self.assertEqual(
- response.headers.get(header_name),
- header_value)
-
- for header_name in test.get("not_in_request_headers", {}).items():
- self.assertNotIn(header_name, response.headers)
-
- for test_descr in tests_descriptions:
- with self.subTest(group_name=test_descr["name"]):
- app = web.Application()
- cors = setup(app, defaults=test_descr["defaults"])
-
- if use_resources:
- resource = cors.add(app.router.add_resource("/resource"))
- cors.add(resource.add_route("GET", handler),
- test_descr["route_config"])
-
- else:
- cors.add(
- app.router.add_route("GET", "/resource", handler),
- test_descr["route_config"])
-
- yield from self.create_server(app)
-
- try:
- for test_data in test_descr["tests"]:
- with self.subTest(name=test_data["name"]):
- yield from run_test(test_data)
- finally:
- yield from self.shutdown_server()
-
- @asynctest
- @asyncio.coroutine
- def test_preflight_default(self):
- """Test CORS preflight requests with a route with the default
- configuration.
-
- The default configuration means that:
- * no credentials are allowed,
- * no headers are exposed,
- * no client headers are allowed.
- """
-
- client1 = "http://client1.example.org"
- client2 = "http://client2.example.org"
-
- tests_descriptions = [
- {
- "name": "default",
- "defaults": None,
- "route_config":
- {
- client1: ResourceOptions(),
- },
- "tests": [
- {
- "name": "no origin",
- "response_status": 403,
- "in_response": "origin header is not specified",
- "not_in_response_headers": {
- hdrs.ACCESS_CONTROL_ALLOW_ORIGIN,
- hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
- hdrs.ACCESS_CONTROL_MAX_AGE,
- hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
- hdrs.ACCESS_CONTROL_ALLOW_METHODS,
- hdrs.ACCESS_CONTROL_ALLOW_HEADERS,
- },
- },
- {
- "name": "no method",
- "request_headers": {
- hdrs.ORIGIN: client1,
- },
- "response_status": 403,
- "in_response": "'Access-Control-Request-Method' "
- "header is not specified",
- "not_in_response_headers": {
- hdrs.ACCESS_CONTROL_ALLOW_ORIGIN,
- hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
- hdrs.ACCESS_CONTROL_MAX_AGE,
- hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
- hdrs.ACCESS_CONTROL_ALLOW_METHODS,
- hdrs.ACCESS_CONTROL_ALLOW_HEADERS,
- },
- },
- {
- "name": "origin and method",
- "request_headers": {
- hdrs.ORIGIN: client1,
- hdrs.ACCESS_CONTROL_REQUEST_METHOD: "GET",
- },
- "in_response_headers": {
- hdrs.ACCESS_CONTROL_ALLOW_ORIGIN: client1,
- hdrs.ACCESS_CONTROL_ALLOW_METHODS: "GET",
- },
- "not_in_response_headers": {
- hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
- hdrs.ACCESS_CONTROL_MAX_AGE,
- hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
- hdrs.ACCESS_CONTROL_ALLOW_HEADERS,
- },
- },
- {
- "name": "disallowed origin",
- "request_headers": {
- hdrs.ORIGIN: client2,
- hdrs.ACCESS_CONTROL_REQUEST_METHOD: "GET",
- },
- "response_status": 403,
- "in_response": "origin '{}' is not allowed".format(
- client2),
- "not_in_response_headers": {
- hdrs.ACCESS_CONTROL_ALLOW_ORIGIN,
- hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
- hdrs.ACCESS_CONTROL_MAX_AGE,
- hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
- hdrs.ACCESS_CONTROL_ALLOW_METHODS,
- hdrs.ACCESS_CONTROL_ALLOW_HEADERS,
- },
- },
- {
- "name": "disallowed method",
- "request_headers": {
- hdrs.ORIGIN: client1,
- hdrs.ACCESS_CONTROL_REQUEST_METHOD: "POST",
- },
- "response_status": 403,
- "in_response": "request method 'POST' is not allowed",
- "not_in_response_headers": {
- hdrs.ACCESS_CONTROL_ALLOW_ORIGIN,
- hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
- hdrs.ACCESS_CONTROL_MAX_AGE,
- hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
- hdrs.ACCESS_CONTROL_ALLOW_METHODS,
- hdrs.ACCESS_CONTROL_ALLOW_HEADERS,
- },
- },
- ],
- },
- ]
-
- yield from self._run_preflight_requests_tests(
- tests_descriptions, False)
- yield from self._run_preflight_requests_tests(
- tests_descriptions, True)
-
- @asynctest
- @asyncio.coroutine
- def test_preflight_request_multiple_routes_with_one_options(self):
- """Test CORS preflight handling on resource that is available through
- several routes.
- """
- app = web.Application()
- cors = setup(app, defaults={
- "*": ResourceOptions(
- allow_credentials=True,
- expose_headers="*",
- allow_headers="*",
- )
- })
-
- cors.add(app.router.add_route("GET", "/{name}", handler))
- cors.add(app.router.add_route("PUT", "/{name}", handler))
-
- yield from self.create_server(app)
-
- response = yield from self.session.request(
- "OPTIONS", self.server_url + "user",
- headers={
- hdrs.ORIGIN: "http://example.org",
- hdrs.ACCESS_CONTROL_REQUEST_METHOD: "PUT"
- }
+ resp = await client.get("/resource")
+ assert resp.status == 200
+ resp_text = await resp.text()
+ assert resp_text == TEST_BODY
+
+ for header_name in {
+ hdrs.ACCESS_CONTROL_ALLOW_ORIGIN,
+ hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
+ hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
+ }:
+ assert header_name not in resp.headers
+
+
+async def test_simple_allowed_origin(aiohttp_client, make_app):
+ app = make_app(None, {"http://client1.example.org":
+ ResourceOptions()})
+
+ client = await aiohttp_client(app)
+
+ resp = await client.get("/resource",
+ headers={hdrs.ORIGIN:
+ 'http://client1.example.org'})
+ assert resp.status == 200
+ resp_text = await resp.text()
+ assert resp_text == TEST_BODY
+
+ for hdr, val in {
+ hdrs.ACCESS_CONTROL_ALLOW_ORIGIN: 'http://client1.example.org',
+ }.items():
+ assert resp.headers.get(hdr) == val
+
+ for header_name in {
+ hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
+ hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
+ }:
+ assert header_name not in resp.headers
+
+
+async def test_simple_not_allowed_origin(aiohttp_client, make_app):
+ app = make_app(None, {"http://client1.example.org":
+ ResourceOptions()})
+
+ client = await aiohttp_client(app)
+
+ resp = await client.get("/resource",
+ headers={hdrs.ORIGIN:
+ 'http://client2.example.org'})
+ assert resp.status == 200
+ resp_text = await resp.text()
+ assert resp_text == TEST_BODY
+
+ for header_name in {
+ hdrs.ACCESS_CONTROL_ALLOW_ORIGIN,
+ hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
+ hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
+ }:
+ assert header_name not in resp.headers
+
+
+async def test_simple_explicit_port(aiohttp_client, make_app):
+ app = make_app(None, {"http://client1.example.org":
+ ResourceOptions()})
+
+ client = await aiohttp_client(app)
+
+ resp = await client.get("/resource",
+ headers={hdrs.ORIGIN:
+ 'http://client1.example.org:80'})
+ assert resp.status == 200
+ resp_text = await resp.text()
+ assert resp_text == TEST_BODY
+
+ for header_name in {
+ hdrs.ACCESS_CONTROL_ALLOW_ORIGIN,
+ hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
+ hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
+ }:
+ assert header_name not in resp.headers
+
+
+async def test_simple_different_scheme(aiohttp_client, make_app):
+ app = make_app(None, {"http://client1.example.org":
+ ResourceOptions()})
+
+ client = await aiohttp_client(app)
+
+ resp = await client.get("/resource",
+ headers={hdrs.ORIGIN:
+ 'https://client1.example.org'})
+ assert resp.status == 200
+ resp_text = await resp.text()
+ assert resp_text == TEST_BODY
+
+ for header_name in {
+ hdrs.ACCESS_CONTROL_ALLOW_ORIGIN,
+ hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
+ hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
+ }:
+ assert header_name not in resp.headers
+
+
+@pytest.fixture(params=[
+ (None,
+ {"http://client1.example.org": ResourceOptions(allow_credentials=True)}),
+ ({"http://client1.example.org": ResourceOptions(allow_credentials=True)},
+ None),
+])
+def app_for_credentials(make_app, request):
+ return make_app(*request.param)
+
+
+async def test_cred_no_origin(aiohttp_client, app_for_credentials):
+ app = app_for_credentials
+
+ client = await aiohttp_client(app)
+
+ resp = await client.get("/resource")
+ assert resp.status == 200
+ resp_text = await resp.text()
+ assert resp_text == TEST_BODY
+
+ for header_name in {
+ hdrs.ACCESS_CONTROL_ALLOW_ORIGIN,
+ hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
+ hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
+ }:
+ assert header_name not in resp.headers
+
+
+async def test_cred_allowed_origin(aiohttp_client, app_for_credentials):
+ app = app_for_credentials
+
+ client = await aiohttp_client(app)
+
+ resp = await client.get("/resource",
+ headers={hdrs.ORIGIN:
+ 'http://client1.example.org'})
+ assert resp.status == 200
+ resp_text = await resp.text()
+ assert resp_text == TEST_BODY
+
+ for hdr, val in {
+ hdrs.ACCESS_CONTROL_ALLOW_ORIGIN: 'http://client1.example.org',
+ hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS: "true"}.items():
+ assert resp.headers.get(hdr) == val
+
+ for header_name in {
+ hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
+ }:
+ assert header_name not in resp.headers
+
+
+async def test_cred_disallowed_origin(aiohttp_client, app_for_credentials):
+ app = app_for_credentials
+
+ client = await aiohttp_client(app)
+
+ resp = await client.get("/resource",
+ headers={hdrs.ORIGIN:
+ 'http://client2.example.org'})
+ assert resp.status == 200
+ resp_text = await resp.text()
+ assert resp_text == TEST_BODY
+
+ for header_name in {
+ hdrs.ACCESS_CONTROL_ALLOW_ORIGIN,
+ hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
+ hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
+ }:
+ assert header_name not in resp.headers
+
+
+async def test_simple_expose_headers_no_origin(aiohttp_client, make_app):
+ app = make_app(None, {"http://client1.example.org":
+ ResourceOptions(
+ expose_headers=(SERVER_CUSTOM_HEADER_NAME,))})
+
+ client = await aiohttp_client(app)
+
+ resp = await client.get("/resource")
+ assert resp.status == 200
+ resp_text = await resp.text()
+ assert resp_text == TEST_BODY
+
+ for header_name in {
+ hdrs.ACCESS_CONTROL_ALLOW_ORIGIN,
+ hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
+ hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
+ }:
+ assert header_name not in resp.headers
+
+
+async def test_simple_expose_headers_allowed_origin(aiohttp_client, make_app):
+ app = make_app(None, {"http://client1.example.org":
+ ResourceOptions(
+ expose_headers=(SERVER_CUSTOM_HEADER_NAME,))})
+
+ client = await aiohttp_client(app)
+
+ resp = await client.get("/resource",
+ headers={hdrs.ORIGIN:
+ 'http://client1.example.org'})
+ assert resp.status == 200
+ resp_text = await resp.text()
+ assert resp_text == TEST_BODY
+
+ for hdr, val in {
+ hdrs.ACCESS_CONTROL_ALLOW_ORIGIN: 'http://client1.example.org',
+ hdrs.ACCESS_CONTROL_EXPOSE_HEADERS:
+ SERVER_CUSTOM_HEADER_NAME}.items():
+ assert resp.headers.get(hdr) == val
+
+ for header_name in {
+ hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
+ }:
+ assert header_name not in resp.headers
+
+
+async def test_simple_expose_headers_not_allowed_origin(aiohttp_client,
+ make_app):
+ app = make_app(None, {"http://client1.example.org":
+ ResourceOptions(
+ expose_headers=(SERVER_CUSTOM_HEADER_NAME,))})
+
+ client = await aiohttp_client(app)
+
+ resp = await client.get("/resource",
+ headers={hdrs.ORIGIN:
+ 'http://client2.example.org'})
+ assert resp.status == 200
+ resp_text = await resp.text()
+ assert resp_text == TEST_BODY
+
+ for header_name in {
+ hdrs.ACCESS_CONTROL_ALLOW_ORIGIN,
+ hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
+ hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
+ }:
+ assert header_name not in resp.headers
+
+
+async def test_preflight_default_no_origin(aiohttp_client, make_app):
+ app = make_app(None, {"http://client1.example.org":
+ ResourceOptions()})
+
+ client = await aiohttp_client(app)
+
+ resp = await client.options("/resource")
+ assert resp.status == 403
+ resp_text = await resp.text()
+ assert "origin header is not specified" in resp_text
+
+ for header_name in {
+ hdrs.ACCESS_CONTROL_ALLOW_ORIGIN,
+ hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
+ hdrs.ACCESS_CONTROL_MAX_AGE,
+ hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
+ hdrs.ACCESS_CONTROL_ALLOW_METHODS,
+ hdrs.ACCESS_CONTROL_ALLOW_HEADERS,
+ }:
+ assert header_name not in resp.headers
+
+
+async def test_preflight_default_no_method(aiohttp_client, make_app):
+
+ app = make_app(None, {"http://client1.example.org":
+ ResourceOptions()})
+
+ client = await aiohttp_client(app)
+
+ resp = await client.options("/resource", headers={
+ hdrs.ORIGIN: "http://client1.example.org",
+ })
+ assert resp.status == 403
+ resp_text = await resp.text()
+ assert "'Access-Control-Request-Method' header is not specified"\
+ in resp_text
+
+ for header_name in {
+ hdrs.ACCESS_CONTROL_ALLOW_ORIGIN,
+ hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
+ hdrs.ACCESS_CONTROL_MAX_AGE,
+ hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
+ hdrs.ACCESS_CONTROL_ALLOW_METHODS,
+ hdrs.ACCESS_CONTROL_ALLOW_HEADERS,
+ }:
+ assert header_name not in resp.headers
+
+
+async def test_preflight_default_origin_and_method(aiohttp_client, make_app):
+
+ app = make_app(None, {"http://client1.example.org":
+ ResourceOptions()})
+
+ client = await aiohttp_client(app)
+
+ resp = await client.options("/resource", headers={
+ hdrs.ORIGIN: "http://client1.example.org",
+ hdrs.ACCESS_CONTROL_REQUEST_METHOD: "GET",
+ })
+ assert resp.status == 200
+ resp_text = await resp.text()
+ assert '' == resp_text
+
+ for hdr, val in {
+ hdrs.ACCESS_CONTROL_ALLOW_ORIGIN: "http://client1.example.org",
+ hdrs.ACCESS_CONTROL_ALLOW_METHODS: "GET"}.items():
+ assert resp.headers.get(hdr) == val
+
+ for header_name in {
+ hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
+ hdrs.ACCESS_CONTROL_MAX_AGE,
+ hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
+ hdrs.ACCESS_CONTROL_ALLOW_HEADERS,
+ }:
+ assert header_name not in resp.headers
+
+
+async def test_preflight_default_disallowed_origin(aiohttp_client, make_app):
+
+ app = make_app(None, {"http://client1.example.org":
+ ResourceOptions()})
+
+ client = await aiohttp_client(app)
+
+ resp = await client.options("/resource", headers={
+ hdrs.ORIGIN: "http://client2.example.org",
+ hdrs.ACCESS_CONTROL_REQUEST_METHOD: "GET",
+ })
+ assert resp.status == 403
+ resp_text = await resp.text()
+ assert "origin 'http://client2.example.org' is not allowed" in resp_text
+
+ for header_name in {
+ hdrs.ACCESS_CONTROL_ALLOW_ORIGIN,
+ hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
+ hdrs.ACCESS_CONTROL_MAX_AGE,
+ hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
+ hdrs.ACCESS_CONTROL_ALLOW_METHODS,
+ hdrs.ACCESS_CONTROL_ALLOW_HEADERS,
+ }:
+ assert header_name not in resp.headers
+
+
+async def test_preflight_default_disallowed_method(aiohttp_client, make_app):
+
+ app = make_app(None, {"http://client1.example.org":
+ ResourceOptions()})
+
+ client = await aiohttp_client(app)
+
+ resp = await client.options("/resource", headers={
+ hdrs.ORIGIN: "http://client1.example.org",
+ hdrs.ACCESS_CONTROL_REQUEST_METHOD: "POST",
+ })
+ assert resp.status == 403
+ resp_text = await resp.text()
+ assert ("request method 'POST' is not allowed for "
+ "'http://client1.example.org' origin" in resp_text)
+
+ for header_name in {
+ hdrs.ACCESS_CONTROL_ALLOW_ORIGIN,
+ hdrs.ACCESS_CONTROL_ALLOW_CREDENTIALS,
+ hdrs.ACCESS_CONTROL_MAX_AGE,
+ hdrs.ACCESS_CONTROL_EXPOSE_HEADERS,
+ hdrs.ACCESS_CONTROL_ALLOW_METHODS,
+ hdrs.ACCESS_CONTROL_ALLOW_HEADERS,
+ }:
+ assert header_name not in resp.headers
+
+
+async def test_preflight_req_multiple_routes_with_one_options(aiohttp_client):
+ """Test CORS preflight handling on resource that is available through
+ several routes.
+ """
+ app = web.Application()
+ cors = _setup(app, defaults={
+ "*": ResourceOptions(
+ allow_credentials=True,
+ expose_headers="*",
+ allow_headers="*",
)
- self.assertEqual(response.status, 200)
+ })
- data = yield from response.text()
- self.assertEqual(data, "")
+ cors.add(app.router.add_route("GET", "/{name}", handler))
+ cors.add(app.router.add_route("PUT", "/{name}", handler))
- @asynctest
- @asyncio.coroutine
- def test_preflight_request_multiple_routes_with_one_options_resource(self):
- """Test CORS preflight handling on resource that is available through
- several routes.
- """
- app = web.Application()
- cors = setup(app, defaults={
- "*": ResourceOptions(
- allow_credentials=True,
- expose_headers="*",
- allow_headers="*",
- )
- })
-
- resource = cors.add(app.router.add_resource("/{name}"))
- cors.add(resource.add_route("GET", handler))
- cors.add(resource.add_route("PUT", handler))
-
- yield from self.create_server(app)
-
- response = yield from self.session.request(
- "OPTIONS", self.server_url + "user",
- headers={
- hdrs.ORIGIN: "http://example.org",
- hdrs.ACCESS_CONTROL_REQUEST_METHOD: "PUT"
- }
+ client = await aiohttp_client(app)
+
+ resp = await client.options(
+ "/user",
+ headers={
+ hdrs.ORIGIN: "http://example.org",
+ hdrs.ACCESS_CONTROL_REQUEST_METHOD: "PUT"
+ }
+ )
+ assert resp.status == 200
+
+ data = await resp.text()
+ assert data == ""
+
+
+async def test_preflight_request_mult_routes_with_one_options_resource(
+ aiohttp_client):
+ """Test CORS preflight handling on resource that is available through
+ several routes.
+ """
+ app = web.Application()
+ cors = _setup(app, defaults={
+ "*": ResourceOptions(
+ allow_credentials=True,
+ expose_headers="*",
+ allow_headers="*",
)
- self.assertEqual(response.status, 200)
+ })
- data = yield from response.text()
- self.assertEqual(data, "")
+ resource = cors.add(app.router.add_resource("/{name}"))
+ cors.add(resource.add_route("GET", handler))
+ cors.add(resource.add_route("PUT", handler))
- @asynctest
- @asyncio.coroutine
- def test_preflight_request_headers_resource(self):
- """Test CORS preflight request handlers handling."""
- app = web.Application()
- cors = setup(app, defaults={
- "*": ResourceOptions(
- allow_credentials=True,
- expose_headers="*",
- allow_headers=("Content-Type", "X-Header"),
- )
- })
-
- cors.add(app.router.add_route("PUT", "/", handler))
-
- yield from self.create_server(app)
-
- response = yield from self.session.request(
- "OPTIONS", self.server_url,
- headers={
- hdrs.ORIGIN: "http://example.org",
- hdrs.ACCESS_CONTROL_REQUEST_METHOD: "PUT",
- hdrs.ACCESS_CONTROL_REQUEST_HEADERS: "content-type",
- }
+ client = await aiohttp_client(app)
+
+ resp = await client.options(
+ "/user",
+ headers={
+ hdrs.ORIGIN: "http://example.org",
+ hdrs.ACCESS_CONTROL_REQUEST_METHOD: "PUT"
+ }
+ )
+ assert resp.status == 200
+
+ data = await resp.text()
+ assert data == ""
+
+
+async def test_preflight_request_max_age_resource(aiohttp_client):
+ """Test CORS preflight handling on resource that is available through
+ several routes.
+ """
+ app = web.Application()
+ cors = _setup(app, defaults={
+ "*": ResourceOptions(
+ allow_credentials=True,
+ expose_headers="*",
+ allow_headers="*",
+ max_age=1200
)
- self.assertEqual((yield from response.text()), "")
- self.assertEqual(response.status, 200)
- # Access-Control-Allow-Headers must be compared in case-insensitive
- # way.
- self.assertEqual(
- response.headers[hdrs.ACCESS_CONTROL_ALLOW_HEADERS].upper(),
- "content-type".upper())
+ })
+
+ resource = cors.add(app.router.add_resource("/{name}"))
+ cors.add(resource.add_route("GET", handler))
+
+ client = await aiohttp_client(app)
+
+ resp = await client.options(
+ "/user",
+ headers={
+ hdrs.ORIGIN: "http://example.org",
+ hdrs.ACCESS_CONTROL_REQUEST_METHOD: "GET"
+ }
+ )
+ assert resp.status == 200
+ assert resp.headers[hdrs.ACCESS_CONTROL_MAX_AGE].upper() == "1200"
+
+ data = await resp.text()
+ assert data == ""
- response = yield from self.session.request(
- "OPTIONS", self.server_url,
- headers={
- hdrs.ORIGIN: "http://example.org",
- hdrs.ACCESS_CONTROL_REQUEST_METHOD: "PUT",
- hdrs.ACCESS_CONTROL_REQUEST_HEADERS: "X-Header,content-type",
- }
+
+async def test_preflight_request_max_age_webview(aiohttp_client):
+ """Test CORS preflight handling on resource that is available through
+ several routes.
+ """
+ app = web.Application()
+ cors = _setup(app, defaults={
+ "*": ResourceOptions(
+ allow_credentials=True,
+ expose_headers="*",
+ allow_headers="*",
+ max_age=1200
)
- self.assertEqual(response.status, 200)
- # Access-Control-Allow-Headers must be compared in case-insensitive
- # way.
- self.assertEqual(
- frozenset(response.headers[hdrs.ACCESS_CONTROL_ALLOW_HEADERS]
- .upper().split(",")),
- {"X-Header".upper(), "content-type".upper()})
- self.assertEqual((yield from response.text()), "")
-
- response = yield from self.session.request(
- "OPTIONS", self.server_url,
- headers={
- hdrs.ORIGIN: "http://example.org",
- hdrs.ACCESS_CONTROL_REQUEST_METHOD: "PUT",
- hdrs.ACCESS_CONTROL_REQUEST_HEADERS: "content-type,Test",
- }
+ })
+
+ class TestView(web.View, CorsViewMixin):
+ async def get(self):
+ resp = web.Response(text=TEST_BODY)
+
+ resp.headers[SERVER_CUSTOM_HEADER_NAME] = \
+ SERVER_CUSTOM_HEADER_VALUE
+
+ return resp
+
+ cors.add(app.router.add_route("*", "/{name}", TestView))
+
+ client = await aiohttp_client(app)
+
+ resp = await client.options(
+ "/user",
+ headers={
+ hdrs.ORIGIN: "http://example.org",
+ hdrs.ACCESS_CONTROL_REQUEST_METHOD: "GET"
+ }
+ )
+ assert resp.status == 200
+ assert resp.headers[hdrs.ACCESS_CONTROL_MAX_AGE].upper() == "1200"
+
+ data = await resp.text()
+ assert data == ""
+
+
+async def test_preflight_request_mult_routes_with_one_options_webview(
+ aiohttp_client):
+ """Test CORS preflight handling on resource that is available through
+ several routes.
+ """
+ app = web.Application()
+ cors = _setup(app, defaults={
+ "*": ResourceOptions(
+ allow_credentials=True,
+ expose_headers="*",
+ allow_headers="*",
)
- self.assertEqual(response.status, 403)
- self.assertNotIn(
- hdrs.ACCESS_CONTROL_ALLOW_HEADERS,
- response.headers)
- self.assertIn(
- "headers are not allowed: TEST",
- (yield from response.text()))
-
- @asynctest
- @asyncio.coroutine
- def test_preflight_request_headers(self):
- """Test CORS preflight request handlers handling."""
- app = web.Application()
- cors = setup(app, defaults={
- "*": ResourceOptions(
- allow_credentials=True,
- expose_headers="*",
- allow_headers=("Content-Type", "X-Header"),
- )
- })
-
- resource = cors.add(app.router.add_resource("/"))
- cors.add(resource.add_route("PUT", handler))
-
- yield from self.create_server(app)
-
- response = yield from self.session.request(
- "OPTIONS", self.server_url,
- headers={
- hdrs.ORIGIN: "http://example.org",
- hdrs.ACCESS_CONTROL_REQUEST_METHOD: "PUT",
- hdrs.ACCESS_CONTROL_REQUEST_HEADERS: "content-type",
- }
+ })
+
+ class TestView(web.View, CorsViewMixin):
+ async def get(self):
+ resp = web.Response(text=TEST_BODY)
+
+ resp.headers[SERVER_CUSTOM_HEADER_NAME] = \
+ SERVER_CUSTOM_HEADER_VALUE
+
+ return resp
+
+ put = get
+
+ cors.add(app.router.add_route("*", "/{name}", TestView))
+
+ client = await aiohttp_client(app)
+
+ resp = await client.options(
+ "/user",
+ headers={
+ hdrs.ORIGIN: "http://example.org",
+ hdrs.ACCESS_CONTROL_REQUEST_METHOD: "PUT"
+ }
+ )
+ assert resp.status == 200
+
+ data = await resp.text()
+ assert data == ""
+
+
+async def test_preflight_request_headers_webview(aiohttp_client):
+ """Test CORS preflight request handlers handling."""
+ app = web.Application()
+ cors = _setup(app, defaults={
+ "*": ResourceOptions(
+ allow_credentials=True,
+ expose_headers="*",
+ allow_headers=("Content-Type", "X-Header"),
)
- self.assertEqual((yield from response.text()), "")
- self.assertEqual(response.status, 200)
- # Access-Control-Allow-Headers must be compared in case-insensitive
- # way.
- self.assertEqual(
- response.headers[hdrs.ACCESS_CONTROL_ALLOW_HEADERS].upper(),
+ })
+
+ class TestView(web.View, CorsViewMixin):
+ async def put(self):
+ response = web.Response(text=TEST_BODY)
+
+ response.headers[SERVER_CUSTOM_HEADER_NAME] = \
+ SERVER_CUSTOM_HEADER_VALUE
+
+ return response
+
+ cors.add(app.router.add_route("*", "/", TestView))
+
+ client = await aiohttp_client(app)
+
+ resp = await client.options(
+ '/',
+ headers={
+ hdrs.ORIGIN: "http://example.org",
+ hdrs.ACCESS_CONTROL_REQUEST_METHOD: "PUT",
+ hdrs.ACCESS_CONTROL_REQUEST_HEADERS: "content-type",
+ }
+ )
+ assert (await resp.text()) == ""
+ assert resp.status == 200
+ # Access-Control-Allow-Headers must be compared in case-insensitive
+ # way.
+ assert (resp.headers[hdrs.ACCESS_CONTROL_ALLOW_HEADERS].upper() ==
"content-type".upper())
- response = yield from self.session.request(
- "OPTIONS", self.server_url,
- headers={
- hdrs.ORIGIN: "http://example.org",
- hdrs.ACCESS_CONTROL_REQUEST_METHOD: "PUT",
- hdrs.ACCESS_CONTROL_REQUEST_HEADERS: "X-Header,content-type",
- }
+ resp = await client.options(
+ '/',
+ headers={
+ hdrs.ORIGIN: "http://example.org",
+ hdrs.ACCESS_CONTROL_REQUEST_METHOD: "PUT",
+ hdrs.ACCESS_CONTROL_REQUEST_HEADERS: "X-Header,content-type",
+ }
+ )
+ assert resp.status == 200
+ # Access-Control-Allow-Headers must be compared in case-insensitive
+ # way.
+ assert (
+ frozenset(resp.headers[hdrs.ACCESS_CONTROL_ALLOW_HEADERS]
+ .upper().split(",")) ==
+ {"X-Header".upper(), "content-type".upper()})
+ assert (await resp.text()) == ""
+
+ resp = await client.options(
+ '/',
+ headers={
+ hdrs.ORIGIN: "http://example.org",
+ hdrs.ACCESS_CONTROL_REQUEST_METHOD: "PUT",
+ hdrs.ACCESS_CONTROL_REQUEST_HEADERS: "content-type,Test",
+ }
+ )
+ assert resp.status == 403
+ assert hdrs.ACCESS_CONTROL_ALLOW_HEADERS not in resp.headers
+ assert "headers are not allowed: TEST" in (await resp.text())
+
+
+async def test_preflight_request_headers_resource(aiohttp_client):
+ """Test CORS preflight request handlers handling."""
+ app = web.Application()
+ cors = _setup(app, defaults={
+ "*": ResourceOptions(
+ allow_credentials=True,
+ expose_headers="*",
+ allow_headers=("Content-Type", "X-Header"),
)
- self.assertEqual(response.status, 200)
- # Access-Control-Allow-Headers must be compared in case-insensitive
- # way.
- self.assertEqual(
- frozenset(response.headers[hdrs.ACCESS_CONTROL_ALLOW_HEADERS]
- .upper().split(",")),
- {"X-Header".upper(), "content-type".upper()})
- self.assertEqual((yield from response.text()), "")
-
- response = yield from self.session.request(
- "OPTIONS", self.server_url,
- headers={
- hdrs.ORIGIN: "http://example.org",
- hdrs.ACCESS_CONTROL_REQUEST_METHOD: "PUT",
- hdrs.ACCESS_CONTROL_REQUEST_HEADERS: "content-type,Test",
- }
+ })
+
+ cors.add(app.router.add_route("PUT", "/", handler))
+
+ client = await aiohttp_client(app)
+
+ resp = await client.options(
+ '/',
+ headers={
+ hdrs.ORIGIN: "http://example.org",
+ hdrs.ACCESS_CONTROL_REQUEST_METHOD: "PUT",
+ hdrs.ACCESS_CONTROL_REQUEST_HEADERS: "content-type",
+ }
+ )
+ assert (await resp.text()) == ""
+ assert resp.status == 200
+ # Access-Control-Allow-Headers must be compared in case-insensitive
+ # way.
+ assert (
+ resp.headers[hdrs.ACCESS_CONTROL_ALLOW_HEADERS].upper() ==
+ "content-type".upper())
+
+ resp = await client.options(
+ '/',
+ headers={
+ hdrs.ORIGIN: "http://example.org",
+ hdrs.ACCESS_CONTROL_REQUEST_METHOD: "PUT",
+ hdrs.ACCESS_CONTROL_REQUEST_HEADERS: "X-Header,content-type",
+ }
+ )
+ assert resp.status == 200
+ # Access-Control-Allow-Headers must be compared in case-insensitive
+ # way.
+ assert (
+ frozenset(resp.headers[hdrs.ACCESS_CONTROL_ALLOW_HEADERS]
+ .upper().split(",")) ==
+ {"X-Header".upper(), "content-type".upper()})
+ assert (await resp.text()) == ""
+
+ resp = await client.options(
+ '/',
+ headers={
+ hdrs.ORIGIN: "http://example.org",
+ hdrs.ACCESS_CONTROL_REQUEST_METHOD: "PUT",
+ hdrs.ACCESS_CONTROL_REQUEST_HEADERS: "content-type,Test",
+ }
+ )
+ assert resp.status == 403
+ assert hdrs.ACCESS_CONTROL_ALLOW_HEADERS not in resp.headers
+ assert "headers are not allowed: TEST" in (await resp.text())
+
+
+async def test_preflight_request_headers(aiohttp_client):
+ """Test CORS preflight request handlers handling."""
+ app = web.Application()
+ cors = _setup(app, defaults={
+ "*": ResourceOptions(
+ allow_credentials=True,
+ expose_headers="*",
+ allow_headers=("Content-Type", "X-Header"),
)
- self.assertEqual(response.status, 403)
- self.assertNotIn(
- hdrs.ACCESS_CONTROL_ALLOW_HEADERS,
- response.headers)
- self.assertIn(
- "headers are not allowed: TEST",
- (yield from response.text()))
-
- @asynctest
- @asyncio.coroutine
- def test_static_route(self):
- """Test a static route with CORS."""
- app = web.Application()
- cors = setup(app, defaults={
- "*": ResourceOptions(
- allow_credentials=True,
- expose_headers="*",
- allow_methods="*",
- allow_headers=("Content-Type", "X-Header"),
- )
- })
-
- test_static_path = pathlib.Path(__file__).parent
- cors.add(app.router.add_static("/static", test_static_path, name='static'))
-
- yield from self.create_server(app)
-
- response = yield from self.session.request(
- "OPTIONS", URL(self.server_url) / "static/test_page.html",
- headers={
- hdrs.ORIGIN: "http://example.org",
- hdrs.ACCESS_CONTROL_REQUEST_METHOD: "OPTIONS",
- hdrs.ACCESS_CONTROL_REQUEST_HEADERS: "content-type",
- }
+ })
+
+ resource = cors.add(app.router.add_resource("/"))
+ cors.add(resource.add_route("PUT", handler))
+
+ client = await aiohttp_client(app)
+
+ resp = await client.options(
+ '/',
+ headers={
+ hdrs.ORIGIN: "http://example.org",
+ hdrs.ACCESS_CONTROL_REQUEST_METHOD: "PUT",
+ hdrs.ACCESS_CONTROL_REQUEST_HEADERS: "content-type",
+ }
+ )
+ assert (await resp.text()) == ""
+ assert resp.status == 200
+ # Access-Control-Allow-Headers must be compared in case-insensitive
+ # way.
+ assert (
+ resp.headers[hdrs.ACCESS_CONTROL_ALLOW_HEADERS].upper() ==
+ "content-type".upper())
+
+ resp = await client.options(
+ '/',
+ headers={
+ hdrs.ORIGIN: "http://example.org",
+ hdrs.ACCESS_CONTROL_REQUEST_METHOD: "PUT",
+ hdrs.ACCESS_CONTROL_REQUEST_HEADERS: "X-Header,content-type",
+ }
+ )
+ assert resp.status == 200
+ # Access-Control-Allow-Headers must be compared in case-insensitive
+ # way.
+ assert (
+ frozenset(resp.headers[hdrs.ACCESS_CONTROL_ALLOW_HEADERS]
+ .upper().split(",")) ==
+ {"X-Header".upper(), "content-type".upper()})
+ assert (await resp.text()) == ""
+
+ resp = await client.options(
+ '/',
+ headers={
+ hdrs.ORIGIN: "http://example.org",
+ hdrs.ACCESS_CONTROL_REQUEST_METHOD: "PUT",
+ hdrs.ACCESS_CONTROL_REQUEST_HEADERS: "content-type,Test",
+ }
+ )
+ assert resp.status == 403
+ assert hdrs.ACCESS_CONTROL_ALLOW_HEADERS not in resp.headers
+ assert "headers are not allowed: TEST" in (await resp.text())
+
+
+async def test_static_route(aiohttp_client):
+ """Test a static route with CORS."""
+ app = web.Application()
+ cors = _setup(app, defaults={
+ "*": ResourceOptions(
+ allow_credentials=True,
+ expose_headers="*",
+ allow_methods="*",
+ allow_headers=("Content-Type", "X-Header"),
)
- data = yield from response.text()
- self.assertEqual(response.status, 200)
- self.assertEqual(data, '')
+ })
+
+ test_static_path = pathlib.Path(__file__).parent
+ cors.add(app.router.add_static("/static", test_static_path,
+ name='static'))
+
+ client = await aiohttp_client(app)
+
+ resp = await client.options(
+ "/static/test_page.html",
+ headers={
+ hdrs.ORIGIN: "http://example.org",
+ hdrs.ACCESS_CONTROL_REQUEST_METHOD: "OPTIONS",
+ hdrs.ACCESS_CONTROL_REQUEST_HEADERS: "content-type",
+ }
+ )
+ data = await resp.text()
+ assert resp.status == 200
+ assert data == ''
# TODO: test requesting resources with not configured CORS.
generated by cgit v1.2.3 (git 2.25.1) at 2024-06-02 14:43:07 +0000