adam/procurement-ingest
1
0
Fork 0
mirror of https://github.com/Sea-Haven-Industries/procurement-ingest.git synced 2026-09-30 09:33:15 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions
procurement-ingest/tests/test_bundle_consistency.py

623 lines
26 KiB
Python
Raw Blame History

"""Bundle-consistency tests for Terraform Lambda packaging.
Verifies that terraform/build_packages.sh actually ships every first-party
sibling module each handler imports. Guards against a regression where the
copy step 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 file reads + ast on handlers -- no AWS/boto3, no Terraform plan, 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 dataclasses import dataclass, field
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[1]
BUILD_PACKAGES = REPO_ROOT / "terraform" / "build_packages.sh"
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"
SHOC_EMITTER_HANDLER = REPO_ROOT / "lambdas" / "wo" / "shoc_emitter" / "handler.py"
SHOC_ROTATOR_HANDLER = REPO_ROOT / "lambdas" / "wo" / "shoc_hmac_rotator" / "handler.py"
# Phase 3: ses_auth/web_ui_auth/email_parsing/emf are single-sourced here and
# shipped into the email-processor bundles via copy_shared_all.
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", "sentry_init"}
)
# Exact top-level .py stems each dir must hold. Pinned as exact sets (both
# bounds): copy_py_dir ships every top-level .py in these dirs, so a stray
# scratch/secrets .py -- or a shadow copy of a shared module -- would silently
# ship into the production zip. 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.
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 dir + shared).
# 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 copy_shared_all
# (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/): same exact-set discipline.
API_PIPELINE_MODULES = frozenset(
{
"handler",
"router",
"pagination",
"serialization",
"wo_repo",
"po_repo",
}
)
# Non-.py files the api package must also copy: the handler serves the spec,
# docs page, and the vendored Redoc bundle from its own package dir, so dropping
# any copy 500s /docs at runtime with green CI.
API_DATA_FILES = (
"api/openapi.json",
"api/docs.html",
"api/redoc.standalone.js",
"api/fonts.css",
)
# SHOC webhook emitter + HMAC rotator. Same exact-set discipline.
SHOC_EMITTER_MODULES = frozenset({"__init__", "handler", "envelope", "delivery"})
SHOC_ROTATOR_MODULES = frozenset({"__init__", "handler"})
@dataclass
class PackageRecipe:
"""What build_packages.sh copies into one terraform/build/<name> dir."""
py_dirs: list[str] = field(default_factory=list)
shared_all: bool = False
src_files: list[str] = field(default_factory=list)
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
packaging 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.
return {
name
for name in names
if (sibling_dir / f"{name}.py").exists() or (SHARED_DIR / f"{name}.py").exists()
}
_FOR_LOOP_RE = re.compile(
r"for\s+(\w+)\s+in\s+([^;]+);\s*do\s*"
r'copy_src_file\s+"([^"]*)\$\{\1\}([^"]*)"\s+"(\$\{BUILD\}/[^"]*)\$\{\1\}([^"]*)"\s*'
r"done",
re.MULTILINE,
)
def _parse_build_packages(script_text: str | None = None) -> dict[str, PackageRecipe]:
"""Parse copy_* calls from build_packages.sh into per-package recipes.
Strips `#` comments so a commented-out copy cannot false-pass. Expands
simple `for f in a b c; do copy_src_file "api/${f}" ...; done` loops used
for the procurement-api static assets.
"""
text = BUILD_PACKAGES.read_text() if script_text is None else script_text
# Drop full-line and trailing comments before parsing.
cleaned_lines = []
for raw_line in text.splitlines():
cleaned_lines.append(raw_line.split("#", 1)[0])
text = "\n".join(cleaned_lines)
recipes: dict[str, PackageRecipe] = {}
def _pkg(dest: str) -> PackageRecipe:
# dest is "${BUILD}/po_email_processor" or "${BUILD}/po_web_ui/web_ui_auth.py"
m = re.match(r"\$\{BUILD\}/([^/\s\"']+)", dest)
if not m:
raise AssertionError(f"unrecognized build dest: {dest!r}")
name = m.group(1)
return recipes.setdefault(name, PackageRecipe())
# Expand for-loops first so ${f} never lands as a literal src path.
for match in _FOR_LOOP_RE.finditer(text):
items = match.group(2).split()
src_prefix, src_suffix = match.group(3), match.group(4)
dest_prefix, dest_suffix = match.group(5), match.group(6)
for item in items:
dest = f"{dest_prefix}{item}{dest_suffix}"
_pkg(dest).src_files.append(f"{src_prefix}{item}{src_suffix}")
text_without_loops = _FOR_LOOP_RE.sub("", text)
for raw_line in text_without_loops.splitlines():
line = raw_line.strip()
if not line:
continue
m = re.match(
r'copy_py_dir\s+"([^"]+)"\s+"(\$\{BUILD\}/[^"]+)"',
line,
)
if m:
_pkg(m.group(2)).py_dirs.append(m.group(1))
continue
m = re.match(r'copy_shared_all\s+"(\$\{BUILD\}/[^"]+)"', line)
if m:
_pkg(m.group(1)).shared_all = True
continue
m = re.match(
r'copy_src_file\s+"([^"]+)"\s+"(\$\{BUILD\}/[^"]+)"',
line,
)
if m:
_pkg(m.group(2)).src_files.append(m.group(1))
continue
return recipes
def _shipped_modules(recipe: PackageRecipe) -> set[str]:
"""Module stems a PackageRecipe would place at the zip root."""
shipped: set[str] = set()
for rel_dir in recipe.py_dirs:
glob_dir = REPO_ROOT / "lambdas" / rel_dir
shipped.update(p.stem for p in glob_dir.glob("*.py"))
if recipe.shared_all:
shipped.update(p.stem for p in SHARED_DIR.glob("*.py"))
for rel in recipe.src_files:
if rel.endswith(".py"):
shipped.add(Path(rel).stem)
return shipped
def _packaging_ships_all(recipe: PackageRecipe, sibling_names: set[str]) -> bool:
"""True if recipe is guaranteed to ship every name in sibling_names."""
if not recipe.py_dirs and not recipe.shared_all and not recipe.src_files:
return False
shipped = _shipped_modules(recipe)
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"
recipes = _parse_build_packages()
recipe = recipes["po_email_processor"]
assert "po/email_processor" in recipe.py_dirs, (
"terraform/build_packages.sh must copy_py_dir po/email_processor into "
"po_email_processor so every first-party sibling handler.py imports "
f"ships automatically. Recipe: {recipe}"
)
assert recipe.shared_all, (
"terraform/build_packages.sh must copy_shared_all into "
"po_email_processor so the single-sourced shared modules ship flat "
f"beside handler.py. Recipe: {recipe}"
)
assert _packaging_ships_all(recipe, siblings), (
f"po_email_processor packaging does not ship all of {sorted(siblings)}: "
f"{recipe}"
)
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"
recipes = _parse_build_packages()
recipe = recipes["wo_email_processor"]
assert "wo/email_processor" in recipe.py_dirs, (
"terraform/build_packages.sh must copy_py_dir wo/email_processor into "
"wo_email_processor so every first-party sibling ships while tests/ "
f"and requirements.txt are excluded. Recipe: {recipe}"
)
assert recipe.shared_all, (
"terraform/build_packages.sh must copy_shared_all into "
"wo_email_processor so the single-sourced shared modules ship flat "
f"beside handler.py. Recipe: {recipe}"
)
assert _packaging_ships_all(recipe, siblings), (
f"wo_email_processor packaging does not ship all of {sorted(siblings)}: "
f"{recipe}"
)
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 packaging ships every module the
handler needs (shipped >= required). This proves the other direction for
the pipeline dir (shipped <= expected): because copy_py_dir copies every
top-level .py in that dir, a stray scratch or secrets .py, OR a shadow
copy of a moved shared module, would silently ship into the zip. 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 -- "
"copy_py_dir would ship exactly these into the Lambda zip. Found "
f"{sorted(top_level)}, expected {sorted(PO_PIPELINE_MODULES)}. A moved "
f"shared module ({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/)."""
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 -- "
"copy_py_dir would ship exactly these into the Lambda zip. Found "
f"{sorted(top_level)}, expected {sorted(WO_PIPELINE_MODULES)}. A moved "
f"shared module ({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).
copy_shared_all 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 single-sourced shared 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 -- copy_shared_all "
"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/."""
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 `_packaging_ships_all` itself.
Simulates the historical regression shape: packaging copies only an
explicit filename set that omits a required sibling. Phase 5 note: the
PR #2 near-miss module (derived_fields) is now a TRANSITIVE import via
enrichment.py; the representative near-miss here is enrichment (PO-only,
a direct handler import).
"""
siblings = _first_party_sibling_imports(PO_HANDLER)
assert "enrichment" in siblings # sanity: a PO-only direct flat-sibling import
incomplete = PackageRecipe(
src_files=[
"po/email_processor/handler.py",
"shared/ses_auth.py",
"po/email_processor/template_parser.py",
]
)
assert not _packaging_ships_all(incomplete, siblings)
# Control: explicit files covering all required direct siblings is complete.
complete = PackageRecipe(
src_files=[
"po/email_processor/handler.py",
"shared/ses_auth.py",
"po/email_processor/template_parser.py",
"shared/email_parsing.py",
"po/email_processor/prompts.py",
"po/email_processor/telemetry.py",
"po/email_processor/extraction.py",
"po/email_processor/enrichment.py",
"po/email_processor/persistence.py",
"shared/sentry_init.py",
]
)
assert _packaging_ships_all(complete, siblings)
def test_detection_logic_rejects_narrowed_py_dir():
"""A recipe that only copies handler.py must not satisfy ships-all."""
siblings = _first_party_sibling_imports(PO_HANDLER)
narrowed = PackageRecipe(src_files=["po/email_processor/handler.py"])
assert not _packaging_ships_all(narrowed, siblings)
def test_detection_logic_rejects_commented_out_shared_copy():
"""A copy_shared_all surviving only in a `#` comment must not count as run."""
script = (
'copy_py_dir "po/email_processor" "${BUILD}/po_email_processor"\n'
'# copy_shared_all "${BUILD}/po_email_processor"\n'
)
recipes = _parse_build_packages(script)
recipe = recipes["po_email_processor"]
assert not recipe.shared_all
po_siblings = _first_party_sibling_imports(PO_HANDLER)
assert not _packaging_ships_all(recipe, po_siblings)
def test_detection_logic_rejects_wrong_pipeline_dir():
"""A copy_py_dir pointing at the WRONG pipeline must not pass ships-all.
A slip that leaves po_email_processor copying wo/email_processor would
stage a bundle missing enrichment.py (PO-only) -- a runtime ImportError.
"""
po_siblings = _first_party_sibling_imports(PO_HANDLER)
assert "enrichment" in po_siblings # lives only under po/email_processor
wrong = PackageRecipe(py_dirs=["wo/email_processor"])
assert not _packaging_ships_all(wrong, po_siblings)
arbitrary = PackageRecipe(py_dirs=["venv/lib"])
assert not _packaging_ships_all(arbitrary, po_siblings)
def test_po_web_ui_bundle_stages_shared_auth_module():
"""The PO web_ui package must copy 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/. Dropping this copy would
ImportError the auth-gated Lambda at cold start with green CI.
"""
recipes = _parse_build_packages()
recipe = recipes["po_web_ui"]
assert "po/web_ui" in recipe.py_dirs, (
f"po_web_ui must copy_py_dir po/web_ui. Recipe: {recipe}"
)
assert "shared/web_ui_auth.py" in recipe.src_files, (
"terraform/build_packages.sh must copy_src_file shared/web_ui_auth.py "
"into po_web_ui so `from web_ui_auth import is_authenticated` resolves "
f"at cold start. Recipe: {recipe}"
)
def test_wo_web_ui_bundle_stages_shared_auth_module():
"""The WO web_ui package must copy shared/web_ui_auth.py."""
recipes = _parse_build_packages()
recipe = recipes["wo_web_ui"]
assert "wo/web_ui" in recipe.py_dirs, (
f"wo_web_ui must copy_py_dir wo/web_ui. Recipe: {recipe}"
)
assert "shared/web_ui_auth.py" in recipe.src_files, (
"terraform/build_packages.sh must copy_src_file shared/web_ui_auth.py "
"into wo_web_ui so `from web_ui_auth import is_authenticated` resolves "
f"at cold start. Recipe: {recipe}"
)
def test_detection_logic_rejects_commented_out_web_ui_auth_cp():
"""A web_ui_auth copy surviving only in a `#` comment must not pass."""
script = (
'copy_py_dir "po/web_ui" "${BUILD}/po_web_ui"\n'
'# copy_src_file "shared/web_ui_auth.py" "${BUILD}/po_web_ui/web_ui_auth.py"\n'
)
recipes = _parse_build_packages(script)
recipe = recipes["po_web_ui"]
assert "shared/web_ui_auth.py" not in recipe.src_files
def test_api_bundling_ships_all_first_party_siblings():
"""The procurement-api package must ship every sibling handler.py imports."""
required = _first_party_sibling_imports(API_HANDLER)
assert required, "expected api/handler.py to import first-party siblings"
recipes = _parse_build_packages()
recipe = recipes["procurement_api"]
assert _packaging_ships_all(recipe, required), (
f"procurement_api packaging does not ship all first-party siblings "
f"{sorted(required)}. Recipe: {recipe}"
)
def test_api_dir_ships_no_unexpected_top_level_modules():
"""Exact-set pin on lambdas/api/*.py -- both bounds."""
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 package must copy shared/web_ui_auth.py (docs-token gate)."""
recipes = _parse_build_packages()
recipe = recipes["procurement_api"]
assert "shared/web_ui_auth.py" in recipe.src_files, (
"terraform/build_packages.sh must copy_src_file shared/web_ui_auth.py "
"into procurement_api so the docs-token gate resolves at cold start. "
f"Recipe: {recipe}"
)
def test_api_bundle_stages_spec_and_docs_page():
"""The api package must copy openapi.json, docs.html, and Redoc assets."""
recipes = _parse_build_packages()
recipe = recipes["procurement_api"]
for rel in API_DATA_FILES:
assert rel in recipe.src_files, (
f"terraform/build_packages.sh must copy_src_file {rel!r} into "
f"procurement_api. Recipe: {recipe}"
)
def test_shoc_emitter_bundling_ships_all_first_party_siblings():
"""The SHOC emitter package must ship every sibling handler.py imports."""
required = _first_party_sibling_imports(SHOC_EMITTER_HANDLER)
assert required == {"envelope", "delivery", "sentry_init"}, (
f"expected the emitter handler's first-party siblings to be envelope + "
f"delivery + sentry_init, found {sorted(required)} — update this pin "
"deliberately"
)
recipes = _parse_build_packages()
recipe = recipes["wo_shoc_emitter"]
assert "wo/shoc_emitter" in recipe.py_dirs, (
f"wo_shoc_emitter must copy_py_dir wo/shoc_emitter. Recipe: {recipe}"
)
assert _packaging_ships_all(recipe, required), (
f"wo_shoc_emitter packaging does not ship all first-party siblings "
f"{sorted(required)}. Recipe: {recipe}"
)
def test_shoc_rotator_bundling_ships_handler():
"""The rotator package must ship handler.py and shared sentry_init."""
required = _first_party_sibling_imports(SHOC_ROTATOR_HANDLER)
assert required == {"sentry_init"}, (
f"the rotator handler first-party siblings {sorted(required)} — "
"expected only sentry_init; extend this ships-all test if more appear"
)
recipes = _parse_build_packages()
recipe = recipes["wo_shoc_hmac_rotator"]
assert "wo/shoc_hmac_rotator" in recipe.py_dirs, (
f"wo_shoc_hmac_rotator must copy_py_dir wo/shoc_hmac_rotator. Recipe: {recipe}"
)
assert _packaging_ships_all(recipe, {"handler", "sentry_init"}), (
f"wo_shoc_hmac_rotator packaging does not ship handler.py + sentry_init. "
f"Recipe: {recipe}"
)
def test_shoc_emitter_dir_ships_no_unexpected_top_level_modules():
"""Exact-set pin on lambdas/wo/shoc_emitter/*.py -- both bounds."""
top_level = {p.stem for p in SHOC_EMITTER_HANDLER.parent.glob("*.py")}
assert top_level == set(SHOC_EMITTER_MODULES), (
f"lambdas/wo/shoc_emitter/ top-level modules {sorted(top_level)} != "
f"pinned {sorted(SHOC_EMITTER_MODULES)}. If a new module is intended, "
"add it to SHOC_EMITTER_MODULES."
)
def test_shoc_rotator_dir_ships_no_unexpected_top_level_modules():
"""Exact-set pin on lambdas/wo/shoc_hmac_rotator/*.py -- both bounds."""
top_level = {p.stem for p in SHOC_ROTATOR_HANDLER.parent.glob("*.py")}
assert top_level == set(SHOC_ROTATOR_MODULES), (
f"lambdas/wo/shoc_hmac_rotator/ top-level modules {sorted(top_level)} "
f"!= pinned {sorted(SHOC_ROTATOR_MODULES)}. If a new module is "
"intended, add it to SHOC_ROTATOR_MODULES."
)
def test_email_processor_packages_do_not_collide_with_web_ui_dirs():
"""Email-processor recipes must not copy web_ui dirs (and vice versa)."""
recipes = _parse_build_packages()
for name in ("po_email_processor", "wo_email_processor"):
recipe = recipes[name]
assert not any("web_ui" in d for d in recipe.py_dirs), (
f"{name} packaging unexpectedly copies a web_ui dir: {recipe}"
)
for name in ("po_web_ui", "wo_web_ui"):
recipe = recipes[name]
assert not any("email_processor" in d for d in recipe.py_dirs), (
f"{name} packaging unexpectedly copies an email_processor dir: {recipe}"
)
def test_non_email_processor_packages_copy_sentry_init():
"""Functions that do not copy_shared_all must copy sentry_init by name."""
recipes = _parse_build_packages()
for name in (
"po_web_ui",
"wo_web_ui",
"procurement_api",
"po_site_extractor",
"wo_shoc_emitter",
"wo_shoc_hmac_rotator",
):
recipe = recipes[name]
assert "shared/sentry_init.py" in recipe.src_files, (
f"terraform/build_packages.sh must copy_src_file shared/sentry_init.py "
f"into {name} so `import sentry_init` resolves at cold start. "
f"Recipe: {recipe}"
)
Reference in a new issue View git blame Copy permalink