"""Rotation-step tests for workorder-shoc-hmac-rotator (plan Phase 5). Exercises the Secrets Manager rotation protocol against a dict-backed fake client (no moto, no AWS): createSecret from the CDK bootstrap value and from a populated list (newest-first, truncated to 2), createSecret idempotency on retried tokens, the intra-hour kid-collision fallback, testSecret's shape/hex validation, finishSecret's stage move + idempotency, the setSecret no-op, and the unknown-step failure. datetime is monkeypatched to a fixed instant so kid assertions are exact and the collision case is deterministic (no wall-clock hour-boundary flake). """ import json import re from datetime import datetime, timezone import pytest from tests.support import load_lambda_module SECRET_ID = "workorder-ingest/shoc-webhook-hmac" HEX_64_RE = re.compile(r"^[0-9a-f]{64}$") # Fixed rotation instant: kid "2026-07-24T15" (intra-hour fallback # "2026-07-24T1507"). FIXED_NOW = datetime(2026, 7, 24, 15, 7, 42, tzinfo=timezone.utc) KEY_A = {"kid": "2026-06-20T00", "secret": "a" * 64} KEY_B = {"kid": "2026-05-21T00", "secret": "b" * 64} class _ResourceNotFound(Exception): pass class _Exceptions: ResourceNotFoundException = _ResourceNotFound class FakeSecretsManager: """Dict-backed Secrets Manager stub recording every mutating call. ``versions`` maps version_id -> {"stages": set[str], "value": str}. """ exceptions = _Exceptions def __init__(self, versions=None): self.versions = { vid: {"stages": set(v["stages"]), "value": v["value"]} for vid, v in (versions or {}).items() } self.put_calls = [] self.stage_calls = [] def describe_secret(self, SecretId): # noqa: N803 (boto3 kwarg names) return { "VersionIdsToStages": { vid: sorted(v["stages"]) for vid, v in self.versions.items() } } def get_secret_value( # noqa: N803 (boto3 kwarg names) self, SecretId, VersionId=None, VersionStage=None ): if VersionId is not None: version = self.versions.get(VersionId) if version is None or ( VersionStage is not None and VersionStage not in version["stages"] ): raise _ResourceNotFound(f"{VersionId} / {VersionStage}") return {"SecretString": version["value"]} stage = VersionStage or "AWSCURRENT" for version in self.versions.values(): if stage in version["stages"]: return {"SecretString": version["value"]} raise _ResourceNotFound(stage) def put_secret_value( # noqa: N803 (boto3 kwarg names) self, SecretId, ClientRequestToken, SecretString, VersionStages ): self.put_calls.append( { "SecretId": SecretId, "ClientRequestToken": ClientRequestToken, "SecretString": SecretString, "VersionStages": list(VersionStages), } ) self.versions[ClientRequestToken] = { "stages": set(VersionStages), "value": SecretString, } def update_secret_version_stage(self, **kwargs): self.stage_calls.append(kwargs) stage = kwargs["VersionStage"] removed = kwargs.get("RemoveFromVersionId") if removed is not None: self.versions[removed]["stages"].discard(stage) self.versions[kwargs["MoveToVersionId"]]["stages"].add(stage) class _FixedDatetime: @classmethod def now(cls, tz=None): return FIXED_NOW @pytest.fixture(scope="module") def rotator(): return load_lambda_module("wo", "shoc_hmac_rotator/handler") @pytest.fixture def fixed_now(monkeypatch, rotator): monkeypatch.setattr(rotator, "datetime", _FixedDatetime) def _run(monkeypatch, rotator, fake, step, token="tok-1"): monkeypatch.setattr(rotator, "secretsmanager_client", fake) rotator.handler( {"SecretId": SECRET_ID, "ClientRequestToken": token, "Step": step}, None ) def _bootstrap_fake(value=None): current = value if value is not None else {"keys": [], "bootstrap_entropy": "x"} return FakeSecretsManager( {"cur-1": {"stages": {"AWSCURRENT"}, "value": json.dumps(current)}} ) # --- createSecret ------------------------------------------------------------ def test_create_from_bootstrap_stages_one_fresh_key(monkeypatch, rotator, fixed_now): fake = _bootstrap_fake() _run(monkeypatch, rotator, fake, "createSecret") assert len(fake.put_calls) == 1 put = fake.put_calls[0] assert put["SecretId"] == SECRET_ID assert put["ClientRequestToken"] == "tok-1" assert put["VersionStages"] == ["AWSPENDING"] staged = json.loads(put["SecretString"]) assert list(staged) == ["keys"] assert len(staged["keys"]) == 1 assert staged["keys"][0]["kid"] == "2026-07-24T15" assert HEX_64_RE.fullmatch(staged["keys"][0]["secret"]) def test_create_prepends_and_truncates_to_two_newest_first( monkeypatch, rotator, fixed_now ): fake = _bootstrap_fake({"keys": [KEY_A, KEY_B]}) _run(monkeypatch, rotator, fake, "createSecret") staged = json.loads(fake.put_calls[0]["SecretString"]) assert len(staged["keys"]) == 2 # Newest first: the fresh key leads, the previous head is retained for # one overlap cycle, the oldest key falls off. assert staged["keys"][0]["kid"] == "2026-07-24T15" assert staged["keys"][1] == KEY_A assert KEY_B["secret"] not in fake.put_calls[0]["SecretString"] def test_create_is_idempotent_when_pending_already_staged( monkeypatch, rotator, fixed_now ): fake = _bootstrap_fake({"keys": [KEY_A]}) fake.versions["tok-1"] = { "stages": {"AWSPENDING"}, "value": json.dumps({"keys": [KEY_A]}), } _run(monkeypatch, rotator, fake, "createSecret") assert fake.put_calls == [] def test_create_is_noop_when_token_already_current(monkeypatch, rotator, fixed_now): fake = FakeSecretsManager( { "tok-1": { "stages": {"AWSCURRENT"}, "value": json.dumps({"keys": [KEY_A]}), } } ) _run(monkeypatch, rotator, fake, "createSecret") assert fake.put_calls == [] def test_create_kid_collision_same_hour_gets_distinct_suffix( monkeypatch, rotator, fixed_now ): # Forced re-rotation within the same hour: the hour-format kid already # exists, so the new kid extends to minutes + a random suffix, guaranteed # distinct from every retained kid (never reissue a kid for a new secret). fake = _bootstrap_fake({"keys": [{"kid": "2026-07-24T15", "secret": "c" * 64}]}) _run(monkeypatch, rotator, fake, "createSecret") staged = json.loads(fake.put_calls[0]["SecretString"]) new_kid = staged["keys"][0]["kid"] assert new_kid.startswith("2026-07-24T1507") assert new_kid != "2026-07-24T15" assert staged["keys"][1]["kid"] == "2026-07-24T15" def test_create_kid_collision_against_second_key_also_avoided( monkeypatch, rotator, fixed_now ): # Uniqueness is checked against ALL retained kids, not just keys[0]: if the # hour-format kid matches keys[1] (a third same-hour rotation), it still # gets a distinct suffix rather than being reissued. fake = _bootstrap_fake( { "keys": [ {"kid": "2026-07-24T1500ab", "secret": "a" * 64}, {"kid": "2026-07-24T15", "secret": "b" * 64}, ] } ) _run(monkeypatch, rotator, fake, "createSecret") new_kid = json.loads(fake.put_calls[0]["SecretString"])["keys"][0]["kid"] assert new_kid not in {"2026-07-24T15", "2026-07-24T1500ab"} def test_create_malformed_current_value_starts_fresh_list( monkeypatch, rotator, fixed_now ): fake = FakeSecretsManager( {"cur-1": {"stages": {"AWSCURRENT"}, "value": "not json {"}} ) _run(monkeypatch, rotator, fake, "createSecret") staged = json.loads(fake.put_calls[0]["SecretString"]) assert len(staged["keys"]) == 1 assert HEX_64_RE.fullmatch(staged["keys"][0]["secret"]) def test_create_transient_current_read_error_propagates_not_swallowed( monkeypatch, rotator, fixed_now ): # A transient AWSCURRENT read failure (throttle/KMS blip) must NOT be # swallowed into an empty key list -- that would drop the overlap key and # strand in-flight deliveries. It re-raises so Secrets Manager fails and # retries the rotation with the prior AWSCURRENT intact. class _ThrottlingFake(FakeSecretsManager): def get_secret_value(self, SecretId, VersionId=None, VersionStage=None): # noqa: N803 if VersionStage == "AWSCURRENT": raise RuntimeError("ThrottlingException") return super().get_secret_value( SecretId, VersionId=VersionId, VersionStage=VersionStage ) fake = _ThrottlingFake( {"cur-1": {"stages": {"AWSCURRENT"}, "value": json.dumps({"keys": []})}} ) with pytest.raises(RuntimeError, match="ThrottlingException"): _run(monkeypatch, rotator, fake, "createSecret") assert fake.put_calls == [] # --- testSecret -------------------------------------------------------------- def _pending_fake(value): return FakeSecretsManager( {"tok-1": {"stages": {"AWSPENDING"}, "value": json.dumps(value)}} ) def test_test_secret_accepts_valid_pending_value(monkeypatch, rotator): fake = _pending_fake({"keys": [{"kid": "2026-07-24T15", "secret": "d" * 64}]}) _run(monkeypatch, rotator, fake, "testSecret") # no raise @pytest.mark.parametrize( "pending_value", [ {"keys": []}, {"keys": [{"kid": "2026-07-24T15", "secret": "d" * 63}]}, {"keys": [{"kid": "2026-07-24T15", "secret": "z" * 64}]}, {"keys": [{"kid": "2026-07-24T15", "secret": "D" * 64}]}, {"keys": [{"secret": "d" * 64}]}, ], ids=["empty-keys", "short-secret", "non-hex", "uppercase-hex", "missing-kid"], ) def test_test_secret_rejects_bad_pending_values(monkeypatch, rotator, pending_value): fake = _pending_fake(pending_value) with pytest.raises(ValueError): _run(monkeypatch, rotator, fake, "testSecret") # --- finishSecret ------------------------------------------------------------ def test_finish_moves_current_stage_from_old_version(monkeypatch, rotator): fake = FakeSecretsManager( { "old-1": { "stages": {"AWSCURRENT"}, "value": json.dumps({"keys": [KEY_A]}), }, "tok-1": { "stages": {"AWSPENDING"}, "value": json.dumps({"keys": [KEY_B, KEY_A]}), }, } ) _run(monkeypatch, rotator, fake, "finishSecret") assert fake.stage_calls == [ { "SecretId": SECRET_ID, "VersionStage": "AWSCURRENT", "MoveToVersionId": "tok-1", "RemoveFromVersionId": "old-1", } ] assert "AWSCURRENT" in fake.versions["tok-1"]["stages"] assert "AWSCURRENT" not in fake.versions["old-1"]["stages"] def test_finish_is_idempotent_when_token_already_current(monkeypatch, rotator): fake = FakeSecretsManager( { "tok-1": { "stages": {"AWSCURRENT", "AWSPENDING"}, "value": json.dumps({"keys": [KEY_A]}), } } ) _run(monkeypatch, rotator, fake, "finishSecret") assert fake.stage_calls == [] # --- Dispatch ---------------------------------------------------------------- def test_set_secret_is_a_noop(monkeypatch, rotator): fake = _bootstrap_fake() _run(monkeypatch, rotator, fake, "setSecret") assert fake.put_calls == [] assert fake.stage_calls == [] def test_unknown_step_raises(monkeypatch, rotator): fake = _bootstrap_fake() with pytest.raises(ValueError, match="Unknown rotation step"): _run(monkeypatch, rotator, fake, "rotateHarder")