test: verify validation decision rules

backend/test_validation_proof.py

@@ -0,0 +1,435 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+from app.database.session import SessionLocal
+from app.models.rule import Rule, RuleType
+from app.repositories.rule_repository import RuleRepository
+from app.agents.validation import RuleValidationAgent
+from app.agents.decision import DecisionAgent
+
+
+print("=" * 60)
+print("DOCWEAVE REAL VALIDATION PROOF")
+print("=" * 60)
+
+db = SessionLocal()
+
+created_rules = []
+
+try:
+    # --------------------------------------------------------------
+    # Find a real workspace
+    # --------------------------------------------------------------
+
+    from app.models.workspace import Workspace
+
+    workspace = db.query(Workspace).first()
+
+    if workspace is None:
+        raise RuntimeError("No workspace found.")
+
+    print("WORKSPACE:", workspace.id)
+
+    # --------------------------------------------------------------
+    # Synthetic proposals
+    #
+    # These are lightweight proposal-shaped objects because the
+    # RuleValidationAgent only needs id, proposal_type and
+    # proposed_changes.
+    # --------------------------------------------------------------
+
+    passing_proposal = SimpleNamespace(
+        id="proposal-pass",
+        proposal_type="CREATE",
+        proposed_changes={
+            "type": "ENTITY",
+            "title": "Congenital Heart Valve Disease",
+            "value": "Congenital Heart Valve Disease",
+            "confidence": 0.95,
+            "evidence": [
+                {
+                    "document_version_id": "synthetic-version",
+                    "page": 1,
+                    "text": "Evidence supporting the entity.",
+                }
+            ],
+        },
+    )
+
+    failing_proposal = SimpleNamespace(
+        id="proposal-fail",
+        proposal_type="CREATE",
+        proposed_changes={
+            "type": "ENTITY",
+            "title": "Low Confidence Entity",
+            "value": "Low Confidence Entity",
+            "confidence": 0.40,
+        },
+    )
+
+    update_proposal = SimpleNamespace(
+        id="proposal-update",
+        proposal_type="UPDATE",
+        proposed_changes={
+            "type": "ENTITY",
+            "title": "Updated Entity",
+            "value": "Updated Entity",
+            "confidence": 0.95,
+            "evidence": [
+                {
+                    "document_version_id": "synthetic-version",
+                    "page": 2,
+                    "text": "Update evidence.",
+                }
+            ],
+        },
+    )
+
+    agent = RuleValidationAgent()
+    decision_agent = DecisionAgent()
+    repository = RuleRepository(db)
+
+    # --------------------------------------------------------------
+    # Helper
+    # --------------------------------------------------------------
+
+    def make_rule(
+        name,
+        operator,
+        configuration,
+        enabled=True,
+    ):
+        rule = Rule(
+            workspace_id=workspace.id,
+            name=name,
+            description="Validation proof rule",
+            rule_type=RuleType.VALIDATION,
+            configuration={
+                "operator": operator,
+                **configuration,
+            },
+            enabled=enabled,
+        )
+
+        db.add(rule)
+        db.flush()
+
+        created_rules.append(rule)
+
+        return rule
+
+    # ==============================================================
+    # 1. PASS
+    # ==============================================================
+
+    print()
+    print("1. TESTING PASS")
+
+    pass_rule = make_rule(
+        "TEST Required Evidence PASS",
+        "required_evidence",
+        {},
+    )
+
+    results = agent.validate(
+        rules=[pass_rule],
+        proposals=[passing_proposal],
+    )
+
+    print("RESULT:", results)
+
+    assert len(results) == 1
+    assert results[0]["status"] == "PASS"
+    assert "proposal-pass" in results[0]["proposal_ids"]
+
+    print("PASS TEST: OK")
+
+    # ==============================================================
+    # 2. FAIL - missing evidence
+    # ==============================================================
+
+    print()
+    print("2. TESTING FAIL")
+
+    evidence_fail_rule = make_rule(
+        "TEST Required Evidence FAIL",
+        "required_evidence",
+        {},
+    )
+
+    results = agent.validate(
+        rules=[evidence_fail_rule],
+        proposals=[failing_proposal],
+    )
+
+    print("RESULT:", results)
+
+    assert len(results) == 1
+    assert results[0]["status"] == "FAIL"
+    assert results[0]["severity"] == "HIGH"
+    assert "proposal-fail" in results[0]["proposal_ids"]
+
+    decision = decision_agent.decide(results)
+
+    print("DECISION:", decision)
+
+    assert decision["decision"] == "REVIEW"
+    assert decision["failure_count"] == 1
+    assert "proposal-fail" in decision["affected_proposal_ids"]
+
+    print("FAIL TEST: OK")
+
+    # ==============================================================
+    # 3. WARNING - malformed min_confidence
+    # ==============================================================
+
+    print()
+    print("3. TESTING WARNING")
+
+    warning_rule = make_rule(
+        "TEST Malformed Confidence Rule",
+        "min_confidence",
+        {
+            "value": "not-a-number",
+        },
+    )
+
+    results = agent.validate(
+        rules=[warning_rule],
+        proposals=[passing_proposal],
+    )
+
+    print("RESULT:", results)
+
+    assert len(results) == 1
+    assert results[0]["status"] == "WARNING"
+    assert results[0]["severity"] == "HIGH"
+
+    decision = decision_agent.decide(results)
+
+    print("DECISION:", decision)
+
+    assert decision["decision"] == "REVIEW"
+    assert decision["warning_count"] == 1
+
+    print("WARNING TEST: OK")
+
+    # ==============================================================
+    # 4. MIN CONFIDENCE FAIL
+    # ==============================================================
+
+    print()
+    print("4. TESTING MIN CONFIDENCE FAIL")
+
+    confidence_rule = make_rule(
+        "TEST Minimum Confidence",
+        "min_confidence",
+        {
+            "value": 0.80,
+        },
+    )
+
+    results = agent.validate(
+        rules=[confidence_rule],
+        proposals=[failing_proposal],
+    )
+
+    print("RESULT:", results)
+
+    assert len(results) == 1
+    assert results[0]["status"] == "FAIL"
+    assert results[0]["operator"] == "min_confidence"
+    assert "proposal-fail" in results[0]["proposal_ids"]
+
+    print("MIN CONFIDENCE TEST: OK")
+
+    # ==============================================================
+    # 5. MIN CONFIDENCE PASS
+    # ==============================================================
+
+    print()
+    print("5. TESTING MIN CONFIDENCE PASS")
+
+    results = agent.validate(
+        rules=[confidence_rule],
+        proposals=[passing_proposal],
+    )
+
+    print("RESULT:", results)
+
+    assert len(results) == 1
+    assert results[0]["status"] == "PASS"
+
+    print("MIN CONFIDENCE PASS TEST: OK")
+
+    # ==============================================================
+    # 6. ALLOWED PROPOSAL TYPES FAIL
+    # ==============================================================
+
+    print()
+    print("6. TESTING ALLOWED PROPOSAL TYPES FAIL")
+
+    type_rule = make_rule(
+        "TEST Allowed Proposal Types",
+        "allowed_proposal_types",
+        {
+            "values": ["UPDATE"],
+        },
+    )
+
+    results = agent.validate(
+        rules=[type_rule],
+        proposals=[passing_proposal],
+    )
+
+    print("RESULT:", results)
+
+    assert len(results) == 1
+    assert results[0]["status"] == "FAIL"
+    assert results[0]["operator"] == "allowed_proposal_types"
+
+    print("PROPOSAL TYPE FAIL TEST: OK")
+
+    # ==============================================================
+    # 7. ALLOWED PROPOSAL TYPES PASS
+    # ==============================================================
+
+    print()
+    print("7. TESTING ALLOWED PROPOSAL TYPES PASS")
+
+    results = agent.validate(
+        rules=[type_rule],
+        proposals=[update_proposal],
+    )
+
+    print("RESULT:", results)
+
+    assert len(results) == 1
+    assert results[0]["status"] == "PASS"
+
+    print("PROPOSAL TYPE PASS TEST: OK")
+
+    # ==============================================================
+    # 8. UNKNOWN OPERATOR → WARNING
+    # ==============================================================
+
+    print()
+    print("8. TESTING UNKNOWN OPERATOR")
+
+    unknown_rule = make_rule(
+        "TEST Unknown Operator",
+        "does_not_exist",
+        {},
+    )
+
+    results = agent.validate(
+        rules=[unknown_rule],
+        proposals=[passing_proposal],
+    )
+
+    print("RESULT:", results)
+
+    assert len(results) == 1
+    assert results[0]["status"] == "WARNING"
+    assert results[0]["operator"] == "does_not_exist"
+
+    print("UNKNOWN OPERATOR TEST: OK")
+
+    # ==============================================================
+    # 9. DISABLED RULE MUST BE IGNORED
+    # ==============================================================
+
+    print()
+    print("9. TESTING DISABLED RULE")
+
+    disabled_rule = make_rule(
+        "TEST Disabled Rule",
+        "required_evidence",
+        {},
+        enabled=False,
+    )
+
+    db.commit()
+
+    enabled_rules = repository.list_enabled(
+        workspace.id
+    )
+
+    enabled_rule_ids = {
+        str(rule.id)
+        for rule in enabled_rules
+    }
+
+    print(
+        "DISABLED RULE:",
+        disabled_rule.id,
+    )
+
+    print(
+        "ENABLED RULE COUNT:",
+        len(enabled_rules),
+    )
+
+    assert str(disabled_rule.id) not in enabled_rule_ids
+
+    print("DISABLED RULE TEST: OK")
+
+    # ==============================================================
+    # 10. NO RULES → PASS
+    # ==============================================================
+
+    print()
+    print("10. TESTING NO ENABLED RULES")
+
+    results = agent.validate(
+        rules=[],
+        proposals=[passing_proposal],
+    )
+
+    print("RESULT:", results)
+
+    assert len(results) == 1
+    assert results[0]["status"] == "PASS"
+    assert results[0]["rule_name"] == "No enabled rules"
+
+    decision = decision_agent.decide(results)
+
+    print("DECISION:", decision)
+
+    assert decision["decision"] == "CONTINUE"
+
+    print("NO RULES TEST: OK")
+
+    # ==============================================================
+    # FINAL CLEANUP
+    # ==============================================================
+
+    for rule in created_rules:
+        db.delete(rule)
+
+    db.commit()
+
+    # ==============================================================
+    # SUCCESS
+    # ==============================================================
+
+    print()
+    print("=" * 60)
+    print("PASS: REAL VALIDATION ENGINE VERIFIED")
+    print("=" * 60)
+    print()
+    print("PASS RULES:              OK")
+    print("FAIL RULES:              OK")
+    print("WARNING RULES:           OK")
+    print("MIN CONFIDENCE:          OK")
+    print("PROPOSAL TYPE RULES:     OK")
+    print("UNKNOWN OPERATOR:        OK")
+    print("DISABLED RULE:           OK")
+    print("NO ENABLED RULES:        OK")
+    print("DECISION ROUTING:        OK")
+    print()
+
+finally:
+    db.rollback()
+    db.close()