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"
PO_STACK = REPO_ROOT / "cdk" / "po_stack.py"
WO_STACK = REPO_ROOT / "cdk" / "wo_stack.py"

# The complete set of top-level modules the PO email_processor ships via the
# non-recursive `cp ./*.py` glob. Pinned as an upper bound: a new top-level
# .py in that dir must be added here deliberately, which is the moment to
# decide whether it SHOULD ship (a real module) or must be excluded (a scratch
# or secrets file that from_asset would otherwise stage into the bundle on a
# local deploy). See test_po_bundle_ships_no_unexpected_top_level_modules.
PO_EXPECTED_TOP_LEVEL_MODULES = frozenset(
    {"handler", "ses_auth", "template_parser", "derived_fields"}
)

# 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|$)"


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
    return {name for name in names if (sibling_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.

    Each stack currently has exactly one bundled function; if a second one
    is ever added, "the" bundling command becomes ambiguous and every
    assertion in this file needs to pick its target explicitly — so demand
    exactly one match rather than silently returning whichever ast.walk
    happens to visit first.
    """
    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)
    if len(commands) != 1:
        raise AssertionError(
            f"expected exactly one bundling command=[...] list in {stack_path}, "
            f"found {len(commands)} — update this test to select the intended "
            "bundling command explicitly"
        )
    return commands[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`). An executed copy ships every top-level .py
    sibling unconditionally in two shapes:
      - a non-recursive glob copy whose (optional) path prefix resolves, under
        the ../lambdas bundling cwd, to a directory that ACTUALLY contains
        every required sibling, e.g. `cp po/email_processor/*.py /asset-output/`
        (PO, Phase 2). The directory is filesystem-checked, not merely
        pattern-matched: a wrong-pipeline or arbitrary-directory glob
        (`cp wo/email_processor/*.py` in po_stack, `cp venv/lib/*.py`) does NOT
        qualify, because that directory does not hold every required sibling --
        so it can never short-circuit to True while dropping a module;
      - a recursive copy of the WHOLE source dir, e.g. `cp -r . /asset-output/`
        -- the source operand must be `.`/`./`, so a narrowed recursive
        copy like `cp -r ./package /asset-output/` does NOT qualify.
    Any other executed `cp` is treated as an explicit filename allowlist (the
    legacy PO form) and is safe only if every required `<name>.py` literally
    appears among the copied filenames -- the branch that must reject a
    reverted allowlist missing a sibling.
    """
    cp_cmds = _executed_cp_commands(command)
    if not cp_cmds:
        return False
    copied_files: set[str] = set()
    for cp in cp_cmds:
        glob_match = re.search(r"(?:^|\s)((?:[\w./-]+/)?)\*\.py(?:\s|$)", cp)
        if glob_match:
            # A `*.py` glob ships every top-level .py in the globbed directory
            # -- but ONLY that directory. Resolve its prefix against the
            # ../lambdas asset root (the Phase 2 bundling cwd) and confirm it
            # actually holds every required sibling, so a wrong-pipeline glob
            # cannot pass the ships-all check on the mere presence of a `*.py`.
            glob_dir = (REPO_ROOT / "lambdas" / glob_match.group(1)).resolve()
            if all((glob_dir / f"{name}.py").exists() for name in sibling_names):
                return True
            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:
            copied_files.update(match.group(1).split())
    return all(f"{name}.py" in copied_files 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}"
    )
    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}"
    )
    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_bundle_ships_no_unexpected_top_level_modules():
    """Upper bound: the PO glob ships EXACTLY the expected top-level modules.

    The sibling-import tests above prove the glob ships every module the
    handler needs (shipped >= required). This proves the other direction
    (shipped <= expected): because `cp ./*.py` copies every top-level .py in
    the dir -- and `Code.from_asset` stages untracked files too (it does not
    honor .gitignore) -- a stray scratch/secrets .py left in this dir would
    silently ship into the production zip on a local deploy. Pinning the set
    forces any new top-level module to be added to
    PO_EXPECTED_TOP_LEVEL_MODULES deliberately, at which point the author
    decides whether it should ship or be excluded from bundling.
    """
    top_level = {p.stem for p in PO_HANDLER.parent.glob("*.py")}
    assert top_level == set(PO_EXPECTED_TOP_LEVEL_MODULES), (
        "unexpected top-level .py set in lambdas/po/email_processor -- the "
        "'cp ./*.py' glob would ship exactly these into the Lambda zip. "
        f"Found {sorted(top_level)}, expected "
        f"{sorted(PO_EXPECTED_TOP_LEVEL_MODULES)}. If a new module is "
        "intended, add it to PO_EXPECTED_TOP_LEVEL_MODULES; if it is a scratch "
        "or secrets file, remove it (or exclude it from bundling) before deploy."
    )


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
    derived_fields.py (the near-miss from PR #2). `_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 "derived_fields" in siblings  # sanity: this is the PR #2 near-miss module

    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 WITH derived_fields.py added back in
    # is correctly recognized as complete, proving the failure above is
    # about the missing file and not a false-positive-prone regex.
    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 derived_fields.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_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)
    assert "derived_fields" 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/")