291 lines
11 KiB
Python
291 lines
11 KiB
Python
import time
|
|
import uuid
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _register_and_login(client: httpx.Client, email: str, password: str = "password123") -> str:
|
|
client.post("/auth/register", json={"email": email, "password": password})
|
|
resp = client.post("/auth/login", json={"email": email, "password": password})
|
|
return resp.json()["access_token"]
|
|
|
|
|
|
def _auth_client(client: httpx.Client, email: str) -> httpx.Client:
|
|
token = _register_and_login(client, email)
|
|
client.headers["Authorization"] = f"Bearer {token}"
|
|
return client
|
|
|
|
|
|
def _wait_for_adventure_status(
|
|
client: httpx.Client,
|
|
adventure_id: str,
|
|
expected_status: str,
|
|
timeout: int = 30,
|
|
) -> dict:
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
resp = client.get(f"/api/adventures/{adventure_id}")
|
|
assert resp.status_code == 200, resp.text
|
|
if resp.json()["status"] == expected_status:
|
|
return resp.json()
|
|
time.sleep(0.5)
|
|
raise TimeoutError(
|
|
f"Adventure {adventure_id} did not reach '{expected_status}' within {timeout}s. "
|
|
f"Last status: {client.get(f'/api/adventures/{adventure_id}').json().get('status')}"
|
|
)
|
|
|
|
|
|
_DEFAULT_ADVENTURE_BODY = {
|
|
"language": "fr",
|
|
"source_language": "en",
|
|
"competencies": ["B1"],
|
|
"genres": ["crime fiction"],
|
|
"setting": ["Paris", "city"],
|
|
"vibes": ["dark"],
|
|
"protagonist": ["male", "late-teens"],
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def user_client(client: httpx.Client) -> httpx.Client:
|
|
email = f"adventure-user-{uuid.uuid4()}@example.com"
|
|
return _auth_client(client, email)
|
|
|
|
|
|
@pytest.fixture
|
|
def second_user_client(client: httpx.Client) -> httpx.Client:
|
|
email = f"adventure-user2-{uuid.uuid4()}@example.com"
|
|
return _auth_client(client, email)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test 1: Adventure creation and first-entry pipeline
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_create_adventure_generates_first_entry(user_client: httpx.Client) -> None:
|
|
resp = user_client.post("/api/adventures", json=_DEFAULT_ADVENTURE_BODY)
|
|
|
|
assert resp.status_code == 201
|
|
body = resp.json()
|
|
adventure_id = body["id"]
|
|
assert body["status"] == "awaiting_first_entry"
|
|
assert body["title"] == "Untitled adventure"
|
|
|
|
adventure = _wait_for_adventure_status(user_client, adventure_id, "active")
|
|
assert adventure["title"] != "Untitled adventure"
|
|
assert adventure["description"] is not None
|
|
|
|
entries_resp = user_client.get(f"/api/adventures/{adventure_id}/entries")
|
|
assert entries_resp.status_code == 200
|
|
entries = entries_resp.json()
|
|
assert len(entries) == 1
|
|
assert entries[0]["status"] == "complete"
|
|
assert entries[0]["entry_index"] == 0
|
|
assert entries[0]["story_text"] is not None
|
|
|
|
detail_resp = user_client.get(f"/api/adventures/{adventure_id}/entries/{entries[0]['id']}")
|
|
assert detail_resp.status_code == 200
|
|
detail = detail_resp.json()
|
|
assert len(detail["choices"]) == 4
|
|
assert detail["translation"] is not None
|
|
assert detail["audio_file_name"] is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test 2: Recording a decision generates the next entry
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_record_decision_generates_next_entry(user_client: httpx.Client) -> None:
|
|
resp = user_client.post("/api/adventures", json=_DEFAULT_ADVENTURE_BODY)
|
|
adventure_id = resp.json()["id"]
|
|
_wait_for_adventure_status(user_client, adventure_id, "active")
|
|
|
|
entries = user_client.get(f"/api/adventures/{adventure_id}/entries").json()
|
|
detail = user_client.get(
|
|
f"/api/adventures/{adventure_id}/entries/{entries[0]['id']}"
|
|
).json()
|
|
choice_id = detail["choices"][0]["id"]
|
|
|
|
decision_resp = user_client.post(
|
|
f"/api/adventures/{adventure_id}/decisions", json={"choice_id": choice_id}
|
|
)
|
|
assert decision_resp.status_code == 201
|
|
assert decision_resp.json()["choice_id"] == choice_id
|
|
|
|
_wait_for_adventure_status(user_client, adventure_id, "active")
|
|
|
|
entries = user_client.get(f"/api/adventures/{adventure_id}/entries").json()
|
|
assert len(entries) == 2
|
|
second = next(e for e in entries if e["entry_index"] == 1)
|
|
assert second["status"] == "complete"
|
|
assert second["generated_from_choice_id"] == choice_id
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test 3: Adventure completes after max_entry_count entries
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_adventure_completes_at_max_entries(user_client: httpx.Client) -> None:
|
|
body = {**_DEFAULT_ADVENTURE_BODY, "max_entry_count": 2}
|
|
resp = user_client.post("/api/adventures", json=body)
|
|
adventure_id = resp.json()["id"]
|
|
_wait_for_adventure_status(user_client, adventure_id, "active")
|
|
|
|
entries = user_client.get(f"/api/adventures/{adventure_id}/entries").json()
|
|
detail = user_client.get(
|
|
f"/api/adventures/{adventure_id}/entries/{entries[0]['id']}"
|
|
).json()
|
|
choice_id = detail["choices"][0]["id"]
|
|
|
|
user_client.post(
|
|
f"/api/adventures/{adventure_id}/decisions", json={"choice_id": choice_id}
|
|
)
|
|
adventure = _wait_for_adventure_status(user_client, adventure_id, "complete")
|
|
assert adventure["status"] == "complete"
|
|
|
|
entries = user_client.get(f"/api/adventures/{adventure_id}/entries").json()
|
|
assert len(entries) == 2
|
|
|
|
final_detail = user_client.get(
|
|
f"/api/adventures/{adventure_id}/entries/{entries[1]['id']}"
|
|
).json()
|
|
assert final_detail["choices"] == []
|
|
|
|
# Decision on a complete adventure returns 409
|
|
extra = user_client.post(
|
|
f"/api/adventures/{adventure_id}/decisions", json={"choice_id": choice_id}
|
|
)
|
|
assert extra.status_code == 409
|
|
assert "not_active" in extra.json()["detail"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test 4: Double-decision on the same entry is rejected
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_cannot_make_second_decision_on_same_entry(user_client: httpx.Client) -> None:
|
|
resp = user_client.post("/api/adventures", json=_DEFAULT_ADVENTURE_BODY)
|
|
adventure_id = resp.json()["id"]
|
|
_wait_for_adventure_status(user_client, adventure_id, "active")
|
|
|
|
entries = user_client.get(f"/api/adventures/{adventure_id}/entries").json()
|
|
detail = user_client.get(
|
|
f"/api/adventures/{adventure_id}/entries/{entries[0]['id']}"
|
|
).json()
|
|
choice_id = detail["choices"][0]["id"]
|
|
other_choice_id = detail["choices"][1]["id"]
|
|
|
|
first = user_client.post(
|
|
f"/api/adventures/{adventure_id}/decisions", json={"choice_id": choice_id}
|
|
)
|
|
assert first.status_code == 201
|
|
|
|
second = user_client.post(
|
|
f"/api/adventures/{adventure_id}/decisions", json={"choice_id": other_choice_id}
|
|
)
|
|
assert second.status_code == 409
|
|
assert "decision_already_made" in second.json()["detail"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test 5: User isolation — one user cannot see or interact with another's adventure
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_user_cannot_access_another_users_adventure(
|
|
user_client: httpx.Client, second_user_client: httpx.Client
|
|
) -> None:
|
|
resp = user_client.post("/api/adventures", json=_DEFAULT_ADVENTURE_BODY)
|
|
adventure_id = resp.json()["id"]
|
|
|
|
assert second_user_client.get(f"/api/adventures/{adventure_id}").status_code == 404
|
|
assert second_user_client.get(f"/api/adventures/{adventure_id}/entries").status_code == 404
|
|
|
|
decision_resp = second_user_client.post(
|
|
f"/api/adventures/{adventure_id}/decisions",
|
|
json={"choice_id": str(uuid.uuid4())},
|
|
)
|
|
assert decision_resp.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test 6: Soft-delete removes adventure from list
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_soft_delete_hides_adventure(user_client: httpx.Client) -> None:
|
|
resp = user_client.post("/api/adventures", json=_DEFAULT_ADVENTURE_BODY)
|
|
adventure_id = resp.json()["id"]
|
|
|
|
delete_resp = user_client.delete(f"/api/adventures/{adventure_id}")
|
|
assert delete_resp.status_code == 204
|
|
|
|
# Should no longer appear in the list
|
|
adventures = user_client.get("/api/adventures").json()
|
|
assert not any(a["id"] == adventure_id for a in adventures)
|
|
|
|
# Direct GET also 404s after deletion
|
|
assert user_client.get(f"/api/adventures/{adventure_id}").status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Unit-level: LLM response parser (no Docker / network required)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _parse(text: str):
|
|
"""Local copy of the parser for isolated unit testing."""
|
|
import re
|
|
|
|
parts = text.split("\n-----\n")
|
|
if len(parts) < 3:
|
|
parts = text.split("-----\n")
|
|
if len(parts) < 3:
|
|
raise ValueError(f"LLM response has {len(parts)} section(s); expected 3")
|
|
|
|
story_text = parts[0].strip()
|
|
options_raw = parts[1].strip()
|
|
gm_notes = "\n-----\n".join(parts[2:]).strip()
|
|
|
|
choices = []
|
|
for line in options_raw.splitlines():
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
m = re.match(r"^(\d+)[.)]\s+(.+)$", line)
|
|
if m:
|
|
choices.append((m.group(1), m.group(2).strip()))
|
|
|
|
if not choices:
|
|
raise ValueError("No choices parsed from LLM response options section")
|
|
|
|
return story_text, choices, gm_notes
|
|
|
|
|
|
def test_parse_valid_three_section_response() -> None:
|
|
text = (
|
|
"Story text here.\n-----\n"
|
|
"1. Option one\n2. Option two\n3. Option three\n4. Option four\n"
|
|
"-----\nno notes"
|
|
)
|
|
story, choices, notes = _parse(text)
|
|
assert story == "Story text here."
|
|
assert len(choices) == 4
|
|
assert choices[0] == ("1", "Option one")
|
|
assert choices[3] == ("4", "Option four")
|
|
assert notes == "no notes"
|
|
|
|
|
|
def test_parse_with_parenthesis_delimiters() -> None:
|
|
text = "Story.\n-----\n1) First\n2) Second\n3) Third\n4) Fourth\n-----\nGM note."
|
|
_, choices, notes = _parse(text)
|
|
assert len(choices) == 4
|
|
assert choices[2] == ("3", "Third")
|
|
assert notes == "GM note."
|
|
|
|
|
|
def test_parse_missing_sections_raises() -> None:
|
|
with pytest.raises(ValueError, match="section"):
|
|
_parse("Only one section here, no separators at all.")
|