test_wo23_allocator_dcp_quarantine.py

"""WO-23 (#991) — DCP proposal intake never accepts the legacy allocator's shape.

No DCP source is changed here (that is CC-1 territory); this test PROVES the
existing intake contract against a fixture of the legacy allocator's output
(`miz-oki-command-center-ui/lib/decisioning/omnichannel.ts::AllocationResult`)
and grep-guards the UI tree so nothing outside the quarantine posts that shape
into the governed pathway. Companion static guard on the UI side:
`miz-oki-command-center-ui/app/api/omnichannel/allocation/quarantine.test.ts`.
"""
from __future__ import annotations

import importlib.util
import os
import re
import sys
import unittest
from pathlib import Path

ROOT = Path(__file__).resolve().parents[3]
DCP_MAIN = ROOT / "services" / "service-decision-control-plane" / "main.py"
UI_ROOT = ROOT / "miz-oki-command-center-ui"
CONTRACTS = ROOT / "contracts"

# The audit's counterexample output, as the allocator emits it (WO-22 shape,
# labelled). Keys are the allocator's, NOT ProposeRequest's.
ALLOCATOR_OUTPUT = {
    "kernel": "default",
    "objective": "maximize_incremental_roas",
    "mode": "simulation",
    "claimLabel": "illustrative scenario",
    "totalBudget": 68000,
    "recommendations": [],
    "portfolio": {
        "currentBudget": 68000,
        "recommendedBudget": 53828.12,
        "allocatedBudget": 53828.12,
        "unallocatedBudget": 14171.88,
        "projectedRevenue": 281835.51,
        "currentConversionRate": None,
        "channels": [],
    },
    "simulationResults": {"scenarios": []},
    "governance": {"requiresManualReview": False, "guardrailsTriggered": []},
}


def _load_dcp():
    os.environ.setdefault("MIZOKI_STORE", "memory")
    os.environ.setdefault("MIZOKI_AUTH_DISABLED", "local-dev-only")
    os.environ.pop("K_SERVICE", None)
    for p in (str(CONTRACTS), str(DCP_MAIN.parent)):
        if p not in sys.path:
            sys.path.insert(0, p)
    name = "wo23_dcp_main"
    if name in sys.modules:
        return sys.modules[name]
    spec = importlib.util.spec_from_file_location(name, DCP_MAIN)
    mod = importlib.util.module_from_spec(spec)
    sys.modules[name] = mod
    spec.loader.exec_module(mod)  # type: ignore[union-attr]
    return mod


class AllocatorNeverEntersDcp(unittest.TestCase):
    def test_wo23_propose_request_rejects_the_allocator_shape(self) -> None:
        try:
            dcp = _load_dcp()
        except Exception as exc:  # pragma: no cover — environment, not contract
            self.skipTest(f"DCP main not importable here: {exc!r}")
        from pydantic import ValidationError

        with self.assertRaises(ValidationError) as ctx:
            dcp.ProposeRequest(**ALLOCATOR_OUTPUT)
        missing = {err["loc"][0] for err in ctx.exception.errors() if err["type"] == "missing"}
        # The governed intake needs a passport-bound, tenant-scoped proposal;
        # the allocator produces none of it.
        self.assertTrue({"tenant_id", "domain", "proposal", "chosen_path_id", "passport_id"} <= missing, missing)

    def test_wo23_allocator_keys_are_not_propose_request_fields(self) -> None:
        try:
            dcp = _load_dcp()
        except Exception as exc:  # pragma: no cover
            self.skipTest(f"DCP main not importable here: {exc!r}")
        fields = set(dcp.ProposeRequest.model_fields)
        overlap = fields & set(ALLOCATOR_OUTPUT)
        self.assertEqual(overlap, set(), f"allocator keys that DCP would read: {sorted(overlap)}")

    def test_wo23_no_ui_file_outside_the_quarantine_posts_allocator_output_to_propose(self) -> None:
        """Grep evidence: a file that imports the allocator never reaches the
        DCP propose adapter, and vice versa (tests excluded)."""
        allocator_re = re.compile(r"@/lib/decisioning/(?:live-)?omnichannel")
        propose_re = re.compile(r"\bpropose\s*\(|/api/v1/propose\b")
        skip = {"node_modules", ".next", "archive", "archive-ui", "backups", "backend", "Cell1toCell10"}
        offenders = []
        for path in UI_ROOT.rglob("*.ts*"):
            if any(part in skip for part in path.parts):
                continue
            if path.name.endswith((".test.ts", ".test.tsx")):
                continue
            rel = path.relative_to(UI_ROOT).as_posix()
            if rel.startswith("lib/decisioning/"):
                continue
            src = path.read_text(encoding="utf-8", errors="ignore")
            if allocator_re.search(src) and propose_re.search(src):
                offenders.append(rel)
        self.assertEqual(offenders, [], offenders)


if __name__ == "__main__":
    unittest.main()