"""Replay re-drives the real work, and refuses to pretend it worked. The properties here are the ones that make a "fix it" button safe to give an admin: * the row's own ``replayable`` flag is the ONLY gate — an earlier draft gated on source and would have re-driven rows the collector had deliberately marked un-replayable; * success never writes ``recovered_at`` — a retry returning 200 is not proof the underlying work succeeded; * one bad row does not stop the batch; * a second click inside the cooldown does nothing; * every attempt leaves evidence, including the ones that failed. """ from __future__ import annotations import uuid from datetime import UTC, datetime, timedelta import pytest from app.domains.automation_exception import replay as replay_mod from app.domains.automation_exception.models import ( AutomationException, AutomationExceptionOccurrence, ) from app.domains.automation_exception.taxonomy import ( SOURCE_IDENTITY_SYNC, SOURCE_INTEGRATION, SOURCE_NOTIFICATION, SOURCE_PROVISIONING, SOURCE_SCHEDULED_JOB, ) pytestmark = pytest.mark.asyncio async def _exception(db, *, source=SOURCE_PROVISIONING, integration_id=None): row = AutomationException( id=uuid.uuid4(), source=source, scope_key="", signature_hash=uuid.uuid4().hex[:32], signature="vendor said no", status="open", severity="normal", root_cause="unknown", cause_source="none", integration_id=integration_id, occurrence_count=1, open_occurrence_count=1, affected_person_count=0, recurrence_count=0, first_seen_at=datetime.now(UTC), last_seen_at=datetime.now(UTC), ) db.add(row) await db.flush() return row async def _occurrence(db, exc, *, source=None, replayable=True, ref=None, **kw): row = AutomationExceptionOccurrence( id=uuid.uuid4(), exception_id=exc.id, source=source or exc.source, source_ref_type="task", source_ref_id=ref or str(uuid.uuid4()), replayable=replayable, occurred_at=kw.get("occurred_at", datetime.now(UTC)), recovered_at=kw.get("recovered_at"), error_text=kw.get("error_text", "boom"), person_id=kw.get("person_id"), ) db.add(row) await db.flush() return row class TestDispatchTableMatchesTheCollectors: """The map must cover exactly the sources that can be replayable.""" def test_every_handler_source_is_one_a_collector_marks_replayable(self): # Sources whose collectors emit ``replayable=True`` today. emitting = {SOURCE_PROVISIONING, SOURCE_NOTIFICATION, SOURCE_IDENTITY_SYNC} assert replay_mod.SOURCES_WITH_HANDLERS == emitting def test_scheduled_job_and_integration_have_no_handler(self): # A scheduled run is a historical event; re-testing a connection # is not re-driving failed work. Both are deliberate. assert SOURCE_SCHEDULED_JOB not in replay_mod.SOURCES_WITH_HANDLERS assert SOURCE_INTEGRATION not in replay_mod.SOURCES_WITH_HANDLERS class TestTheRowFlagIsTheGate: async def test_a_non_replayable_row_is_skipped_even_with_a_handler( self, db_session, monkeypatch, ): """The defect this pins: gating on source would have re-driven rows the collector deliberately marked un-replayable.""" called: list[str] = [] async def _spy(db, occ, *, actor_id, trace_id): called.append(str(occ.id)) return True, None monkeypatch.setitem( replay_mod._REPLAY_BY_SOURCE, SOURCE_PROVISIONING, _spy, ) exc = await _exception(db_session) occ = await _occurrence(db_session, exc, replayable=False) outcome = await replay_mod.replay_occurrences( db_session, [occ], actor_id=uuid.uuid4(), trace_id="t", ) assert called == [], "a non-replayable row must never reach a handler" assert outcome.attempted == 0 assert outcome.skipped_unsupported == 1 class TestReplayBehaviour: async def test_success_does_not_mark_the_row_recovered( self, db_session, monkeypatch, ): """A retry that returned 200 is not proof the work succeeded. Recovery is the collectors' call, from their own evidence. If this ever writes ``recovered_at``, the one table built to hold the truth about failures starts holding a lie. """ async def _ok(db, occ, *, actor_id, trace_id): return True, None monkeypatch.setitem(replay_mod._REPLAY_BY_SOURCE, SOURCE_PROVISIONING, _ok) exc = await _exception(db_session) occ = await _occurrence(db_session, exc) outcome = await replay_mod.replay_occurrences( db_session, [occ], actor_id=uuid.uuid4(), trace_id="t", ) assert outcome.succeeded == 1 assert occ.recovered_at is None, "replay must not invent recovery" async def test_one_bad_row_does_not_stop_the_batch( self, db_session, monkeypatch, ): seen: list[str] = [] async def _every_other(db, occ, *, actor_id, trace_id): seen.append(occ.source_ref_id) if len(seen) == 1: return False, "vendor still down" return True, None monkeypatch.setitem( replay_mod._REPLAY_BY_SOURCE, SOURCE_PROVISIONING, _every_other, ) exc = await _exception(db_session) rows = [await _occurrence(db_session, exc) for _ in range(3)] outcome = await replay_mod.replay_occurrences( db_session, rows, actor_id=uuid.uuid4(), trace_id="t", ) assert len(seen) == 3, "a failure must not abort the remaining rows" assert outcome.failed == 1 assert outcome.succeeded == 2 assert "vendor still down" in outcome.errors async def test_a_failed_attempt_still_leaves_evidence( self, db_session, monkeypatch, ): """An attempt nobody can see is an attempt that gets silently repeated.""" async def _fail(db, occ, *, actor_id, trace_id): return False, "nope" monkeypatch.setitem(replay_mod._REPLAY_BY_SOURCE, SOURCE_PROVISIONING, _fail) exc = await _exception(db_session) occ = await _occurrence(db_session, exc) await replay_mod.replay_occurrences( db_session, [occ], actor_id=uuid.uuid4(), trace_id="t", ) assert occ.last_replay_at is not None assert occ.replay_attempt_count == 1 assert occ.last_replay_error == "nope" async def test_a_success_clears_the_previous_error( self, db_session, monkeypatch, ): async def _ok(db, occ, *, actor_id, trace_id): return True, None monkeypatch.setitem(replay_mod._REPLAY_BY_SOURCE, SOURCE_PROVISIONING, _ok) exc = await _exception(db_session) occ = await _occurrence(db_session, exc) occ.last_replay_error = "an older failure" await db_session.flush() await replay_mod.replay_occurrences( db_session, [occ], actor_id=uuid.uuid4(), trace_id="t", ) assert occ.last_replay_error is None async def test_a_second_click_inside_the_cooldown_does_nothing( self, db_session, monkeypatch, ): calls: list[int] = [] async def _ok(db, occ, *, actor_id, trace_id): calls.append(1) return True, None monkeypatch.setitem(replay_mod._REPLAY_BY_SOURCE, SOURCE_PROVISIONING, _ok) exc = await _exception(db_session) occ = await _occurrence(db_session, exc) moment = datetime.now(UTC) await replay_mod.replay_occurrences( db_session, [occ], actor_id=uuid.uuid4(), trace_id="t", now=moment, ) second = await replay_mod.replay_occurrences( db_session, [occ], actor_id=uuid.uuid4(), trace_id="t", now=moment + timedelta(seconds=30), ) assert len(calls) == 1, "the double-click must not double vendor load" assert second.skipped_cooldown == 1 async def test_past_the_cooldown_it_tries_again( self, db_session, monkeypatch, ): calls: list[int] = [] async def _ok(db, occ, *, actor_id, trace_id): calls.append(1) return True, None monkeypatch.setitem(replay_mod._REPLAY_BY_SOURCE, SOURCE_PROVISIONING, _ok) exc = await _exception(db_session) occ = await _occurrence(db_session, exc) moment = datetime.now(UTC) await replay_mod.replay_occurrences( db_session, [occ], actor_id=uuid.uuid4(), trace_id="t", now=moment, ) await replay_mod.replay_occurrences( db_session, [occ], actor_id=uuid.uuid4(), trace_id="t", now=moment + replay_mod.REPLAY_COOLDOWN + timedelta(seconds=1), ) assert len(calls) == 2 assert occ.replay_attempt_count == 2 async def test_the_batch_is_bounded(self, db_session, monkeypatch): """A vendor that just came back does not need a bad deploy's worth of retries in one second.""" calls: list[int] = [] async def _ok(db, occ, *, actor_id, trace_id): calls.append(1) return True, None monkeypatch.setitem(replay_mod._REPLAY_BY_SOURCE, SOURCE_PROVISIONING, _ok) monkeypatch.setattr(replay_mod, "MAX_REPLAY_PER_RUN", 3) exc = await _exception(db_session) rows = [await _occurrence(db_session, exc) for _ in range(6)] await replay_mod.replay_occurrences( db_session, rows, actor_id=uuid.uuid4(), trace_id="t", ) assert len(calls) == 3 class TestIntegrationScopedReplay: async def test_it_picks_up_open_replayable_rows_under_that_integration( self, db_session, monkeypatch, ): from app.integrations.models import Integration seen: list[str] = [] async def _ok(db, occ, *, actor_id, trace_id): seen.append(occ.source_ref_id) return True, None monkeypatch.setitem(replay_mod._REPLAY_BY_SOURCE, SOURCE_PROVISIONING, _ok) # Column names taken from the model, not assumed: an earlier # draft used ``key``/``display_name``, neither of which exists. integration = Integration( id=uuid.uuid4(), kind="provisioning", provider="zoom", name=f"zoom-{uuid.uuid4().hex[:8]}", slug=f"zoom-{uuid.uuid4().hex[:8]}", enabled=True, ) db_session.add(integration) await db_session.flush() mine = await _exception(db_session, integration_id=integration.id) theirs = await _exception(db_session, integration_id=None) wanted = await _occurrence(db_session, mine, ref="wanted") await _occurrence(db_session, theirs, ref="other-integration") await _occurrence( db_session, mine, ref="already-recovered", recovered_at=datetime.now(UTC), ) await _occurrence(db_session, mine, ref="not-replayable", replayable=False) outcome = await replay_mod.replay_for_integration( db_session, integration.id, actor_id=uuid.uuid4(), trace_id="t", ) assert seen == [wanted.source_ref_id] assert outcome.succeeded == 1 class TestTheCooldownSurvivesAStaleIdentityMap: """The cooldown must read the DATABASE, not this session's memory. Written after a peer session hit the same trap independently: in this repo, "I already read it" is precisely the condition that makes every later re-read return the old value. ``select()`` and ``Session.get()`` both hand back whatever is in the identity map, and ``expire_on_commit=False`` means a commit does not expire it either. Only ``populate_existing`` refreshes. Without it, two admins clicking at the same moment both read ``last_replay_at IS NULL`` and both fire — doubling exactly the vendor load the cooldown exists to prevent. """ async def test_the_entry_point_refreshes_a_stale_in_memory_row( self, db_session, monkeypatch, ): from sqlalchemy import update calls: list[str] = [] async def _ok(db, occ, *, actor_id, trace_id): calls.append(occ.source_ref_id) return True, None monkeypatch.setitem(replay_mod._REPLAY_BY_SOURCE, SOURCE_PROVISIONING, _ok) exc = await _exception(db_session) occ = await _occurrence(db_session, exc) # Someone else replays this row. Written with a Core UPDATE and # ``synchronize_session=False`` so the ORM does NOT helpfully # write the new value back into our in-memory instance — that # write-back is what silently empties this kind of test. await db_session.execute( update(AutomationExceptionOccurrence) .where(AutomationExceptionOccurrence.id == occ.id) .values(last_replay_at=datetime.now(UTC), replay_attempt_count=1) .execution_options(synchronize_session=False), ) # The assertion that keeps this test honest: our copy really is # stale right now. Without it, a future change that made the # in-memory row fresh would leave the test green while testing # nothing at all. assert occ.last_replay_at is None, ( "precondition: the in-memory copy must still be stale here" ) outcome = await replay_mod.replay_for_exception( db_session, exc.id, actor_id=uuid.uuid4(), trace_id="t", ) assert calls == [], ( "the entry point must read the row's CURRENT state, not the " "value this session happened to load earlier" ) assert outcome.skipped_cooldown == 1 class TestOutboxReplayClosesTheSourceRow: """Regression from the PR5 review. ``resend_outbox_row`` clones the dead row into a fresh dispatch and deliberately leaves the original untouched — and ``notification_source`` scans exactly "dead and unacknowledged". So without closing the source row the same outbox id is a candidate on every tick forever: the occurrence never recovers, the parent never resolves, nobody gets the recovery echo, and the UI meanwhile reported "1 re-driven". """ async def test_a_refused_resend_is_not_counted_as_success( self, db_session, monkeypatch, ): """``resend_outbox_row`` REFUSES some rows without raising.""" async def _refused(self, outbox_id, *, actor_user_id, trace_id=""): return { "status": "skipped", "reason": "recipient departed / deleted — resend refused", } monkeypatch.setattr( "app.notifications.service.NotificationService.resend_outbox_row", _refused, ) exc = await _exception(db_session, source=SOURCE_NOTIFICATION) occ = await _occurrence( db_session, exc, source=SOURCE_NOTIFICATION, ref=str(uuid.uuid4()), ) outcome = await replay_mod.replay_occurrences( db_session, [occ], actor_id=uuid.uuid4(), trace_id="t", ) assert outcome.succeeded == 0, ( "a refused resend must not be reported as re-driven" ) assert outcome.failed == 1 assert occ.last_replay_error is not None assert "departed" in occ.last_replay_error async def test_a_successful_resend_closes_the_original_dead_row( self, db_session, monkeypatch, ): from app.notifications.models import ACK_NOTE_PREFIX closed: list[str] = [] async def _ok(self, outbox_id, *, actor_user_id, trace_id=""): return {"status": "sent"} async def _cancel(self, outbox_id, *, actor_user_id, reason=None): closed.append(str(outbox_id)) return None monkeypatch.setattr( "app.notifications.service.NotificationService.resend_outbox_row", _ok, ) monkeypatch.setattr( "app.notifications.service.NotificationService.cancel_outbox", _cancel, ) # A real dead row, unacknowledged — the state the collector # keeps listing. from app.notifications.models import NotificationOutbox outbox = NotificationOutbox( id=uuid.uuid4(), recipient_person_id=None, title="t", body_text="b", channels="feishu", status="dead", last_error="feishu 500", ) db_session.add(outbox) await db_session.flush() assert not (outbox.last_error or "").startswith(ACK_NOTE_PREFIX) exc = await _exception(db_session, source=SOURCE_NOTIFICATION) occ = await _occurrence( db_session, exc, source=SOURCE_NOTIFICATION, ref=str(outbox.id), ) outcome = await replay_mod.replay_occurrences( db_session, [occ], actor_id=uuid.uuid4(), trace_id="t", ) assert outcome.succeeded == 1 assert closed == [str(outbox.id)], ( "the source row must be closed, or the collector lists it " "again forever and the exception never resolves" )