tests/core/test_mcp_push.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 """Tests for the MCP push channel (core 1e Phase 6): GET -> SSE over the hub.16 17 Real objects, no mocks: a real ``AsgiServer`` (storage on tmp_path) with an18 ``McpApplication`` mounted, driven at the ASGI level. The ``sse_request``19 fixture (conftest) keeps the GET open like a real SSE client; frames are20 awaited, then the connection is cancelled (the client-gone path, exercising21 the hub unsubscribe). The A<->C bridge is tested end-to-end: a spool task22 carrying the launching ``session_id`` surfaces as ``data:`` frames while it23 runs.24 """25 26 from __future__ import annotations27 28 import json29 from pathlib import Path30 from typing import Any31 32 import pytest33 from genro_routes import route34 35 from tests.storage_support import site_storage36 37 from genro_asgi import AsgiServer, McpApplication, RoutedApplication38 from genro_asgi.tasks import new_descriptor39 from genro_asgi.types import Message, Scope40 41 42 class Primary(RoutedApplication):43 """The primary app: one handler a spool task can run."""44 45 @route()46 def sum_sync(self, a: int = 0, b: int = 0) -> int:47 return a + b48 49 50 @pytest.fixture51 def server(tmp_path: Path) -> AsgiServer:52 """A real server: Primary + McpApplication at ``/mcp``, storage on tmp_path."""53 srv = AsgiServer(54 applications=[Primary(mount=""), McpApplication(code="mcp")],55 storage=site_storage(tmp_path),56 )57 return srv58 59 60 async def drive(server: Any, path: str, *, method: str = "GET",61 headers: list[tuple[bytes, bytes]] | None = None,62 body: bytes = b"") -> list[Message]:63 """One plain (non-streaming) request through the server."""64 scope: Scope = {"type": "http", "method": method, "path": path,65 "query_string": b"", "headers": list(headers or [])}66 sent: list[Message] = []67 68 async def receive() -> Message:69 return {"type": "http.request", "body": body, "more_body": False}70 71 async def send(message: Message) -> None:72 sent.append(message)73 74 await server(scope, receive, send)75 return sent76 77 78 def start_headers(sent: list[Message]) -> dict[bytes, bytes]:79 start = next(m for m in sent if m["type"] == "http.response.start")80 return dict(start["headers"])81 82 83 def payload(frame: bytes) -> dict[str, Any]:84 """Decode the JSON payload of one ``data:`` SSE frame."""85 text = frame.decode()86 data_lines = [line[6:] for line in text.splitlines() if line.startswith("data: ")]87 return json.loads("\n".join(data_lines))88 89 90 def stage(server: AsgiServer, task_id: str, session_id: str | None) -> None:91 """Create a pending task on the primary, launched by ``session_id``."""92 descriptor = new_descriptor(task_id, owner="alice", mount="", node_path="sum_sync",93 session_id=session_id)94 server.tasks.spool.create(descriptor, {"a": 2, "b": 3})95 96 97 class TestSessionId:98 """GET mints or echoes the Mcp-Session-Id and streams text/event-stream."""99 100 async def test_get_mints_session_id(self, server: AsgiServer, sse_request) -> None:101 conn = await sse_request(server, "/mcp")102 try:103 headers = start_headers(conn.sent)104 assert headers[b"content-type"].startswith(b"text/event-stream")105 assert len(headers[b"mcp-session-id"]) > 20 # minted token106 finally:107 await conn.close()108 109 async def test_get_echoes_supplied_session_id(self, server: AsgiServer, sse_request) -> None:110 conn = await sse_request(server, "/mcp", headers=[(b"mcp-session-id", b"sess-1")])111 try:112 assert start_headers(conn.sent)[b"mcp-session-id"] == b"sess-1"113 finally:114 await conn.close()115 116 async def test_get_without_tasks_is_405(self, tmp_path: Path) -> None:117 srv = AsgiServer(118 applications=[Primary(mount=""), McpApplication(code="mcp")],119 tasks=False,120 storage=site_storage(tmp_path),121 )122 sent = await drive(srv, "/mcp", method="GET")123 start = next(m for m in sent if m["type"] == "http.response.start")124 assert start["status"] == 405125 126 127 class TestLiveFeed:128 """hub.publish surfaces as a ``data:`` frame on the subscribed stream."""129 130 async def test_publish_becomes_frame(self, server: AsgiServer, sse_request) -> None:131 conn = await sse_request(server, "/mcp", headers=[(b"mcp-session-id", b"sess-2")])132 try:133 server.tasks.hub.publish("sess-2", {"type": "progress", "task_id": "t1",134 "data": {"pct": 10}})135 frames = await conn.wait_frames(1)136 assert payload(frames[0]) == {"type": "progress", "task_id": "t1",137 "data": {"pct": 10}}138 finally:139 await conn.close()140 141 async def test_other_session_not_delivered(self, server: AsgiServer, sse_request) -> None:142 conn = await sse_request(server, "/mcp", headers=[(b"mcp-session-id", b"sess-3")])143 try:144 server.tasks.hub.publish("other", {"type": "progress", "task_id": "x"})145 server.tasks.hub.publish("sess-3", {"type": "marker"})146 frames = await conn.wait_frames(1)147 assert payload(frames[0]) == {"type": "marker"} # only own session148 finally:149 await conn.close()150 151 async def test_close_unsubscribes(self, server: AsgiServer, sse_request) -> None:152 conn = await sse_request(server, "/mcp", headers=[(b"mcp-session-id", b"sess-4")])153 await conn.close()154 assert "sess-4" not in server.tasks.hub._subscribers # finally ran155 156 157 class TestExecutorBridge:158 """A task carrying session_id publishes started/settled while it runs."""159 160 async def test_lifecycle_events_stream(self, server: AsgiServer, sse_request) -> None:161 stage(server, "t-bridge", "sess-5")162 conn = await sse_request(server, "/mcp", headers=[(b"mcp-session-id", b"sess-5")])163 try:164 server.tasks.spool.assign("t-bridge", "local")165 outcome = await server.tasks.executor.execute("t-bridge", "local")166 assert outcome == "ok"167 frames = await conn.wait_frames(2)168 events = [payload(f) for f in frames]169 assert events[0] == {"task_id": "t-bridge", "type": "started"}170 assert events[1]["type"] == "settled" and events[1]["outcome"] == "ok"171 finally:172 await conn.close()173 174 async def test_no_session_no_publish(self, server: AsgiServer) -> None:175 stage(server, "t-silent", None) # no push channel176 server.tasks.spool.assign("t-silent", "local")177 outcome = await server.tasks.executor.execute("t-silent", "local")178 assert outcome == "ok" # publish was a no-op179 assert server.tasks.hub._subscribers == {}180 181 182 class TestPublishProgressSeam:183 """manager.publish_progress pairs the spool write with the hub publish."""184 185 async def test_paired_write_and_publish(self, server: AsgiServer, sse_request) -> None:186 stage(server, "t-prog", "sess-6")187 server.tasks.spool.assign("t-prog", "local")188 conn = await sse_request(server, "/mcp", headers=[(b"mcp-session-id", b"sess-6")])189 try:190 server.tasks.publish_progress("t-prog", {"pct": 40})191 assert server.tasks.spool.read_progress("t-prog") == {"pct": 40} # spool wrote192 frames = await conn.wait_frames(1) # hub carried193 assert payload(frames[0]) == {"type": "progress", "task_id": "t-prog",194 "data": {"pct": 40}}195 finally:196 await conn.close()197 198 def test_not_active_raises(self, server: AsgiServer) -> None:199 stage(server, "t-pending", "sess-7") # pending, not active200 with pytest.raises(LookupError, match="not active"):201 server.tasks.publish_progress("t-pending", {"pct": 1})202 with pytest.raises(LookupError, match="not active"):203 server.tasks.publish_progress("t-ghost", {"pct": 1})204 205 206 class TestBaseline:207 """Last-Event-ID replays the session's progress snapshots, then live."""208 209 async def test_reconnect_replays_snapshot(self, server: AsgiServer, sse_request) -> None:210 stage(server, "t-base", "sess-8")211 server.tasks.spool.assign("t-base", "local")212 server.tasks.publish_progress("t-base", {"pct": 70}) # no subscriber yet213 conn = await sse_request(server, "/mcp", headers=[(b"mcp-session-id", b"sess-8"),214 (b"last-event-id", b"0")])215 try:216 frames = await conn.wait_frames(1)217 assert payload(frames[0]) == {"type": "progress", "task_id": "t-base",218 "data": {"pct": 70}}219 finally:220 await conn.close()221 222 async def test_fresh_connect_has_no_baseline(self, server: AsgiServer, sse_request) -> None:223 stage(server, "t-fresh", "sess-9")224 server.tasks.spool.assign("t-fresh", "local")225 server.tasks.publish_progress("t-fresh", {"pct": 70})226 conn = await sse_request(server, "/mcp", headers=[(b"mcp-session-id", b"sess-9")])227 try:228 server.tasks.hub.publish("sess-9", {"type": "marker"})229 frames = await conn.wait_frames(1)230 assert payload(frames[0]) == {"type": "marker"} # live only, no replay231 finally:232 await conn.close()233 234 235 class TestInitializeAdvertisesPush:236 """POST initialize advertises the push capability; POST path unchanged."""237 238 async def test_capability_advertised(self, server: AsgiServer) -> None:239 envelope = {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}}240 sent = await drive(server, "/mcp", method="POST",241 headers=[(b"content-type", b"application/json")],242 body=json.dumps(envelope).encode())243 body = b"".join(m.get("body", b"") for m in sent244 if m["type"] == "http.response.body")245 result = json.loads(body)["result"]246 assert result["capabilities"]["experimental"] == {"push": {}}247 assert result["capabilities"]["tools"] == {} # unchanged248 249 async def test_notification_still_202(self, server: AsgiServer) -> None:250 envelope = {"jsonrpc": "2.0", "method": "initialize"}251 sent = await drive(server, "/mcp", method="POST",252 headers=[(b"content-type", b"application/json")],253 body=json.dumps(envelope).encode())254 start = next(m for m in sent if m["type"] == "http.response.start")255 assert start["status"] == 202