Alexander Whitestone
08f1d0bc8d
Some checks failed
Contributor Attribution Check / check-attribution (pull_request) Failing after 44s
Docker Build and Publish / build-and-push (pull_request) Has been skipped
Supply Chain Audit / Scan PR for supply chain risks (pull_request) Successful in 45s
Tests / e2e (pull_request) Successful in 4m57s
Tests / test (pull_request) Failing after 43m26s
Tests for duplicate detection, window expiry, stats. Refs #756
58 lines
1.9 KiB
Python
58 lines
1.9 KiB
Python
"""
|
|
Tests for message deduplication (#756).
|
|
"""
|
|
|
|
import pytest
|
|
import time
|
|
from gateway.message_dedup import MessageDeduplicator
|
|
|
|
|
|
class TestMessageDeduplicator:
|
|
def test_first_message_allowed(self):
|
|
dedup = MessageDeduplicator()
|
|
assert dedup.should_send("Hello") is True
|
|
|
|
def test_duplicate_suppressed(self):
|
|
dedup = MessageDeduplicator()
|
|
dedup.record_message("Hello", "session1", "telegram")
|
|
assert dedup.should_send("Hello", "session1", "telegram") is False
|
|
|
|
def test_different_session_allowed(self):
|
|
dedup = MessageDeduplicator()
|
|
dedup.record_message("Hello", "session1", "telegram")
|
|
assert dedup.should_send("Hello", "session2", "telegram") is True
|
|
|
|
def test_different_platform_allowed(self):
|
|
dedup = MessageDeduplicator()
|
|
dedup.record_message("Hello", "session1", "telegram")
|
|
assert dedup.should_send("Hello", "session1", "discord") is True
|
|
|
|
def test_different_content_allowed(self):
|
|
dedup = MessageDeduplicator()
|
|
dedup.record_message("Hello", "session1", "telegram")
|
|
assert dedup.should_send("World", "session1", "telegram") is True
|
|
|
|
def test_window_expiry(self):
|
|
dedup = MessageDeduplicator(window_seconds=1)
|
|
dedup.record_message("Hello", "session1", "telegram")
|
|
time.sleep(1.1)
|
|
assert dedup.should_send("Hello", "session1", "telegram") is True
|
|
|
|
def test_record_returns_uuid(self):
|
|
dedup = MessageDeduplicator()
|
|
msg_id = dedup.record_message("Hello")
|
|
assert msg_id is not None
|
|
assert len(msg_id) == 36 # UUID format
|
|
|
|
def test_stats(self):
|
|
dedup = MessageDeduplicator()
|
|
dedup.record_message("Hello")
|
|
dedup.record_message("Hello") # duplicate
|
|
stats = dedup.get_stats()
|
|
assert stats["total_records"] == 1
|
|
assert stats["suppressed_count"] == 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__])
|