From 924864f9646288bb0abff1be23ab8d7fa2a00697 Mon Sep 17 00:00:00 2001 From: Alex Date: Mon, 10 Aug 2026 14:51:33 +0100 Subject: [PATCH] fix: tests --- .../repositories/test_truncate_supersede.py | 35 ------------- tests/test_message_journal.py | 49 +++++++++++++++++++ 2 files changed, 49 insertions(+), 35 deletions(-) diff --git a/tests/storage/db/repositories/test_truncate_supersede.py b/tests/storage/db/repositories/test_truncate_supersede.py index 81717e5e..8eb800f2 100644 --- a/tests/storage/db/repositories/test_truncate_supersede.py +++ b/tests/storage/db/repositories/test_truncate_supersede.py @@ -118,38 +118,3 @@ class TestTruncateAfterSupersede: def test_invalid_conversation_id_is_rejected(self, pg_conn): repo = ConversationsRepository(pg_conn) assert repo.truncate_after("not-a-uuid", keep_up_to=0) == 0 - - -class TestJournalParentMissingLatch: - """One WARN per stream, not one per flush.""" - - def test_latches_after_first_foreign_key_failure(self, caplog): - import logging - - from application.streaming.message_journal import BatchedJournalWriter - - writer = BatchedJournalWriter( - "99999999-9999-9999-9999-999999999999", batch_size=1, - ) - - with caplog.at_level(logging.WARNING): - for seq in range(25): - writer.record(seq, "answer", {"answer": "x"}) - - warnings = [ - r for r in caplog.records - if "no conversation_messages row" in r.getMessage() - ] - assert len(warnings) == 1, ( - f"expected a single latched warning, got {len(warnings)}" - ) - - def test_record_returns_false_once_latched(self): - from application.streaming.message_journal import BatchedJournalWriter - - writer = BatchedJournalWriter( - "88888888-8888-8888-8888-888888888888", batch_size=1, - ) - writer.record(0, "answer", {"answer": "x"}) - - assert writer.record(1, "answer", {"answer": "y"}) is False diff --git a/tests/test_message_journal.py b/tests/test_message_journal.py index 7461b096..147e18a9 100644 --- a/tests/test_message_journal.py +++ b/tests/test_message_journal.py @@ -648,6 +648,55 @@ class TestBatchedJournalWriter: # Buffer cleared so close() is a no-op and memory stays bounded. assert writer._buffer == [] + def test_missing_parent_is_latched_after_the_first_failure(self, caplog): + """One WARN per stream, not one per flush. + + A parent row deleted mid-stream (retry/edit truncation) never comes + back, so every later flush FK-fails identically. Un-latched, a single + 3.5-minute research stream logged 342 identical WARNs in production + and was that day's entire WARN volume. + """ + import logging + + session_cm, readonly_cm, repo_cls_p, topic_cls_p = self._patch_io() + with session_cm as mock_session, readonly_cm, repo_cls_p as mock_repo_cls, topic_cls_p as mock_topic_cls: + mock_session.return_value.__enter__.return_value = MagicMock() + repo = MagicMock(name="repo") + repo.bulk_record.side_effect = _integrity_error("23503") + mock_repo_cls.return_value = repo + mock_topic_cls.return_value = MagicMock() + + writer = BatchedJournalWriter("msg-1", batch_size=1) + with caplog.at_level( + logging.WARNING, logger="application.streaming.message_journal" + ): + for seq in range(25): + writer.record(seq, "answer", {"text": "x"}) + + warnings = [ + r for r in caplog.records + if "no conversation_messages row" in r.getMessage() + ] + assert len(warnings) == 1, ( + f"expected a single latched warning, got {len(warnings)}" + ) + # And it stopped hitting the DB after the first failure. + assert repo.bulk_record.call_count == 1 + + def test_record_returns_false_once_parent_is_known_missing(self): + session_cm, readonly_cm, repo_cls_p, topic_cls_p = self._patch_io() + with session_cm as mock_session, readonly_cm, repo_cls_p as mock_repo_cls, topic_cls_p as mock_topic_cls: + mock_session.return_value.__enter__.return_value = MagicMock() + repo = MagicMock(name="repo") + repo.bulk_record.side_effect = _integrity_error("23503") + mock_repo_cls.return_value = repo + mock_topic_cls.return_value = MagicMock() + + writer = BatchedJournalWriter("msg-1", batch_size=1) + + assert writer.record(0, "answer", {"text": "a"}) is True + assert writer.record(1, "answer", {"text": "b"}) is False + def test_bulk_collision_23505_still_falls_back_to_per_row(self): """An explicit 23505 (unique/PK collision) keeps the per-row fallback — the parent exists, only a ``sequence_no`` collided.