feat(07-05): admin AI-config endpoints + load_provider_config_by_id + atomic is_active flip
- Add GET/PUT /api/admin/ai-config + GET /api/admin/ai-config/test-connection - _ai_config_to_dict whitelist excludes api_key_enc (T-07-01) - PUT: HKDF-encrypt api_key, atomic UPDATE SET is_active = (provider_id = target) (T-07-03) - PUT: upsert with PROVIDER_DEFAULTS for omitted fields; audit log records fields_changed only (T-07-14) - All 3 endpoints require get_current_admin (T-07-15) - Add load_provider_config_by_id to services/ai_config.py (ignores is_active) - Promote 3 xfail tests in test_admin_ai_config.py — all 3 pass
This commit is contained in:
@@ -1,25 +1,134 @@
|
||||
"""
|
||||
Wave 0 xfail stubs for Phase 7 admin AI config endpoint tests.
|
||||
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)
|
||||
- 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)
|
||||
|
||||
Each function is a placeholder to be promoted in Plan 07-05 once the admin
|
||||
AI config endpoints exist.
|
||||
|
||||
Stub policy (STATE.md decision: xfail(strict=False) for Wave 0):
|
||||
- Body is a single pytest.xfail() call — no assertion code.
|
||||
- strict=False so unexpected passes (xpass) never break CI.
|
||||
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
|
||||
|
||||
|
||||
@pytest.mark.xfail(strict=False, reason="Wave 0 stub — promoted in Plan 07-05")
|
||||
async def test_get_never_returns_key():
|
||||
pytest.xfail("not implemented yet — Plan 07-05")
|
||||
# ── 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")
|
||||
|
||||
|
||||
@pytest.mark.xfail(strict=False, reason="Wave 0 stub — promoted in Plan 07-05")
|
||||
async def test_put_writes_active_provider():
|
||||
pytest.xfail("not implemented yet — Plan 07-05")
|
||||
# ── 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}"
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user