Spaces:
Running
Running
| 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 | |
| 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()) | |