procurement-ingest/tests/test_bundle_consistency.py

"""AST-based bundle-consistency test for the email-processor Lambdas.

Verifies that each pipeline's CDK bundling `command` actually ships every
first-party sibling module handler.py imports into /asset-output. This
guards against a regression where the bundling `cp` step (whether an
explicit filename allowlist or a glob) silently drops a module the handler
depends on -- history: PR #105 shipped without template_parser.py, and PR #2
nearly shipped without derived_fields.py, both allowlist-maintenance misses
that would ImportError at runtime.

Pure ast + file reads -- no AWS/boto3/CDK synth, no handler import, no moto.
Fast and has no dependency on the moto-before-handler import-order invariant
that the rest of the suite relies on.
"""

import ast
import re
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parents[1]

PO_HANDLER = REPO_ROOT / "lambdas" / "po" / "email_processor" / "handler.py"
WO_HANDLER = REPO_ROOT / "lambdas" / "wo" / "email_processor" / "handler.py"
API_HANDLER = REPO_ROOT / "lambdas" / "api" / "handler.py"
PO_STACK = REPO_ROOT / "cdk" / "po_stack.py"
WO_STACK = REPO_ROOT / "cdk" / "wo_stack.py"
API_STACK = REPO_ROOT / "cdk" / "procurement_api_stack.py"
# Phase 3: ses_auth/web_ui_auth/email_parsing/emf are single-sourced here and
# shipped into the email-processor bundles via `cp shared/*.py`.
SHARED_DIR = REPO_ROOT / "lambdas" / "shared"

# The names Phase 3 single-sourced under lambdas/shared/. After the move NONE
# of these may reappear as a top-level .py in either email-processor pipeline
# dir: every handler loader resolves a pipeline-local copy FIRST
# (tests/conftest.py load_handler checks path.parent before _SHARED_DIR;
# lambdas/wo/email_processor/tests/_wo_parser_support.py inserts the pipeline
# dir ahead of shared/ on sys.path), so a stray reappearance would silently
# SHADOW the single shared source in every handler-loaded test while
# test_ses_auth / test_parse_raw_email keep exercising shared/ -- divergence
# undetected. This is exactly the future-drift invariant the retired
# byte-identity ses_auth fixture used to guard; it is now enforced by policing
# the pipeline dirs and shared/ SEPARATELY (never as a union, which would let
# the same name already expected in shared/ mask a pipeline-dir stray) --
# see test_no_shared_module_shadow_in_pipeline_dirs and the per-dir exact-set
# pins below.
SHARED_MODULES = frozenset({"ses_auth", "email_parsing", "emf", "web_ui_auth"})

# Exact top-level .py stems each dir must hold. Pinned as exact sets (both
# bounds): the bundling globs (`cp <pipeline>/email_processor/*.py` +
# `cp shared/*.py`) ship every top-level .py in these dirs, and
# `Code.from_asset` stages untracked files too (it does not honor .gitignore),
# so a stray scratch/secrets .py -- or a shadow copy of a shared module -- would
# silently ship into the production zip on a local deploy. Any new top-level
# module must be added here deliberately, the moment to decide whether it SHOULD
# ship (a real module) or must be excluded.
# Phase 5 decomposed each God-handler into flat siblings (constraint 6): the
# non-recursive `cp <pipeline>/email_processor/*.py` glob auto-ships them, but
# the exact-set pin must list every new sibling or the ships-no-unexpected test
# fails -- keeping the teeth (a sibling not added here, or a commented-out cp,
# still fails). PO gained prompts/extraction/enrichment/telemetry/persistence;
# WO the same minus enrichment (it has no enrichment stage).
PO_PIPELINE_MODULES = frozenset(
    {
        "handler",
        "template_parser",
        "derived_fields",
        "extraction",
        "enrichment",
        "telemetry",
        "persistence",
        "prompts",
    }
)
WO_PIPELINE_MODULES = frozenset(
    {
        "__init__",
        "handler",
        "template_parser",
        "extraction",
        "telemetry",
        "persistence",
        "prompts",
    }
)

# Full shipped set of each email-processor bundle (pipeline glob + shared glob).
# Derived from the per-dir pins above so it stays consistent with them; policed
# per-dir (NOT via this union) so a shared-name shadow in a pipeline dir cannot
# be masked. web_ui_auth is a deliberate, harmless ride-along of the pinned
# `cp shared/*.py` (constraint #8) -- never imported by the email handlers, but
# it ships, so it is part of the shipped set.
PO_EXPECTED_TOP_LEVEL_MODULES = PO_PIPELINE_MODULES | SHARED_MODULES
WO_EXPECTED_TOP_LEVEL_MODULES = WO_PIPELINE_MODULES | SHARED_MODULES

