"""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] _SHARED_DIR = REPO_ROOT / "lambdas" / "shared" # Sibling modules imported by bare name from the handlers (the Lambda runtime # puts each function's own directory on sys.path; the CDK bundling then cp's the # shared modules in flat beside handler.py so those bare imports resolve too). # template_parser/derived_fields are duplicated PER PIPELINE, so their 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. # ses_auth/email_parsing/emf are now single-sourced under lambdas/shared/ (Phase # 3); the loop below resolves them from there via a shared-dir fallback. _SIBLING_MODULES = ( "ses_auth", "template_parser", "derived_fields", "email_parsing", "emf", ) 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: # Per-pipeline siblings (template_parser/derived_fields) resolve next to # the handler; the shared, single-sourced siblings (ses_auth/ # email_parsing/emf) fall back to lambdas/shared/. No ambiguity: post # Phase 3 the shared names exist ONLY under shared/, the per-pipeline # names ONLY next to the handler. sibling_path = path.parent / f"{sibling}.py" if not sibling_path.exists(): sibling_path = _SHARED_DIR / 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 def ses_auth(): """The single-sourced ses_auth module (lambdas/shared/, Phase 3). Previously parameterized over the two per-pipeline copies to prove they stayed byte-identical; now there is exactly one copy, so this loads it once -- halving the test_ses_auth run. """ return load_handler("lambdas/shared/ses_auth.py", "shared_ses_auth")