"""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 `/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 `.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/")