# procurement-api (lambdas/api/): a fourth bundled function, in its own stack.
# Same exact-set discipline as the pipeline dirs -- `cp api/*.py` ships every
# top-level .py here (untracked strays included), so any new module is a
# deliberate addition to this pin.
API_PIPELINE_MODULES = frozenset(
    {
        "handler",
        "router",
        "pagination",
        "serialization",
        "wo_repo",
        "po_repo",
    }
)
# Non-.py files the api bundle must also EXECUTE cps for: the handler serves
# the spec, docs page, and the vendored Swagger UI assets from its own package
# dir, so dropping any cp 500s /docs at runtime with green CI.
API_DATA_FILES_CP_RES = (
    r"(?:^|\s)api/openapi\.json(?:\s|$)",
    r"(?:^|\s)api/docs\.html(?:\s|$)",
    r"(?:^|\s)api/swagger-ui-bundle\.js(?:\s|$)",
    r"(?:^|\s)api/swagger-ui-standalone-preset\.js(?:\s|$)",
    r"(?:^|\s)api/swagger-ui\.css(?:\s|$)",
)

# Pipeline-scoped glob shapes the per-stack ships-all pins accept. Phase 2
# widened the bundling cwd to the shared ../lambdas asset root, so the executed
# glob carries its own pipeline's path prefix (`po/email_processor/*.py`); a
# bare `*.py`/`./*.py` prefix is still accepted for the legacy in-dir cwd. A
# WRONG-pipeline prefix (`wo/email_processor/*.py` in po_stack) or an arbitrary
# directory (`venv/lib/*.py`) is deliberately NOT accepted: it would false-pass
# the ships-all shape check while staging a bundle missing a required sibling
# (e.g. derived_fields.py, which lives only under po/) -- a runtime ImportError
# that green CI must never wave through. Do NOT relax these to an any-prefix
# pattern: that reintroduces exactly the wrong-pipeline false-pass this pins out.
PO_SCOPED_GLOB_RE = r"(?:^|\s)(?:\./|po/email_processor/)?\*\.py(?:\s|$)"
WO_SCOPED_GLOB_RE = r"(?:^|\s)(?:\./|wo/email_processor/)?\*\.py(?:\s|$)"

# Phase 3: both email-processor bundles gain a second glob copying the
# single-sourced shared modules flat into the zip. This must be the EXECUTED
# form (`_executed_cp_commands` strips comments), so a commented-out shared cp
# cannot false-pass the pin.
SHARED_CP_RE = r"(?:^|\s)shared/\*\.py(?:\s|$)"

# Phase 3: each stack's web_ui function stages ONLY shared/web_ui_auth.py flat
# beside its handler (NOT all of shared/ -- the email-only modules must not
# bloat the web UI zip). The auth-gated web_ui handlers do a module-top-level
# `from web_ui_auth import is_authenticated`, so dropping/commenting this cp
# ImportErrors the Lambda at cold start with green CI -- the PR #105 ImportError
# class the AST test exists to prevent, but for a surface (web_ui) the
# email-processor selector deliberately excludes. Checked as the EXECUTED form
# so a commented-out cp cannot false-pass.
WEB_UI_AUTH_CP_RE = r"(?:^|\s)shared/web_ui_auth\.py(?:\s|$)"


def _first_party_sibling_imports(handler_path: Path) -> set[str]:
    """Top-level module names handler.py imports that are first-party siblings.

    Parses only top-level (module-body) `import X` / `from X import ...`
    statements -- not imports nested in functions -- and keeps a name only
    if `<sibling_dir>/X.py` exists, which filters out stdlib/third-party
    imports (json, os, boto3, ...) and keeps exactly the modules the
    bundling step is obligated to ship.
    """
    tree = ast.parse(handler_path.read_text())
    names: set[str] = set()
    for node in tree.body:
        if isinstance(node, ast.Import):
            for alias in node.names:
                names.add(alias.name.split(".")[0])
        elif isinstance(node, ast.ImportFrom):
            if node.module:
                names.add(node.module.split(".")[0])
    sibling_dir = handler_path.parent
    # A first-party sibling resolves either next to the handler (per-pipeline:
    # template_parser/derived_fields) or under lambdas/shared/ (single-sourced:
    # ses_auth/email_parsing/emf, Phase 3). Both must count, or the ships-all
    # pin would silently drop the shared siblings (ses_auth was silently
    # dropped, email_parsing/emf never seen, before this resolved shared/).
    return {
        name
        for name in names
        if (sibling_dir / f"{name}.py").exists() or (SHARED_DIR / f"{name}.py").exists()
    }


