mirror of
https://github.com/Sea-Haven-Industries/procurement-ingest.git
synced 2026-10-07 16:08:54 +00:00
Adam Moussa
042a03b3a4
Bring the two post-fork main merges into the feature branch: #97 (PO revision/cancellation conditional-write logic, web-UI auth/XSS, KMS-grant removal, Secrets Manager hardening) and #98 (fail-closed SES sender authentication). Conflicts resolved by union so all prior work survives: - wo/po email_processor handlers: authenticate first (#98), then parse; WO keeps template-first parse + Bedrock fallback + EMF metric + issue #23 comment_id fix; PO keeps #97 conditional writes on top of the Bedrock provider switch. The retired Anthropic Secrets Manager client is dropped (Bedrock migration) rather than hardened. - cdk stacks: union of #97 web-ui secret handling + KMS-grant removal, #98 ALLOWED_DKIM_DOMAINS env + sender-auth-rejected alarm, and the branch's Bedrock dual-ARN grant, anthropic-secret removal, and template-fallback alarm. Restore the aws_secretsmanager import that the auto-merge silently dropped from both stacks. - README, conftest, requirements, .gitignore: union of all suites/docs. - test_bedrock_fallback: bypass the new INFRA-107 auth gate in the two dispatch tests (auth is covered by test_ses_auth). Refs: #99
84 lines
2.9 KiB
Python
84 lines
2.9 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]
|
|
|
|
|
|
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.
|
|
|
|
The Lambda runtime puts each function's own directory on ``sys.path``,
|
|
so handler modules import their siblings by bare name (e.g.
|
|
``from ses_auth import authenticate_inbound_email``). The function
|
|
directory is added to ``sys.path`` here so the module executes the
|
|
same way under test as it does in the Lambda.
|
|
"""
|
|
path = REPO_ROOT / relative_path
|
|
# Put the handler's own directory on sys.path so its sibling-module imports
|
|
# (e.g. the WO handler's ``from template_parser import ...`` and the
|
|
# ``from ses_auth import ...`` in both handlers) resolve.
|
|
handler_dir = str(path.parent)
|
|
if handler_dir not in sys.path:
|
|
sys.path.insert(0, handler_dir)
|
|
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
|
|
|
|
|
|
@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",
|
|
)
|