74 lines
3.3 KiB
Python
74 lines
3.3 KiB
Python
import pytest
|
|
import asyncio
|
|
from fastapi.testclient import TestClient
|
|
from thinkstorm.main import app
|
|
from thinkstorm.database import init_db, get_db, get_utc_now
|
|
from thinkstorm.processors.pipeline import execute_work_track_workflow
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def setup_database():
|
|
init_db()
|
|
|
|
client = TestClient(app)
|
|
|
|
def test_work_track_artifact_versioning_and_deletion():
|
|
# 1. Login as admin
|
|
login_res = client.post("/api/auth/login", json={"username": "admin", "password": "admin-thinkstorm-pass-2026"})
|
|
assert login_res.status_code == 200
|
|
|
|
# 2. Submit an idea
|
|
sub_res = client.post("/api/ideas", json={"text": "Test Idea for Artifact Versioning and Deletion System"})
|
|
assert sub_res.status_code == 201
|
|
idea_id = sub_res.json()["id"]
|
|
|
|
# 3. Create a work track
|
|
track_res = client.post(
|
|
f"/api/ideas/{idea_id}/work-tracks",
|
|
json={"work_type_id": "ARTICLE", "name": "Article Test Track", "model_override": "auto/best-fast"}
|
|
)
|
|
assert track_res.status_code == 200
|
|
track_id = track_res.json()["id"]
|
|
assert track_res.json()["model_override"] == "auto/best-fast"
|
|
|
|
# 4. Execute work track workflow (Run 1 -> Version 1)
|
|
asyncio.run(execute_work_track_workflow(track_id, model_override="auto/best-fast"))
|
|
|
|
with get_db() as conn:
|
|
outs_v1 = conn.execute("SELECT id, name, version, is_current, model_used FROM work_track_outputs WHERE work_track_id = ? ORDER BY version DESC", (track_id,)).fetchall()
|
|
assert len(outs_v1) >= 2
|
|
for o in outs_v1:
|
|
assert o["version"] == 1
|
|
assert o["is_current"] == 1
|
|
|
|
# 5. Execute work track workflow again (Run 2 -> Version 2)
|
|
asyncio.run(execute_work_track_workflow(track_id, model_override="auto/best-reasoning"))
|
|
|
|
with get_db() as conn:
|
|
outs_v2 = conn.execute("SELECT id, name, version, is_current, model_used FROM work_track_outputs WHERE work_track_id = ? ORDER BY name ASC, version DESC", (track_id,)).fetchall()
|
|
# Should now have 4 records (2 for article.md: v2 and v1; 2 for outline.md: v2 and v1)
|
|
assert len(outs_v2) == 4
|
|
article_v2 = [o for o in outs_v2 if o["name"] == "article.md" and o["version"] == 2][0]
|
|
article_v1 = [o for o in outs_v2 if o["name"] == "article.md" and o["version"] == 1][0]
|
|
assert article_v2["is_current"] == 1
|
|
assert article_v1["is_current"] == 0
|
|
|
|
# 6. Verify GET /api/ideas/{id} serializes version metadata
|
|
detail_res = client.get(f"/api/ideas/{idea_id}")
|
|
assert detail_res.status_code == 200
|
|
data = detail_res.json()
|
|
track_data = [t for t in data["work_tracks"] if t["id"] == track_id][0]
|
|
assert len(track_data["outputs"]) == 4
|
|
output_ids = [o["id"] for o in track_data["outputs"]]
|
|
|
|
# 7. Delete artifact version 2 of article.md
|
|
del_res = client.delete(f"/api/ideas/work-tracks/outputs/{article_v2['id']}")
|
|
assert del_res.status_code == 200
|
|
assert "deleted successfully" in del_res.json()["message"]
|
|
|
|
# 8. Verify article_v1 is now promoted to is_current = 1
|
|
with get_db() as conn:
|
|
remaining_article = conn.execute("SELECT id, version, is_current FROM work_track_outputs WHERE work_track_id = ? AND name = 'article.md'", (track_id,)).fetchall()
|
|
assert len(remaining_article) == 1
|
|
assert remaining_article[0]["version"] == 1
|
|
assert remaining_article[0]["is_current"] == 1
|