def _extract_bundling_command(stack_path: Path) -> str:
    """Extract the bash -c bundling command string from a CDK stack file.

    Locates the `command=[...]` keyword argument to BundlingOptions and
    returns its final list element's string value. Adjacent string-literal
    concatenation (used in both stacks to keep the command readable across
    lines, with `#` comments interspersed) is folded by the parser itself,
    so this returns the exact single command string CDK will hand to bash.

    Since Phase 3 each stack has TWO bundled functions -- the email processor
    and the web_ui function (which now also stages a shared module). "The"
    bundling command this test cares about is the EMAIL-processor one, so we
    select the command whose text mentions ``email_processor`` (its first cp
    is ``cp <pipeline>/email_processor/*.py``) and demand exactly one match --
    the web_ui command (``cp -r <pipeline>/web_ui/.``) and site_extractor do
    not match.
    """
    tree = ast.parse(stack_path.read_text())
    commands: list[str] = []
    for node in ast.walk(tree):
        if isinstance(node, ast.keyword) and node.arg == "command":
            list_node = node.value
            if isinstance(list_node, ast.List) and list_node.elts:
                last = list_node.elts[-1]
                if isinstance(last, ast.Constant) and isinstance(last.value, str):
                    commands.append(last.value)
    email_cmds = [c for c in commands if "email_processor" in c]
    if len(email_cmds) != 1:
        raise AssertionError(
            f"expected exactly one email-processor bundling command=[...] list in "
            f"{stack_path}, found {len(email_cmds)} (of {len(commands)} total "
            "command lists) — update this test to select the intended bundling "
            "command explicitly"
        )
    return email_cmds[0]


def _extract_web_ui_command(stack_path: Path) -> str:
    """Extract the web_ui function's bash -c bundling command string.

    Mirrors _extract_bundling_command but selects the command whose text
    mentions ``web_ui`` (its first cp is ``cp -r <pipeline>/web_ui/.``). The
    email-processor command (``cp <pipeline>/email_processor/*.py``, plus
    ``cp shared/*.py``) and site_extractor do not contain ``web_ui`` in the
    folded command STRING (the explanatory ``# ... web_ui_auth ...`` comments
    around the email command are Python comments, not string content, so they
    are not part of the folded literal). Demands exactly one match so an
    ambiguity surfaces loudly instead of silently checking the wrong command.
    """
    tree = ast.parse(stack_path.read_text())
    commands: list[str] = []
    for node in ast.walk(tree):
        if isinstance(node, ast.keyword) and node.arg == "command":
            list_node = node.value
            if isinstance(list_node, ast.List) and list_node.elts:
                last = list_node.elts[-1]
                if isinstance(last, ast.Constant) and isinstance(last.value, str):
                    commands.append(last.value)
    web_ui_cmds = [c for c in commands if "web_ui" in c]
    if len(web_ui_cmds) != 1:
        raise AssertionError(
            f"expected exactly one web_ui bundling command=[...] list in "
            f"{stack_path}, found {len(web_ui_cmds)} (of {len(commands)} total "
            "command lists) — update this test to select the intended bundling "
            "command explicitly"
        )
    return web_ui_cmds[0]


def _executed_cp_commands(command: str) -> list[str]:
    """The `cp ...` invocations bash would actually execute, comment-stripped.

    Splits the bundling command on shell separators (`&&`, `;`, `|`, newline),
    drops any trailing `#` comment from each segment, and returns the segments
    whose command word is `cp`. This is deliberately stricter than a substring
    match: a glob or `cp -r` string that survives only inside a comment
    (e.g. `cp handler.py /asset-output/  # was: cp ./*.py /asset-output/`) is
    NOT returned, because bash would not execute it -- so a revert that strips
    the real copy but leaves the old text commented cannot false-pass.
    """
    segments = re.split(r"&&|\|\||;|\||\n", command)
    cp_cmds: list[str] = []
    for seg in segments:
        seg = seg.split("#", 1)[0].strip()
        if re.match(r"cp\b", seg):
            cp_cmds.append(seg)
    return cp_cmds


def _bundling_ships_all(command: str, sibling_names: set[str]) -> bool:
    """True if `command` is guaranteed to ship every name in `sibling_names`.

    Inspects only the `cp` commands bash actually executes (comments stripped
    via `_executed_cp_commands`) and ACCUMULATES the set of shipped module
    stems across ALL of them -- because since Phase 3 the required siblings
    ship via TWO globs, not one: `cp <pipeline>/email_processor/*.py` covers
    the per-pipeline siblings and `cp shared/*.py` covers the single-sourced
    ones (ses_auth/email_parsing/emf). A single-cp "one glob ships everything"
    check would wrongly report False now that coverage is split.

    Each executed cp contributes to the shipped set as follows:
      - a non-recursive `*.py` glob ships every top-level .py in the globbed
        directory -- but ONLY that directory. Its (optional) path prefix is
        resolved against the ../lambdas asset root (the Phase 2 bundling cwd)
        and every `.py` stem actually present there is added. A wrong-pipeline
        or arbitrary-directory glob (`cp wo/email_processor/*.py` in po_stack,
        `cp venv/lib/*.py`) therefore contributes only what that dir really
        holds (or nothing, if it does not exist) and can never cover a sibling
        that lives elsewhere;
      - a recursive copy of the WHOLE source dir, e.g. `cp -r . /asset-output/`
        (`.`/`./` source only), ships everything -> short-circuit True;
      - any other executed `cp` is an explicit filename allowlist (the legacy
        PO form): each copied `<name>.py` stem is added, so a reverted
        allowlist missing a sibling leaves that sibling out of the set.
    Returns True only if every required sibling ended up in the accumulated set.
    """
    cp_cmds = _executed_cp_commands(command)
    if not cp_cmds:
        return False
    shipped: set[str] = set()
    for cp in cp_cmds:
        glob_match = re.search(r"(?:^|\s)((?:[\w./-]+/)?)\*\.py(?:\s|$)", cp)
        if glob_match:
            glob_dir = (REPO_ROOT / "lambdas" / glob_match.group(1)).resolve()
            shipped.update(p.stem for p in glob_dir.glob("*.py"))
            continue
        if re.search(r"cp\s+-r\s+\.\/?\s+/asset-output", cp):
            return True
        # Explicit-allowlist shape: collect the copied source filenames,
        # ignoring any cp flags and the trailing /asset-output destination.
        match = re.search(
            r"cp\s+(?:-\S+\s+)*(.*?)\s*/asset-output/?\"?\s*$", cp.strip()
        )
        if match:
            shipped.update(Path(f).stem for f in match.group(1).split())
    return all(name in shipped for name in sibling_names)


