tests/core/test_demux.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """Demux, uvicorn boot and empty-websocket tests (SPECIFICATION.md §4, D3/D7).16 17 The http tests boot the server on port 0 in a background thread, discover the18 bound port from the uvicorn server state, and hit it with real httpx requests.19 The websocket test drives ``BaseServer.__call__`` directly at the ASGI level:20 the empty socket exists and closes cleanly without needing a websocket client.21 """22 23 from __future__ import annotations24 25 import threading26 import time27 from collections.abc import Iterator28 from contextlib import contextmanager29 30 import httpx31 32 from genro_asgi import BaseServer33 34 from ..throwaway_app import ThrowawayApp35 36 37 @contextmanager38 def running_server(server: BaseServer) -> Iterator[int]:39 """Boot ``server`` on port 0 in a background thread; yield the bound port."""40 thread = threading.Thread(target=lambda: server.serve(host="127.0.0.1", port=0), daemon=True)41 thread.start()42 deadline = time.monotonic() + 543 while server.uvicorn_server is None or not server.uvicorn_server.started:44 if time.monotonic() > deadline:45 raise RuntimeError("server did not start in time")46 time.sleep(0.01)47 port = server.uvicorn_server.servers[0].sockets[0].getsockname()[1]48 try:49 yield port50 finally:51 server.uvicorn_server.should_exit = True52 thread.join(timeout=5)53 54 55 @contextmanager56 def booted() -> Iterator[int]:57 """A server with a root app and a secondary mounted as ``api`` (both throwaway)."""58 server = BaseServer(59 applications=[60 ThrowawayApp(mount="", name="root"),61 ThrowawayApp(name="api", code="api"),62 ]63 )64 with running_server(server) as port:65 yield port66 67 68 class TestDemux:69 def test_root_goes_to_the_root_app(self) -> None:70 with booted() as port:71 r = httpx.get(f"http://127.0.0.1:{port}/")72 assert r.status_code == 20073 assert r.text == "root:/"74 75 def test_unclaimed_first_segment_goes_to_the_root_app(self) -> None:76 with booted() as port:77 r = httpx.get(f"http://127.0.0.1:{port}/nothing/claimed")78 assert r.status_code == 20079 assert r.text == "root:/nothing/claimed"80 81 def test_mounted_first_segment_reaches_its_app_with_path_stripped(self) -> None:82 with booted() as port:83 r = httpx.get(f"http://127.0.0.1:{port}/api/echo")84 assert r.status_code == 20085 assert r.text == "api:/echo"86 87 def test_raising_route_returns_500_and_server_survives(self) -> None:88 with booted() as port:89 boom = httpx.get(f"http://127.0.0.1:{port}/boom")90 assert boom.status_code == 50091 healthy = httpx.get(f"http://127.0.0.1:{port}/")92 assert healthy.status_code == 20093 assert healthy.text == "root:/"94 95 async def test_double_slash_before_a_mount_forwards_the_remainder(self) -> None:96 # the forwarded path is rebuilt from the same remainder used to find97 # the segment: //api/x reaches the mount as /x (driven at ASGI level)98 server = BaseServer(99 applications=[100 ThrowawayApp(mount="", name="root"),101 ThrowawayApp(name="api", code="api"),102 ]103 )104 sent: list[dict[str, object]] = []105 106 async def receive() -> dict[str, object]:107 return {"type": "http.request"}108 109 async def send(message: dict[str, object]) -> None:110 sent.append(message)111 112 await server({"type": "http", "path": "//api/x"}, receive, send)113 114 body = next(m["body"] for m in sent if m["type"] == "http.response.body")115 assert body == b"api:/x"116 117 118 class TestServerOfMountsOnly:119 """The three branches a server without a root application answers with."""120 121 async def drive(self, server: BaseServer, path: str, query: bytes = b"") -> dict[str, object]:122 """Drive one GET at the ASGI level; return the ``http.response.start``."""123 sent: list[dict[str, object]] = []124 125 async def receive() -> dict[str, object]:126 return {"type": "http.request"}127 128 async def send(message: dict[str, object]) -> None:129 sent.append(message)130 131 scope = {"type": "http", "path": path, "query_string": query}132 await server(scope, receive, send)133 return next(m for m in sent if m["type"] == "http.response.start")134 135 def mounts_only(self, **kwargs: object) -> BaseServer:136 """A server serving ``/api`` and nothing on the root."""137 return BaseServer(applications=[ThrowawayApp(name="api", code="api")], **kwargs)138 139 async def test_a_claimed_segment_still_reaches_its_app(self) -> None:140 start = await self.drive(self.mounts_only(), "/api/echo")141 assert start["status"] == 200142 143 async def test_an_unclaimed_path_is_404(self) -> None:144 start = await self.drive(self.mounts_only(), "/nothing/claimed")145 assert start["status"] == 404146 147 async def test_the_root_is_404_without_a_default(self) -> None:148 start = await self.drive(self.mounts_only(), "/")149 assert start["status"] == 404150 151 async def test_the_root_redirects_to_the_default_with_a_307(self) -> None:152 start = await self.drive(self.mounts_only(default="api"), "/")153 assert start["status"] == 307154 assert dict(start["headers"])[b"location"] == b"/api/"155 156 async def test_the_redirect_carries_the_query_string_over(self) -> None:157 start = await self.drive(self.mounts_only(default="api"), "/", query=b"q=moka&n=2")158 assert dict(start["headers"])[b"location"] == b"/api/?q=moka&n=2"159 160 async def test_an_unclaimed_path_is_404_even_with_a_default(self) -> None:161 # the default answers the ROOT only: it is not a catch-all162 start = await self.drive(self.mounts_only(default="api"), "/nothing")163 assert start["status"] == 404164 165 166 class TestTheWebsocketBranch:167 """The websocket scope reaches the motor (#68 phase 2).168 169 Until phase 2 this branch was the empty socket of D7: it consumed the170 connect and closed 1000. It now lives one whole connection; what that171 connection does is `tests/test_wsx_connection.py`'s subject, and what is172 asserted here is only that `__call__` hands it over.173 """174 175 async def test_a_handshake_on_a_served_path_is_accepted(self) -> None:176 server = BaseServer(applications=[ThrowawayApp(mount="")])177 sent: list[dict[str, object]] = []178 incoming: list[dict[str, object]] = [179 {"type": "websocket.connect"},180 {"type": "websocket.disconnect", "code": 1000},181 ]182 183 async def receive() -> dict[str, object]:184 return incoming.pop(0) if incoming else {"type": "websocket.disconnect", "code": 1006}185 186 async def send(message: dict[str, object]) -> None:187 sent.append(message)188 189 await server({"type": "websocket", "path": "/", "headers": []}, receive, send)190 assert sent == [{"type": "websocket.accept"}]