DocWeave / backend /test_e2e_validation_decision.py
shak3008's picture
feat: complete end-to-end platform with all standout behaviors
fcacf10
Raw
History Blame Contribute Delete
4.44 kB
"""
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,
)
# === SCENARIO 1: Stakeholder scenario ===
# Rule: min_confidence 0.95, Proposal confidence 0.9 -> FAIL -> REVIEW
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")
# === SCENARIO 2: Confidence passes ===
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")
# === SCENARIO 3: No rules -> auto-passes ===
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")
# === SCENARIO 4: Multiple rules, one fails ===
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")
# === SCENARIO 5: Warning-only triggers review ===
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("===================================")