125 lines
4.9 KiB
Python
125 lines
4.9 KiB
Python
import asyncio
|
|
from urllib.parse import urlsplit
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from pwdlib import PasswordHash
|
|
from sqlalchemy import delete, select
|
|
|
|
from ai_platform.bootstrap.app import create_app
|
|
from ai_platform.infrastructure.config.settings import Settings
|
|
from ai_platform.infrastructure.persistence.database import make_session_factory
|
|
from ai_platform.infrastructure.persistence.models import (
|
|
AdminSession,
|
|
AdminUser,
|
|
AIProfile,
|
|
AIProfileAuditEvent,
|
|
)
|
|
|
|
|
|
def test_admin_selects_provider_with_csrf_and_audit():
|
|
loaded = Settings.load()
|
|
parsed = urlsplit(loaded.database_url)
|
|
if parsed.path != "/ai_platform_test" or parsed.hostname not in {"localhost", "127.0.0.1"}:
|
|
pytest.skip("requires isolated local test database")
|
|
settings = Settings(
|
|
database_url=loaded.database_url,
|
|
environment="development",
|
|
aihub_base_url="http://localhost:8080",
|
|
openai_base_url="http://127.0.0.1:8317",
|
|
)
|
|
factory = make_session_factory(settings.database_url)
|
|
version = f"provider-admin-{uuid4().hex}"
|
|
email = f"provider-admin-{uuid4().hex}@example.invalid"
|
|
password = "test-only-password"
|
|
original_active: set[str] = set()
|
|
|
|
async def setup() -> None:
|
|
nonlocal original_active
|
|
async with factory() as db:
|
|
original_active = {
|
|
row.version for row in (await db.scalars(select(AIProfile))).all() if row.active
|
|
}
|
|
db.add(AdminUser(email=email, password_hash=PasswordHash.recommended().hash(password)))
|
|
await db.commit()
|
|
await factory.kw["bind"].dispose()
|
|
|
|
async def cleanup() -> None:
|
|
try:
|
|
async with factory() as db:
|
|
await db.execute(
|
|
delete(AIProfileAuditEvent).where(AIProfileAuditEvent.version == version)
|
|
)
|
|
await db.execute(delete(AIProfile).where(AIProfile.version == version))
|
|
await db.execute(delete(AdminSession).where(AdminSession.admin_email == email))
|
|
await db.execute(delete(AdminUser).where(AdminUser.email == email))
|
|
for row in (await db.scalars(select(AIProfile))).all():
|
|
row.active = row.version in original_active
|
|
await db.commit()
|
|
finally:
|
|
await factory.kw["bind"].dispose()
|
|
|
|
asyncio.run(setup())
|
|
try:
|
|
profile = {
|
|
"version": version,
|
|
"provider": "openai_compatible",
|
|
"base_url": "http://127.0.0.1:8317",
|
|
"model": "gemini-3.5-flash-lite",
|
|
"system_prompt": "Chỉ dùng dữ liệu có trong snapshot.",
|
|
"timeout_seconds": 30,
|
|
}
|
|
with TestClient(create_app(settings)) as client:
|
|
assert client.put("/v1/admin/ai-profiles", json=profile).status_code == 401
|
|
login = client.post("/v1/admin/login", json={"email": email, "password": password})
|
|
assert login.status_code == 200
|
|
assert client.put("/v1/admin/ai-profiles", json=profile).status_code == 403
|
|
csrf = {"x-csrf-token": login.json()["csrf_token"]}
|
|
assert (
|
|
client.put(
|
|
"/v1/admin/ai-profiles", json={**profile, "provider": "unknown"}, headers=csrf
|
|
).status_code
|
|
== 422
|
|
)
|
|
created = client.put("/v1/admin/ai-profiles", json=profile, headers=csrf)
|
|
assert created.status_code == 200
|
|
assert created.json()["provider"] == "openai_compatible"
|
|
assert created.json()["base_url"] == profile["base_url"]
|
|
assert (
|
|
client.put(
|
|
"/v1/admin/ai-profiles",
|
|
json={**profile, "base_url": "http://127.0.0.1:8318"},
|
|
headers=csrf,
|
|
).status_code
|
|
== 409
|
|
)
|
|
assert (
|
|
client.put(
|
|
"/v1/admin/ai-profiles",
|
|
json={**profile, "provider": "aihub", "base_url": None},
|
|
headers=csrf,
|
|
).status_code
|
|
== 409
|
|
)
|
|
listed = client.get("/v1/admin/ai-profiles")
|
|
assert listed.status_code == 200
|
|
assert (
|
|
next(p for p in listed.json()["profiles"] if p["version"] == version)["provider"]
|
|
== "openai_compatible"
|
|
)
|
|
assert "api_key" not in listed.text.lower()
|
|
|
|
async def verify() -> None:
|
|
async with factory() as db:
|
|
event = await db.scalar(
|
|
select(AIProfileAuditEvent).where(AIProfileAuditEvent.version == version)
|
|
)
|
|
assert event is not None
|
|
assert (event.actor, event.provider) == (email, "openai_compatible")
|
|
await factory.kw["bind"].dispose()
|
|
|
|
asyncio.run(verify())
|
|
finally:
|
|
asyncio.run(cleanup())
|