"""The kernel routes, pinned before they are moved. `Handler._api` is one method of ~2,100 lines and the agreed direction is to carve route groups out of it. The kernel group is the first slice, and an audit of what would catch a mistake found that it is largely unwatched: * `GET /frames/{id}/execution`, `/kernel`, `/status` and `/environments` had no frozen response shape and no route-level test at all. The one test that mentions `/execution` asserts a 404 raised by an *upstream* guard, so it passes with the handler deleted outright. * `kernel/restart`, `/stop`, `/start` and `/env` had a frozen shape for the 403 "REPL disabled" envelope only. The guard was pinned; what the route actually does was not. Swapping the restart and stop bodies during the move would have passed pytest, passed the response-shape gate, and passed the browser smoke run (restart and start have no call site in either). So these are not tests of the extraction. They are the tests that make the extraction checkable, and they belong before it rather than after, when they would be written to match whatever the moved code happens to do. Two behaviours here look like bugs and are deliberately pinned as they are. Ten of the twelve routes answer 200 for a frame id that does not exist, and `kernel/install` is deliberately *not* gated by `notebook_repl` while its six siblings are. Changing either is a behaviour change; this file's job is to notice if the move makes one by accident. """ from __future__ import annotations from types import SimpleNamespace import pytest from openai4s.config import Config, LLMConfig from openai4s.server import gateway as gateway_mod class _Hub: def emitter(self, root_frame_id): return lambda event: None def broadcast(self, root_frame_id, event): return None def has_subscriber(self, root_frame_id): return False def drop_frame(self, root_frame_id): return None def _setup(tmp_path, *, notebook_repl=False): config = Config( data_dir=tmp_path, llm=LLMConfig(provider="deepseek", api_key="test-key") ) if notebook_repl: # The flag is read through cfg at request time, so overriding the # attribute is enough and avoids a second SessionRunner. config.notebook_repl = True hub = _Hub() runner = gateway_mod.SessionRunner(config, hub, start_idle_sweeper=False) frame_id = runner.store.new_frame( kind="turn", project_id="proj-kernel", status="ready" ) handler = object.__new__(gateway_mod.make_handler(config, hub, runner)) return runner, handler, frame_id def _call(handler, method, path, *, body=None, query=None): replies: list[tuple] = [] handler._query = lambda: query or {} handler._body = lambda: body or {} handler._json = lambda value, code=200: replies.append((code, value)) handler._send = ( lambda code, data, content_type, extra=None, security=None: replies.append( (code, data, content_type, extra or {}) ) ) handler._api(method, path) return replies[-1] if replies else None # -------------------------------------------------------------------------- # the four routes with no net at all # -------------------------------------------------------------------------- @pytest.mark.stubbed_backend def test_execution_snapshot_reaches_the_coordinator(tmp_path): """The handler is three lines and nothing checked that they ran. The only existing test asserts a 404 that an upstream guard produces, so it passes with this body deleted.""" runner, handler, fid = _setup(tmp_path) seen = [] runner.executions = SimpleNamespace( snapshot=lambda f: seen.append(f) or {"session_id": f, "queue": []} ) code, payload = _call(handler, "GET", f"/frames/{fid}/execution") assert code == 200 assert seen == [fid], "the frame id has to reach the coordinator unchanged" assert payload["session_id"] == fid def test_an_unknown_frame_is_refused_by_the_upstream_guard(tmp_path): """This 404 does NOT come from the handler. It comes from the `workbench` guard several hundred lines earlier, and it is the reason the extracted module cannot be called before that guard runs.""" _runner, handler, _fid = _setup(tmp_path) with pytest.raises(gateway_mod.GatewayError) as raised: _call(handler, "GET", "/frames/no-such-frame/execution") assert raised.value.code == 404 @pytest.mark.stubbed_backend def test_kernel_state_is_whatever_the_runner_reports(tmp_path): runner, handler, fid = _setup(tmp_path) runner.kernel_status = lambda f: {"state": "idle", "alive": True, "frame": f} code, payload = _call(handler, "GET", f"/frames/{fid}/kernel") assert code == 200 assert payload == {"state": "idle", "alive": True, "frame": fid} @pytest.mark.stubbed_backend def test_status_combines_the_turn_and_the_kernel(tmp_path): """Three fields assembled inline in the route body. Nothing verified the assembly, so a moved copy could drop or rename one silently.""" runner, handler, fid = _setup(tmp_path) runner.is_running = lambda f: True runner.kernel_status = lambda f: {"state": "busy"} code, payload = _call(handler, "GET", f"/frames/{fid}/status") assert code == 200 assert payload == { "frame_id": fid, "running": True, # The frame's own state, which `running` cannot express: `false` covers # completed, cancelled and failed alike, and a client reopening a # session needs to know which before it can restore anything. "status": "ready", "kernel": {"state": "busy"}, } @pytest.mark.stubbed_backend def test_environments_lists_what_the_runner_offers(tmp_path): runner, handler, fid = _setup(tmp_path) runner.list_environments = lambda f: {"environments": [{"name": "python"}]} code, payload = _call(handler, "GET", f"/frames/{fid}/environments") assert code == 200 assert payload["environments"] == [{"name": "python"}] @pytest.mark.parametrize("route", ["kernel", "status", "environments"]) @pytest.mark.stubbed_backend def test_these_three_answer_200_for_a_frame_that_does_not_exist(tmp_path, route): """Pinned as-is, not endorsed. Ten of the twelve kernel routes answer for an unknown id; only `/execution` and `/kernel/variables` refuse. Tidying one uniform frame check across the group during the move would be a behaviour change wearing a refactor's clothes, so record the asymmetry here where a reviewer can see it.""" runner, handler, _fid = _setup(tmp_path) runner.kernel_status = lambda f: {"state": "none"} runner.is_running = lambda f: False runner.list_environments = lambda f: {"environments": []} code, _payload = _call(handler, "GET", f"/frames/ghost-frame/{route}") assert code == 200 # -------------------------------------------------------------------------- # the four routes where only the 403 guard was pinned # -------------------------------------------------------------------------- @pytest.mark.parametrize( "action,attr", [ ("restart", "restart_kernel"), ("stop", "stop_kernel"), ("start", "start_kernel"), ], ) @pytest.mark.stubbed_backend def test_each_lifecycle_action_calls_its_own_runner_method(tmp_path, action, attr): """The gap that mattered most. With only the 403 pinned, swapping the restart and stop bodies passed every gate in the repo.""" runner, handler, fid = _setup(tmp_path, notebook_repl=True) called = [] for name in ("restart_kernel", "stop_kernel", "start_kernel"): setattr( runner, name, lambda frame, project, _n=name: called.append((_n, frame, project)) or {"ok": _n}, ) code, payload = _call(handler, "POST", f"/frames/{fid}/kernel/{action}") assert code == 200 assert [c[0] for c in called] == [attr], f"{action} must call {attr}, alone" assert called[0][1] == fid assert called[0][2] == "proj-kernel", "the frame's project, not a constant" assert payload["ok"] == attr @pytest.mark.stubbed_backend def test_a_lifecycle_action_on_an_unknown_frame_falls_back_to_the_default_project( tmp_path, ): """`store.get_frame(fid) or {}` then `.get("project_id") or "default"`. The fallback is load-bearing and repeated across six routes, so a move that drops the `or {}` turns a 200 into an AttributeError.""" runner, handler, _fid = _setup(tmp_path, notebook_repl=True) seen = [] runner.start_kernel = lambda frame, project: seen.append((frame, project)) or {} code, _payload = _call(handler, "POST", "/frames/ghost/kernel/start") assert code == 200 assert seen == [("ghost", "default")] @pytest.mark.stubbed_backend def test_selecting_an_environment_passes_the_requested_name_through(tmp_path): runner, handler, fid = _setup(tmp_path, notebook_repl=True) seen = [] runner.set_env = lambda frame, name, project: seen.append( (frame, name, project) ) or {"selected": name} code, payload = _call( handler, "POST", f"/frames/{fid}/kernel/env", body={"name": "r"} ) assert code == 200 assert seen == [(fid, "r", "proj-kernel")] assert payload["selected"] == "r" @pytest.mark.stubbed_backend def test_the_environment_name_may_arrive_under_either_key(tmp_path): """`b.get("env") or b.get("name")`. Two spellings, one route -- a detail that is invisible unless someone writes it down before moving it.""" runner, handler, fid = _setup(tmp_path, notebook_repl=True) seen = [] runner.set_env = lambda frame, name, project: seen.append(name) or {} _call(handler, "POST", f"/frames/{fid}/kernel/env", body={"env": "struct"}) _call(handler, "POST", f"/frames/{fid}/kernel/env", body={"name": "r"}) _call(handler, "POST", f"/frames/{fid}/kernel/env", body={}) assert seen == ["struct", "r", ""] @pytest.mark.parametrize( "action", ["execute", "env", "restart", "stop", "start", "interrupt"] ) def test_the_six_repl_routes_stay_gated_when_the_notebook_is_read_only( tmp_path, action ): """Already covered elsewhere, repeated here because the extraction moves all six gates at once and the group needs its own regression.""" runner, handler, fid = _setup(tmp_path) assert runner.cfg.notebook_repl is False code, payload = _call(handler, "POST", f"/frames/{fid}/kernel/{action}") assert code == 403 assert "disabled" in payload["error"] @pytest.mark.stubbed_backend def test_install_is_deliberately_not_gated_by_the_notebook_flag(tmp_path): """The asymmetry the comment in the route body explains: installing into a prebuilt environment is a Compute affordance, not the code REPL. Unifying the six gates into one during the move must not sweep this in.""" runner, handler, fid = _setup(tmp_path) assert runner.cfg.notebook_repl is False seen = {} def install(packages, **kwargs): seen.update({"packages": packages, **kwargs}) return {"installed": packages} runner.install_packages = install code, payload = _call( handler, "POST", f"/frames/{fid}/kernel/install", body={"package": "numpy"} ) assert code == 200, "install must answer even with the REPL disabled" assert seen["packages"] == ["numpy"] assert seen["root_frame_id"] == fid assert seen["restart"] is True, "restart defaults to True when unstated" @pytest.mark.stubbed_backend def test_install_accepts_a_list_as_well_as_a_single_package(tmp_path): runner, handler, fid = _setup(tmp_path) seen = {} runner.install_packages = lambda packages, **kw: seen.update( {"packages": packages} ) or {"installed": packages} _call( handler, "POST", f"/frames/{fid}/kernel/install", body={"packages": ["numpy", "scipy"]}, ) assert seen["packages"] == ["numpy", "scipy"] # -------------------------------------------------------------------------- # the query string, and the fallthrough # -------------------------------------------------------------------------- @pytest.mark.stubbed_backend def test_variable_inspection_reads_the_language_from_the_query(tmp_path): """`q` is an `_api` local, read on exactly one line of the 222 being moved. An extraction whose signature omits it raises NameError here and nowhere else -- including on the default path, because `q.get` is evaluated before the "python" fallback applies.""" runner, handler, fid = _setup(tmp_path) seen = [] runner.variables = SimpleNamespace( inspect=lambda frame, language: seen.append((frame, language)) or {"variables": []} ) _call( handler, "GET", f"/frames/{fid}/kernel/variables", query={"language": ["r"]}, ) assert seen == [(fid, "r")] seen.clear() _call(handler, "GET", f"/frames/{fid}/kernel/variables") assert seen == [(fid, "python")], "no query means python, not a crash" @pytest.mark.stubbed_backend def test_an_unsupported_inspection_language_is_refused(tmp_path): runner, handler, fid = _setup(tmp_path) runner.variables = SimpleNamespace( inspect=lambda frame, language: pytest.fail("must not reach the kernel") ) code, payload = _call( handler, "GET", f"/frames/{fid}/kernel/variables", query={"language": ["julia"]}, ) assert code == 400 assert "python or r" in payload["error"] def test_a_matched_path_with_the_wrong_method_falls_through_to_404(tmp_path): """Every branch is `if m and method == ...`, so a matched regex is not a handled request. An extracted module that returns `bool(regex_matched)` swallows these twelve 404s into an empty response.""" _runner, handler, fid = _setup(tmp_path) code, payload = _call(handler, "GET", f"/frames/{fid}/kernel/execute") assert code == 404 assert payload["path"] == f"/frames/{fid}/kernel/execute" assert payload["method"] == "GET" # -------------------------------------------------------------------------- # the extraction itself # -------------------------------------------------------------------------- def test_the_registry_is_the_exact_runtime_match_chain(monkeypatch): """Every declared route must be consulted once, in declaration order. The inventory trusts ``ROUTES`` while runtime dispatch remains the explicit chain in ``handle``. Recording every matcher consultation catches the registry-only direction: a route declared but never consulted would be documented and always 404. It cannot catch the other direction on its own -- a branch that matches by some means other than ``RouteSpec.match`` never reaches the recorder -- so the idiom count in the next test is what covers a handler-only route. Identity, not equality: ``RouteSpec`` is a frozen dataclass compared by value, so ``==`` would also accept a locally-built look-alike. """ from openai4s.server import contract, kernel_routes consulted = [] def record_match(spec, method, path): consulted.append((spec, method, path)) return None monkeypatch.setattr(contract.RouteSpec, "match", record_match) handled = kernel_routes.handle( None, "OPTIONS", # Must clear the `/frames/` prefix guard, or the chain short-circuits # before consulting anything. "/frames/__route_registry_probe__/__nope__", {}, None, None, ) assert handled is False assert [spec for spec, _, _ in consulted] == list(kernel_routes.ROUTES) assert all( spec is declared for (spec, _, _), declared in zip(consulted, kernel_routes.ROUTES) ) # Arguments in the declared order. `_ENV.match(sub, method)` would otherwise # record the right spec in the right slot while the route is dead. assert {(method, path) for _, method, path in consulted} == { ("OPTIONS", "/frames/__route_registry_probe__/__nope__") } def test_every_matcher_in_the_handler_is_a_declared_route(): """The handler-only direction, which consultation-recording cannot see. A thirteenth branch written the legacy way -- ``re.fullmatch(r"/frames/...", sub)`` -- is a live route that never calls ``RouteSpec.match``, so the chain test above stays green while the route is absent from ``http_routes()`` and from both captured contracts. Counting the route-matching idioms in the source is what notices, and it names the real cause rather than accusing `_route_sources` of having stopped reading a file. """ import inspect import re as _re from openai4s.server import kernel_routes body = inspect.getsource(kernel_routes.handle) spec_matches = len(_re.findall(r"\.match\(method, sub\)", body)) raw_matches = _re.findall(r'fullmatch\(\s*r?"(/[^"]*)"', body) assert spec_matches == len(kernel_routes.ROUTES), ( f"handle() consults {spec_matches} RouteSpec matchers but ROUTES " f"declares {len(kernel_routes.ROUTES)}" ) assert raw_matches == [], ( "these paths are matched directly instead of through a RouteSpec, so " f"they are live but absent from the contract inventory: {raw_matches}" ) def test_the_group_reports_when_it_does_not_own_a_path(): """The tri-state contract, checked directly rather than through the chain. A module that answered `bool(regex_matched)` would claim these.""" from openai4s.server import kernel_routes for path in ("/projects", "/frames/f-1/messages", "/artifacts/a-1"): assert ( kernel_routes.handle(None, "GET", path, {}, None, None) is False ), f"{path} is not a kernel route" @pytest.mark.stubbed_backend def test_a_matched_path_with_the_wrong_method_is_reported_unhandled(tmp_path): """Matched-but-not-handled has to reach the caller as False, or the twelve wrong-method 404s become empty responses.""" from openai4s.server import kernel_routes _runner, handler, fid = _setup(tmp_path) emitted = [] handler._json = lambda value, code=200: emitted.append((code, value)) handled = kernel_routes.handle( handler, "GET", f"/frames/{fid}/kernel/execute", {}, None, None ) assert handled is False assert emitted == [], "nothing may be emitted for a path it did not handle" def test_the_moved_routes_are_still_contract_inventory(): """The failure prerequisite 3 exists to prevent: routes that leave gateway.py must not leave the documented surface with it.""" from openai4s.server.contract import http_routes routes = http_routes() for path in ( "/frames/([^/]+)/kernel", "/frames/([^/]+)/kernel/execute", "/frames/([^/]+)/kernel/variables", "/frames/([^/]+)/status", "/frames/([^/]+)/environments", "/frames/([^/]+)/execution", ): assert path in routes, f"{path} fell out of the inventory when it moved" def _mutating_route_ids(): from openai4s.server import kernel_routes return [spec.name for spec in kernel_routes.ROUTES if spec.mutates] def _mutating_routes(): from openai4s.server import kernel_routes return [spec for spec in kernel_routes.ROUTES if spec.mutates] @pytest.mark.parametrize("spec", _mutating_routes(), ids=_mutating_route_ids()) def test_the_quarantine_guard_still_covers_the_mutating_routes(spec, tmp_path): """The position dependency, verified rather than asserted in a comment. Nothing inside the extracted module re-checks writability, so a call site moved above that guard would make the code-execution endpoint live on a session imported from an untrusted archive and marked view-only. Parametrised over `ROUTES`, not over a hand-written list. Asserting only `kernel/execute` left the other six mutating routes -- restart, stop, interrupt, start, install, env -- with no quarantine assertion anywhere in the suite, while this module's docstring and `README.md` both claim the dependency is tested rather than merely commented. Deriving the cases from `spec.mutates` also gives that field its first consumer outside a test that restates it: a route declared `mutates=True` must now actually be write-gated, and a new one joins this test by existing. """ from openai4s.server.session_package import session_import_quarantine_key runner, handler, fid = _setup(tmp_path, notebook_repl=True) runner.store.set_setting(session_import_quarantine_key(fid), "1") path = spec.pattern.replace(r"([^/]+)", fid) with pytest.raises(gateway_mod.GatewayError) as raised: _call(handler, spec.method, path, body={"code": "1"}) assert raised.value.code == 423, f"{spec.name} is not write-protected"