Unraid-AppOps-Gateway-Public/backend/tests/test_mcp_hub_policy_reconciliation.py

from __future__ import annotations

from sqlmodel import Session, SQLModel, create_engine, select

from backend.app.db.models import Policy
from backend.app.services.discovery_scheduler import (
    _reconcile_mcp_hub_exact_policies,
)


def _session(tmp_path) -> Session:
    engine = create_engine(f"sqlite:///{tmp_path / 'mcp-hub-policies.db'}")
    SQLModel.metadata.create_all(engine)
    return Session(engine)


def test_reconciliation_creates_only_exact_apply_and_rollback_policies(tmp_path):
    with _session(tmp_path) as session:
        result = _reconcile_mcp_hub_exact_policies(session)
        rows = session.exec(
            select(Policy).where(Policy.target == "itworx-mcp-hub")
        ).all()

    assert result["status"] == "ready"
    assert result["created"] == 2
    assert {row.action for row in rows} == {
        "deployment.apply",
        "deployment.rollback",
    }
    assert all(row.manual == "allowed" for row in rows)
    assert all(row.voice == "blocked" for row in rows)
    assert all(
        row.provenance_json.get("createdBy") == "mcp-hub-state-reconciliation/v1"
        for row in rows
    )


def test_reconciliation_is_idempotent(tmp_path):
    with _session(tmp_path) as session:
        first = _reconcile_mcp_hub_exact_policies(session)
        second = _reconcile_mcp_hub_exact_policies(session)
        rows = session.exec(
            select(Policy).where(Policy.target == "itworx-mcp-hub")
        ).all()

    assert first["created"] == 2
    assert second["status"] == "ready"
    assert second["created"] == 0
    assert len(rows) == 2


def test_reconciliation_never_overwrites_blocked_exact_policy(tmp_path):
    with _session(tmp_path) as session:
        session.add(
            Policy(
                id="existing-block",
                action="deployment.apply",
                target="itworx-mcp-hub",
                voice="blocked",
                manual="blocked",
                risk="high",
                reason="Explicit local block.",
            )
        )
        session.commit()

        result = _reconcile_mcp_hub_exact_policies(session)
        rows = session.exec(
            select(Policy).where(Policy.target == "itworx-mcp-hub")
        ).all()

    assert result == {
        "status": "conflict",
        "action": "deployment.apply",
        "created": 0,
        "policyIds": [],
    }
    assert len(rows) == 1
    assert rows[0].manual == "blocked"

def test_reconciliation_refuses_weakened_allowed_exact_policy(tmp_path):
    with _session(tmp_path) as session:
        session.add(
            Policy(
                id="existing-weak",
                action="deployment.apply",
                target="itworx-mcp-hub",
                voice="allowed",
                manual="allowed",
                risk="medium",
                reason="Unsafe test policy.",
            )
        )
        session.commit()

        result = _reconcile_mcp_hub_exact_policies(session)
        rows = session.exec(
            select(Policy).where(Policy.target == "itworx-mcp-hub")
        ).all()

    assert result == {
        "status": "conflict",
        "action": "deployment.apply",
        "created": 0,
        "policyIds": [],
    }
    assert len(rows) == 1
    assert rows[0].voice == "allowed"
    assert rows[0].risk == "medium"

def test_reconciliation_refuses_duplicate_exact_policies(tmp_path):
    with _session(tmp_path) as session:
        for policy_id in ("duplicate-a", "duplicate-b"):
            session.add(
                Policy(
                    id=policy_id,
                    action="deployment.rollback",
                    target="itworx-mcp-hub",
                    voice="blocked",
                    manual="allowed",
                    risk="high",
                    reason="Duplicate exact policy.",
                )
            )
        session.commit()

        result = _reconcile_mcp_hub_exact_policies(session)
        rows = session.exec(
            select(Policy).where(
                (Policy.action == "deployment.rollback")
                & (Policy.target == "itworx-mcp-hub")
            )
        ).all()

    assert result == {
        "status": "conflict",
        "action": "deployment.rollback",
        "created": 0,
        "policyIds": [],
    }
    assert len(rows) == 2

def test_reconciliation_refuses_deterministic_policy_id_collision(tmp_path):
    with _session(tmp_path) as session:
        session.add(
            Policy(
                id="mcp-hub-reconciled-apply",
                action="container.restart",
                target="unrelated-container",
                voice="blocked",
                manual="allowed",
                risk="medium",
                reason="Unrelated existing policy ID.",
            )
        )
        session.commit()

        result = _reconcile_mcp_hub_exact_policies(session)

    assert result == {
        "status": "conflict",
        "action": "deployment.apply",
        "created": 0,
        "policyIds": [],
    }