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