def test_po_bundling_ships_all_first_party_siblings():
    siblings = _first_party_sibling_imports(PO_HANDLER)
    # Sanity: PO handler.py is known to import ses_auth, template_parser,
    # and derived_fields as bare-name siblings. If this ever collapses to
    # an empty set, the test below would vacuously pass -- guard against that.
    assert siblings, "expected first-party sibling imports in PO handler.py"

    command = _extract_bundling_command(PO_STACK)

    # Pinned per the Phase 0 healthcheck/bundling contract: PO's bundling
    # command must EXECUTE the non-recursive glob, not a hand-maintained
    # allowlist. Checked against the executed cp (comments stripped) so the
    # old glob text surviving only in a comment cannot satisfy this.
    executed_cps = _executed_cp_commands(command)
    assert any(re.search(PO_SCOPED_GLOB_RE, cp) for cp in executed_cps), (
        "cdk/po_stack.py bundling command must EXECUTE the PO-scoped non-recursive "
        "glob 'cp po/email_processor/*.py /asset-output/' so every first-party "
        "sibling handler.py imports ships automatically. A wrong-pipeline or "
        "arbitrary-directory glob is rejected here on purpose. "
        f"Executed cp commands: {executed_cps}"
    )
    # Phase 3: the shared siblings (ses_auth/email_parsing/emf) ship via a
    # SECOND executed glob, `cp shared/*.py /asset-output/`. Require it to be
    # actually executed (comment-stripped), so commenting it out fails here;
    # combined with ships-all below (those siblings resolve only under shared/)
    # a removed/commented shared cp fails BOTH assertions.
    assert any(re.search(SHARED_CP_RE, cp) for cp in executed_cps), (
        "cdk/po_stack.py email-processor bundling command must EXECUTE "
        "'cp shared/*.py /asset-output/' so the single-sourced shared modules "
        f"ship flat beside handler.py. Executed cp commands: {executed_cps}"
    )
    assert _bundling_ships_all(command, siblings), (
        f"cdk/po_stack.py bundling command does not ship all of {sorted(siblings)}: "
        f"{command!r}"
    )


def test_wo_bundling_ships_all_first_party_siblings():
    siblings = _first_party_sibling_imports(WO_HANDLER)
    assert siblings, "expected first-party sibling imports in WO handler.py"

    command = _extract_bundling_command(WO_STACK)

    # Phase 2: WO now ships via the same scoped non-recursive glob as PO,
    # copying only wo/email_processor/*.py from the widened ../lambdas asset
    # root. This deliberately drops tests/, __pycache__, and requirements.txt
    # from the production zip (docs/refactor-evaluation.md Phase 2). Checked
    # against the comment-stripped executed cp so an old `cp -r .` surviving
    # only in a comment cannot satisfy this, and a revert to the whole-dir
    # copy (which would re-ship tests/) fails loudly.
    executed_cps = _executed_cp_commands(command)
    assert any(re.search(WO_SCOPED_GLOB_RE, cp) for cp in executed_cps), (
        "cdk/wo_stack.py bundling command must EXECUTE the WO-scoped non-recursive "
        "glob 'cp wo/email_processor/*.py /asset-output/' so every first-party "
        "sibling ships while tests/ and requirements.txt are excluded. A "
        "wrong-pipeline or arbitrary-directory glob is rejected here on purpose. "
        f"Executed cp commands: {executed_cps}"
    )
    # Phase 3: shared siblings ship via the second executed glob `cp shared/*.py`.
    assert any(re.search(SHARED_CP_RE, cp) for cp in executed_cps), (
        "cdk/wo_stack.py email-processor bundling command must EXECUTE "
        "'cp shared/*.py /asset-output/' so the single-sourced shared modules "
        f"ship flat beside handler.py. Executed cp commands: {executed_cps}"
    )
    assert _bundling_ships_all(command, siblings), (
        f"cdk/wo_stack.py bundling command does not ship all of {sorted(siblings)}: "
        f"{command!r}"
    )


