yunke/yunke-server/tests/test_api_auth.py

"""Tests for the shared-token API auth middleware (Task 2.9).

Trust model being pinned down here (per design doc §8 + plan Task 2.9):

* **No token configured** (``API_TOKEN`` unset / empty) -> auth is DISABLED.
  This is a local/intranet tool; when no token is set we trust the network
  boundary and every request passes through.
* **Token configured** -> every request must present the same token, either as
  ``Authorization: Bearer <token>`` or ``?token=<token>``. Missing / wrong
  tokens get ``401 Unauthorized``.
* ``GET /api/health`` is always reachable (liveness probes must not need to
  know the token).

The app is driven over an in-process httpx transport (``ASGITransport``) with
the ``get_session`` dependency overridden to an isolated in-memory SQLite
session. ``Settings`` is manipulated by mutating the environment plus
``get_settings.cache_clear()`` so the next request re-reads the config.
"""

from __future__ import annotations

import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.pool import StaticPool

from app.api import deps
from app.api.main import app
from app.config import get_settings
from app.db.base import Base
from app.db.models import Customer


@pytest.fixture
async def client_factory(monkeypatch):
    """Return a factory that builds an ``AsyncClient`` against the in-memory app.

    The factory takes an optional ``token``: when given, ``API_TOKEN`` is set in
    the environment and the ``get_settings`` cache is cleared so the running app
    picks up the new value on the next request. When ``None``, any leftover
    ``API_TOKEN`` env var is removed (also cache-cleared) so auth is disabled.

    Each call seeds a fresh in-memory SQLite DB with one ``Customer`` row so
    ``GET /api/customers`` has something to return beyond an empty page.
    Resources (client + engine) are torn down at fixture finalization.
    """
    created: list[tuple[AsyncClient, object]] = []

    async def _factory(token: str | None = None) -> AsyncClient:
        # Reconfigure settings for this call: clear cache first so the next
        # get_settings() inside require_auth observes the env change.
        get_settings.cache_clear()
        monkeypatch.delenv("API_TOKEN", raising=False)
        if token is not None:
            monkeypatch.setenv("API_TOKEN", token)
        # Second cache_clear guarantees the *next* get_settings() call rebuilds.
        get_settings.cache_clear()

        engine = create_async_engine(
            "sqlite+aiosqlite:///:memory:",
            poolclass=StaticPool,
            connect_args={"check_same_thread": False},
        )
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)

        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            session.add(
                Customer(
                    friend_id="wx_a",
                    channel="wechat",
                    owner_account_id="wx_owner_1",
                    user_id="u1",
                    friend_nickname="Alice",
                    last_chat_time=1000,
                    version="v1",
                )
            )
            await session.commit()

        async def _override_session():
            async with SessionLocal() as s:
                yield s

        app.dependency_overrides[deps.get_session] = _override_session
        ac = AsyncClient(transport=ASGITransport(app=app), base_url="http://test")
        await ac.__aenter__()
        created.append((ac, engine))
        return ac

    try:
        yield _factory
    finally:
        for ac, engine in created:
            await ac.aclose()
            await engine.dispose()
        app.dependency_overrides.clear()


async def test_no_token_configured_allows_access(client_factory):
    """Default (empty api_token) -> auth disabled, GET /api/customers is 200."""
    client = await client_factory(token=None)
    r = await client.get("/api/customers")
    assert r.status_code == 200
    assert r.json()["total"] == 1


async def test_correct_bearer_token_allows_access(client_factory):
    """Token configured + matching ``Authorization: Bearer`` -> 200."""
    client = await client_factory(token="test-secret")
    r = await client.get(
        "/api/customers", headers={"Authorization": "Bearer test-secret"}
    )
    assert r.status_code == 200
    assert r.json()["total"] == 1


async def test_correct_token_via_query_param_allows_access(client_factory):
    """Token configured + matching ``?token=`` query param -> 200."""
    client = await client_factory(token="test-secret")
    r = await client.get("/api/customers", params={"token": "test-secret"})
    assert r.status_code == 200


async def test_missing_token_returns_401(client_factory):
    """Token configured + no header/query -> 401."""
    client = await client_factory(token="test-secret")
    r = await client.get("/api/customers")
    assert r.status_code == 401


async def test_wrong_bearer_token_returns_401(client_factory):
    """Token configured + wrong bearer value -> 401."""
    client = await client_factory(token="test-secret")
    r = await client.get(
        "/api/customers", headers={"Authorization": "Bearer wrong-value"}
    )
    assert r.status_code == 401


async def test_malformed_authorization_header_returns_401(client_factory):
    """Token configured + header without ``Bearer`` scheme -> 401."""
    client = await client_factory(token="test-secret")
    r = await client.get(
        "/api/customers", headers={"Authorization": "test-secret"}
    )
    assert r.status_code == 401


async def test_health_accessible_without_token_when_auth_enabled(client_factory):
    """``/api/health`` is exempt: 200 even when a token is configured."""
    client = await client_factory(token="test-secret")
    r = await client.get("/api/health")
    assert r.status_code == 200
    assert r.json() == {"status": "ok"}


async def test_health_accessible_when_auth_disabled(client_factory):
    """``/api/health`` is 200 when auth is disabled (sanity check)."""
    client = await client_factory(token=None)
    r = await client.get("/api/health")
    assert r.status_code == 200
    assert r.json() == {"status": "ok"}