from __future__ import annotations import io import json import os import runpy from email.message import Message from pathlib import Path import pytest import solwebd ROOT = Path(__file__).resolve().parents[1] def handler(headers: list[tuple[str, str]], body: bytes = b"") -> solwebd.Handler: instance = object.__new__(solwebd.Handler) message = Message() for name, value in headers: message[name] = value instance.headers = message instance.rfile = io.BytesIO(body) instance.close_connection = False return instance @pytest.mark.parametrize("headers", [ [], [("Content-Length", "-1")], [("Content-Length", "01")], [("Content-Length", "1"), ("Content-Length", "1")], [("Content-Length", "0"), ("Transfer-Encoding", "chunked")], [("Content-Length", str(solwebd.MAX_REQUEST_BODY + 1))], ]) def test_post_body_framing_rejects_noncanonical_or_unbounded_lengths(headers): with pytest.raises(ValueError): handler(headers)._content_length() def test_post_body_reader_requires_the_declared_bytes(): instance = handler([("Content-Length", "4")], b"abc") with pytest.raises(ValueError, match="ended early"): instance._read_body(4) assert instance.close_connection is True def test_streamed_ask_error_preserves_typed_provider_failure(): instance = handler([("Content-Length", "0")]) instance.send_response = lambda *_: None instance.send_header = lambda *_: None instance.end_headers = lambda: None events: list[dict] = [] instance._write_event = events.append failure = { "kind": "capped", "detail": "daily limit reached", "retry_after_seconds": 900, "resume_at": "2026-08-11T09:00:00Z", } instance._run = lambda *_args, **_kwargs: (429, {"error": failure}, {"Retry-After": "900"}) instance._stream_ask({}) assert events == [{ "type": "error", "phase": "attaching", "class": "capped", "detail": "daily limit reached", "provider_failure": failure, "partial": {"phase": "attaching", "thinking": "", "answer": "", "saved_artifacts": []}, }] def test_gpt_provider_has_one_extension_owned_registration(): models = json.loads((ROOT.parent / "workstation/pi/agent/models.json").read_text()) assert "gpt" not in models["providers"] extension = (ROOT.parent / "workstation/pi/agent/extensions/gpt-sol.ts").read_text() assert extension.count('registerProvider("gpt"') == 1 @pytest.mark.parametrize("isolation_flag", ["-s"]) def test_solwebctl_accepts_its_exec_wrapped_daemon(monkeypatch, isolation_flag): namespace = runpy.run_path(str(ROOT / "bin/solwebctl")) monkeypatch.setattr(Path, "resolve", lambda self, strict=False: Path(os.sys.executable).resolve() if str(self).endswith("/exe") else self) monkeypatch.setattr(Path, "read_bytes", lambda self: f"python3\0{isolation_flag}\0{ROOT / 'solwebd.py'}\0".encode()) assert namespace["owned_command"](123) is True def test_solwebctl_accepts_exact_isolated_launcher(monkeypatch): namespace = runpy.run_path(str(ROOT / "bin/solwebctl")) monkeypatch.setattr(Path, "resolve", lambda self, strict=False: Path(os.sys.executable).resolve() if str(self).endswith("/exe") else self) command = "import runpy,sys;root=sys.argv.pop(1);script=sys.argv.pop(1);sys.path.insert(0,root);sys.argv[0]=script;runpy.run_path(script,run_name='__main__')" argv = ["python3", "-I", "-c", command, str(ROOT), str(ROOT / "solwebd.py"), "--seats", "1"] monkeypatch.setattr(Path, "read_bytes", lambda self: ("\0".join(argv) + "\0").encode()) assert namespace["owned_command"](123) is True def test_solwebctl_marks_owned_daemon_unverified_when_proc_exe_is_unreadable(monkeypatch): namespace = runpy.run_path(str(ROOT / "bin/solwebctl")) command = "import runpy,sys;root=sys.argv.pop(1);script=sys.argv.pop(1);sys.path.insert(0,root);sys.argv[0]=script;runpy.run_path(script,run_name='__main__')" argv = ["/spoofable/path/python", "-I", "-c", command, str(ROOT), str(ROOT / "solwebd.py"), "--seats", "1"] monkeypatch.setattr(Path, "read_bytes", lambda self: ("\0".join(argv) + "\0").encode()) original_resolve = Path.resolve def resolve(self, strict=False): if str(self).endswith("/exe"): raise PermissionError("ptrace policy blocks sibling /proc exe reads") return original_resolve(self, strict=strict) monkeypatch.setattr(Path, "resolve", resolve) assert namespace["owned_command"](123) is None def test_solwebctl_rejects_trusted_launcher_command_string(monkeypatch): namespace = runpy.run_path(str(ROOT / "bin/solwebctl")) monkeypatch.setattr(Path, "resolve", lambda self, strict=False: Path(os.sys.executable).resolve() if str(self).endswith("/exe") else self) argv = ["python3", "-I", "-c", "launcher", str(ROOT / "solwebd.py"), str(ROOT), "--seats", "1"] monkeypatch.setattr(Path, "read_bytes", lambda self: ("\0".join(argv) + "\0").encode()) assert namespace["owned_command"](123) is False @pytest.mark.parametrize("argv", [ ["python3", "-c", "pass", str(ROOT / "solwebd.py")], ["python3", "-m", "other", str(ROOT / "solwebd.py")], ["python3", "-I", "/tmp/other.py", str(ROOT / "solwebd.py")], ]) def test_solwebctl_rejects_daemon_path_as_unexecuted_argument(monkeypatch, argv): namespace = runpy.run_path(str(ROOT / "bin/solwebctl")) monkeypatch.setattr(Path, "resolve", lambda self, strict=False: Path(os.sys.executable).resolve() if str(self).endswith("/exe") else self) monkeypatch.setattr(Path, "read_bytes", lambda self: ("\0".join(argv) + "\0").encode()) assert namespace["owned_command"](123) is False def test_solwebctl_rejects_an_unrelated_live_process(): namespace = runpy.run_path(str(ROOT / "bin/solwebctl")) assert namespace["owned_command"](os.getpid()) is False def test_ensure_removes_dead_runtime_receipt_before_start(monkeypatch, tmp_path): namespace = runpy.run_path(str(ROOT / "bin/solwebctl")) runtime = tmp_path / "runtime.json" runtime.write_text(json.dumps({"pid": 99999999})) monkeypatch.setitem(namespace["ensure"].__globals__, "RUNTIME", runtime) monkeypatch.setitem(namespace["ensure"].__globals__, "STATE", tmp_path) monkeypatch.setitem(namespace["ensure"].__globals__, "IDENTITY", tmp_path / "identity.json") monkeypatch.setitem(namespace["ensure"].__globals__, "TOKEN", tmp_path / "token") monkeypatch.setitem(namespace["ensure"].__globals__, "identity", lambda: {}) monkeypatch.setitem(namespace["ensure"].__globals__, "token", lambda: "token") monkeypatch.setitem(namespace["ensure"].__globals__, "health", lambda _legacy: {"ok": False}) monkeypatch.setitem(namespace["ensure"].__globals__, "proc_ticks", lambda _pid: None) monkeypatch.setitem(namespace["ensure"].__globals__, "time", type("Clock", (), {"sleep": staticmethod(lambda _: (_ for _ in ()).throw(RuntimeError("started")))})()) monkeypatch.setitem(namespace["ensure"].__globals__["subprocess"].__dict__, "Popen", lambda *_args, **_kwargs: None) with pytest.raises(RuntimeError, match="started"): namespace["ensure"](1) assert not runtime.exists()