From 44b2dbac0c32ee1bc7c651c239da533b8096da93 Mon Sep 17 00:00:00 2001 From: Latte Date: Sat, 4 Jul 2026 18:22:01 +0200 Subject: [PATCH] test(serve): cover HTTP bearer auth, start refusal, config, write default, headless login - 401 without/with a bad token (incl. WWW-Authenticate: Bearer), 200 with a valid one, driven through the ASGI app in-process. - configure_http refuses to start when the token is unset or blank. - FAMILIAR_HTTP_* config defaults and overrides; http_token() strip/empty. - HTTP write default disabled by default, env still wins, stdio stays enabled. - parse_callback (URL / bare code / empty) and the manual_login exchange flow. --- tests/test_serve.py | 216 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 216 insertions(+) create mode 100644 tests/test_serve.py diff --git a/tests/test_serve.py b/tests/test_serve.py new file mode 100644 index 0000000..7acda01 --- /dev/null +++ b/tests/test_serve.py @@ -0,0 +1,216 @@ +"""Tests for the opt-in HTTP transport, bearer auth, and headless login. + +The stdio path and every existing tool stay untouched; these only exercise the +new `familiar serve` surface: the bearer verifier (401 without / with a bad +token, through with a good one), the start-refusal when no token is set, the +FAMILIAR_HTTP_* config defaults, the write-default flip in HTTP mode, and the +headless-login callback parsing. +""" + +from __future__ import annotations + +import asyncio +import importlib + +import httpx +import pytest + +from conftest import make_jwt + +from familiar import auth, config, server, writes + + +# --------------------------------------------------------------------------- # +# Bearer auth on the HTTP transport +# --------------------------------------------------------------------------- # +def _initialize_request() -> dict: + return { + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "protocolVersion": "2025-06-18", + "capabilities": {}, + "clientInfo": {"name": "test", "version": "1"}, + }, + } + + +async def _status_for(app, authorization: str | None) -> int: + """POST an MCP initialize to the ASGI app and return the HTTP status.""" + headers = {"Accept": "application/json, text/event-stream"} + if authorization is not None: + headers["Authorization"] = authorization + transport = httpx.ASGITransport(app=app) + async with app.router.lifespan_context(app): + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + resp = await client.post("/mcp", json=_initialize_request(), headers=headers) + return resp.status_code, resp.headers.get("www-authenticate") + + +@pytest.fixture +def http_server(monkeypatch): + """Configure the shared mcp for HTTP with a known token, then reset it.""" + monkeypatch.setenv("FAMILIAR_HTTP_TOKEN", "s3cret-token") + original_auth = server.mcp.auth + token = server.configure_http() + app = server.mcp.http_app(path="/mcp") + yield token, app + server.mcp.auth = original_auth + writes.set_http_context(False) + + +def test_missing_token_is_401_with_www_authenticate(http_server): + _, app = http_server + status, www_auth = asyncio.run(_status_for(app, None)) + assert status == 401 + assert www_auth and www_auth.lower().startswith("bearer") + + +def test_wrong_token_is_401(http_server): + _, app = http_server + status, _ = asyncio.run(_status_for(app, "Bearer nope")) + assert status == 401 + + +def test_valid_token_passes_handshake(http_server): + token, app = http_server + status, _ = asyncio.run(_status_for(app, f"Bearer {token}")) + assert status == 200 + + +# --------------------------------------------------------------------------- # +# Start refusal without a token +# --------------------------------------------------------------------------- # +def test_configure_http_refuses_without_token(monkeypatch): + monkeypatch.delenv("FAMILIAR_HTTP_TOKEN", raising=False) + original_auth = server.mcp.auth + try: + with pytest.raises(SystemExit) as excinfo: + server.configure_http() + assert "FAMILIAR_HTTP_TOKEN" in str(excinfo.value) + finally: + server.mcp.auth = original_auth + writes.set_http_context(False) + + +def test_configure_http_refuses_on_blank_token(monkeypatch): + monkeypatch.setenv("FAMILIAR_HTTP_TOKEN", " ") + original_auth = server.mcp.auth + try: + with pytest.raises(SystemExit): + server.configure_http() + finally: + server.mcp.auth = original_auth + writes.set_http_context(False) + + +# --------------------------------------------------------------------------- # +# Config defaults for the new env vars +# --------------------------------------------------------------------------- # +def test_http_config_defaults(monkeypatch): + for var in ("FAMILIAR_HTTP_HOST", "FAMILIAR_HTTP_PORT", "FAMILIAR_HTTP_PATH"): + monkeypatch.delenv(var, raising=False) + try: + importlib.reload(config) + assert config.HTTP_HOST == "127.0.0.1" + assert config.HTTP_PORT == 8080 + assert config.HTTP_PATH == "/mcp" + finally: + importlib.reload(config) + + +def test_http_config_overrides(monkeypatch): + monkeypatch.setenv("FAMILIAR_HTTP_HOST", "0.0.0.0") + monkeypatch.setenv("FAMILIAR_HTTP_PORT", "9001") + monkeypatch.setenv("FAMILIAR_HTTP_PATH", "/eve") + try: + importlib.reload(config) + assert config.HTTP_HOST == "0.0.0.0" + assert config.HTTP_PORT == 9001 + assert config.HTTP_PATH == "/eve" + finally: + for var in ("FAMILIAR_HTTP_HOST", "FAMILIAR_HTTP_PORT", "FAMILIAR_HTTP_PATH"): + monkeypatch.delenv(var, raising=False) + importlib.reload(config) + + +def test_http_token_reads_and_strips_env(monkeypatch): + monkeypatch.setenv("FAMILIAR_HTTP_TOKEN", " padded ") + assert config.http_token() == "padded" + monkeypatch.delenv("FAMILIAR_HTTP_TOKEN", raising=False) + assert config.http_token() == "" + + +# --------------------------------------------------------------------------- # +# Write default in the HTTP context +# --------------------------------------------------------------------------- # +def test_http_context_disables_writes_by_default(monkeypatch, tmp_path): + monkeypatch.setenv("FAMILIAR_DATA_DIR", str(tmp_path)) + monkeypatch.delenv("FAMILIAR_ENABLE_WRITES", raising=False) + try: + writes.set_http_context(True) + assert writes.writes_enabled() is False + finally: + writes.set_http_context(False) + + +def test_http_context_write_env_still_wins(monkeypatch, tmp_path): + monkeypatch.setenv("FAMILIAR_DATA_DIR", str(tmp_path)) + monkeypatch.setenv("FAMILIAR_ENABLE_WRITES", "1") + try: + writes.set_http_context(True) + assert writes.writes_enabled() is True + finally: + writes.set_http_context(False) + + +def test_stdio_writes_stay_enabled_by_default(monkeypatch, tmp_path): + monkeypatch.setenv("FAMILIAR_DATA_DIR", str(tmp_path)) + monkeypatch.delenv("FAMILIAR_ENABLE_WRITES", raising=False) + writes.set_http_context(False) + assert writes.writes_enabled() is True + + +# --------------------------------------------------------------------------- # +# Headless login callback parsing + flow +# --------------------------------------------------------------------------- # +def test_parse_callback_full_url(): + code, state = auth.parse_callback("http://localhost:8765/callback?code=abc123&state=xyz") + assert code == "abc123" + assert state == "xyz" + + +def test_parse_callback_bare_code(): + code, state = auth.parse_callback(" justthecode ") + assert code == "justthecode" + assert state is None + + +def test_parse_callback_rejects_empty(): + with pytest.raises(auth.AuthError): + auth.parse_callback("") + + +def test_manual_login_exchanges_and_saves(monkeypatch, tmp_path): + monkeypatch.setenv("FAMILIAR_DATA_DIR", str(tmp_path)) + saved: dict = {} + + async def fake_exchange(code, verifier, **kwargs): + assert code == "the-code" + return { + "access_token": make_jwt({"sub": "CHARACTER:EVE:42", "name": "Pilot", "scp": []}), + "refresh_token": "refresh-xyz", + "expires_in": 1200, + } + + monkeypatch.setattr(auth, "exchange_code", fake_exchange) + monkeypatch.setattr(auth, "_save_character", lambda bundle: saved.update(bundle)) + + bundle = auth.manual_login( + read_input=lambda _prompt: "http://localhost:8765/callback?code=the-code" + ) + + assert bundle["character_id"] == 42 + assert bundle["character_name"] == "Pilot" + assert saved["character_id"] == 42