diff --git a/oxyroute/asgi.py b/oxyroute/asgi.py index 6a6f094..e135dbd 100644 --- a/oxyroute/asgi.py +++ b/oxyroute/asgi.py @@ -8,6 +8,7 @@ from __future__ import annotations import asyncio +import os from collections.abc import Callable from typing import Any @@ -246,6 +247,23 @@ def response_empty(self, status: int, headers: list) -> None: ) +def _max_body_bytes() -> int: + # Keep semantics aligned with Rust path (`src/form.rs`): default 8 MiB, `0` means unlimited. + default = 8 * 1024 * 1024 + raw = os.getenv("OXYROUTE_MAX_BODY_BYTES", "").strip() + if not raw: + return default + try: + n = int(raw) + except ValueError: + return default + if n == 0: + return 2**63 - 1 + if n < 0: + return default + return n + + async def asgi_to_rsgi( app_rsgi: Callable[[Any, Any], Any], app_ws: Callable[[dict[str, Any], Any, Any], Any] | None, @@ -262,12 +280,32 @@ async def asgi_to_rsgi( return if st != "http": return - body = b"" + max_body = _max_body_bytes() + body = bytearray() while True: m = await receive() t = m.get("type", "") if t == "http.request": - body += m.get("body", b"") + chunk = m.get("body", b"") + if chunk: + body.extend(chunk) + if len(body) > max_body: + await send( + { + "type": "http.response.start", + "status": 413, + "headers": [ + (b"content-type", b"application/json; charset=utf-8"), + ], + } + ) + await send( + { + "type": "http.response.body", + "body": b'{"error":"payload too large"}', + } + ) + return if not m.get("more_body", False): break elif t == "http.disconnect": @@ -297,7 +335,7 @@ async def _drain_outgoing() -> None: await send(msg) drain_task = asyncio.create_task(_drain_outgoing()) - proto = _RsgiProtocol(body, queue, loop) + proto = _RsgiProtocol(bytes(body), queue, loop) run_exc: BaseException | None = None try: await loop.run_in_executor( diff --git a/tests/test_app_units.py b/tests/test_app_units.py new file mode 100644 index 0000000..97e137d --- /dev/null +++ b/tests/test_app_units.py @@ -0,0 +1,50 @@ +"""Extra App wrapper branch coverage.""" + +from __future__ import annotations + +import asyncio + +from oxyroute.app import App, Depends, _norm_dependencies, _unwrap_dep + + +def test_depends_and_unwrap_helpers() -> None: + def dep() -> int: + return 1 + + d = Depends(dep) + assert _unwrap_dep(d) is dep + assert _unwrap_dep(dep) is dep + assert _norm_dependencies(None) is None + assert _norm_dependencies([]) is None + norm = _norm_dependencies([("x", d)]) + assert norm is not None + assert norm[0][0] == "x" + assert norm[0][1] is dep + + +def test_app_freeze_and_base_rsgi_noops() -> None: + app = App() + app.freeze() + assert asyncio.run(app.__rsgi_init__()) is None + assert asyncio.run(app.__rsgi_del__()) is None + + +def test_put_and_delete_decorators_register_routes() -> None: + app = App() + + @app.put("/u") + def upd() -> str: + return "ok" + + @app.delete("/u") + def rem() -> str: + return "ok" + + assert upd() == "ok" + assert rem() == "ok" + + +def test_rsgi_forwarder_methods_exist() -> None: + app = App() + assert callable(app.handle_rsgi) + assert callable(app.__rsgi__) diff --git a/tests/test_asgi.py b/tests/test_asgi.py index 85c62c2..4afeeb8 100644 --- a/tests/test_asgi.py +++ b/tests/test_asgi.py @@ -117,3 +117,107 @@ async def _run() -> None: asyncio.run(_run()) finally: asgi_mod._run_handle_rsgi_blocking = original + + +def test_asgi_chunked_request_body_joins_without_quadratic_path_assumption() -> None: + app = App() + + @app.patch("/x") + def patch_x(json: dict) -> str: + return f"p:{json.get('a')}" + + scope = { + "type": "http", + "http_version": "1.1", + "scheme": "http", + "method": "PATCH", + "path": "/x", + "query_string": b"", + "headers": [(b"content-type", b"application/json")], + } + + incoming = iter( + [ + {"type": "http.request", "body": b'{"a":', "more_body": True}, + {"type": "http.request", "body": b"7}", "more_body": False}, + ] + ) + sent: list[dict] = [] + + async def receive() -> dict: + return next(incoming) + + async def send(message: dict) -> None: + sent.append(message) + + async def _run() -> None: + await app(scope, receive, send) + + asyncio.run(_run()) + assert sent[0]["type"] == "http.response.start" + assert sent[0]["status"] == 200 + assert sent[1]["type"] == "http.response.body" + assert sent[1]["body"] == b"p:7" + + +def test_asgi_body_limit_enforced_during_chunk_read() -> None: + app = App() + + @app.post("/x") + def x(body: bytes) -> str: + return body.decode("utf-8") + + scope = { + "type": "http", + "http_version": "1.1", + "scheme": "http", + "method": "POST", + "path": "/x", + "query_string": b"", + "headers": [], + } + + incoming = iter( + [ + {"type": "http.request", "body": b"abc", "more_body": True}, + {"type": "http.request", "body": b"de", "more_body": False}, + ] + ) + sent: list[dict] = [] + + async def receive() -> dict: + return next(incoming) + + async def send(message: dict) -> None: + sent.append(message) + + original = asgi_mod._max_body_bytes + asgi_mod._max_body_bytes = lambda: 4 + + async def _run() -> None: + await app(scope, receive, send) + + try: + asyncio.run(_run()) + finally: + asgi_mod._max_body_bytes = original + + assert sent[0]["type"] == "http.response.start" + assert sent[0]["status"] == 413 + assert sent[1]["type"] == "http.response.body" + assert sent[1]["body"] == b'{"error":"payload too large"}' + + +def test_asgi_non_http_non_websocket_scope_is_ignored() -> None: + app = App() + sent: list[dict] = [] + + async def receive() -> dict: + return {"type": "lifespan.startup"} + + async def send(message: dict) -> None: + sent.append(message) + + scope = {"type": "lifespan"} + asyncio.run(app(scope, receive, send)) + assert sent == [] diff --git a/tests/test_asgi_body_limits.py b/tests/test_asgi_body_limits.py new file mode 100644 index 0000000..0676a3c --- /dev/null +++ b/tests/test_asgi_body_limits.py @@ -0,0 +1,170 @@ +"""Extended ASGI body-limit and chunk-read coverage.""" + +from __future__ import annotations + +import asyncio +import os +from collections import deque +from typing import Any + +import oxyroute.asgi as asgi_mod +from oxyroute import App + + +def test_max_body_bytes_default() -> None: + old = os.environ.pop("OXYROUTE_MAX_BODY_BYTES", None) + try: + assert asgi_mod._max_body_bytes() == 8 * 1024 * 1024 + finally: + if old is not None: + os.environ["OXYROUTE_MAX_BODY_BYTES"] = old + + +def test_max_body_bytes_invalid_and_negative_fallback_to_default() -> None: + old = os.environ.get("OXYROUTE_MAX_BODY_BYTES") + try: + os.environ["OXYROUTE_MAX_BODY_BYTES"] = "abc" + assert asgi_mod._max_body_bytes() == 8 * 1024 * 1024 + os.environ["OXYROUTE_MAX_BODY_BYTES"] = "-1" + assert asgi_mod._max_body_bytes() == 8 * 1024 * 1024 + finally: + if old is None: + os.environ.pop("OXYROUTE_MAX_BODY_BYTES", None) + else: + os.environ["OXYROUTE_MAX_BODY_BYTES"] = old + + +def test_max_body_bytes_zero_means_unlimited() -> None: + old = os.environ.get("OXYROUTE_MAX_BODY_BYTES") + try: + os.environ["OXYROUTE_MAX_BODY_BYTES"] = "0" + assert asgi_mod._max_body_bytes() > 10**12 + finally: + if old is None: + os.environ.pop("OXYROUTE_MAX_BODY_BYTES", None) + else: + os.environ["OXYROUTE_MAX_BODY_BYTES"] = old + + +def test_asgi_body_limit_exact_boundary_passes() -> None: + app = App() + + @app.post("/raw", read_json_body=False) + def raw(body: bytes) -> str: + return body.decode("utf-8") + + scope = { + "type": "http", + "http_version": "1.1", + "scheme": "http", + "method": "POST", + "path": "/raw", + "query_string": b"", + "headers": [], + } + incoming = deque( + [ + {"type": "http.request", "body": b"ab", "more_body": True}, + {"type": "http.request", "body": b"cd", "more_body": False}, + ] + ) + sent: list[dict[str, Any]] = [] + + async def receive() -> dict[str, Any]: + return incoming.popleft() + + async def send(msg: dict[str, Any]) -> None: + sent.append(msg) + + old = os.environ.get("OXYROUTE_MAX_BODY_BYTES") + try: + os.environ["OXYROUTE_MAX_BODY_BYTES"] = "4" + asyncio.run(app(scope, receive, send)) + finally: + if old is None: + os.environ.pop("OXYROUTE_MAX_BODY_BYTES", None) + else: + os.environ["OXYROUTE_MAX_BODY_BYTES"] = old + + assert sent[0]["status"] == 200 + assert sent[1]["body"] == b"abcd" + + +def test_asgi_body_limit_over_boundary_returns_413_and_json_ct() -> None: + app = App() + + @app.post("/raw", read_json_body=False) + def raw(body: bytes) -> str: # pragma: no cover + return body.decode("utf-8") + + scope = { + "type": "http", + "http_version": "1.1", + "scheme": "http", + "method": "POST", + "path": "/raw", + "query_string": b"", + "headers": [], + } + incoming = deque( + [ + {"type": "http.request", "body": b"ab", "more_body": True}, + {"type": "http.request", "body": b"cde", "more_body": False}, + ] + ) + sent: list[dict[str, Any]] = [] + + async def receive() -> dict[str, Any]: + return incoming.popleft() + + async def send(msg: dict[str, Any]) -> None: + sent.append(msg) + + old = os.environ.get("OXYROUTE_MAX_BODY_BYTES") + try: + os.environ["OXYROUTE_MAX_BODY_BYTES"] = "4" + asyncio.run(app(scope, receive, send)) + finally: + if old is None: + os.environ.pop("OXYROUTE_MAX_BODY_BYTES", None) + else: + os.environ["OXYROUTE_MAX_BODY_BYTES"] = old + + assert sent[0]["type"] == "http.response.start" + assert sent[0]["status"] == 413 + assert (b"content-type", b"application/json; charset=utf-8") in sent[0]["headers"] + assert sent[1]["body"] == b'{"error":"payload too large"}' + + +def test_asgi_http_disconnect_during_read_sends_nothing() -> None: + app = App() + + @app.post("/raw", read_json_body=False) + def raw(body: bytes) -> str: # pragma: no cover + return body.decode("utf-8") + + scope = { + "type": "http", + "http_version": "1.1", + "scheme": "http", + "method": "POST", + "path": "/raw", + "query_string": b"", + "headers": [], + } + incoming = deque( + [ + {"type": "http.request", "body": b"abc", "more_body": True}, + {"type": "http.disconnect"}, + ] + ) + sent: list[dict[str, Any]] = [] + + async def receive() -> dict[str, Any]: + return incoming.popleft() + + async def send(msg: dict[str, Any]) -> None: + sent.append(msg) + + asyncio.run(app(scope, receive, send)) + assert sent == [] diff --git a/tests/test_asgi_internals.py b/tests/test_asgi_internals.py new file mode 100644 index 0000000..7936255 --- /dev/null +++ b/tests/test_asgi_internals.py @@ -0,0 +1,218 @@ +"""Unit coverage for ASGI bridge helper internals.""" + +from __future__ import annotations + +import asyncio + +import oxyroute.asgi as asgi_mod +import pytest + + +def test_norm_headers_asgi_filters_invalid_pairs() -> None: + out = asgi_mod._norm_headers_asgi( + [ + ("A", "1"), + (b"B", b"2"), + ("C", b"3"), + ("D", 4), + ("bad",), + "x", + ] + ) + assert out == [(b"A", b"1"), (b"B", b"2"), (b"C", b"3")] + + +def test_hdr_from_asgi_lowercases_and_get_lax() -> None: + hv = asgi_mod._hdr_from_asgi([(b"Origin", b"https://a.example"), (b"X", b"1")]) + assert hv.get("origin") == "https://a.example" + assert hv.get("ORIGIN") == "https://a.example" + assert hv.get("missing", "d") == "d" + + +def test_rsgi_scope_shape() -> None: + hv = asgi_mod._hdr_from_asgi([]) + s = asgi_mod._RsgiScope("http", "GET", "/x", "a=1", hv) + assert s.proto == "http" + assert s.http_version == "1.1" + assert s.method == "GET" + assert s.path == "/x" + assert s.query_string == "a=1" + assert s.headers.get("none", "") == "" + + +def test_websocket_send_and_close_idempotent() -> None: + sent: list[dict] = [] + + async def receive() -> dict: + return {"type": "websocket.connect"} + + async def send(msg: dict) -> None: + sent.append(msg) + + ws = asgi_mod.WebSocket(receive, send) + + async def _run() -> None: + await ws.send_text("a") + await ws.send_bytes(b"x") + await ws.close(code=1001) + await ws.send_text("ignored") + await ws.close(code=1000) + + asyncio.run(_run()) + assert sent[0] == {"type": "websocket.send", "text": "a"} + assert sent[1] == {"type": "websocket.send", "bytes": b"x"} + assert sent[2] == {"type": "websocket.close", "code": 1001} + assert len(sent) == 3 + + +def test_websocket_receive_returns_raw_message() -> None: + incoming = iter([{"type": "websocket.receive", "text": "ok"}]) + + async def receive() -> dict: + return next(incoming) + + async def send(_msg: dict) -> None: + return None + + ws = asgi_mod.WebSocket(receive, send) + out = asyncio.run(ws.receive()) + assert out["type"] == "websocket.receive" + assert out["text"] == "ok" + + +def test_websocket_accept_subprotocol_and_idempotent_second_accept() -> None: + sent: list[dict] = [] + incoming = iter( + [ + {"type": "websocket.connect"}, + {"type": "websocket.receive", "text": "msg"}, + ] + ) + + async def receive() -> dict: + return next(incoming) + + async def send(msg: dict) -> None: + sent.append(msg) + + async def _run() -> None: + ws = asgi_mod.WebSocket(receive, send) + await ws.accept(subprotocol="chat") + await ws.accept(subprotocol="chat") # no-op + assert await ws.receive_text() == "msg" + + asyncio.run(_run()) + assert sent == [{"type": "websocket.accept", "subprotocol": "chat"}] + + +def test_websocket_accept_raises_on_unexpected_pre_accept_event() -> None: + async def receive() -> dict: + return {"type": "websocket.receive", "text": "x"} + + async def send(_msg: dict) -> None: + return None + + async def _run() -> None: + ws = asgi_mod.WebSocket(receive, send) + with pytest.raises(RuntimeError, match="unexpected websocket event before accept"): + await ws.accept() + + asyncio.run(_run()) + + +def test_websocket_receive_text_disconnect_raises() -> None: + async def receive() -> dict: + return {"type": "websocket.disconnect"} + + async def send(_msg: dict) -> None: + return None + + async def _run() -> None: + ws = asgi_mod.WebSocket(receive, send) + with pytest.raises(RuntimeError, match="disconnected"): + await ws.receive_text() + + asyncio.run(_run()) + + +def test_rsgi_protocol_response_bytes_and_empty_enqueue() -> None: + async def _run() -> None: + loop = asyncio.get_running_loop() + q: asyncio.Queue[dict | None] = asyncio.Queue() + p = asgi_mod._RsgiProtocol(b"", q, loop) + p.response_bytes(201, [("x", "1")], b"abc") + p.response_empty(204, []) + msg1 = await q.get() + msg2 = await q.get() + msg3 = await q.get() + msg4 = await q.get() + assert msg1 == { + "type": "http.response.start", + "status": 201, + "headers": [(b"x", b"1")], + } + assert msg2 == {"type": "http.response.body", "body": b"abc"} + assert msg3 == {"type": "http.response.start", "status": 204, "headers": []} + assert msg4 == {"type": "http.response.body"} + + asyncio.run(_run()) + + +def test_build_asgi_caller_non_callable_ws_handler_closes() -> None: + class _Framework: + async def handle_rsgi(self, _scope: object, _proto: object) -> None: + return None + + _handle_asgi_websocket = "not-callable" + + app = asgi_mod.build_asgi_caller(_Framework()) + sent: list[dict] = [] + + async def receive() -> dict: + return {"type": "websocket.connect"} + + async def send(msg: dict) -> None: + sent.append(msg) + + scope = {"type": "websocket", "path": "/x", "headers": [], "query_string": b""} + asyncio.run(app(scope, receive, send)) + assert sent == [{"type": "websocket.close", "code": 1000}] + + +def test_build_asgi_caller_uses_inner_app_when_handle_rsgi_missing() -> None: + called: list[str] = [] + + class _Inner: + def handle_rsgi(self, _scope: object, _proto: object) -> object: + called.append("inner") + + async def _noop() -> None: + return None + + return _noop() + + class _Framework: + def __init__(self) -> None: + self._app = _Inner() + + app = asgi_mod.build_asgi_caller(_Framework()) + sent: list[dict] = [] + incoming = iter([{"type": "http.request", "body": b"", "more_body": False}]) + + async def receive() -> dict: + return next(incoming) + + async def send(msg: dict) -> None: + sent.append(msg) + + scope = { + "type": "http", + "http_version": "1.1", + "scheme": "http", + "method": "GET", + "path": "/x", + "query_string": b"", + "headers": [], + } + asyncio.run(app(scope, receive, send)) + assert called == ["inner"] diff --git a/tests/test_cors_units.py b/tests/test_cors_units.py new file mode 100644 index 0000000..556da39 --- /dev/null +++ b/tests/test_cors_units.py @@ -0,0 +1,177 @@ +"""Extra branch coverage for CORS configuration logic.""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass + +import httpx +from oxyroute import App +from oxyroute.cors import CORSConfig, apply_cors + + +@dataclass +class _Scope: + method: str = "GET" + headers: dict[str, str] | None = None + + def __post_init__(self) -> None: + if self.headers is None: + self.headers = {} + + +def test_select_allow_origin_branches() -> None: + cfg = CORSConfig(allow_origins=["*"], allow_credentials=False) + assert cfg._select_allow_origin("") is None + assert cfg._select_allow_origin("https://x.example") == "*" + + cfg2 = CORSConfig( + allow_origins=["https://a.example"], + allow_credentials=True, + ) + assert cfg2._select_allow_origin("https://a.example") == "https://a.example" + assert cfg2._select_allow_origin("https://b.example") is None + + cfg3 = CORSConfig(allow_origins=["https://a.example"], allow_credentials=False) + assert cfg3._select_allow_origin("https://b.example") is None + assert cfg3._select_allow_origin("https://a.example") == "https://a.example" + + +def test_response_header_pairs_origin_not_allowed_is_empty() -> None: + cfg = CORSConfig(allow_origins=["https://a.example"]) + pairs = cfg.response_header_pairs(_Scope(headers={"origin": "https://b.example"})) + assert pairs == [] + + +def test_preflight_non_options_and_missing_acrm_return_none() -> None: + cfg = CORSConfig() + assert cfg.preflight_response(_Scope(method="GET", headers={})) is None + assert ( + cfg.preflight_response(_Scope(method="OPTIONS", headers={"origin": "https://a.example"})) + is None + ) + + +def test_preflight_origin_disallowed_and_success_headers() -> None: + cfg = CORSConfig( + allow_origins=["https://a.example"], + allow_methods=["GET", "POST"], + allow_headers=["X-A", "X-B"], + allow_credentials=True, + max_age=600, + ) + bad = cfg.preflight_response( + _Scope( + method="OPTIONS", + headers={ + "origin": "https://b.example", + "access-control-request-method": "GET", + }, + ) + ) + assert bad is not None + assert bad.status == 400 + assert bad.body == "CORS origin not allowed" + + ok = cfg.preflight_response( + _Scope( + method="OPTIONS", + headers={ + "origin": "https://a.example", + "access-control-request-method": "POST", + "access-control-request-headers": "X-A, X-B", + }, + ) + ) + assert ok is not None + assert ok.status == 204 + assert ok.body is None + assert ok.headers is not None + assert ok.headers["Access-Control-Allow-Origin"] == "https://a.example" + assert ok.headers["Access-Control-Allow-Credentials"] == "true" + assert ok.headers["Access-Control-Allow-Methods"] == "GET, POST" + assert ok.headers["Access-Control-Allow-Headers"] == "X-A, X-B" + assert ok.headers["Access-Control-Max-Age"] == "600" + assert ok.headers["Vary"] == "Origin" + + +def test_preflight_allow_headers_empty_parts_and_reject_unknown() -> None: + cfg = CORSConfig(allow_headers=["X-A", "X-B"]) + # Empty request header list -> configured allow-list value + assert cfg._preflight_access_control_allow_headers_value(" , ,, ") == "X-A, X-B" + # Unknown header is rejected + assert cfg._preflight_access_control_allow_headers_value("X-A, X-C") is None + + +def test_apply_cors_chain_passthrough_and_preflight_short_circuit() -> None: + app = App() + called: list[str] = [] + cfg = CORSConfig(allow_origins=["https://a.example"], allow_methods=["GET"]) + + def chain(scope: object, _protocol: object) -> None: + called.append(f"{getattr(scope, 'method', '')}") + return None + + apply_cors(app, cfg, chain=chain) + + # Validate branch behavior through equivalent closure logic used by apply_cors. + def _cors_mw(scope: object, protocol: object) -> object: + p = cfg.preflight_response(scope) + if p is not None: + return p + return chain(scope, protocol) + + out = _cors_mw(_Scope(method="GET", headers={}), object()) + assert out is None + assert called == ["GET"] + + # Preflight: should short-circuit and not call chain. + called.clear() + out2 = _cors_mw( + _Scope( + method="OPTIONS", + headers={ + "origin": "https://a.example", + "access-control-request-method": "GET", + }, + ), + object(), + ) + assert out2 is not None + assert called == [] + + +def test_apply_cors_chain_called_for_non_preflight_integration() -> None: + app = App() + called: list[str] = [] + + def chain(scope: object, _protocol: object) -> None: + called.append(getattr(scope, "method", "")) + return None + + apply_cors( + app, + CORSConfig(allow_origins=["https://a.example"], allow_methods=["GET"]), + chain=chain, + ) + + @app.get("/hi") + def hi() -> str: + return "ok" + + async def _run() -> None: + tr = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=tr, base_url="http://t") as c: + r = await c.get("/hi", headers={"origin": "https://a.example"}) + assert r.status_code == 200 + assert r.text == "ok" + + asyncio.run(_run()) + assert called == ["GET"] + + +def test_cors_select_origin_with_wildcard_plus_specific_list() -> None: + cfg = CORSConfig(allow_origins=["*", "https://a.example"], allow_credentials=False) + # not the exact ["*"] list, so explicit origin membership branch is used + assert cfg._select_allow_origin("https://a.example") == "https://a.example" + assert cfg._select_allow_origin("https://b.example") is None diff --git a/tests/test_csrf_units.py b/tests/test_csrf_units.py new file mode 100644 index 0000000..6c9bfc9 --- /dev/null +++ b/tests/test_csrf_units.py @@ -0,0 +1,53 @@ +"""Extra CSRF helper branch coverage.""" + +from __future__ import annotations + +from dataclasses import dataclass + +from oxyroute.csrf import CSRFConfig, _read_cookie, _read_header, apply_csrf + + +@dataclass +class _Scope: + method: str = "GET" + headers: dict[str, str] | None = None + + +def test_read_helpers_with_missing_headers_object() -> None: + s = _Scope(method="POST", headers=None) + assert _read_cookie(s, "x") == "" + assert _read_header(s, "x") == "" + + +def test_apply_csrf_chain_runs_after_guard_pass() -> None: + cfg = CSRFConfig(cookie_name="ck", header_name="X-CSRF") + called: list[str] = [] + + def chain(scope: object, _protocol: object) -> None: + called.append(getattr(scope, "method", "")) + return None + + class _AppStub: + def __init__(self) -> None: + self.middleware = None + + def set_middleware(self, mw): # type: ignore[no-untyped-def] + self.middleware = mw + + app = _AppStub() + apply_csrf(app, cfg, chain=chain) # type: ignore[arg-type] + assert app.middleware is not None + + # safe method: guard passes, chain executes + out = app.middleware(_Scope(method="GET", headers={}), object()) + assert out is None + assert called == ["GET"] + + # unsafe method with mismatch: guard blocks before chain + called.clear() + out2 = app.middleware( + _Scope(method="POST", headers={"cookie": "ck=a", "X-CSRF": "b"}), + object(), + ) + assert out2 is not None + assert called == [] diff --git a/tests/test_framework_units.py b/tests/test_framework_units.py new file mode 100644 index 0000000..5629413 --- /dev/null +++ b/tests/test_framework_units.py @@ -0,0 +1,145 @@ +"""Unit-level coverage for config/helper modules.""" + +from __future__ import annotations + +from dataclasses import dataclass + +import pytest +from oxyroute.cors import CORSConfig +from oxyroute.csrf import CSRFConfig, _read_cookie, _read_header, csrf_layer +from oxyroute.exceptions import HTTPException, _http_exception_payload +from oxyroute.security_headers import SecurityHeadersConfig +from oxyroute.sse import SSEEvent, _to_sse_chunk, sse_done + + +@dataclass +class _Scope: + method: str = "GET" + scheme: str = "http" + headers: dict[str, str] | None = None + + def __post_init__(self) -> None: + if self.headers is None: + self.headers = {} + + +def test_cors_select_allow_origin_and_response_pairs() -> None: + cfg = CORSConfig( + allow_origins=["https://a.example", "https://b.example"], + expose_headers=["X-Id"], + allow_credentials=True, + ) + s = _Scope(headers={"origin": "https://a.example"}) + pairs = cfg.response_header_pairs(s) + assert ("Access-Control-Allow-Origin", "https://a.example") in pairs + assert ("Access-Control-Allow-Credentials", "true") in pairs + assert ("Access-Control-Expose-Headers", "X-Id") in pairs + assert ("Vary", "Origin") in pairs + + +def test_cors_preflight_disallowed_method_and_headers() -> None: + cfg = CORSConfig( + allow_origins=["https://a.example"], + allow_methods=["GET"], + allow_headers=["X-Ok"], + ) + bad_method = _Scope( + method="OPTIONS", + headers={ + "origin": "https://a.example", + "access-control-request-method": "POST", + }, + ) + r1 = cfg.preflight_response(bad_method) + assert r1 is not None and r1.status == 400 and r1.body == "Disallowed CORS method" + + bad_headers = _Scope( + method="OPTIONS", + headers={ + "origin": "https://a.example", + "access-control-request-method": "GET", + "access-control-request-headers": "X-Bad", + }, + ) + r2 = cfg.preflight_response(bad_headers) + assert r2 is not None and r2.status == 400 and r2.body == "Disallowed CORS request headers" + + +def test_cors_preflight_wildcard_headers_value() -> None: + cfg = CORSConfig(allow_headers=["*"]) + assert cfg._preflight_access_control_allow_headers_value("X-A, X-B") == "*" + + +def test_csrf_issue_cookie_and_guard_safe_method() -> None: + cfg = CSRFConfig(cookie_name="ck", header_name="X-CSRF", secure=True) + token = cfg.issue_token() + cookie = cfg.set_cookie_value(token) + assert cookie.startswith("ck=") and "; Secure" in cookie and "SameSite=Lax" in cookie + assert cfg.guard(_Scope(method="GET", headers={})) is None + + +def test_csrf_guard_missing_and_mismatch_and_layer() -> None: + cfg = CSRFConfig(cookie_name="ck", header_name="X-CSRF") + missing = cfg.guard(_Scope(method="POST", headers={})) + assert missing is not None and missing.status == 403 and missing.body["detail"] == "missing" + + mismatch = cfg.guard(_Scope(method="POST", headers={"cookie": "ck=a", "X-CSRF": "b"})) + assert mismatch is not None and mismatch.status == 403 and mismatch.body["detail"] == "mismatch" + + layer = csrf_layer(cfg) + assert layer(_Scope(method="GET", headers={}), object()) is None + + +def test_csrf_header_cookie_read_helpers() -> None: + s = _Scope(headers={"Cookie": 'a=1; ck="v.t"; z=9', "x-csrf-token": " tok "}) + assert _read_cookie(s, "ck") == "v.t" + assert _read_header(s, "X-CSRF-Token") == "tok" + + +def test_http_exception_payload_variants() -> None: + assert _http_exception_payload(ValueError("x")) is None + assert _http_exception_payload(HTTPException(99, "x")) is None + + st, body, headers = _http_exception_payload(HTTPException(404, None)) or (0, b"", []) + assert st == 404 and b"Not Found" in body and headers == [] + + st2, body2, _ = _http_exception_payload(HTTPException(422, {"x": 1})) or (0, b"", []) + assert st2 == 422 and body2 == b'{"x": 1}' + + st3, body3, headers3 = _http_exception_payload( + HTTPException(400, "bad", headers={"X-A": "1"}) + ) or (0, b"", []) + assert st3 == 400 and b'"detail": "bad"' in body3 and ("X-A", "1") in headers3 + + +def test_security_headers_pairs_and_extra() -> None: + cfg = SecurityHeadersConfig( + hsts="max-age=60", + x_content_type_options="nosniff", + x_frame_options="DENY", + referrer_policy="strict-origin-when-cross-origin", + content_security_policy="default-src 'none'", + permissions_policy="geolocation=()", + extra={"X-Custom": "1"}, + ) + https_pairs = cfg.response_header_pairs(_Scope(scheme="https")) + assert ("Strict-Transport-Security", "max-age=60") in https_pairs + assert ("Content-Security-Policy", "default-src 'none'") in https_pairs + assert ("Permissions-Policy", "geolocation=()") in https_pairs + assert ("X-Custom", "1") in https_pairs + + http_pairs = cfg.response_header_pairs(_Scope(scheme="http")) + assert not any(k == "Strict-Transport-Security" for k, _ in http_pairs) + + +def test_sse_helpers_and_done_marker() -> None: + chunk = _to_sse_chunk(SSEEvent(data="a\nb", event="tick", id="7", retry=1000)) + assert "event: tick" in chunk and "id: 7" in chunk and "retry: 1000" in chunk + assert "data: a" in chunk and "data: b" in chunk and chunk.endswith("\n\n") + + assert "data: hello" in _to_sse_chunk("hello") + with pytest.raises(TypeError): + _to_sse_chunk(123) # type: ignore[arg-type] + + done = sse_done() + assert getattr(done, "__oxyroute_stream_done__", False) is True diff --git a/tests/test_router_units.py b/tests/test_router_units.py new file mode 100644 index 0000000..260d84a --- /dev/null +++ b/tests/test_router_units.py @@ -0,0 +1,43 @@ +"""Unit-level coverage for APIRouter registration options.""" + +from __future__ import annotations + +from oxyroute import APIRouter + + +def _h() -> str: + return "ok" + + +def test_router_join_path_variants() -> None: + from oxyroute.router import join_path + + assert join_path(" /v1/ ", " users ") == "/v1/users" + assert join_path("", "") == "/" + assert join_path("v1", "") == "/v1/" + + +def test_router_collects_all_method_options() -> None: + r = APIRouter() + + r.get("/g", require_jwt=True, jwt_secret="k", algorithms=["HS256"], jwt_cookie="tok")(_h) + r.post( + "/p", + read_json_body=False, + read_form_body=True, + body_schema={"type": "object"}, + )(_h) + r.put("/u", jwt_issuer="iss", jwt_audience="aud", jwt_leeway=5)(_h) + r.patch("/pa", dependencies=[("x", lambda: 1)])(_h) + r.delete("/d", require_jwt=True, jwt_secret="k")(_h) + r.options("/o")(_h) + + methods = [m for m, _, _, _ in r._routes] + assert methods == ["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"] + + by_path = {p: opts for _, p, _, opts in r._routes} + assert by_path["/g"]["jwt_cookie"] == "tok" + assert by_path["/p"]["read_form_body"] is True + assert by_path["/u"]["jwt_issuer"] == "iss" + assert by_path["/u"]["jwt_audience"] == "aud" + assert by_path["/u"]["jwt_leeway"] == 5 diff --git a/tests/test_sse_units.py b/tests/test_sse_units.py new file mode 100644 index 0000000..8232bbd --- /dev/null +++ b/tests/test_sse_units.py @@ -0,0 +1,70 @@ +"""Unit-level coverage for SSE helper fallback paths.""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass + +from oxyroute.sse import SSEEvent, send_sse + + +@dataclass +class _Stream: + out: list[str] + + async def send_str(self, value: str) -> None: + self.out.append(value) + + +class _ProtoWithStream: + def __init__(self) -> None: + self.status_headers: tuple[int, list[tuple[str, str]]] | None = None + self.out: list[str] = [] + + def response_stream(self, status: int, headers: list[tuple[str, str]]) -> _Stream: + self.status_headers = (status, headers) + return _Stream(self.out) + + +class _ProtoFallback: + def __init__(self) -> None: + self.calls: list[tuple[int, list[tuple[str, str]], str]] = [] + + def response_str(self, status: int, headers: list[tuple[str, str]], body: str) -> None: + self.calls.append((status, headers, body)) + + +def test_send_sse_stream_path_with_async_iterable() -> None: + p = _ProtoWithStream() + + async def events() -> object: + for item in ["a", SSEEvent(data="b", event="tick")]: + yield item + + async def _run() -> None: + done = await send_sse(p, events(), status=201, headers=[("x-test", "1")]) + assert getattr(done, "__oxyroute_stream_done__", False) + + asyncio.run(_run()) + assert p.status_headers is not None + st, headers = p.status_headers + assert st == 201 + assert ("x-test", "1") in headers + assert any("data: a" in chunk for chunk in p.out) + assert any("event: tick" in chunk for chunk in p.out) + + +def test_send_sse_fallback_path_with_list_events() -> None: + p = _ProtoFallback() + + async def _run() -> None: + done = await send_sse(p, ["hello", SSEEvent(data="w", id="1")]) + assert getattr(done, "__oxyroute_stream_done__", False) + + asyncio.run(_run()) + assert len(p.calls) == 1 + status, headers, body = p.calls[0] + assert status == 200 + assert ("content-type", "text/event-stream; charset=utf-8") in headers + assert "data: hello" in body + assert "id: 1" in body diff --git a/tests/test_websocket_asgi.py b/tests/test_websocket_asgi.py index acb402d..745e636 100644 --- a/tests/test_websocket_asgi.py +++ b/tests/test_websocket_asgi.py @@ -7,6 +7,7 @@ from collections import deque from typing import Any +import pytest from oxyroute import App @@ -93,3 +94,85 @@ async def _run() -> None: asyncio.run(_run()) assert sent[0]["type"] == "websocket.accept" assert sent[1]["type"] == "websocket.close" + + +def test_asgi_websocket_unknown_path_closes() -> None: + app = App() + sent: list[dict[str, Any]] = [] + + async def receive() -> dict[str, Any]: + return {"type": "websocket.connect"} + + async def send(message: dict[str, Any]) -> None: + sent.append(message) + + scope = { + "type": "websocket", + "path": "/missing", + "headers": [], + "query_string": b"", + "scheme": "ws", + } + + asyncio.run(app(scope, receive, send)) + assert sent == [{"type": "websocket.close", "code": 1000}] + + +def test_asgi_websocket_accept_disconnect_before_connect_raises() -> None: + app = App() + + @app.websocket("/ws") + async def ws(sock: Any) -> None: + await sock.accept() + + incoming: deque[dict[str, Any]] = deque([{"type": "websocket.disconnect", "code": 1000}]) + + async def receive() -> dict[str, Any]: + return incoming.popleft() + + async def send(_message: dict[str, Any]) -> None: + return None + + scope = { + "type": "websocket", + "path": "/ws", + "headers": [], + "query_string": b"", + "scheme": "ws", + } + + with pytest.raises(RuntimeError, match="disconnected before accept"): + asyncio.run(app(scope, receive, send)) + + +def test_asgi_websocket_receive_text_non_text_frame_raises() -> None: + app = App() + + @app.websocket("/ws") + async def ws(sock: Any) -> None: + await sock.accept() + await sock.receive_text() + + incoming: deque[dict[str, Any]] = deque( + [ + {"type": "websocket.connect"}, + {"type": "websocket.receive", "bytes": b"x"}, + ] + ) + + async def receive() -> dict[str, Any]: + return incoming.popleft() + + async def send(_message: dict[str, Any]) -> None: + return None + + scope = { + "type": "websocket", + "path": "/ws", + "headers": [], + "query_string": b"", + "scheme": "ws", + } + + with pytest.raises(RuntimeError, match="expected text websocket frame"): + asyncio.run(app(scope, receive, send))