test: add tests for email loop prevention
Add tests/test_email_loop_prevention.py (47 tests) covering: - detect_automated_message(): sender patterns (mailer-daemon, postmaster, noreply, …), RFC headers (Auto-Submitted, X-Auto-Response-Suppress, multipart/report, X-Loop, Precedence), subject heuristics (German and English bounce/OOO phrases), and negative cases for normal senders - EmailAgent.handle_automated_message(): state creation, loop_escalated flag, one-shot admin alert, silent drop on repeat, notifier failure safety - EmailAgent.process_message() message-count cap: at-limit still processes, over-limit returns "", escalation called once, duplicate prevention, notifier failure safety, constant value assertion - AdminNotifier.notify_loop_escalation(): SMTP dispatch, [WARNUNG] subject, body content, no-CC guard, no-SMTP dev-mode guard Add TestConversationStateLoopEscalated to test_models.py (5 tests): default value, to_dict inclusion, True round-trip, from_dict backward compatibility with persisted conversations missing the key. All 168 tests pass. https://claude.ai/code/session_01KwvR5hDPjSuJg4kvw5b5e5
This commit is contained in:
@@ -150,3 +150,45 @@ class TestConversationStateSerialization:
|
||||
d = state_with_messages.to_dict()
|
||||
assert d["messages"][0]["role"] == "user"
|
||||
assert "Hallo" in d["messages"][0]["content"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ConversationState — loop_escalated field
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestConversationStateLoopEscalated:
|
||||
def test_default_loop_escalated_is_false(self, fresh_state):
|
||||
assert fresh_state.loop_escalated is False
|
||||
|
||||
def test_to_dict_includes_loop_escalated(self, fresh_state):
|
||||
d = fresh_state.to_dict()
|
||||
assert "loop_escalated" in d
|
||||
assert d["loop_escalated"] is False
|
||||
|
||||
def test_to_dict_reflects_true_when_set(self, fresh_state):
|
||||
fresh_state.loop_escalated = True
|
||||
d = fresh_state.to_dict()
|
||||
assert d["loop_escalated"] is True
|
||||
|
||||
def test_from_dict_restores_loop_escalated_true(self, fresh_state):
|
||||
fresh_state.loop_escalated = True
|
||||
restored = ConversationState.from_dict(fresh_state.to_dict())
|
||||
assert restored.loop_escalated is True
|
||||
|
||||
def test_from_dict_defaults_to_false_when_key_missing(self):
|
||||
"""Older persisted conversations without the key deserialise safely."""
|
||||
data = {
|
||||
"conversation_id": "old@example.com",
|
||||
"parent_email": "old@example.com",
|
||||
"language": "de",
|
||||
"flow_step": "greeting",
|
||||
"registration": {},
|
||||
"messages": [],
|
||||
"completed": False,
|
||||
"reminder_count": 0,
|
||||
"last_inbound_message_id": "",
|
||||
# loop_escalated intentionally absent
|
||||
}
|
||||
state = ConversationState.from_dict(data)
|
||||
assert state.loop_escalated is False
|
||||
|
||||
Reference in New Issue
Block a user