mirror of
https://github.com/Sea-Haven-Industries/procurement-ingest.git
synced 2026-10-07 16:08:54 +00:00
Adam Moussa
14fad0c317
132 tests: golden-file comparison for all 25 positive fixtures (17 single-line new-PO + 8 cancellations, Decimal-exact via parse_float=Decimal), every fail-closed gate reason code covered (body-level triggers via 17 synthetic adversarial .eml mutations, candidate-level via direct validate() unit tests), real multi-line and comment/non-Coupa fallback fixtures, dual line-ending parse identity, two-path enrich/save parity (site-extractor stream guard), V10 URL-id corpus sweep, fixture hygiene (ses_auth pass + scrub-marker leak sweep), and Bedrock dispatch/EMF assertions. The suite loads handler/template_parser via importlib under unique module names and binds the handler's bare sibling imports around exec (tests/conftest.py load_handler gets the same treatment) -- the WO suite caches bare 'handler'/'template_parser' names in sys.modules, and bare imports here would silently bind to the wrong pipeline. moto is imported before the handler so its botocore stubber hook precedes boto3 session creation (the PO conftest chain now loads at pytest session start). Fixtures are scrubbed real S3 samples: transport/auth header values replaced with same-shape placeholders (structure kept so ses_auth still passes), per-file digit ciphers, amounts remapped with sum==total re-established. The .gitignore exception is scoped to the PO fixtures path only. Refs: #99
119 lines
4.4 KiB
Python
119 lines
4.4 KiB
Python
"""Shared pytest configuration for the procurement-ingest test suite.
|
|
|
|
The Lambda handlers create boto3 clients at module import time, so a
|
|
region and dummy credentials must be present in the environment before
|
|
any handler module is imported. Setting them here at conftest import
|
|
time guarantees they exist before test collection touches a handler.
|
|
"""
|
|
|
|
import importlib.util
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
os.environ.setdefault("AWS_DEFAULT_REGION", "us-east-1")
|
|
os.environ.setdefault("AWS_ACCESS_KEY_ID", "testing")
|
|
os.environ.setdefault("AWS_SECRET_ACCESS_KEY", "testing")
|
|
os.environ.setdefault("AWS_SESSION_TOKEN", "testing")
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
# Sibling modules that exist per-pipeline and are imported by bare name from
|
|
# the handlers (the Lambda runtime puts each function's own directory on
|
|
# sys.path). Both pipelines duplicate these filenames, so the bare names MUST
|
|
# be bound to the right pipeline's file around each handler exec -- relying on
|
|
# sys.path ordering (or on whatever a previously collected suite left in
|
|
# sys.modules) silently binds a handler to the OTHER pipeline's sibling.
|
|
_SIBLING_MODULES = ("ses_auth", "template_parser")
|
|
|
|
|
|
def _load_module(path, module_name):
|
|
if module_name in sys.modules:
|
|
return sys.modules[module_name]
|
|
spec = importlib.util.spec_from_file_location(module_name, path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[module_name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def load_handler(relative_path, module_name):
|
|
"""Load a Lambda handler module by file path under a unique name.
|
|
|
|
The handler files all share the basename ``handler.py`` and are not
|
|
importable as packages, so a plain ``import handler`` would collide
|
|
across Lambdas. The same loader serves the ``ses_auth.py`` modules,
|
|
which are likewise duplicated per pipeline and not importable as
|
|
packages.
|
|
|
|
Handler modules import their siblings by bare name (e.g. ``from
|
|
template_parser import try_deterministic_parse``). Each sibling is loaded
|
|
from the handler's own directory under a unique module name and registered
|
|
under its bare name only for the duration of the handler exec, then the
|
|
previous binding is restored -- so this loader is deterministic regardless
|
|
of collection order and of what the per-Lambda test suites (which put
|
|
their own module dir on sys.path) have already cached in sys.modules.
|
|
"""
|
|
path = REPO_ROOT / relative_path
|
|
if module_name in sys.modules:
|
|
return sys.modules[module_name]
|
|
# Keep the handler dir on sys.path for parity with the Lambda runtime.
|
|
handler_dir = str(path.parent)
|
|
if handler_dir not in sys.path:
|
|
sys.path.insert(0, handler_dir)
|
|
if path.name != "handler.py":
|
|
# Leaf modules (e.g. ses_auth.py itself) have no sibling imports.
|
|
return _load_module(path, module_name)
|
|
saved = {}
|
|
for sibling in _SIBLING_MODULES:
|
|
sibling_path = path.parent / f"{sibling}.py"
|
|
if not sibling_path.exists():
|
|
continue
|
|
saved[sibling] = sys.modules.get(sibling)
|
|
sys.modules[sibling] = _load_module(sibling_path, f"{module_name}__{sibling}")
|
|
try:
|
|
module = _load_module(path, module_name)
|
|
finally:
|
|
for sibling, previous in saved.items():
|
|
if previous is not None:
|
|
sys.modules[sibling] = previous
|
|
else:
|
|
sys.modules.pop(sibling, None)
|
|
return module
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def po_handler():
|
|
"""The PO email processor handler module."""
|
|
return load_handler(
|
|
"lambdas/po/email_processor/handler.py",
|
|
"po_email_processor_handler",
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def wo_handler():
|
|
"""The WO email processor handler module."""
|
|
return load_handler(
|
|
"lambdas/wo/email_processor/handler.py",
|
|
"wo_email_processor_handler",
|
|
)
|
|
|
|
|
|
@pytest.fixture(params=["po_handler", "wo_handler"])
|
|
def email_handler(request):
|
|
"""Parametrized fixture yielding each email processor handler module."""
|
|
return request.getfixturevalue(request.param)
|
|
|
|
|
|
@pytest.fixture(params=["wo", "po"])
|
|
def ses_auth(request):
|
|
"""The ses_auth module of each pipeline (duplicated file, kept in sync)."""
|
|
pipeline = request.param
|
|
return load_handler(
|
|
f"lambdas/{pipeline}/email_processor/ses_auth.py",
|
|
f"{pipeline}_ses_auth",
|
|
)
|