File size: 5,737 Bytes
94f31ec e382248 94f31ec e382248 94f31ec e382248 94f31ec e382248 94f31ec e382248 94f31ec e382248 94f31ec e382248 94f31ec e382248 94f31ec e382248 94f31ec e382248 94f31ec e382248 94f31ec e382248 94f31ec e382248 94f31ec e382248 94f31ec e382248 94f31ec e382248 94f31ec e382248 94f31ec e382248 94f31ec e382248 94f31ec e382248 94f31ec e382248 94f31ec e382248 94f31ec e382248 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 | """
Integration tests for the PRD generation flow.
Tests the full lifecycle: start → chat → doc → status → sections.
Uses real NVIDIA LLM API calls.
**Why a fixture rather than test ordering.** These tests previously passed the
session id between themselves via `pytest.session_id`, falling back to an unused
random UUID when it was absent. A failed `/prd/start` therefore left the
remaining tests asserting against a session that had never existed - and they
passed, because the endpoints answer 200 for unknown sessions. The session is now
created by a module-scoped fixture, so a failure there ERRORs every dependent
test instead of quietly weakening it.
"""
import pytest
from tests.integration.conftest import _require_env, require_ok
@pytest.fixture(scope="module")
def prd_session(live_client) -> dict:
"""Start one PRD session shared by every test in this module."""
_require_env("NVIDIA_API_KEY")
response = live_client.post(
"/prd/start",
json={
"description": (
"A task management app for remote teams with real-time collaboration, "
"Kanban boards, and time tracking."
),
"user_id": 1,
},
)
require_ok(response, "POST /prd/start")
return response.json()
class TestPRDFlow:
"""Test the complete PRD generation lifecycle with live LLM."""
def test_prd_start_session(self, prd_session):
"""POST /prd/start creates a new PRD session with real LLM evaluation."""
assert prd_session["session_id"]
assert len(prd_session["message"]) > 10, (
f"expected a real LLM reply, got {prd_session['message']!r}"
)
def test_prd_chat_response(self, live_client, prd_session):
"""POST /prd/chat gets a real LLM response."""
_require_env("NVIDIA_API_KEY")
response = live_client.post(
"/prd/chat",
json={
"session_id": prd_session["session_id"],
"message": (
"The app should support multiple projects, assign tasks to team "
"members, and send email notifications for deadlines."
),
},
)
require_ok(response, "POST /prd/chat")
data = response.json()
assert "phase" in data
assert "needs_more" in data
assert len(data["agent_response"]) > 20, (
f"expected real LLM content, got {data['agent_response']!r}"
)
def test_prd_status(self, live_client, prd_session):
"""GET /prd/status/{session_id} returns session status."""
session_id = prd_session["session_id"]
response = live_client.get(f"/prd/status/{session_id}")
require_ok(response, f"GET /prd/status/{session_id}")
data = response.json()
assert data["phase"] != "not_found", (
"the session created by this module is not recognised by /prd/status"
)
assert "requirements_status" in data
assert "follow_up_count" in data
def test_prd_doc_retrieval(self, live_client, prd_session):
"""GET /prd/doc/{session_id} retrieves the PRD document."""
session_id = prd_session["session_id"]
response = live_client.get(f"/prd/doc/{session_id}")
require_ok(response, f"GET /prd/doc/{session_id}")
data = response.json()
assert data["session_id"] == session_id
assert "requirements_status" in data
def test_prd_sections(self, live_client, prd_session):
"""GET /prd/sections/{session_id} returns structured PRD sections."""
session_id = prd_session["session_id"]
response = live_client.get(f"/prd/sections/{session_id}")
require_ok(response, f"GET /prd/sections/{session_id}")
data = response.json()
assert data["session_id"] == session_id
assert "full_text" in data
assert "key_features" in data
def test_prd_missing_session(self, live_client):
"""GET /prd/status/{missing_id} handles missing session gracefully."""
response = live_client.get("/prd/status/00000000-0000-0000-0000-000000000000")
assert response.status_code == 200, response.text
data = response.json()
assert data["phase"] == "not_found"
class TestPRDDownload:
"""Test PRD download endpoints."""
def test_download_markdown(self, live_client, prd_session):
"""GET /prd/download/{session_id}?format=markdown returns markdown.
The endpoint 404s until `generated_prd` is populated, and this module's
conversation may not have reached generation. Rather than accept
`status_code in (200, 404)` - which passes whatever happens - the
expectation is derived from `/prd/sections`, whose `full_text` is empty
exactly when the PRD has not been generated. Both branches are checked.
"""
session_id = prd_session["session_id"]
sections = live_client.get(f"/prd/sections/{session_id}")
require_ok(sections, f"GET /prd/sections/{session_id}")
prd_generated = bool(sections.json()["full_text"])
response = live_client.get(
f"/prd/download/{session_id}",
params={"format": "markdown"},
)
if prd_generated:
assert response.status_code == 200, response.text
assert response.json()["markdown"], "download returned an empty PRD"
else:
assert response.status_code == 404, (
f"PRD was not generated, so download must 404; got "
f"{response.status_code}: {response.text[:300]}"
)
assert response.json()["error"] == "PRD not found or not generated"
|