def test_po_email_processor_dir_ships_no_unexpected_top_level_modules():
    """Upper bound on the PO pipeline dir alone (NOT unioned with shared/).

    The sibling-import tests above prove the glob ships every module the
    handler needs (shipped >= required). This proves the other direction for
    the pipeline dir (shipped <= expected): because `cp po/email_processor/*.py`
    copies every top-level .py in that dir -- and `Code.from_asset` stages
    untracked files too (it does not honor .gitignore) -- a stray scratch or
    secrets .py, OR a shadow copy of a moved shared module, would silently ship
    into the zip on a local deploy. Policing this dir SEPARATELY from shared/
    (rather than as a union with it) is what makes a strayed-back ses_auth.py /
    email_parsing.py / emf.py FAIL here instead of being masked by the same name
    already being expected in shared/.
    """
    top_level = {p.stem for p in PO_HANDLER.parent.glob("*.py")}
    assert top_level == set(PO_PIPELINE_MODULES), (
        "unexpected top-level .py set in lambdas/po/email_processor -- the "
        "'cp po/email_processor/*.py' glob would ship exactly these into the "
        f"Lambda zip. Found {sorted(top_level)}, expected "
        f"{sorted(PO_PIPELINE_MODULES)}. A moved shared module "
        f"({sorted(SHARED_MODULES)}) reappearing here would SHADOW the single "
        "shared source in every handler-loaded test -- remove it. If a new "
        "per-pipeline module is intended, add it to PO_PIPELINE_MODULES."
    )


def test_wo_email_processor_dir_ships_no_unexpected_top_level_modules():
    """Upper bound on the WO pipeline dir alone (NOT unioned with shared/).

    The WO equivalent of the PO exact-set pin -- previously MISSING entirely,
    so a stray .py (or a shadow copy of a moved shared module) in
    lambdas/wo/email_processor was policed by nothing. Same rationale as the PO
    pin: `cp wo/email_processor/*.py` ships every top-level .py in this dir, and
    a strayed-back ses_auth/email_parsing/emf would shadow the shared source in
    the WO handler-loaded tests.
    """
    top_level = {p.stem for p in WO_HANDLER.parent.glob("*.py")}
    assert top_level == set(WO_PIPELINE_MODULES), (
        "unexpected top-level .py set in lambdas/wo/email_processor -- the "
        "'cp wo/email_processor/*.py' glob would ship exactly these into the "
        f"Lambda zip. Found {sorted(top_level)}, expected "
        f"{sorted(WO_PIPELINE_MODULES)}. A moved shared module "
        f"({sorted(SHARED_MODULES)}) reappearing here would SHADOW the single "
        "shared source in every handler-loaded test -- remove it. If a new "
        "per-pipeline module is intended, add it to WO_PIPELINE_MODULES."
    )


def test_shared_dir_ships_no_unexpected_top_level_modules():
    """Upper bound on lambdas/shared/ alone (NOT unioned with a pipeline dir).

    `cp shared/*.py` ships every top-level .py under lambdas/shared/ into BOTH
    email-processor bundles, so a stray scratch/secrets .py here would leak into
    both production zips. Pinned to exactly the four single-sourced modules.
    """
    top_level = {p.stem for p in SHARED_DIR.glob("*.py")}
    assert top_level == set(SHARED_MODULES), (
        "unexpected top-level .py set in lambdas/shared -- the 'cp shared/*.py' "
        "glob would ship exactly these into BOTH email-processor Lambda zips. "
        f"Found {sorted(top_level)}, expected {sorted(SHARED_MODULES)}. If a new "
        "shared module is intended, add it to SHARED_MODULES; if it is a scratch "
        "or secrets file, remove it before deploy."
    )


def test_no_shared_module_shadow_in_pipeline_dirs():
    """The moved shared names must live ONLY under lambdas/shared/.

    Replaces the future-drift guard the retired byte-identity ses_auth fixture
    used to provide. A stray reappearance of a moved shared module in either
    email-processor pipeline dir would be resolved FIRST by every handler loader
    -- tests/conftest.py load_handler checks `path.parent / f"{sibling}.py"`
    before the _SHARED_DIR fallback, and
    lambdas/wo/email_processor/tests/_wo_parser_support.py inserts the pipeline
    dir ahead of shared/ on sys.path -- silently SHADOWING the single shared
    source in every handler-loaded test, while test_ses_auth / test_parse_raw_email
    keep exercising shared/. Divergence would go undetected. Police the pipeline
    dirs and shared/ SEPARATELY so no union can mask the shadow.
    """
    for name in sorted(SHARED_MODULES):
        assert (SHARED_DIR / f"{name}.py").exists(), (
            f"{name}.py must exist under lambdas/shared/ (single source of truth)"
        )
        assert not (PO_HANDLER.parent / f"{name}.py").exists(), (
            f"{name}.py reappeared in lambdas/po/email_processor -- it would "
            "SHADOW lambdas/shared/{name}.py in every PO handler-loaded test "
            "(the loader resolves the pipeline-local copy first). Delete it; "
            "the single source lives under lambdas/shared/."
        )
        assert not (WO_HANDLER.parent / f"{name}.py").exists(), (
            f"{name}.py reappeared in lambdas/wo/email_processor -- it would "
            "SHADOW lambdas/shared/{name}.py in every WO handler-loaded test "
            "(_wo_parser_support inserts the pipeline dir ahead of shared/ on "
            "sys.path). Delete it; the single source lives under lambdas/shared/."
        )


