127 lines
3.9 KiB
Python
127 lines
3.9 KiB
Python
"""Tests for prompt profile import endpoint (Phase PX.2).
|
|
|
|
Covers:
|
|
- POST /api/v1/prompts/profiles/{name}/import — import with validation
|
|
- Format validation, step validation, profile name validation
|
|
- Import overwrites existing prompts, does not change active profile
|
|
"""
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.core.sqlite_db import init_prompts_db, seed_default_profiles, _get_db
|
|
from app.routers.prompts import router
|
|
|
|
|
|
_VALID_STEPS = {
|
|
"decompose", "filter", "generate",
|
|
"generate_per_subq", "filter_intro", "filter_section", "filter_outro",
|
|
}
|
|
|
|
_IMPORT_PAYLOAD = {
|
|
"format": "legco-reranker-profile/v1",
|
|
"profile_name": "A",
|
|
"prompts": {
|
|
"decompose": "imported decompose",
|
|
"filter": "imported filter",
|
|
"generate": "imported generate",
|
|
"generate_per_subq": "imported generate_per_subq",
|
|
"filter_intro": "imported filter_intro",
|
|
"filter_section": "imported filter_section",
|
|
"filter_outro": "imported filter_outro",
|
|
},
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def client(tmp_path, monkeypatch):
|
|
prompts_path = str(tmp_path / "prompts.db")
|
|
monkeypatch.setenv("PROMPTS_DB_PATH", prompts_path)
|
|
monkeypatch.setenv("HISTORY_DB_PATH", str(tmp_path / "history.db"))
|
|
|
|
from app.core.config import get_settings
|
|
get_settings.cache_clear()
|
|
from app.core.dependencies import get_settings_cached
|
|
get_settings_cached.cache_clear()
|
|
|
|
conn = _get_db(prompts_path)
|
|
init_prompts_db(conn)
|
|
seed_default_profiles(conn)
|
|
conn.close()
|
|
|
|
test_app = FastAPI()
|
|
test_app.include_router(router)
|
|
|
|
yield TestClient(test_app)
|
|
|
|
get_settings_cached.cache_clear()
|
|
get_settings.cache_clear()
|
|
|
|
|
|
def test_import_valid(client):
|
|
resp = client.post("/api/v1/prompts/profiles/B/import", json=_IMPORT_PAYLOAD)
|
|
assert resp.status_code == 200
|
|
|
|
data = resp.json()
|
|
assert data["status"] == "ok"
|
|
assert data["profile"] == "B"
|
|
assert data["imported_steps"] == 7
|
|
assert data["source_profile"] == "A"
|
|
|
|
|
|
def test_import_missing_step(client):
|
|
payload = {
|
|
"format": "legco-reranker-profile/v1",
|
|
"profile_name": "A",
|
|
"prompts": {k: "x" for k in list(_VALID_STEPS) if k != "decompose"},
|
|
}
|
|
resp = client.post("/api/v1/prompts/profiles/B/import", json=payload)
|
|
assert resp.status_code == 400
|
|
assert "decompose" in resp.json()["detail"]
|
|
|
|
|
|
def test_import_unknown_step(client):
|
|
payload = {
|
|
"format": "legco-reranker-profile/v1",
|
|
"profile_name": "A",
|
|
"prompts": {**{k: "x" for k in _VALID_STEPS}, "extra_step": "x"},
|
|
}
|
|
resp = client.post("/api/v1/prompts/profiles/B/import", json=payload)
|
|
assert resp.status_code == 400
|
|
assert "extra_step" in resp.json()["detail"]
|
|
|
|
|
|
def test_import_invalid_format(client):
|
|
payload = {**_IMPORT_PAYLOAD, "format": "v999"}
|
|
resp = client.post("/api/v1/prompts/profiles/B/import", json=payload)
|
|
assert resp.status_code == 400
|
|
assert "v999" in resp.json()["detail"]
|
|
|
|
|
|
def test_import_invalid_target(client):
|
|
resp = client.post("/api/v1/prompts/profiles/X/import", json=_IMPORT_PAYLOAD)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_import_overwrites_existing(client):
|
|
resp = client.get("/api/v1/prompts/profiles/B")
|
|
original = resp.json()["prompts"]["decompose"]
|
|
|
|
client.post("/api/v1/prompts/profiles/B/import", json=_IMPORT_PAYLOAD)
|
|
|
|
resp = client.get("/api/v1/prompts/profiles/B")
|
|
assert resp.json()["prompts"]["decompose"] == "imported decompose"
|
|
assert resp.json()["prompts"]["decompose"] != original
|
|
|
|
|
|
def test_import_does_not_change_active(client):
|
|
resp = client.get("/api/v1/prompts/profiles")
|
|
active_before = {p["name"]: p["is_active"] for p in resp.json()["profiles"]}
|
|
|
|
client.post("/api/v1/prompts/profiles/B/import", json=_IMPORT_PAYLOAD)
|
|
|
|
resp = client.get("/api/v1/prompts/profiles")
|
|
active_after = {p["name"]: p["is_active"] for p in resp.json()["profiles"]}
|
|
assert active_before == active_after
|