procurement-ingest/tests/conftest.py

"""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", "derived_fields")


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",
    )