def test_detection_logic_catches_allowlist_missing_a_sibling():
    """Unit-level check on `_bundling_ships_all` itself.

    Simulates the historical regression shape directly: someone reverts the
    PO glob back to an explicit filename allowlist that omits a required sibling.
    Phase 5 note: the PR #2 near-miss module (derived_fields) is now a TRANSITIVE
    import via enrichment.py, so it is no longer among handler.py's DIRECT
    first-party imports; the representative near-miss here is enrichment (PO-only,
    a direct handler import). `_bundling_ships_all` must detect the gap so that,
    combined with the "cp ./*.py" pin above,
    test_po_bundling_ships_all_first_party_siblings fails loudly on any such
    revert rather than silently passing.
    """
    siblings = _first_party_sibling_imports(PO_HANDLER)
    assert "enrichment" in siblings  # sanity: a PO-only direct flat-sibling import

    reverted_allowlist_command = (
        "pip install --platform manylinux2014_aarch64 --only-binary=:all: "
        "-r requirements.txt -t /asset-output && "
        "cp handler.py ses_auth.py template_parser.py /asset-output/"
    )

    assert not _bundling_ships_all(reverted_allowlist_command, siblings)

    # Control: the same allowlist shape listing ALL required direct siblings
    # (the Phase 3 shared ses_auth/email_parsing + the Phase 5 flat siblings
    # prompts/telemetry/extraction/enrichment/persistence) is correctly
    # recognized as complete under the accumulate model, proving the failure
    # above is about the missing files and not a regex artifact.
    complete_allowlist_command = (
        "pip install --platform manylinux2014_aarch64 --only-binary=:all: "
        "-r requirements.txt -t /asset-output && "
        "cp handler.py ses_auth.py template_parser.py email_parsing.py "
        "prompts.py telemetry.py extraction.py enrichment.py persistence.py "
        "/asset-output/"
    )
    assert _bundling_ships_all(complete_allowlist_command, siblings)


def test_detection_logic_rejects_narrowed_scoped_glob():
    """A narrowed single-file cp (no `*`) must not satisfy the scoped-glob pin.

    Phase 2 widened the glob's source prefix to `po/email_processor/*.py`
    (resp. `wo/email_processor/*.py`) against the ../lambdas asset root.
    Guard against a narrowing regression -- e.g. a bad find/replace that
    keeps the path prefix but drops the `*` and copies only handler.py --
    from silently passing as ships-all. `cp po/email_processor/handler.py`
    has a path prefix but no glob star, so the scoped-glob regex must not
    match it, and it also fails the explicit-allowlist fallback because it
    omits ses_auth/template_parser/derived_fields.
    """
    siblings = _first_party_sibling_imports(PO_HANDLER)

    narrowed_command = (
        "pip install --platform manylinux2014_aarch64 --only-binary=:all: "
        "-r po/email_processor/requirements.txt -t /asset-output && "
        "cp po/email_processor/handler.py /asset-output/"
    )
    assert not _bundling_ships_all(narrowed_command, siblings)


def test_detection_logic_rejects_commented_out_scoped_glob():
    """A scoped glob surviving only in a `#` comment must not pass.

    Simulates a revert that narrows the executed `cp` to a single file but
    leaves the old scoped-glob text trailing as a comment (e.g. a
    half-finished revert). `_executed_cp_commands` strips `#` comments
    before any regex sees the segment, so the glob text alone -- never
    actually executed by bash -- cannot false-pass the pin.
    """
    siblings = _first_party_sibling_imports(WO_HANDLER)

    commented_out_command = (
        "pip install --platform manylinux2014_aarch64 --only-binary=:all: "
        "-r wo/email_processor/requirements.txt -t /asset-output && "
        "cp wo/email_processor/handler.py /asset-output/  "
        "# was: cp wo/email_processor/*.py /asset-output/"
    )
    assert not _bundling_ships_all(commented_out_command, siblings)


