| 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: |
| |
| |
| |
|
|
| from app.models.workspace import Workspace |
|
|
| workspace = db.query(Workspace).first() |
|
|
| if workspace is None: |
| raise RuntimeError("No workspace found.") |
|
|
| print("WORKSPACE:", workspace.id) |
|
|
| |
| |
| |
| |
| |
| |
| |
|
|
| 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) |
|
|
| |
| |
| |
|
|
| 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 |
|
|
| |
| |
| |
|
|
| 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") |
|
|
| |
| |
| |
|
|
| 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") |
|
|
| |
| |
| |
|
|
| 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") |
|
|
| |
| |
| |
|
|
| 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") |
|
|
| |
| |
| |
|
|
| 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") |
|
|
| |
| |
| |
|
|
| 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") |
|
|
| |
| |
| |
|
|
| 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") |
|
|
| |
| |
| |
|
|
| 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") |
|
|
| |
| |
| |
|
|
| 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") |
|
|
| |
| |
| |
|
|
| 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") |
|
|
| |
| |
| |
|
|
| for rule in created_rules: |
| db.delete(rule) |
|
|
| db.commit() |
|
|
| |
| |
| |
|
|
| 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() |
|
|
