"""The echo tells the waiting person once, and never tells them something they cannot act on. Each test pins a property whose opposite is a real harm: * told twice ⇒ they stop believing the first message; * a new notification category ⇒ everyone silently gains a ping they never agreed to, bypassing every preference the notification governance work established; * the failure narrative sent to the subject ⇒ a solved problem becomes a new worry they cannot act on; * a failed SEND stamped as done ⇒ the person is never told at all. """ from __future__ import annotations import uuid from datetime import UTC, datetime, timedelta import pytest from app.domains.automation_exception import echo as echo_mod from app.domains.automation_exception.models import ( AutomationException, AutomationExceptionOccurrence, ) from app.notifications.categories import CAT_LIFECYCLE, classify pytestmark = pytest.mark.asyncio async def _resolved_exception(db, *, recurrence=0, owner_id=None, notified=None): row = AutomationException( id=uuid.uuid4(), source="provisioning", scope_key="", signature_hash=uuid.uuid4().hex[:32], signature="vendor said no", status="resolved", severity="normal", root_cause="credential", cause_source="rule", occurrence_count=1, open_occurrence_count=0, affected_person_count=1, recurrence_count=recurrence, first_seen_at=datetime.now(UTC) - timedelta(hours=2), last_seen_at=datetime.now(UTC) - timedelta(hours=1), resolved_at=datetime.now(UTC), owner_person_id=owner_id, subject_notified_at=notified, ) db.add(row) await db.flush() return row async def _occurrence(db, exc, *, person_id=None): row = AutomationExceptionOccurrence( id=uuid.uuid4(), exception_id=exc.id, source=exc.source, source_ref_type="task", source_ref_id=str(uuid.uuid4()), replayable=True, occurred_at=datetime.now(UTC) - timedelta(hours=2), recovered_at=datetime.now(UTC), error_text="boom", person_id=person_id, ) db.add(row) await db.flush() return row class TestRoutingUsesAnExistingCategory: """A new category would silently hand everyone a ping they never agreed to. Both templates must land in an existing, mutable, employee-governable category.""" @pytest.mark.parametrize( "key", [echo_mod.TPL_SUBJECT_RECOVERED, echo_mod.TPL_OWNER_RECURRED], ) def test_the_template_routes_through_lifecycle(self, key): category = classify(key) assert category.key == CAT_LIFECYCLE assert category.mutable, "an employee must be able to turn this off" assert category.employee_facing class TestSubjectEcho: async def test_the_waiting_person_is_told_once( self, db_session, employee_user, monkeypatch, ): sent: list[tuple[str, str]] = [] async def _capture(self, env, **kw): sent.append((env.recipient.work_email, env.template_key or "")) return None monkeypatch.setattr( "app.notifications.service.NotificationService.dispatch", _capture, ) person = employee_user[0] exc = await _resolved_exception(db_session) await _occurrence(db_session, exc, person_id=person.id) first = await echo_mod.send_recovery_echoes(db_session) assert first.subjects_notified == 1 assert exc.subject_notified_at is not None # Second tick: the stamp makes it a transition, not a state. second = await echo_mod.send_recovery_echoes(db_session) assert second.subjects_notified == 0 assert len(sent) == 1, "being told twice destroys trust in the first" async def test_the_subject_is_not_told_how_badly_it_failed( self, db_session, employee_user, monkeypatch, ): """Their problem is solved; the 47 failures are the admin's.""" bodies: list[str] = [] async def _capture(self, env, **kw): bodies.append(env.body_text) return None monkeypatch.setattr( "app.notifications.service.NotificationService.dispatch", _capture, ) person = employee_user[0] exc = await _resolved_exception(db_session) exc.occurrence_count = 47 exc.signature = "HTTP 500 from vendor xyz" await _occurrence(db_session, exc, person_id=person.id) await echo_mod.send_recovery_echoes(db_session) assert bodies, "someone should have been told" body = bodies[0] assert "47" not in body assert "500" not in body assert exc.signature not in body async def test_an_exception_nobody_waited_on_is_stamped_not_re_examined( self, db_session, monkeypatch, ): async def _capture(self, env, **kw): raise AssertionError("nobody was waiting; nothing should be sent") monkeypatch.setattr( "app.notifications.service.NotificationService.dispatch", _capture, ) exc = await _resolved_exception(db_session) await _occurrence(db_session, exc, person_id=None) outcome = await echo_mod.send_recovery_echoes(db_session) assert outcome.skipped_no_person == 1 assert exc.subject_notified_at is not None, ( "a finished state must not be re-walked on every tick" ) async def test_a_failed_send_is_retried_rather_than_stamped( self, db_session, employee_user, monkeypatch, ): """Stamping a failed send means the person is never told.""" async def _boom(self, env, **kw): raise RuntimeError("feishu is down") monkeypatch.setattr( "app.notifications.service.NotificationService.dispatch", _boom, ) person = employee_user[0] exc = await _resolved_exception(db_session) await _occurrence(db_session, exc, person_id=person.id) outcome = await echo_mod.send_recovery_echoes(db_session) assert outcome.subjects_notified == 0 assert exc.subject_notified_at is None, ( "a failed send must remain a candidate for the next tick" ) async def test_a_leaver_is_not_told( self, db_session, employee_user, monkeypatch, ): sent: list[str] = [] async def _capture(self, env, **kw): sent.append(env.recipient.work_email) return None monkeypatch.setattr( "app.notifications.service.NotificationService.dispatch", _capture, ) # Inserted already departed — flipping a live person to ``left`` # trips the leaver-blocked write guard, which is the platform # working correctly. from app.identity.person_models import IdentityPerson gone = IdentityPerson( hris_person_id=f"P{uuid.uuid4().hex[:8]}", work_email=f"gone-{uuid.uuid4().hex[:8]}@pingcap.com", display_name="Departed", employee_no=f"P{uuid.uuid4().hex[:5]}", employment_status="left", source_system="successfactors", ) db_session.add(gone) await db_session.flush() exc = await _resolved_exception(db_session) await _occurrence(db_session, exc, person_id=gone.id) outcome = await echo_mod.send_recovery_echoes(db_session) assert sent == [] assert outcome.subjects_notified == 0 assert exc.subject_notified_at is not None class TestRecurrenceEscalation: async def test_one_recurrence_is_noise_two_is_a_pattern( self, db_session, admin_user, monkeypatch, ): sent: list[str] = [] async def _capture(self, env, **kw): sent.append(env.template_key or "") return None monkeypatch.setattr( "app.notifications.service.NotificationService.dispatch", _capture, ) owner = admin_user[0] once = await _resolved_exception(db_session, recurrence=1, owner_id=owner.id) once.resolved_at = None twice = await _resolved_exception(db_session, recurrence=2, owner_id=owner.id) twice.resolved_at = None await db_session.flush() outcome = await echo_mod.escalate_recurrences(db_session) assert outcome.owners_escalated == 1 assert once.escalated_at is None assert twice.escalated_at is not None async def test_the_owner_is_told_the_next_step_not_just_the_fact( self, db_session, admin_user, monkeypatch, ): """"It came back" without "here is what to check" is the dead end this epic removes.""" bodies: list[str] = [] async def _capture(self, env, **kw): bodies.append(env.body_text) return None monkeypatch.setattr( "app.notifications.service.NotificationService.dispatch", _capture, ) owner = admin_user[0] exc = await _resolved_exception(db_session, recurrence=3, owner_id=owner.id) exc.resolved_at = None await db_session.flush() await echo_mod.escalate_recurrences(db_session) assert bodies from app.domains.automation_exception.taxonomy import remedy_for assert remedy_for("credential") in bodies[0] async def test_it_escalates_once_per_episode( self, db_session, admin_user, monkeypatch, ): async def _capture(self, env, **kw): return None monkeypatch.setattr( "app.notifications.service.NotificationService.dispatch", _capture, ) owner = admin_user[0] exc = await _resolved_exception(db_session, recurrence=5, owner_id=owner.id) exc.resolved_at = None await db_session.flush() await echo_mod.escalate_recurrences(db_session) second = await echo_mod.escalate_recurrences(db_session) assert second.owners_escalated == 0 async def test_an_unowned_exception_escalates_to_nobody( self, db_session, monkeypatch, ): """Not a silent drop: an unowned row is a gap the workbench already shows. Paging a random admin would be worse.""" async def _capture(self, env, **kw): raise AssertionError("there is no owner to escalate to") monkeypatch.setattr( "app.notifications.service.NotificationService.dispatch", _capture, ) exc = await _resolved_exception(db_session, recurrence=4, owner_id=None) exc.resolved_at = None await db_session.flush() outcome = await echo_mod.escalate_recurrences(db_session) assert outcome.owners_escalated == 0 class TestReopenStartsANewEpisode: """Regression: the echo must be once per EPISODE, not once ever. Found in the PR5 review. Reopen cleared ``escalated_at`` but left ``subject_notified_at`` set, so a person whose problem recurred was never told about the second fix — the exact silence this epic exists to remove. """ async def test_reopening_clears_the_echo_marker(self, db_session): from app.domains.automation_exception.collectors.base import ( Candidate, CollectorResult, ) from app.domains.automation_exception.ingest import ingest_source from app.domains.automation_exception.signature import ( canonical_signature, signature_hash, ) exc = await _resolved_exception( db_session, notified=datetime.now(UTC) - timedelta(days=1), ) # The identity must be the one ingest will COMPUTE, not a random # digest: with a mismatched hash ingest creates a second cluster # and the reopen path is never taken — the test would then pass # or fail for a reason unrelated to what it claims to check. error_text = "vendor said no" exc.signature = canonical_signature(error_text) exc.signature_hash = signature_hash( source=exc.source, scope_key=exc.scope_key or None, signature=exc.signature, ) await db_session.flush() # Preconditions that keep this test honest. assert exc.subject_notified_at is not None assert exc.resolved_at is not None # The same signature fails again. stats = await ingest_source( db_session, source=exc.source, result=CollectorResult( candidates=[ Candidate( source_ref_type="task", source_ref_id=str(uuid.uuid4()), scope_key=exc.scope_key, error_text=error_text, occurred_at=datetime.now(UTC), replayable=True, ), ], absence_means_recovery=False, ), now=datetime.now(UTC), ) assert stats.exceptions_reopened == 1, ( "precondition: the recurrence must land on the EXISTING " "cluster, otherwise the reopen path was never exercised" ) await db_session.refresh(exc) assert exc.subject_notified_at is None, ( "a reopen starts a new episode; the next recovery must be " "announced, not swallowed by the previous episode's stamp" )