familiar/tests/test_serve.py

"""Tests for transport selection, HTTP settings, and the bearer-auth gate."""

from __future__ import annotations

import asyncio

import pytest

from familiar import config, serve, server

_HTTP_ENV = [
    "FAMILIAR_TRANSPORT",
    "FAMILIAR_HTTP_HOST",
    "FAMILIAR_HTTP_PORT",
    "FAMILIAR_HTTP_PATH",
    "FAMILIAR_HTTP_AUTH_TOKEN",
    "FAMILIAR_ALLOW_INSECURE_BIND",
]


@pytest.fixture
def clean_env(monkeypatch):
    """Strip all transport-related env vars so defaults are exercised."""
    for name in _HTTP_ENV:
        monkeypatch.delenv(name, raising=False)
    return monkeypatch


# --------------------------------------------------------------------------- #
# config.http_settings
# --------------------------------------------------------------------------- #
def test_http_settings_defaults(clean_env):
    s = config.http_settings()
    assert s.transport == "stdio"
    assert s.host == "127.0.0.1"
    assert s.port == 8000
    assert s.path == "/mcp"
    assert s.auth_token is None
    assert s.allow_insecure_bind is False
    assert s.is_loopback is True


def test_http_settings_overrides(clean_env):
    clean_env.setenv("FAMILIAR_TRANSPORT", "HTTP")  # case-insensitive
    clean_env.setenv("FAMILIAR_HTTP_HOST", "0.0.0.0")
    clean_env.setenv("FAMILIAR_HTTP_PORT", "9001")
    clean_env.setenv("FAMILIAR_HTTP_PATH", "/eve")
    clean_env.setenv("FAMILIAR_HTTP_AUTH_TOKEN", "  secret  ")
    clean_env.setenv("FAMILIAR_ALLOW_INSECURE_BIND", "yes")
    s = config.http_settings()
    assert s.transport == "http"
    assert s.host == "0.0.0.0"
    assert s.port == 9001
    assert s.path == "/eve"
    assert s.auth_token == "secret"  # trimmed
    assert s.allow_insecure_bind is True
    assert s.is_loopback is False


# --------------------------------------------------------------------------- #
# serve.validate — fail-closed on public binds
# --------------------------------------------------------------------------- #
def _settings(**kw) -> config.HttpSettings:
    base = dict(
        transport="http",
        host="0.0.0.0",
        port=8000,
        path="/mcp",
        auth_token=None,
        allow_insecure_bind=False,
    )
    base.update(kw)
    return config.HttpSettings(**base)


def test_validate_stdio_is_noop():
    serve.validate(_settings(transport="stdio", host="0.0.0.0"))  # no raise


def test_validate_loopback_needs_no_token():
    serve.validate(_settings(host="127.0.0.1"))  # no raise


def test_validate_public_without_token_raises():
    with pytest.raises(serve.InsecureBind):
        serve.validate(_settings(host="0.0.0.0"))


def test_validate_public_with_token_ok():
    serve.validate(_settings(host="0.0.0.0", auth_token="secret"))  # no raise


def test_validate_public_with_insecure_optout_ok():
    serve.validate(_settings(host="0.0.0.0", allow_insecure_bind=True))  # no raise


# --------------------------------------------------------------------------- #
# server.build_http_auth — bearer token verifier
# --------------------------------------------------------------------------- #
def test_build_http_auth_none_without_token():
    assert server.build_http_auth(None) is None
    assert server.build_http_auth("") is None


def test_build_http_auth_verifies_matching_token():
    verifier = server.build_http_auth("s3cret")
    assert verifier is not None
    good = asyncio.run(verifier.verify_token("s3cret"))
    bad = asyncio.run(verifier.verify_token("wrong"))
    assert good is not None
    assert bad is None


# --------------------------------------------------------------------------- #
# serve.run — transport dispatch
# --------------------------------------------------------------------------- #
def test_run_stdio(clean_env, monkeypatch):
    calls = {}
    monkeypatch.setattr(server.mcp, "run", lambda *a, **k: calls.setdefault("args", (a, k)))
    serve.run()
    assert calls["args"] == ((), {})  # plain stdio run


def test_run_http_passes_transport_kwargs(clean_env, monkeypatch):
    clean_env.setenv("FAMILIAR_TRANSPORT", "http")
    clean_env.setenv("FAMILIAR_HTTP_AUTH_TOKEN", "secret")
    clean_env.setenv("FAMILIAR_HTTP_PORT", "9100")
    captured = {}
    monkeypatch.setattr(server.mcp, "run", lambda *a, **k: captured.update(k))
    serve.run()
    assert captured["transport"] == "http"
    assert captured["host"] == "127.0.0.1"
    assert captured["port"] == 9100
    assert captured["path"] == "/mcp"


def test_run_http_public_without_token_refuses(clean_env, monkeypatch):
    clean_env.setenv("FAMILIAR_TRANSPORT", "http")
    clean_env.setenv("FAMILIAR_HTTP_HOST", "0.0.0.0")
    monkeypatch.setattr(server.mcp, "run", lambda *a, **k: pytest.fail("must not run"))
    with pytest.raises(serve.InsecureBind):
        serve.run()


def test_run_unknown_transport_exits(clean_env, monkeypatch):
    clean_env.setenv("FAMILIAR_TRANSPORT", "carrier-pigeon")
    monkeypatch.setattr(server.mcp, "run", lambda *a, **k: pytest.fail("must not run"))
    with pytest.raises(SystemExit):
        serve.run()


# --------------------------------------------------------------------------- #
# /healthz custom route
# --------------------------------------------------------------------------- #
def test_healthz_route():
    resp = asyncio.run(server.healthz(None))
    assert resp.status_code == 200
    import json

    body = json.loads(bytes(resp.body))
    assert body["status"] == "ok"
    assert body["server"] == "familiar"