woggioni
031d933204
pwo.async_test runs every test in a fresh event loop and TortoiseMixin builds one TortoiseContext per loop. Dropping the previous context when the loop changed orphaned its aiosqlite connections; their non-daemon worker threads then blocked threading._shutdown forever, so the suite printed OK but the interpreter never exited. Add TortoiseMixin.aclose() and a tests.helpers.async_test wrapper that closes the context in a finally on the test's own loop, and switch the test modules over to it.
130 lines
5.3 KiB
Python
130 lines
5.3 KiB
Python
"""Integration tests for the CORS configuration in :mod:`tavolo.app`.
|
|
|
|
The mixin under test is kaya-cors' :class:`~kaya.cors.CorsMixin`; these
|
|
tests only verify that :func:`tavolo.app.cors_mixin_from_settings` maps the
|
|
environment-driven :class:`~tavolo.config.Settings` onto it correctly. A
|
|
minimal ``KayaApp`` is used instead of the global ``app`` so the tests do
|
|
not depend on the environment the suite was imported with.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from httpx import ASGITransport, AsyncClient
|
|
from kaya.core import HttpContext, KayaApp
|
|
|
|
from tavolo.app import cors_mixin_from_settings
|
|
from tavolo.config import Settings
|
|
from tests.helpers import async_test
|
|
|
|
ORIGIN = "https://cards.example"
|
|
|
|
|
|
def _settings(env: dict) -> Settings:
|
|
with patch.dict(os.environ, env, clear=True):
|
|
return Settings.from_env()
|
|
|
|
|
|
def _app(settings: Settings) -> KayaApp:
|
|
mixin = cors_mixin_from_settings(settings)
|
|
assert mixin is not None
|
|
app = KayaApp(mixins=[mixin])
|
|
|
|
@app.GET("/api/health")
|
|
async def health(ctx: HttpContext) -> None:
|
|
await ctx.send_str(200, "ok")
|
|
|
|
return app
|
|
|
|
|
|
class CorsMixinFromSettingsTests(unittest.TestCase):
|
|
def test_disabled_when_unconfigured(self):
|
|
self.assertIsNone(cors_mixin_from_settings(_settings({})))
|
|
|
|
def test_enabled_by_allow_origins(self):
|
|
self.assertIsNotNone(cors_mixin_from_settings(
|
|
_settings({"CORS_ALLOW_ORIGINS": ORIGIN})))
|
|
|
|
def test_enabled_by_allow_origin_regex_alone(self):
|
|
self.assertIsNotNone(cors_mixin_from_settings(
|
|
_settings({"CORS_ALLOW_ORIGIN_REGEX": r"https://.*\.example\.com"})))
|
|
|
|
|
|
class CorsBehaviorTests(unittest.TestCase):
|
|
@async_test
|
|
async def test_request_without_origin_is_untouched(self) -> None:
|
|
app = _app(_settings({"CORS_ALLOW_ORIGINS": ORIGIN}))
|
|
async with AsyncClient(transport=ASGITransport(app=app),
|
|
base_url="http://127.0.0.1") as client:
|
|
response = await client.get("/api/health")
|
|
self.assertEqual(200, response.status_code)
|
|
self.assertNotIn("access-control-allow-origin", response.headers)
|
|
|
|
@async_test
|
|
async def test_simple_request_with_allowed_origin(self) -> None:
|
|
app = _app(_settings({"CORS_ALLOW_ORIGINS": ORIGIN}))
|
|
async with AsyncClient(transport=ASGITransport(app=app),
|
|
base_url="http://127.0.0.1") as client:
|
|
response = await client.get("/api/health", headers={"Origin": ORIGIN})
|
|
self.assertEqual(200, response.status_code)
|
|
self.assertEqual(ORIGIN, response.headers["access-control-allow-origin"])
|
|
|
|
@async_test
|
|
async def test_simple_request_with_disallowed_origin(self) -> None:
|
|
app = _app(_settings({"CORS_ALLOW_ORIGINS": ORIGIN}))
|
|
async with AsyncClient(transport=ASGITransport(app=app),
|
|
base_url="http://127.0.0.1") as client:
|
|
response = await client.get(
|
|
"/api/health", headers={"Origin": "https://mallory.example"})
|
|
self.assertEqual(200, response.status_code)
|
|
self.assertNotIn("access-control-allow-origin", response.headers)
|
|
|
|
@async_test
|
|
async def test_preflight_allowed(self) -> None:
|
|
app = _app(_settings({
|
|
"CORS_ALLOW_ORIGINS": ORIGIN,
|
|
"CORS_ALLOW_METHODS": "GET,POST",
|
|
"CORS_MAX_AGE": "3600",
|
|
}))
|
|
async with AsyncClient(transport=ASGITransport(app=app),
|
|
base_url="http://127.0.0.1") as client:
|
|
response = await client.options("/api/health", headers={
|
|
"Origin": ORIGIN,
|
|
"Access-Control-Request-Method": "POST",
|
|
})
|
|
self.assertEqual(200, response.status_code)
|
|
self.assertEqual(ORIGIN, response.headers["access-control-allow-origin"])
|
|
self.assertEqual("GET, POST", response.headers["access-control-allow-methods"])
|
|
self.assertEqual("3600", response.headers["access-control-max-age"])
|
|
|
|
@async_test
|
|
async def test_preflight_disallowed_origin(self) -> None:
|
|
app = _app(_settings({"CORS_ALLOW_ORIGINS": ORIGIN}))
|
|
async with AsyncClient(transport=ASGITransport(app=app),
|
|
base_url="http://127.0.0.1") as client:
|
|
response = await client.options("/api/health", headers={
|
|
"Origin": "https://mallory.example",
|
|
"Access-Control-Request-Method": "GET",
|
|
})
|
|
self.assertEqual(400, response.status_code)
|
|
self.assertIn("Disallowed CORS", response.text)
|
|
|
|
@async_test
|
|
async def test_credentials_echo_origin_and_set_flag(self) -> None:
|
|
app = _app(_settings({
|
|
"CORS_ALLOW_ORIGINS": "*",
|
|
"CORS_ALLOW_CREDENTIALS": "true",
|
|
}))
|
|
async with AsyncClient(transport=ASGITransport(app=app),
|
|
base_url="http://127.0.0.1") as client:
|
|
response = await client.get("/api/health", headers={"Origin": ORIGIN})
|
|
self.assertEqual(200, response.status_code)
|
|
self.assertEqual(ORIGIN, response.headers["access-control-allow-origin"])
|
|
self.assertEqual("true", response.headers["access-control-allow-credentials"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|