| """ |
| End-to-end test: Validation -> Decision -> routing. |
| Simulates the real stakeholder scenario: |
| Rule: min_confidence 0.95 |
| Proposal confidence: 0.9 |
| Expected: FAIL -> REVIEW -> human_review required |
| """ |
| from types import SimpleNamespace |
| from uuid import uuid4 |
|
|
| from app.agents.validation import RuleValidationAgent |
| from app.agents.decision import DecisionAgent |
|
|
|
|
| validation_agent = RuleValidationAgent() |
| decision_agent = DecisionAgent() |
|
|
|
|
| def make_rule(name, configuration): |
| return SimpleNamespace( |
| id=uuid4(), |
| name=name, |
| description="test", |
| rule_type="VALIDATION", |
| configuration=configuration, |
| enabled=True, |
| ) |
|
|
|
|
| def make_proposal(confidence=0.90, evidence=True, ptype="CREATE"): |
| proposed = {"confidence": confidence} |
| if evidence: |
| proposed["evidence"] = {"source": "resume.pdf", "page": 1} |
| return SimpleNamespace( |
| id=uuid4(), |
| proposal_type=ptype, |
| proposed_changes=proposed, |
| ) |
|
|
|
|
| |
| |
| print("SCENARIO 1: min_confidence 0.95, proposal confidence 0.9") |
| print("=" * 60) |
|
|
| rule = make_rule("Minimum Confidence", {"operator": "min_confidence", "value": 0.95}) |
| proposal = make_proposal(confidence=0.9) |
|
|
| validation_results = validation_agent.validate(rules=[rule], proposals=[proposal]) |
| print(f" Validation result: {validation_results[0]['status']}") |
| assert validation_results[0]["status"] == "FAIL" |
|
|
| decision = decision_agent.decide(validation_results) |
| print(f" Decision: {decision['decision']}") |
| assert decision["decision"] == "REVIEW" |
| print(" -> Human review REQUIRED. Workflow would enter WAITING_FOR_REVIEW.") |
| print(" PASS") |
|
|
| |
| print("\nSCENARIO 2: min_confidence 0.95, proposal confidence 0.98") |
| print("=" * 60) |
|
|
| proposal2 = make_proposal(confidence=0.98) |
| validation_results2 = validation_agent.validate(rules=[rule], proposals=[proposal2]) |
| print(f" Validation result: {validation_results2[0]['status']}") |
| assert validation_results2[0]["status"] == "PASS" |
|
|
| decision2 = decision_agent.decide(validation_results2) |
| print(f" Decision: {decision2['decision']}") |
| assert decision2["decision"] == "CONTINUE" |
| print(" -> Auto-continues. Workflow would proceed to COMPLETED.") |
| print(" PASS") |
|
|
| |
| print("\nSCENARIO 3: No rules configured") |
| print("=" * 60) |
|
|
| validation_results3 = validation_agent.validate(rules=[], proposals=[proposal]) |
| print(f" Validation result: {validation_results3[0]['status']}") |
| assert validation_results3[0]["status"] == "PASS" |
|
|
| decision3 = decision_agent.decide(validation_results3) |
| print(f" Decision: {decision3['decision']}") |
| assert decision3["decision"] == "CONTINUE" |
| print(" -> Auto-continues. No rules = no review needed.") |
| print(" PASS") |
|
|
| |
| print("\nSCENARIO 4: Multiple rules, evidence rule fails") |
| print("=" * 60) |
|
|
| rules = [ |
| make_rule("Minimum Confidence", {"operator": "min_confidence", "value": 0.80}), |
| make_rule("Required Evidence", {"operator": "required_evidence"}), |
| ] |
| proposal_no_evidence = make_proposal(confidence=0.9, evidence=False) |
|
|
| validation_results4 = validation_agent.validate(rules=rules, proposals=[proposal_no_evidence]) |
| statuses4 = [r["status"] for r in validation_results4] |
| print(f" Validation statuses: {statuses4}") |
| assert "PASS" in statuses4 and "FAIL" in statuses4 |
|
|
| decision4 = decision_agent.decide(validation_results4) |
| print(f" Decision: {decision4['decision']}") |
| assert decision4["decision"] == "REVIEW" |
| print(" -> Any FAIL triggers REVIEW regardless of other PASS results.") |
| print(" PASS") |
|
|
| |
| print("\nSCENARIO 5: Unknown operator -> WARNING -> REVIEW") |
| print("=" * 60) |
|
|
| rules5 = [make_rule("Custom", {"operator": "custom_unknown_op"})] |
| validation_results5 = validation_agent.validate(rules=rules5, proposals=[proposal]) |
| print(f" Validation result: {validation_results5[0]['status']}") |
| assert validation_results5[0]["status"] == "WARNING" |
|
|
| decision5 = decision_agent.decide(validation_results5) |
| print(f" Decision: {decision5['decision']}") |
| assert decision5["decision"] == "REVIEW" |
| print(" -> WARNING also triggers REVIEW.") |
| print(" PASS") |
|
|
| print("\n===================================") |
| print("ALL 5 E2E SCENARIOS PASSED") |
| print("===================================") |
|
|