Spaces:
Running
Running
File size: 2,069 Bytes
0cd3dc5 | 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 | import sys
import os
import pytest
import asyncio
import json
# Add project root to path
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from core.graph_workflow import app as graph_app
@pytest.mark.asyncio
async def test_end_to_end_workflow():
"""Test full LangGraph execution for a standard query"""
query = "Explain Article 14 and its relation to equality"
initial_state = {"query": query, "retry_count": 0}
print(f"\n🚀 Starting E2E Workflow Test: '{query}'")
try:
result = await graph_app.ainvoke(initial_state)
# 1. Check if execution completed
assert result is not None, "Workflow returned None"
# 2. Check Trace
trace = result.get("trace", [])
print(f"Workflow Trace: {trace}")
assert "Classify" in trace
assert "GraphPlan" in trace
assert "Retrieve" in trace
assert "Reason" in trace
assert "Validate" in trace
# 3. Check State Outputs
assert result.get("classification"), "Missing classification"
assert result.get("graph_plan"), "Missing graph plan"
assert result.get("qdrant_scope"), "Missing Qdrant scope (Tree Router failed)"
assert len(result.get("retrieved_chunks", [])) > 0, "No chunks retrieved"
# 4. Check Answer
draft_answer = result.get("draft_answer", {})
assert draft_answer.get("answer"), "No answer generated"
assert draft_answer.get("constitutional_status"), "No status generated"
# 5. Check Critique
critique = result.get("critique", {})
print(f"Quality Grade: {critique.get('quality_grade')}")
assert critique.get("final_confidence") is not None
print("\n✅ End-to-End Workflow: SUCCESS")
print(f"Final Answer Preview: {draft_answer.get('answer')[:100]}...")
except Exception as e:
pytest.fail(f"Workflow Execution Failed: {e}")
if __name__ == "__main__":
asyncio.run(test_end_to_end_workflow())
|