procurement-ingest/tests/test_shoc_emitter_delivery.py

"""Signing + delivery tests for the SHOC webhook emitter (plan Phase 5).

Golden vectors: docs/shoc-webhook-test-vectors.json is the shared handoff
artifact -- Luby's receiver verifies against the same vectors -- and BOTH
producer-side sign_body implementations (the emitter's delivery module and
the replay script) are pinned here against every vector, so they can never
drift from each other or from the published vectors.

Also covers the contract section 7 response-classification matrix (2xx /
429+5xx / timeout+connection error / other 4xx) with a monkeypatched urllib
opener, and the Secrets Manager key cache: 5-minute TTL via time.monotonic,
keys[0] selection, empty-keys (bootstrap) -> retryable.
"""

import importlib.util
import io
import json
import urllib.error
from pathlib import Path

import pytest

from tests.support import REPO_ROOT, load_lambda_module

_VECTORS_PATH = Path(REPO_ROOT) / "docs" / "shoc-webhook-test-vectors.json"
VECTORS = json.loads(_VECTORS_PATH.read_text(encoding="utf-8"))["vectors"]

TEST_KEYS = [
    {"kid": "2026-07-20T00", "secret": "ab" * 32},
    {"kid": "2026-06-20T00", "secret": "cd" * 32},
]

ENVELOPE = {
    "schema_version": 1,
    "delivery_id": "evt-1",
    "event_type": "work_order.created",
    "occurred_at": "2026-07-16T14:03:22.114208+00:00",
    "source": "procurement-ingest/workorder-shoc-emitter",
    "replay": False,
    "data": {"work_order_id": "11144580730"},
}


@pytest.fixture(scope="module")
def delivery():
    return load_lambda_module("wo", "shoc_emitter/delivery")


def _load_replay_script():
    # scripts/ is not a package and not on sys.path; load by file path
    # (mirrors tests/test_reprocess_contract.py). Import is side-effect-free:
    # the script builds its boto3 session inside main().
    path = Path(REPO_ROOT) / "scripts" / "replay_shoc_webhooks.py"
    spec = importlib.util.spec_from_file_location(
        "replay_shoc_webhooks_for_vectors", path
    )
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


# --- Golden vectors ----------------------------------------------------------


def test_vector_file_covers_required_shapes():
    # The handoff artifact itself must keep its coverage promises: at least
    # one non-ASCII UTF-8 body (multi-byte signing) and one empty-object body.
    assert len(VECTORS) >= 4
    assert any(any(ord(ch) > 127 for ch in v["body"]) for v in VECTORS)
    assert any(v["body"] == "{}" for v in VECTORS)
    for vector in VECTORS:
        assert set(vector) == {
            "kid",
            "secret_hex",
            "timestamp",
            "body",
            "expected_signature",
        }


def test_delivery_sign_body_matches_golden_vectors(delivery):
    for vector in VECTORS:
        signature = delivery.sign_body(
            vector["secret_hex"],
            vector["timestamp"],
            vector["body"].encode("utf-8"),
        )
        assert signature == vector["expected_signature"], (
            f"delivery.sign_body drifted from golden vector kid="
            f"{vector['kid']} ts={vector['timestamp']}"
        )


def test_replay_sign_body_matches_golden_vectors():
    replay = _load_replay_script()
    for vector in VECTORS:
        signature = replay.sign_body(
            vector["secret_hex"],
            vector["timestamp"],
            vector["body"].encode("utf-8"),
        )
        assert signature == vector["expected_signature"], (
            f"replay sign_body drifted from golden vector kid="
            f"{vector['kid']} ts={vector['timestamp']}"
        )


# --- Response classification matrix (contract section 7) ---------------------


class _FakeResponse:
    def __init__(self, status):
        self.status = status

    def __enter__(self):
        return self

    def __exit__(self, *exc_info):
        return False


@pytest.fixture
def signing_keys(monkeypatch, delivery):
    monkeypatch.setattr(delivery, "_get_hmac_keys", lambda: TEST_KEYS)


