kite/backend/tests/test_admin_ai_config.py

"""
Integration tests for Phase 7 admin AI config endpoints.

Promoted from Wave 0 xfail stubs in Plan 07-05.

Covers:
  - T-07-01: GET /api/admin/ai-config never returns api_key_enc (D-05/D-08)
  - D-08:    PUT /api/admin/ai-config writes active provider (admin AI Providers panel)
  - T-07-15: PUT /api/admin/ai-config is admin-only (non-admin gets 403)

All tests run against the in-memory SQLite fixture; no live services required.
"""
import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession


# ── Helpers ───────────────────────────────────────────────────────────────────

async def _get_admin_client(db_session: AsyncSession) -> AsyncClient:
    """Return an AsyncClient with the DB dependency overridden to db_session."""
    from deps.db import get_db
    from main import app

    app.dependency_overrides[get_db] = lambda: db_session
    return AsyncClient(transport=ASGITransport(app=app), base_url="http://test")


# ── Tests ─────────────────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_get_never_returns_key(
    async_client: AsyncClient,
    admin_user: dict,
    db_session: AsyncSession,
):
    """GET /api/admin/ai-config must never include api_key_enc or a plaintext key.

    Mitigates T-07-01: admin panel response whitelist (D-08).
    """
    # First PUT a provider config with an API key so the DB row exists
    put_resp = await async_client.put(
        "/api/admin/ai-config",
        json={"provider_id": "openai", "api_key": "sk-test-secret"},
        headers=admin_user["headers"],
    )
    assert put_resp.status_code == 200

    # GET must not expose the key in any form
    get_resp = await async_client.get(
        "/api/admin/ai-config",
        headers=admin_user["headers"],
    )
    assert get_resp.status_code == 200
    body_text = get_resp.text

    # Security assertions (T-07-01)
    assert "api_key_enc" not in body_text, "api_key_enc must never appear in GET response"
    assert "sk-test-secret" not in body_text, "plaintext API key must never appear in GET response"

    providers = get_resp.json()["providers"]
    assert len(providers) > 0, "providers list must not be empty"

    openai_row = next((p for p in providers if p["provider_id"] == "openai"), None)
    assert openai_row is not None, "openai provider must appear in response"
    assert "api_key_enc" not in openai_row, "api_key_enc key must not be present in provider dict"
    assert openai_row["has_api_key"] is True, "has_api_key must be True after saving a key"


@pytest.mark.asyncio
async def test_put_writes_active_provider(
    async_client: AsyncClient,
    admin_user: dict,
    db_session: AsyncSession,
):
    """PUT /api/admin/ai-config with is_active=True atomically flips active provider.

    Mitigates T-07-03: no dual-active race (D-08).
    After two sequential PUTs with is_active=True on different providers,
    exactly one row must have is_active=True.
    """
    from sqlalchemy import func, select
    from db.models import SystemSettings

    # PUT openai as active
    r1 = await async_client.put(
        "/api/admin/ai-config",
        json={"provider_id": "openai", "api_key": "sk-test", "is_active": True},
        headers=admin_user["headers"],
    )
    assert r1.status_code == 200
    assert r1.json()["has_api_key"] is True
    assert r1.json()["is_active"] is True

    # PUT anthropic as active (should atomically flip openai to inactive)
    r2 = await async_client.put(
        "/api/admin/ai-config",
        json={"provider_id": "anthropic", "is_active": True},
        headers=admin_user["headers"],
    )
    assert r2.status_code == 200
    assert r2.json()["is_active"] is True

    # DB assertion: exactly one row must be active
    count_result = await db_session.execute(
        select(func.count(SystemSettings.id)).where(
            SystemSettings.is_active.is_(True)
        )
    )
    active_count = count_result.scalar_one()
    assert active_count == 1, (
        f"Expected exactly 1 active provider after two PUTs with is_active=True, "
        f"got {active_count}"
    )


@pytest.mark.asyncio
async def test_put_admin_only(
    async_client: AsyncClient,
    auth_user: dict,
):
    """PUT /api/admin/ai-config by a regular (non-admin) user must return 403.

    Mitigates T-07-15: admin-only enforcement (D-08).
    """
    resp = await async_client.put(
        "/api/admin/ai-config",
        json={"provider_id": "openai", "api_key": "sk-test"},
        headers=auth_user["headers"],
    )
    assert resp.status_code == 403, (
        f"Expected 403 Forbidden for non-admin user, got {resp.status_code}"
    )