feat(#89): IMP-89 89-a u1~u5 Layer A render path activation (B4→mapper source-of-truth switch, default-OFF flag)
PHASE_Z_B4_MAPPER_SOURCE env flag (default OFF) switches slot_payload source-of-truth from legacy mapper-only / V4 rank-1 to B4 PlacementPlan .selected_template_id at the single switch site in the runtime loop. OFF preserves final.html SHA byte-equivalence (u4 parity guard, mdx 01-05). ON requires Layer A render-active path; BLOCKED exits on B4 no-cover and on B4-selected FitError (IMP-87 honesty gate pattern — NO silent fallback). Distinct from PHASE_Z_B4_GATEKEEPER (mismatch render-skip). Units (1 commit = 1 axis per Stage 1 scope_lock): u1 — _b4_mapper_source_enabled() flag reader (default OFF) u2 — _select_mapper_template_id() selector wired at the switch site u3 — _b4_mapper_source_blocked_exit() for b4_no_cover / b4_selected_fit_error u4 — render SHA parity regression (tests/regression/ baseline mdx 01-05) u5 — slot_payload byte-equivalence (matches_mapper=True axis, mdx 01-05) Targeted 89-a suite 63 PASS; Phase Z regression 323 PASS; IMP-87 mirror 20 PASS. Demo activation via .env only (no vite.config hardcoding). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
157
tests/phase_z2/test_b4_mapper_source_blocked.py
Normal file
157
tests/phase_z2/test_b4_mapper_source_blocked.py
Normal file
@@ -0,0 +1,157 @@
|
||||
"""IMP-89 89-a u3 — BLOCKED exit unit tests for Layer A render path.
|
||||
|
||||
Stage 2 plan (u3): when PHASE_Z_B4_MAPPER_SOURCE=ON and the Layer A render
|
||||
path cannot resolve a covering frame, the runtime MUST sys.exit(1) instead of
|
||||
silently degrading to adapter_needed or to the legacy V4 rank-1 mapper input.
|
||||
|
||||
Locked semantics (Stage 1 Q2 lock; IMP-87 honesty gate pattern):
|
||||
flag OFF → legacy adapter_needed path
|
||||
(silent fallback preserved)
|
||||
flag ON + B4 no-cover → BLOCKED (sys.exit 1)
|
||||
flag ON + FitError on B4-selected → BLOCKED (sys.exit 1)
|
||||
flag ON + matches_mapper + FitError → BLOCKED (explicit no-silent
|
||||
fallback even when V4 rank-1
|
||||
equals B4 pick)
|
||||
|
||||
These tests target the `_b4_mapper_source_blocked_exit()` helper directly
|
||||
plus contract-level assertions of its stderr output. The runtime call-sites
|
||||
inside `run_phase_z2_mvp1` are guarded by `_b4_mapper_source_enabled()`
|
||||
checks; u3 changes ZERO behavior under the default-OFF path.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from src.phase_z2_pipeline import (
|
||||
_b4_mapper_source_blocked_exit,
|
||||
_b4_mapper_source_enabled,
|
||||
)
|
||||
|
||||
FLAG = "PHASE_Z_B4_MAPPER_SOURCE"
|
||||
|
||||
|
||||
def test_blocked_exit_no_cover_exits_with_code_1(
|
||||
capsys: pytest.CaptureFixture[str],
|
||||
) -> None:
|
||||
"""b4_no_cover reason → SystemExit(1), no silent fallback."""
|
||||
with pytest.raises(SystemExit) as exc:
|
||||
_b4_mapper_source_blocked_exit(
|
||||
"b4_no_cover",
|
||||
position="top",
|
||||
context={
|
||||
"unit": "source_section_ids=['01-1'] merge_type=raw",
|
||||
"v4_rank1": "F13",
|
||||
"b4_pick": None,
|
||||
},
|
||||
)
|
||||
assert exc.value.code == 1
|
||||
|
||||
|
||||
def test_blocked_exit_fit_error_exits_with_code_1(
|
||||
capsys: pytest.CaptureFixture[str],
|
||||
) -> None:
|
||||
"""b4_selected_fit_error reason → SystemExit(1)."""
|
||||
with pytest.raises(SystemExit) as exc:
|
||||
_b4_mapper_source_blocked_exit(
|
||||
"b4_selected_fit_error",
|
||||
position="bottom_l",
|
||||
context={
|
||||
"template": "F29 (B4 selected)",
|
||||
"unit": "source_section_ids=['02-2']",
|
||||
"v4_rank1": "F13",
|
||||
"fit_error": "slot 'title' missing",
|
||||
},
|
||||
)
|
||||
assert exc.value.code == 1
|
||||
|
||||
|
||||
def test_blocked_exit_stderr_carries_reason_and_position(
|
||||
capsys: pytest.CaptureFixture[str],
|
||||
) -> None:
|
||||
"""Header line surfaces the locked reason enum + zone position."""
|
||||
with pytest.raises(SystemExit):
|
||||
_b4_mapper_source_blocked_exit(
|
||||
"b4_no_cover",
|
||||
position="bottom_r",
|
||||
context={"v4_rank1": "F13"},
|
||||
)
|
||||
err = capsys.readouterr().err
|
||||
assert "[Phase Z-2 IMP-89 89-a u3] BLOCKED" in err
|
||||
assert "b4_no_cover" in err
|
||||
assert "zone--bottom_r" in err
|
||||
|
||||
|
||||
def test_blocked_exit_stderr_carries_honesty_policy_line(
|
||||
capsys: pytest.CaptureFixture[str],
|
||||
) -> None:
|
||||
"""Policy banner names PHASE_Z_B4_MAPPER_SOURCE + IMP-87 honesty pattern."""
|
||||
with pytest.raises(SystemExit):
|
||||
_b4_mapper_source_blocked_exit(
|
||||
"b4_selected_fit_error",
|
||||
position="top",
|
||||
context={"fit_error": "x"},
|
||||
)
|
||||
err = capsys.readouterr().err
|
||||
assert "PHASE_Z_B4_MAPPER_SOURCE=ON" in err
|
||||
assert "NO silent fallback" in err
|
||||
assert "IMP-87 honesty gate pattern" in err
|
||||
|
||||
|
||||
def test_blocked_exit_stderr_carries_all_context_fields(
|
||||
capsys: pytest.CaptureFixture[str],
|
||||
) -> None:
|
||||
"""Each context dict entry surfaces on its own stderr line."""
|
||||
with pytest.raises(SystemExit):
|
||||
_b4_mapper_source_blocked_exit(
|
||||
"b4_selected_fit_error",
|
||||
position="top",
|
||||
context={
|
||||
"template": "F29 (B4 selected)",
|
||||
"unit": "source_section_ids=['02-2']",
|
||||
"v4_rank1": "F13",
|
||||
"fit_error": "slot 'title' missing",
|
||||
},
|
||||
)
|
||||
err = capsys.readouterr().err
|
||||
assert "template" in err
|
||||
assert "F29 (B4 selected)" in err
|
||||
assert "unit" in err
|
||||
assert "source_section_ids=['02-2']" in err
|
||||
assert "v4_rank1" in err
|
||||
assert "F13" in err
|
||||
assert "fit_error" in err
|
||||
assert "slot 'title' missing" in err
|
||||
|
||||
|
||||
def test_blocked_exit_ignores_flag_state(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
capsys: pytest.CaptureFixture[str],
|
||||
) -> None:
|
||||
"""Helper is unconditional — flag-gating is the call-site's responsibility.
|
||||
|
||||
The runtime checks `_b4_mapper_source_enabled()` BEFORE invoking this
|
||||
helper, so once invoked the helper always exits. This keeps the helper
|
||||
behavior orthogonal to env state and makes the call-sites the
|
||||
single-source-of-truth for ON/OFF policy.
|
||||
"""
|
||||
monkeypatch.delenv(FLAG, raising=False)
|
||||
with pytest.raises(SystemExit) as exc:
|
||||
_b4_mapper_source_blocked_exit(
|
||||
"b4_no_cover",
|
||||
position="top",
|
||||
context={"v4_rank1": "F13"},
|
||||
)
|
||||
assert exc.value.code == 1
|
||||
|
||||
|
||||
def test_default_off_flag_state_does_not_invoke_blocked_helper(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""Under default-OFF, `_b4_mapper_source_enabled()` is False, which is
|
||||
the precondition the runtime checks before calling the helper. This test
|
||||
locks the contract that the flag reader returns False by default — any
|
||||
accidental flip would break the byte-identity guarantee of the legacy
|
||||
adapter_needed path.
|
||||
"""
|
||||
monkeypatch.delenv(FLAG, raising=False)
|
||||
assert _b4_mapper_source_enabled() is False
|
||||
426
tests/phase_z2/test_b4_mapper_source_equivalence.py
Normal file
426
tests/phase_z2/test_b4_mapper_source_equivalence.py
Normal file
@@ -0,0 +1,426 @@
|
||||
"""IMP-89 89-a u5 — slot_payload byte-equivalence when B4 matches mapper.
|
||||
|
||||
Stage 2 u5 contract (verbatim)::
|
||||
|
||||
slot_payload byte-equivalent (PHASE_Z_B4_MAPPER_SOURCE ON + matches_mapper=True)
|
||||
vs OFF, across mdx 01-05
|
||||
|
||||
Why this is load-bearing
|
||||
========================
|
||||
|
||||
u4 freezes the FULL pipeline ``final.html`` SHA under flag OFF. u5 isolates
|
||||
the *mapper-input* axis: when B4 ``PlacementPlan.selected_template_id``
|
||||
equals the legacy mapper input (``unit.frame_template_id`` — V4 rank-1),
|
||||
the selector at ``src/phase_z2_pipeline.py:223-242`` returns the same
|
||||
template id under either flag state. The mapper is a pure function of
|
||||
``(MdxSection, template_id)`` (deterministic dispatch via
|
||||
``map_with_contract`` → named ``PAYLOAD_BUILDERS`` — verified at
|
||||
``src/phase_z2_mapper.py:894-919``), so identical inputs → identical
|
||||
``slot_payload`` dicts → identical JSON-canonical bytes.
|
||||
|
||||
This is the *cross-axis* proof complementing u4:
|
||||
|
||||
* u4 = on-disk ``final.html`` SHA parity, default-OFF only (legacy
|
||||
preservation guard).
|
||||
* u5 = ``slot_payload`` byte equivalence, *flag ON ↔ flag OFF* (Layer A
|
||||
render-active behavior-preserving proof under matches_mapper).
|
||||
|
||||
The negative case (``test_slot_payload_diverges_when_b4_mismatches_under_flag_on``)
|
||||
locks the fact that ``slot_payload`` actually *depends* on the
|
||||
``template_id`` selector output — without it, the equivalence test could
|
||||
trivially pass even if the selector were a no-op.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from dataclasses import asdict, dataclass
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
import pytest
|
||||
|
||||
from src.phase_z2_mapper import (
|
||||
FitError,
|
||||
get_contract,
|
||||
load_frame_contracts,
|
||||
map_with_contract,
|
||||
)
|
||||
from src.phase_z2_pipeline import (
|
||||
_b4_mapper_source_enabled,
|
||||
_select_mapper_template_id,
|
||||
extract_content_objects,
|
||||
parse_mdx,
|
||||
)
|
||||
from src.phase_z2_placement_planner import plan_placement
|
||||
|
||||
|
||||
@dataclass
|
||||
class _StubPlan:
|
||||
"""Minimal placement-plan stand-in for selector unit checks.
|
||||
|
||||
``_select_mapper_template_id`` reads ONLY ``selected_template_id``
|
||||
(verified at ``src/phase_z2_pipeline.py:240-242``). Constructing the
|
||||
real ``PlacementPlan`` with placeholder slot/region lists would force
|
||||
the test to track schema drift on fields the selector never touches.
|
||||
"""
|
||||
|
||||
selected_template_id: Optional[str]
|
||||
|
||||
FLAG = "PHASE_Z_B4_MAPPER_SOURCE"
|
||||
_REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||
_SAMPLES_DIR = _REPO_ROOT / "samples" / "mdx_batch"
|
||||
_MDX_BATCH = ("01.mdx", "02.mdx", "03.mdx", "04.mdx", "05.mdx")
|
||||
|
||||
|
||||
def _canonical_bytes(payload: dict) -> bytes:
|
||||
"""Stable JSON canonical encoding for byte-level dict comparison.
|
||||
|
||||
``sort_keys`` removes dict-ordering noise; ``ensure_ascii=False`` keeps
|
||||
Korean text from being mangled into ``\\uXXXX`` escapes (which would
|
||||
still compare equal but would silently mask any encoding regression in
|
||||
the mapper).
|
||||
"""
|
||||
return json.dumps(payload, sort_keys=True, ensure_ascii=False).encode(
|
||||
"utf-8"
|
||||
)
|
||||
|
||||
|
||||
def _matches_mapper_cases() -> list[tuple[str, str, object, str]]:
|
||||
"""Enumerate (mdx_file, section_id, section, template_id) tuples where
|
||||
the matches_mapper scenario is reachable.
|
||||
|
||||
"matches_mapper=True" in production is the predicate
|
||||
``placement_plan.selected_template_id == unit.frame_template_id``. To
|
||||
cover it at the unit-test level without driving the full Type B
|
||||
coordinator, we treat each B4-selected template as the *simulated*
|
||||
legacy mapper input — i.e. we force matches_mapper=True by construction
|
||||
via ``mapper_template_id := plan.selected_template_id``.
|
||||
|
||||
Only sections where (a) B4 finds a covering frame AND (b) the mapper
|
||||
accepts that frame (no FitError) are byte-equivalence-eligible. Under
|
||||
flag ON the BLOCKED u3 path would otherwise fire — that axis is
|
||||
covered by ``test_b4_mapper_source_blocked.py`` and is out of scope
|
||||
here.
|
||||
"""
|
||||
frame_contracts = list(load_frame_contracts().values())
|
||||
cases: list[tuple[str, str, object, str]] = []
|
||||
for mdx_file in _MDX_BATCH:
|
||||
mdx_path = _SAMPLES_DIR / mdx_file
|
||||
_title, sections, _footer = parse_mdx(mdx_path)
|
||||
for section in sections:
|
||||
content_objects = extract_content_objects(
|
||||
section, source_shape=None
|
||||
)
|
||||
plan = plan_placement(
|
||||
content_objects=content_objects,
|
||||
frame_contracts=frame_contracts,
|
||||
section_id=section.section_id,
|
||||
)
|
||||
template_id = plan.selected_template_id
|
||||
if template_id is None:
|
||||
continue
|
||||
contract = get_contract(template_id)
|
||||
if contract is None:
|
||||
continue
|
||||
try:
|
||||
map_with_contract(section, contract)
|
||||
except FitError:
|
||||
continue
|
||||
cases.append((mdx_file, section.section_id, section, template_id))
|
||||
return cases
|
||||
|
||||
|
||||
# Frozen at collection time so a parametrize zero-iteration cannot silently
|
||||
# pass the byte-equivalence assertion (additional coverage lock below).
|
||||
_MATCHES_CASES = _matches_mapper_cases()
|
||||
|
||||
|
||||
def _slot_payload_via_selector(
|
||||
section, plan, mapper_input: str
|
||||
) -> tuple[dict, str]:
|
||||
"""Compose ``_select_mapper_template_id → map_mdx_to_slots`` once.
|
||||
|
||||
Mirrors the exact runtime path at
|
||||
``src/phase_z2_pipeline.py:4771-4797`` minus the BLOCKED u3 gate
|
||||
(which is out of scope for u5 byte equivalence — covered by u3).
|
||||
Returns ``(slot_payload, resolved_template_id)`` so per-case asserts
|
||||
can verify *both* axes (input + output) match.
|
||||
"""
|
||||
resolved = _select_mapper_template_id(plan, mapper_input)
|
||||
assert resolved is not None, (
|
||||
"u5 fixture invariant violated: resolved template_id is None even "
|
||||
"though the case was pre-filtered for B4 cover. Re-check "
|
||||
"_matches_mapper_cases()."
|
||||
)
|
||||
contract = get_contract(resolved)
|
||||
assert contract is not None, (
|
||||
f"u5 fixture invariant violated: no contract for resolved="
|
||||
f"{resolved!r} (case was pre-filtered for catalog membership)."
|
||||
)
|
||||
return map_with_contract(section, contract), resolved
|
||||
|
||||
|
||||
# ─── algebraic precondition (no pipeline / no mapper run) ──────────────
|
||||
|
||||
|
||||
def test_selector_returns_same_value_under_flag_flip_when_matches_mapper(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""Pure-function property: when ``plan.selected_template_id == T`` the
|
||||
selector returns ``T`` under either flag state.
|
||||
|
||||
This is the algebra that makes the end-to-end byte equivalence below
|
||||
hold mathematically. If this property breaks, every parametrized
|
||||
equivalence assertion would also break — this test localizes the
|
||||
failure to the selector helper itself.
|
||||
"""
|
||||
plan = _StubPlan(selected_template_id="F13")
|
||||
legacy_input = "F13" # matches_mapper=True by construction
|
||||
|
||||
monkeypatch.setenv(FLAG, "1")
|
||||
assert _b4_mapper_source_enabled() is True
|
||||
on_value = _select_mapper_template_id(plan, legacy_input)
|
||||
|
||||
monkeypatch.delenv(FLAG, raising=False)
|
||||
assert _b4_mapper_source_enabled() is False
|
||||
off_value = _select_mapper_template_id(plan, legacy_input)
|
||||
|
||||
assert on_value == off_value == "F13"
|
||||
|
||||
|
||||
# ─── end-to-end byte equivalence (parametrized over real mdx data) ────
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.parametrize(
|
||||
("mdx_file", "section_id", "section", "template_id"),
|
||||
_MATCHES_CASES,
|
||||
ids=lambda case: (
|
||||
case if isinstance(case, str) else getattr(case, "section_id", "_")
|
||||
),
|
||||
)
|
||||
def test_slot_payload_byte_equivalent_when_matches_mapper(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
mdx_file: str,
|
||||
section_id: str,
|
||||
section,
|
||||
template_id: str,
|
||||
) -> None:
|
||||
"""Per-section byte equivalence proof under matches_mapper=True.
|
||||
|
||||
Recomputes ``PlacementPlan`` from scratch inside the test (fixture
|
||||
enumeration cached only the section + B4 pick) and asserts that the
|
||||
mapper output is JSON-canonical-byte-identical between flag ON and
|
||||
flag OFF, given the same mapper input.
|
||||
"""
|
||||
frame_contracts = list(load_frame_contracts().values())
|
||||
content_objects = extract_content_objects(section, source_shape=None)
|
||||
plan = plan_placement(
|
||||
content_objects=content_objects,
|
||||
frame_contracts=frame_contracts,
|
||||
section_id=section.section_id,
|
||||
)
|
||||
assert plan.selected_template_id == template_id, (
|
||||
f"u5 invariant: B4 selection drifted between enumeration and "
|
||||
f"test execution for {mdx_file} {section_id}: enumerated="
|
||||
f"{template_id!r} live={plan.selected_template_id!r}"
|
||||
)
|
||||
|
||||
# Under matches_mapper=True the legacy mapper input equals plan pick.
|
||||
legacy_mapper_input = template_id
|
||||
|
||||
monkeypatch.delenv(FLAG, raising=False)
|
||||
plan_snapshot_off = asdict(plan) # type: ignore[call-overload]
|
||||
payload_off, resolved_off = _slot_payload_via_selector(
|
||||
section, plan, legacy_mapper_input
|
||||
)
|
||||
plan_after_off = asdict(plan) # type: ignore[call-overload]
|
||||
|
||||
monkeypatch.setenv(FLAG, "1")
|
||||
payload_on, resolved_on = _slot_payload_via_selector(
|
||||
section, plan, legacy_mapper_input
|
||||
)
|
||||
plan_after_on = asdict(plan) # type: ignore[call-overload]
|
||||
|
||||
assert resolved_off == resolved_on == template_id, (
|
||||
f"selector returned different template_id under matches_mapper for "
|
||||
f"{mdx_file} {section_id}: off={resolved_off!r} on={resolved_on!r}"
|
||||
)
|
||||
assert _canonical_bytes(payload_off) == _canonical_bytes(payload_on), (
|
||||
f"slot_payload byte equivalence broken for {mdx_file} {section_id} "
|
||||
f"(template_id={template_id}): mapper output diverged between "
|
||||
f"flag OFF and flag ON despite identical mapper input. This means "
|
||||
f"either map_with_contract gained nondeterminism or a hidden "
|
||||
f"selector-side effect crept in."
|
||||
)
|
||||
assert plan_snapshot_off == plan_after_off == plan_after_on, (
|
||||
f"PlacementPlan mutated by selector / mapper call for {mdx_file} "
|
||||
f"{section_id} — u5 byte equivalence relies on the selector being "
|
||||
f"a pure read of plan.selected_template_id."
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_matches_mapper_corpus_coverage_is_non_empty() -> None:
|
||||
"""Lock: the parametrized equivalence test above must have iterated at
|
||||
least once.
|
||||
|
||||
Without this guard a pytest parametrize zero-iteration (e.g. all
|
||||
sections rejected by B4 or all FitError-raising) would let the byte
|
||||
equivalence test silently pass with zero work. mdx 01-05 is rich
|
||||
enough that at least one matches_mapper case is always reachable.
|
||||
"""
|
||||
assert _MATCHES_CASES, (
|
||||
"u5 byte equivalence had zero matches_mapper cases — every section "
|
||||
"across mdx 01-05 was either B4-uncovered or raised FitError. "
|
||||
"Either the corpus shrank, B4 algorithm regressed, or the mapper "
|
||||
"now rejects every B4 pick. Investigate before re-locking."
|
||||
)
|
||||
seen_files = {case[0] for case in _MATCHES_CASES}
|
||||
assert len(seen_files) >= 1, (
|
||||
f"u5 coverage too narrow: {seen_files} — at least one mdx file "
|
||||
f"must yield a matches_mapper case for the equivalence proof to "
|
||||
f"be load-bearing."
|
||||
)
|
||||
|
||||
|
||||
# ─── negative case — bytes MUST diverge when B4 mismatches ─────────────
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_slot_payload_diverges_when_b4_mismatches_under_flag_on(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""Anti-vacuous proof: when B4 picks a template DIFFERENT from the
|
||||
legacy mapper input AND flag ON, the resulting ``slot_payload``
|
||||
differs from the flag-OFF case.
|
||||
|
||||
Without this assertion the equivalence test would pass even if the
|
||||
selector were a no-op that always returned the legacy input — i.e.
|
||||
the equivalence test would be load-bearing in the wrong direction.
|
||||
This test proves the mapper output genuinely depends on the selector's
|
||||
template_id choice, so equivalence under matches_mapper is a real
|
||||
behavioral guarantee rather than a tautology.
|
||||
|
||||
Strategy: find a section where the mapper accepts *both* the B4 pick
|
||||
AND a distinct alternative template (a frame the mapper also covers
|
||||
with a different builder/source_shape). Compare slot_payload bytes
|
||||
across the two — they MUST differ.
|
||||
"""
|
||||
frame_contracts = list(load_frame_contracts().values())
|
||||
diverging_case: tuple | None = None
|
||||
|
||||
for mdx_file in _MDX_BATCH:
|
||||
mdx_path = _SAMPLES_DIR / mdx_file
|
||||
_title, sections, _footer = parse_mdx(mdx_path)
|
||||
for section in sections:
|
||||
content_objects = extract_content_objects(
|
||||
section, source_shape=None
|
||||
)
|
||||
plan = plan_placement(
|
||||
content_objects=content_objects,
|
||||
frame_contracts=frame_contracts,
|
||||
section_id=section.section_id,
|
||||
)
|
||||
b4_pick = plan.selected_template_id
|
||||
if b4_pick is None:
|
||||
continue
|
||||
b4_contract = get_contract(b4_pick)
|
||||
if b4_contract is None:
|
||||
continue
|
||||
try:
|
||||
b4_payload = map_with_contract(section, b4_contract)
|
||||
except FitError:
|
||||
continue
|
||||
# Hunt for a *different* template the mapper also accepts on
|
||||
# this same section. Iterate the catalog in declaration order
|
||||
# so the search is deterministic.
|
||||
for alt in frame_contracts:
|
||||
alt_id = alt.get("template_id")
|
||||
if not alt_id or alt_id == b4_pick:
|
||||
continue
|
||||
try:
|
||||
alt_payload = map_with_contract(section, alt)
|
||||
except FitError:
|
||||
continue
|
||||
if _canonical_bytes(b4_payload) != _canonical_bytes(
|
||||
alt_payload
|
||||
):
|
||||
diverging_case = (
|
||||
mdx_file,
|
||||
section.section_id,
|
||||
b4_pick,
|
||||
alt_id,
|
||||
b4_payload,
|
||||
alt_payload,
|
||||
)
|
||||
break
|
||||
if diverging_case is not None:
|
||||
break
|
||||
if diverging_case is not None:
|
||||
break
|
||||
|
||||
assert diverging_case is not None, (
|
||||
"Could not find a section across mdx 01-05 where the mapper "
|
||||
"accepts two distinct templates with divergent slot_payload. "
|
||||
"Without such a case the equivalence test above is tautological."
|
||||
)
|
||||
|
||||
(
|
||||
mdx_file,
|
||||
section_id,
|
||||
b4_pick,
|
||||
alt_id,
|
||||
b4_payload,
|
||||
alt_payload,
|
||||
) = diverging_case
|
||||
|
||||
# Now drive the selector path under flag ON with B4 picking ``b4_pick``
|
||||
# while the legacy mapper input is ``alt_id`` — i.e. B4 mismatches the
|
||||
# legacy input. Flag ON → selector returns b4_pick → mapper produces
|
||||
# b4_payload. Flag OFF → selector returns alt_id → mapper produces
|
||||
# alt_payload. The two MUST differ.
|
||||
plan = _StubPlan(selected_template_id=b4_pick)
|
||||
|
||||
mdx_path = _SAMPLES_DIR / mdx_file
|
||||
_title, sections, _footer = parse_mdx(mdx_path)
|
||||
section = next(s for s in sections if s.section_id == section_id)
|
||||
|
||||
monkeypatch.setenv(FLAG, "1")
|
||||
on_payload, on_resolved = _slot_payload_via_selector(
|
||||
section, plan, alt_id
|
||||
)
|
||||
monkeypatch.delenv(FLAG, raising=False)
|
||||
off_payload, off_resolved = _slot_payload_via_selector(
|
||||
section, plan, alt_id
|
||||
)
|
||||
|
||||
assert on_resolved == b4_pick
|
||||
assert off_resolved == alt_id
|
||||
assert _canonical_bytes(on_payload) != _canonical_bytes(off_payload), (
|
||||
f"Negative case failed: selector flip from {alt_id} (OFF) to "
|
||||
f"{b4_pick} (ON) produced byte-identical slot_payload for "
|
||||
f"{mdx_file} {section_id}. The mapper appears to ignore "
|
||||
f"template_id, which would make the equivalence test tautological."
|
||||
)
|
||||
|
||||
|
||||
# ─── selector default-state lock (mirror of u4 sanity check) ───────────
|
||||
|
||||
|
||||
def test_selector_default_state_returns_legacy_under_b4_mismatch(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""Final sanity: even when B4 would pick something different, the
|
||||
flag-OFF default selector returns the legacy mapper input verbatim.
|
||||
|
||||
This is the property that makes u4 SHA parity hold and the negative
|
||||
test above meaningful. Repeated here at the u5 axis so a single test
|
||||
file change cannot accidentally hide the regression signal across
|
||||
both u4 and u5.
|
||||
"""
|
||||
plan = _StubPlan(selected_template_id="F29")
|
||||
monkeypatch.delenv(FLAG, raising=False)
|
||||
assert _b4_mapper_source_enabled() is False
|
||||
assert _select_mapper_template_id(plan, "F13") == "F13"
|
||||
54
tests/phase_z2/test_b4_mapper_source_flag.py
Normal file
54
tests/phase_z2/test_b4_mapper_source_flag.py
Normal file
@@ -0,0 +1,54 @@
|
||||
"""IMP-89 89-a u1 — PHASE_Z_B4_MAPPER_SOURCE flag reader unit tests.
|
||||
|
||||
Stage 2 plan (u1): adds an env flag reader helper (default OFF) distinct
|
||||
from PHASE_Z_B4_GATEKEEPER. u1 only locks reader semantics — u2 wires it
|
||||
into the slot_payload source-of-truth switch and u3 layers BLOCKED exits
|
||||
for B4 no-cover and B4-selected FitError under flag ON.
|
||||
|
||||
Truthy contract (mirrors PHASE_Z_B4_GATEKEEPER /
|
||||
PHASE_Z_B4_SOURCE_SHAPE_ENABLED at src/phase_z2_pipeline.py:4625,4662):
|
||||
case-insensitive + leading/trailing whitespace stripped; truthy set
|
||||
= {'1', 'true', 'yes'}. Everything else (including '0', '', 'no',
|
||||
'false', missing env var) is OFF.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from src.phase_z2_pipeline import _b4_mapper_source_enabled
|
||||
|
||||
FLAG = "PHASE_Z_B4_MAPPER_SOURCE"
|
||||
|
||||
|
||||
def test_default_off_when_env_unset(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.delenv(FLAG, raising=False)
|
||||
assert _b4_mapper_source_enabled() is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize("value", ["1", "true", "yes", "TRUE", "Yes", " true ", " 1\t"])
|
||||
def test_truthy_values_enable_flag(
|
||||
monkeypatch: pytest.MonkeyPatch, value: str
|
||||
) -> None:
|
||||
monkeypatch.setenv(FLAG, value)
|
||||
assert _b4_mapper_source_enabled() is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize("value", ["", "0", "no", "false", "off", "2", "on", "y"])
|
||||
def test_non_truthy_values_keep_flag_off(
|
||||
monkeypatch: pytest.MonkeyPatch, value: str
|
||||
) -> None:
|
||||
monkeypatch.setenv(FLAG, value)
|
||||
assert _b4_mapper_source_enabled() is False
|
||||
|
||||
|
||||
def test_flag_distinct_from_gatekeeper(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""PHASE_Z_B4_GATEKEEPER ON must not flip the mapper-source flag.
|
||||
|
||||
Locks Stage 2 design decision (Stage 1 Q1 resolution): the new flag
|
||||
governs slot_payload source-of-truth; PHASE_Z_B4_GATEKEEPER retains
|
||||
its mismatch render-skip semantics. They must be independently
|
||||
toggleable.
|
||||
"""
|
||||
monkeypatch.setenv("PHASE_Z_B4_GATEKEEPER", "1")
|
||||
monkeypatch.delenv(FLAG, raising=False)
|
||||
assert _b4_mapper_source_enabled() is False
|
||||
96
tests/phase_z2/test_b4_mapper_source_switch.py
Normal file
96
tests/phase_z2/test_b4_mapper_source_switch.py
Normal file
@@ -0,0 +1,96 @@
|
||||
"""IMP-89 89-a u2 — slot_payload source-of-truth switch unit tests.
|
||||
|
||||
Stage 2 plan (u2): wires the u1 PHASE_Z_B4_MAPPER_SOURCE flag into the
|
||||
single slot_payload construction site at src/phase_z2_pipeline.py:4702
|
||||
via the _select_mapper_template_id() selector helper.
|
||||
|
||||
Locked semantics (Stage 1 Q1 / Stage 2 u2):
|
||||
flag ON → mapper input = placement_plan.selected_template_id (B4)
|
||||
flag OFF → mapper input = unit.frame_template_id (legacy mapper-only)
|
||||
|
||||
u3 will add BLOCKED exits for (selected_template_id is None OR FitError
|
||||
on B4-selected) under flag ON — NO silent fallback. u4 guards default-OFF
|
||||
final.html SHA parity for mdx 01-05.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional
|
||||
|
||||
import pytest
|
||||
|
||||
from src.phase_z2_pipeline import _select_mapper_template_id
|
||||
|
||||
FLAG = "PHASE_Z_B4_MAPPER_SOURCE"
|
||||
|
||||
|
||||
@dataclass
|
||||
class _StubPlan:
|
||||
"""Minimal PlacementPlan stand-in — only selected_template_id is read."""
|
||||
|
||||
selected_template_id: Optional[str]
|
||||
|
||||
|
||||
def test_flag_off_returns_unit_frame_template_id(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""Default-OFF preserves legacy mapper input (V4 rank-1)."""
|
||||
monkeypatch.delenv(FLAG, raising=False)
|
||||
plan = _StubPlan(selected_template_id="B4_PICK")
|
||||
assert _select_mapper_template_id(plan, "V4_PICK") == "V4_PICK"
|
||||
|
||||
|
||||
def test_flag_on_returns_placement_plan_selected_template_id(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""Flag ON routes mapper input to B4 PlacementPlan."""
|
||||
monkeypatch.setenv(FLAG, "1")
|
||||
plan = _StubPlan(selected_template_id="B4_PICK")
|
||||
assert _select_mapper_template_id(plan, "V4_PICK") == "B4_PICK"
|
||||
|
||||
|
||||
def test_flag_on_with_matching_b4_returns_same_value(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""When B4-selected == mapper, switch is behavior-preserving."""
|
||||
monkeypatch.setenv(FLAG, "true")
|
||||
plan = _StubPlan(selected_template_id="F13")
|
||||
assert _select_mapper_template_id(plan, "F13") == "F13"
|
||||
|
||||
|
||||
def test_flag_on_with_no_b4_cover_returns_none(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""Flag ON + B4 no-cover surfaces None — u3 will BLOCK on this signal."""
|
||||
monkeypatch.setenv(FLAG, "yes")
|
||||
plan = _StubPlan(selected_template_id=None)
|
||||
assert _select_mapper_template_id(plan, "V4_PICK") is None
|
||||
|
||||
|
||||
def test_flag_off_with_no_b4_cover_still_returns_legacy(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""Default-OFF ignores B4 None — legacy mapper input always honored."""
|
||||
monkeypatch.delenv(FLAG, raising=False)
|
||||
plan = _StubPlan(selected_template_id=None)
|
||||
assert _select_mapper_template_id(plan, "V4_PICK") == "V4_PICK"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("non_truthy", ["", "0", "no", "false", "off", "2"])
|
||||
def test_non_truthy_env_values_keep_legacy_source(
|
||||
monkeypatch: pytest.MonkeyPatch, non_truthy: str
|
||||
) -> None:
|
||||
"""Non-truthy env values mirror u1 flag-reader contract — legacy source."""
|
||||
monkeypatch.setenv(FLAG, non_truthy)
|
||||
plan = _StubPlan(selected_template_id="B4_PICK")
|
||||
assert _select_mapper_template_id(plan, "V4_PICK") == "V4_PICK"
|
||||
|
||||
|
||||
def test_gatekeeper_flag_does_not_flip_mapper_source(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""PHASE_Z_B4_GATEKEEPER ON alone must NOT route mapper to B4 (Stage 1 Q1)."""
|
||||
monkeypatch.setenv("PHASE_Z_B4_GATEKEEPER", "1")
|
||||
monkeypatch.delenv(FLAG, raising=False)
|
||||
plan = _StubPlan(selected_template_id="B4_PICK")
|
||||
assert _select_mapper_template_id(plan, "V4_PICK") == "V4_PICK"
|
||||
Reference in New Issue
Block a user