def _patch_urlopen(monkeypatch, delivery, fn):
    # deliver() posts through the no-redirect opener, not the module-level
    # urlopen, so patch the opener's open method.
    monkeypatch.setattr(delivery._opener, "open", fn)


@pytest.mark.parametrize("status", [200, 204])
def test_2xx_is_delivered(monkeypatch, delivery, signing_keys, status):
    _patch_urlopen(
        monkeypatch, delivery, lambda request, timeout: _FakeResponse(status)
    )
    assert delivery.deliver(ENVELOPE) == ("delivered", status)


@pytest.mark.parametrize("status", [429, 500, 503])
def test_429_and_5xx_raise_retryable(monkeypatch, delivery, signing_keys, status):
    def _raise(request, timeout):
        raise urllib.error.HTTPError(
            delivery.SHOC_WEBHOOK_URL, status, "boom", None, io.BytesIO(b"")
        )

    _patch_urlopen(monkeypatch, delivery, _raise)
    with pytest.raises(delivery.RetryableDeliveryError) as exc:
        delivery.deliver(ENVELOPE)
    assert exc.value.status_code == status


@pytest.mark.parametrize(
    "error",
    [urllib.error.URLError(OSError("connection refused")), TimeoutError()],
    ids=["connection-error", "timeout"],
)
def test_connection_failures_raise_retryable(
    monkeypatch, delivery, signing_keys, error
):
    def _raise(request, timeout):
        raise error

    _patch_urlopen(monkeypatch, delivery, _raise)
    with pytest.raises(delivery.RetryableDeliveryError) as exc:
        delivery.deliver(ENVELOPE)
    assert exc.value.status_code is None


@pytest.mark.parametrize("status", [400, 404, 422])
def test_other_4xx_is_rejected_not_raised(monkeypatch, delivery, signing_keys, status):
    def _raise(request, timeout):
        raise urllib.error.HTTPError(
            delivery.SHOC_WEBHOOK_URL, status, "bad", None, io.BytesIO(b"")
        )

    _patch_urlopen(monkeypatch, delivery, _raise)
    assert delivery.deliver(ENVELOPE) == ("rejected", status)


@pytest.mark.parametrize("status", [401, 403])
def test_auth_failures_are_retryable_and_invalidate_cache(
    monkeypatch, delivery, signing_keys, status
):
    # A transient auth failure (stale cached key mid-rotation, receiver
    # secret-fetch blip, clock skew) must retry in order -- NOT park -- and
    # drop the key cache so the retry re-signs with the current secret.
    invalidated = {"called": False}

    def _mark(*_a, **_k):
        invalidated["called"] = True

    monkeypatch.setattr(delivery, "_invalidate_hmac_keys", _mark)

    def _raise(request, timeout):
        raise urllib.error.HTTPError(
            delivery.SHOC_WEBHOOK_URL, status, "unauthorized", None, io.BytesIO(b"")
        )

    _patch_urlopen(monkeypatch, delivery, _raise)
    with pytest.raises(delivery.RetryableDeliveryError) as exc:
        delivery.deliver(ENVELOPE)
    assert exc.value.status_code == status
    assert invalidated["called"] is True


def test_redirects_are_not_followed():
    # The opener must refuse 3xx so auth headers are never forwarded to a
    # receiver-chosen Location. redirect_request returning None makes urllib
    # raise instead of following.
    delivery = load_lambda_module("wo", "shoc_emitter/delivery")
    handler = delivery._NoRedirectHandler()
    assert handler.redirect_request(None, None, 302, "Found", {}, "http://evil") is None


def test_request_signed_with_keys0_and_contract_headers(
    monkeypatch, delivery, signing_keys
):
    captured = {}

    def _capture(request, timeout):
        captured["request"] = request
        captured["timeout"] = timeout
        return _FakeResponse(200)

    _patch_urlopen(monkeypatch, delivery, _capture)
    assert delivery.deliver(ENVELOPE) == ("delivered", 200)

    request = captured["request"]
    assert captured["timeout"] == delivery.POST_TIMEOUT_SECONDS
    assert request.get_method() == "POST"
    assert request.data == json.dumps(ENVELOPE).encode("utf-8")
    assert request.get_header("Content-type") == "application/json; charset=utf-8"
    assert request.get_header("User-agent") == "workorder-shoc-emitter/1"
    # The producer always signs with keys[0] (contract section 6.1).
    assert request.get_header("X-sh-key-id") == TEST_KEYS[0]["kid"]
    timestamp = request.get_header("X-sh-timestamp")
    assert timestamp.isdigit()
    expected = delivery.sign_body(TEST_KEYS[0]["secret"], int(timestamp), request.data)
    assert request.get_header("X-sh-signature") == f"v1={expected}"


