python/tests/integration/test_lean_model_alignment.py

# python/tests/integration/test_lean_model_alignment.py
# Automated Verification: Ensures actual backend states and permissions align with Lean 4 theorems.

import os
import re


def get_lean_file_content():
    """Reads the Lean 4 specification file AuditParity.lean."""
    # Handle paths for both Host and Docker environments
    paths = [
        "../lean_proofs/LeanProofs/AuditParity.lean",
        "lean_proofs/LeanProofs/AuditParity.lean",
        "/app/lean_proofs/LeanProofs/AuditParity.lean"
    ]
    for p in paths:
        if os.path.exists(p):
            with open(p, encoding="utf-8") as f:
                return f.read()
    return None


def test_lean_rbac_roles_alignment():
    """
    Verifies that the roles defined in Lean 4 (Role inductive type)
    match the core transactional roles.
    """
    content = get_lean_file_content()
    assert content is not None, "Lean 4 specification file not found."

    # Parse Role enum from Lean file
    role_block_match = re.search(r"inductive Role\s*(.*?)(?=\n\n|\ninductive|\ndef)", content, re.DOTALL)
    assert role_block_match is not None, "Could not locate inductive Role block in Lean file."

    lean_roles = set(re.findall(r"\|\s*([a-zA-Z0-9_]+)", role_block_match.group(1)))

    # We allow the Lean proof model to be a simplified representative subset of system roles.
    # But it MUST contain the core transactional human/agent roles.
    core_roles = {"system_admin", "manager", "sales", "marketing", "employee", "ai_agent"}

    for r in core_roles:
        assert r in lean_roles, f"Core role '{r}' missing from Lean 4 specifications!"


def test_lean_workflow_states_alignment():
    """
    Verifies that WorkflowState count and state mapping in Lean 4
    matches the transactional approval stages in the backend database schemas.
    """
    content = get_lean_file_content()
    assert content is not None, "Lean 4 specification file not found."

    # Parse WorkflowState from Lean file
    state_block_match = re.search(r"inductive WorkflowState\s*(.*?)(?=\n\n|\ninductive|\ndef)", content, re.DOTALL)
    assert state_block_match is not None, "Could not locate inductive WorkflowState block in Lean file."

    lean_states = set(re.findall(r"\|\s*([a-zA-Z0-9_]+)", state_block_match.group(1)))

    # Core workflow states required for business audit
    expected_states = {"empty", "pending_approval", "approved", "published"}
    assert lean_states == expected_states, f"Lean WorkflowState {lean_states} mismatch with expected business stages!"


def test_lean_dual_judge_short_circuit_logic():
    """
    Ensures that the short-circuit logic defined in Lean's dual_judge
    is strictly matching the conceptual design (data check overrides visual check).
    """
    content = get_lean_file_content()
    assert content is not None, "Lean 4 specification file not found."

    # Find the def dual_judge block in Lean
    judge_match = re.search(r"def dual_judge.*?:=\s*(.*?)\n", content)
    assert judge_match is not None, "Could not locate dual_judge definition in Lean."

    expression = judge_match.group(1).strip()
    # Ensure it implements data_check first (short-circuiting logic)
    assert expression.startswith("data_check"), f"Lean dual_judge does not prioritize data_check! Expression: {expression}"


def test_lean_agent_topology_nodes_alignment():
    """
    Verifies that the Agent nodes defined in Lean (AgentNode)
    match the core transactional Bots in the system.
    """
    content = get_lean_file_content()
    assert content is not None, "Lean 4 specification file not found."

    # Parse AgentNode enum from Lean file
    agent_block_match = re.search(r"inductive AgentNode\s*(.*?)(?=\n\n|\ninductive|\ndef)", content, re.DOTALL)
    assert agent_block_match is not None, "Could not locate inductive AgentNode block in Lean file."

    lean_nodes = set(re.findall(r"\|\s*([a-zA-Z0-9_]+)", agent_block_match.group(1)))

    expected_agents = {"supervisor", "devbot", "librarian", "marketbot"}
    assert lean_nodes == expected_agents, f"Lean AgentNodes {lean_nodes} mismatch with expected multi-agent roles!"


def test_python_rag_threshold_alignment():
    """
    Ensures that the default RAG similarity threshold in python search strategy
    is hardened to 0.30 as defined in the system specifications and Lean 4 model context.
    """
    from src.server.services.search.base_search_strategy import SIMILARITY_THRESHOLD
    assert SIMILARITY_THRESHOLD == 0.30, f"Hardened RAG SIMILARITY_THRESHOLD should be 0.30, found {SIMILARITY_THRESHOLD}"


def test_python_budget_limit_default_alignment():
    """
    Ensures that the default budget limit parameter matches system spec defaults.
    """
    # Verify the fallback budget limit is 10.0 USD
    # Reading lines in budget_guard.py to verify fallback assignment budget_limit = 10.0
    import inspect

    from src.server.middleware.budget_guard import BudgetGuardMiddleware
    source = inspect.getsource(BudgetGuardMiddleware)
    assert "budget_limit = 10.0" in source, "Default budget limit fallback in middleware should be 10.0 USD"