def test_detection_logic_rejects_commented_out_shared_cp():
    """A `cp shared/*.py` surviving only in a `#` comment must not count as run.

    Simulates a half-finished revert that drops the executed shared cp but
    leaves its text trailing as a comment. `_executed_cp_commands` strips `#`
    comments before any regex sees the segment, so the shared-cp text alone --
    never executed by bash -- is not among the executed cp's. Combined with the
    fixed ships-all (ses_auth/email_parsing/emf resolve ONLY under shared/), a
    removed or commented shared cp fails both the SHARED_CP_RE pin and ships-all.
    """
    commented_out_command = (
        "cp po/email_processor/*.py /asset-output/  # cp shared/*.py /asset-output/"
    )
    executed = _executed_cp_commands(commented_out_command)
    assert not any(re.search(SHARED_CP_RE, cp) for cp in executed)
    # And ships-all is False: the shared siblings never got shipped.
    po_siblings = _first_party_sibling_imports(PO_HANDLER)
    assert not _bundling_ships_all(commented_out_command, po_siblings)


def test_detection_logic_rejects_wrong_pipeline_glob():
    """A glob pointing at the WRONG pipeline (or any other dir) must not pass.

    Phase 2 widened the bundling cwd to the shared ../lambdas asset root, so a
    copy-paste slip that leaves po_stack copying `wo/email_processor/*.py`
    would stage a bundle missing derived_fields.py (which lives only under
    po/email_processor) -- a runtime ImportError. `_bundling_ships_all`
    resolves the globbed directory against the lambdas root and confirms it
    actually holds every required sibling, so a wrong-pipeline or
    arbitrary-directory glob is rejected rather than short-circuiting to True
    on the mere presence of a `*.py` token. This is the missing-sibling class
    (PR #105 / PR #2) the AST test exists to catch at CI time.
    """
    po_siblings = _first_party_sibling_imports(PO_HANDLER)
    # enrichment is a direct PO handler import that lives ONLY under
    # po/email_processor (WO has no enrichment stage) -- Phase 5's clean
    # stand-in for derived_fields (now only a transitive import) as the
    # sibling a wrong-pipeline `wo/email_processor/*.py` glob would miss.
    assert "enrichment" in po_siblings  # lives only under po/email_processor

    wrong_pipeline_command = (
        "pip install --platform manylinux2014_aarch64 --only-binary=:all: "
        "-r po/email_processor/requirements.txt -t /asset-output && "
        "cp wo/email_processor/*.py /asset-output/"
    )
    assert not _bundling_ships_all(wrong_pipeline_command, po_siblings)

    arbitrary_dir_command = (
        "pip install --platform manylinux2014_aarch64 --only-binary=:all: "
        "-r po/email_processor/requirements.txt -t /asset-output && "
        "cp venv/lib/*.py /asset-output/"
    )
    assert not _bundling_ships_all(arbitrary_dir_command, po_siblings)

    # The per-stack shape-check pins reject the wrong prefix too: the PO pin
    # matches its own scoped glob but not the WO one, and vice versa.
    assert re.search(PO_SCOPED_GLOB_RE, "cp po/email_processor/*.py /asset-output/")
    assert not re.search(PO_SCOPED_GLOB_RE, "cp wo/email_processor/*.py /asset-output/")
    assert re.search(WO_SCOPED_GLOB_RE, "cp wo/email_processor/*.py /asset-output/")
    assert not re.search(WO_SCOPED_GLOB_RE, "cp po/email_processor/*.py /asset-output/")


def test_po_web_ui_bundle_stages_shared_auth_module():
    """The PO web_ui bundle must EXECUTE `cp shared/web_ui_auth.py`.

    Phase 3 removed the inline auth block from lambdas/po/web_ui/handler.py,
    which now does a module-top-level `from web_ui_auth import is_authenticated`;
    web_ui_auth.py lives ONLY under lambdas/shared/. No test imports the web_ui
    handler, and the email-processor AST pins deliberately exclude the web_ui
    command -- so without this pin, dropping/commenting the web_ui cp would
    ImportError the auth-gated Lambda at cold start with green CI. Checked
    against the comment-stripped executed cp so the old text surviving only in a
    comment cannot satisfy it.
    """
    command = _extract_web_ui_command(PO_STACK)
    executed_cps = _executed_cp_commands(command)
    assert any(re.search(WEB_UI_AUTH_CP_RE, cp) for cp in executed_cps), (
        "cdk/po_stack.py web_ui bundling command must EXECUTE "
        "'cp shared/web_ui_auth.py /asset-output/' so the shared auth module "
        "ships flat beside handler.py and `from web_ui_auth import "
        f"is_authenticated` resolves at cold start. Executed cp commands: "
        f"{executed_cps}"
    )


def test_wo_web_ui_bundle_stages_shared_auth_module():
    """The WO web_ui bundle must EXECUTE `cp shared/web_ui_auth.py`.

    WO equivalent of test_po_web_ui_bundle_stages_shared_auth_module -- same
    ImportError-at-cold-start hazard for lambdas/wo/web_ui/handler.py.
    """
    command = _extract_web_ui_command(WO_STACK)
    executed_cps = _executed_cp_commands(command)
    assert any(re.search(WEB_UI_AUTH_CP_RE, cp) for cp in executed_cps), (
        "cdk/wo_stack.py web_ui bundling command must EXECUTE "
        "'cp shared/web_ui_auth.py /asset-output/' so the shared auth module "
        "ships flat beside handler.py and `from web_ui_auth import "
        f"is_authenticated` resolves at cold start. Executed cp commands: "
        f"{executed_cps}"
    )