# --- Secret cache ------------------------------------------------------------


class _FakeSecretsManager:
    """Returns payloads in sequence (last one repeats); counts fetches."""

    def __init__(self, payloads):
        self.payloads = list(payloads)
        self.calls = 0
        self.secret_ids = []

    def get_secret_value(self, SecretId):  # noqa: N803 (boto3 kwarg name)
        self.calls += 1
        self.secret_ids.append(SecretId)
        payload = self.payloads.pop(0) if len(self.payloads) > 1 else self.payloads[0]
        return {"SecretString": json.dumps(payload)}


@pytest.fixture
def cache_reset(monkeypatch, delivery):
    monkeypatch.setattr(delivery, "_hmac_keys_cache", None)
    monkeypatch.setattr(delivery, "_hmac_keys_cached_at", 0.0)
    monkeypatch.setattr(
        delivery,
        "HMAC_SECRET_ARN",
        "arn:aws:secretsmanager:us-east-1:011934824531:secret:"
        "workorder-ingest/shoc-webhook-hmac-AbCdEf",
    )


def _wire_cache(monkeypatch, delivery, fake, clock):
    monkeypatch.setattr(delivery.boto3, "client", lambda service: fake)
    monkeypatch.setattr(delivery.time, "monotonic", lambda: clock["t"])


def test_cache_honors_ttl_and_refreshes_after_expiry(
    monkeypatch, delivery, cache_reset
):
    rotated = [
        {"keys": [{"kid": "2026-08-20T00", "secret": "ef" * 32}] + TEST_KEYS[:1]}
    ]
    fake = _FakeSecretsManager([{"keys": TEST_KEYS}] + rotated)
    clock = {"t": 1000.0}
    _wire_cache(monkeypatch, delivery, fake, clock)

    assert delivery._get_hmac_keys()[0]["kid"] == "2026-07-20T00"
    assert fake.calls == 1

    # Inside the 300s TTL: served from cache, no refetch.
    clock["t"] = 1000.0 + 299.0
    assert delivery._get_hmac_keys()[0]["kid"] == "2026-07-20T00"
    assert fake.calls == 1

    # TTL expired: refetch picks up the rotated keys[0] (cache invalidation
    # is what makes 30-day rotation propagate within 5 minutes).
    clock["t"] = 1000.0 + 300.5
    assert delivery._get_hmac_keys()[0]["kid"] == "2026-08-20T00"
    assert fake.calls == 2
    assert fake.secret_ids[0] == delivery.HMAC_SECRET_ARN


def test_empty_or_missing_keys_is_retryable_bootstrap_state(
    monkeypatch, delivery, cache_reset
):
    clock = {"t": 5000.0}
    for payload in ({"keys": []}, {"keys": [], "bootstrap_entropy": "seed"}, {}):
        monkeypatch.setattr(delivery, "_hmac_keys_cache", None)
        monkeypatch.setattr(delivery, "_hmac_keys_cached_at", 0.0)
        fake = _FakeSecretsManager([payload])
        _wire_cache(monkeypatch, delivery, fake, clock)
        with pytest.raises(delivery.RetryableDeliveryError):
            delivery._get_hmac_keys()


def test_secret_fetch_failure_is_retryable(monkeypatch, delivery, cache_reset):
    class _Boom:
        def get_secret_value(self, SecretId):  # noqa: N803 (boto3 kwarg name)
            raise RuntimeError("throttled")

    clock = {"t": 9000.0}
    _wire_cache(monkeypatch, delivery, _Boom(), clock)
    with pytest.raises(delivery.RetryableDeliveryError):
        delivery._get_hmac_keys()