backend/test_validation_proof.py

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()