def test_detection_logic_rejects_commented_out_web_ui_auth_cp():
    """A `cp shared/web_ui_auth.py` surviving only in a `#` comment must not pass.

    Mirror of test_detection_logic_rejects_commented_out_shared_cp for the
    web_ui staging: a half-finished revert that drops the executed cp but leaves
    its text trailing as a comment must NOT register as executed, since bash
    would never run it.
    """
    commented_out_command = (
        "cp -r po/web_ui/. /asset-output/  # cp shared/web_ui_auth.py /asset-output/"
    )
    executed = _executed_cp_commands(commented_out_command)
    assert not any(re.search(WEB_UI_AUTH_CP_RE, cp) for cp in executed)


def _extract_api_command(stack_path: Path) -> str:
    """Extract the procurement-api function's bash -c bundling command string.

    Mirrors _extract_bundling_command but selects the command whose text
    mentions ``api/`` (its first cp is ``cp api/*.py``). procurement_api_stack
    has exactly one bundled function today; the exactly-one demand makes any
    future second bundle in that stack surface loudly instead of silently
    checking the wrong command.
    """
    tree = ast.parse(stack_path.read_text())
    commands: list[str] = []
    for node in ast.walk(tree):
        if isinstance(node, ast.keyword) and node.arg == "command":
            list_node = node.value
            if isinstance(list_node, ast.List) and list_node.elts:
                last = list_node.elts[-1]
                if isinstance(last, ast.Constant) and isinstance(last.value, str):
                    commands.append(last.value)
    api_cmds = [c for c in commands if "api/" in c]
    if len(api_cmds) != 1:
        raise AssertionError(
            f"expected exactly one api bundling command=[...] list in "
            f"{stack_path}, found {len(api_cmds)} (of {len(commands)} total "
            "command lists) — update this test to select the intended bundling "
            "command explicitly"
        )
    return api_cmds[0]


def test_api_bundling_ships_all_first_party_siblings():
    """The procurement-api bundle must ship every sibling handler.py imports.

    Same PR #105 ImportError class as the email processors: the api handler
    imports router/pagination/serialization/wo_repo/po_repo (next to it) and
    web_ui_auth (lambdas/shared/); a narrowed glob or dropped shared cp would
    cold-start ImportError with green CI.
    """
    required = _first_party_sibling_imports(API_HANDLER)
    assert required, "expected api/handler.py to import first-party siblings"
    command = _extract_api_command(API_STACK)
    assert _bundling_ships_all(command, required), (
        f"cdk/procurement_api_stack.py bundling does not ship all first-party "
        f"siblings {sorted(required)}. Executed cp commands: "
        f"{_executed_cp_commands(command)}"
    )


def test_api_dir_ships_no_unexpected_top_level_modules():
    """Exact-set pin on lambdas/api/*.py -- both bounds.

    `cp api/*.py` ships every top-level .py in the dir (untracked strays
    included, since Code.from_asset does not honor .gitignore), so a stray
    scratch/secrets module would silently ship into the production zip. Any
    new module must be deliberately added to API_PIPELINE_MODULES.
    """
    api_dir = REPO_ROOT / "lambdas" / "api"
    top_level = {p.stem for p in api_dir.glob("*.py")}
    assert top_level == set(API_PIPELINE_MODULES), (
        f"lambdas/api/ top-level modules {sorted(top_level)} != pinned "
        f"{sorted(API_PIPELINE_MODULES)}. If a new module is intended, add it "
        "to API_PIPELINE_MODULES."
    )


def test_api_bundle_stages_shared_auth_module():
    """The api bundle must EXECUTE `cp shared/web_ui_auth.py` (docs-token gate)."""
    command = _extract_api_command(API_STACK)
    executed_cps = _executed_cp_commands(command)
    assert any(re.search(WEB_UI_AUTH_CP_RE, cp) for cp in executed_cps), (
        "cdk/procurement_api_stack.py bundling command must EXECUTE "
        "'cp shared/web_ui_auth.py /asset-output/' so the docs-token gate "
        f"resolves at cold start. Executed cp commands: {executed_cps}"
    )


def test_api_bundle_stages_spec_and_docs_page():
    """The api bundle must EXECUTE cps for openapi.json and docs.html.

    The handler serves both from its package dir; the .py glob does not cover
    them, so each needs its own executed cp or /docs 500s at runtime.
    """
    command = _extract_api_command(API_STACK)
    executed_cps = _executed_cp_commands(command)
    for pattern in API_DATA_FILES_CP_RES:
        assert any(re.search(pattern, cp) for cp in executed_cps), (
            f"cdk/procurement_api_stack.py bundling command must EXECUTE a cp "
            f"matching {pattern!r}. Executed cp commands: {executed_cps}"
        )