feat: add prompt history management
This commit is contained in:
@@ -0,0 +1,98 @@
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from sqlalchemy import create_engine, select
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
from app.api.admin_settings import (
|
||||
DEFAULT_PROMPT,
|
||||
get_prompt,
|
||||
prompt_history,
|
||||
prompt_history_detail,
|
||||
reset_prompt,
|
||||
restore_prompt,
|
||||
save_prompt,
|
||||
)
|
||||
from app.models import Base
|
||||
from app.models.admin import Admin
|
||||
from app.models.ai_config import Prompt
|
||||
from app.schemas.admin import PromptSaveRequest
|
||||
|
||||
|
||||
def _database() -> Session:
|
||||
engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool)
|
||||
Base.metadata.create_all(engine)
|
||||
return Session(engine)
|
||||
|
||||
|
||||
def _admin() -> Admin:
|
||||
return Admin(id=1, username="admin", password="hash", name="系统管理员", status=1)
|
||||
|
||||
|
||||
def test_prompt_history_is_paginated_and_excludes_full_content_from_list():
|
||||
with _database() as db:
|
||||
admin = _admin()
|
||||
db.add(admin)
|
||||
started = datetime(2026, 1, 1)
|
||||
db.add_all([
|
||||
Prompt(
|
||||
id=index,
|
||||
prompt_content=f"版本 {index} " + ("x" * 500),
|
||||
change_type="save",
|
||||
updated_by=admin.id,
|
||||
updated_at=started + timedelta(minutes=index),
|
||||
)
|
||||
for index in range(1, 16)
|
||||
])
|
||||
db.commit()
|
||||
|
||||
result = prompt_history(page=2, pageSize=10, db=db, current_admin=admin)["data"]
|
||||
|
||||
assert result["total"] == 15
|
||||
assert len(result["items"]) == 5
|
||||
assert result["items"][0]["id"] == 5
|
||||
assert result["items"][0]["isCurrent"] is False
|
||||
assert "promptContent" not in result["items"][0]
|
||||
assert len(result["items"][0]["preview"]) <= 140
|
||||
|
||||
|
||||
def test_prompt_history_detail_and_restore_create_a_new_current_version():
|
||||
with _database() as db:
|
||||
admin = _admin()
|
||||
db.add(admin)
|
||||
db.add_all([
|
||||
Prompt(id=1, prompt_content="旧版主提示词", change_type="save", updated_by=admin.id),
|
||||
Prompt(id=2, prompt_content="当前主提示词", change_type="save", updated_by=admin.id),
|
||||
])
|
||||
db.commit()
|
||||
|
||||
detail = prompt_history_detail(1, db=db, current_admin=admin)["data"]
|
||||
restored = restore_prompt(1, db=db, current_admin=admin)["data"]
|
||||
current = get_prompt(db=db, current_admin=admin)["data"]
|
||||
|
||||
assert detail["promptContent"] == "旧版主提示词"
|
||||
assert restored["id"] not in {1, 2}
|
||||
assert restored["changeType"] == "restore"
|
||||
assert restored["sourcePromptId"] == 1
|
||||
assert current["id"] == restored["id"]
|
||||
assert current["promptContent"] == "旧版主提示词"
|
||||
assert db.scalar(select(Prompt).where(Prompt.id == 2)).prompt_content == "当前主提示词"
|
||||
|
||||
|
||||
def test_save_and_reset_each_create_an_auditable_prompt_version():
|
||||
with _database() as db:
|
||||
admin = _admin()
|
||||
db.add(admin)
|
||||
db.commit()
|
||||
|
||||
saved = save_prompt(PromptSaveRequest(promptContent="新主提示词"), db=db, current_admin=admin)["data"]
|
||||
reset = reset_prompt(db=db, current_admin=admin)["data"]
|
||||
versions = db.scalars(select(Prompt).order_by(Prompt.id)).all()
|
||||
|
||||
assert saved["changeType"] == "save"
|
||||
assert reset["changeType"] == "reset"
|
||||
assert reset["promptContent"] == DEFAULT_PROMPT
|
||||
assert [(item.change_type, item.prompt_content) for item in versions] == [
|
||||
("save", "新主提示词"),
|
||||
("reset", DEFAULT_PROMPT),
|
||||
]
|
||||
Reference in New Issue
Block a user