Add pytest test suite (92 tests, all passing)
Covers every module in src/ with unit tests:
- tests/conftest.py shared fixtures (complete_registration, fresh_state, …)
- tests/test_models.py RegistrationData.is_complete(), to_dict/from_dict round-trips
- tests/test_storage.py normalize_email, _diff_registrations, ConversationStore CRUD,
registration versioning
- tests/test_llm.py litellm wrapper — message construction, model passthrough,
error propagation
- tests/test_agent.py EmailAgent — new/existing conversations, registration
completion, admin notification, fallback on LLM error,
JSON parsing, _apply_updates
- tests/test_notifier.py AdminNotifier routing, fee calculation, SMTP dispatch
- tests/test_knowledge_base.py KnowledgeBase loading and reload
All external I/O (litellm, SMTP, filesystem) is mocked. Tests run fast (~6s)
with no network access required.
https://claude.ai/code/session_01HaUFs7SaLD5SoiuGCY27Tw
This commit is contained in:
Claude
@@ -0,0 +1,61 @@
|
||||
"""Shared pytest fixtures."""
|
||||
|
||||
import pytest
|
||||
|
||||
from src.models.conversation import ConversationState, ChatMessage
|
||||
from src.models.registration import (
|
||||
RegistrationData,
|
||||
ChildInfo,
|
||||
ParentGuardian,
|
||||
EmergencyContact,
|
||||
Booking,
|
||||
BookingDay,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def complete_registration() -> RegistrationData:
|
||||
"""A fully populated RegistrationData that passes is_complete()."""
|
||||
return RegistrationData(
|
||||
child=ChildInfo(
|
||||
full_name="Lena Muster",
|
||||
date_of_birth="2022-03-15",
|
||||
special_needs="None",
|
||||
),
|
||||
parent_guardian=ParentGuardian(
|
||||
full_name="Anna Muster",
|
||||
street_address="Hauptstrasse 1",
|
||||
postal_code="8117",
|
||||
city="Fällanden",
|
||||
phone="044 123 45 67",
|
||||
email="anna.muster@example.com",
|
||||
),
|
||||
emergency_contact=EmergencyContact(
|
||||
full_name="Hans Muster",
|
||||
phone="079 123 45 67",
|
||||
),
|
||||
booking=Booking(
|
||||
playgroup_types=["indoor"],
|
||||
selected_days=[BookingDay(day="monday", type="indoor")],
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fresh_state() -> ConversationState:
|
||||
"""A brand-new ConversationState for a parent email."""
|
||||
return ConversationState(
|
||||
conversation_id="anna.muster@example.com",
|
||||
parent_email="anna.muster@example.com",
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def state_with_messages(fresh_state) -> ConversationState:
|
||||
"""A ConversationState with a couple of chat turns."""
|
||||
fresh_state.messages = [
|
||||
ChatMessage(role="user", content="Hallo, ich möchte mein Kind anmelden."),
|
||||
ChatMessage(role="assistant", content="Hallo! Wie heisst dein Kind?"),
|
||||
ChatMessage(role="user", content="Lena Muster"),
|
||||
]
|
||||
return fresh_state
|
||||
@@ -0,0 +1,271 @@
|
||||
"""Tests for EmailAgent — the conversation orchestrator."""
|
||||
|
||||
import json
|
||||
import pytest
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from src.agent.core import EmailAgent
|
||||
from src.models.conversation import ConversationState
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers / fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
VALID_LLM_REPLY = json.dumps({
|
||||
"reply": "Wie heisst dein Kind?",
|
||||
"updates": {},
|
||||
"next_step": "child_name",
|
||||
"registration_complete": False,
|
||||
"language": "de",
|
||||
})
|
||||
|
||||
COMPLETION_LLM_REPLY = json.dumps({
|
||||
"reply": "Vielen Dank, dein Kind ist angemeldet!",
|
||||
"updates": {},
|
||||
"next_step": "done",
|
||||
"registration_complete": True,
|
||||
"language": "de",
|
||||
})
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_kb():
|
||||
kb = MagicMock()
|
||||
kb.get_all.return_value = "# FAQ\nSome knowledge base content."
|
||||
return kb
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_store():
|
||||
store = MagicMock()
|
||||
store.load.return_value = None # no prior conversation by default
|
||||
store.save_registration.return_value = ("anna.muster@example.com", 1)
|
||||
return store
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_notifier():
|
||||
return MagicMock()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def agent(mock_kb, mock_store, mock_notifier):
|
||||
return EmailAgent(
|
||||
model="anthropic/claude-opus-4-6",
|
||||
kb=mock_kb,
|
||||
store=mock_store,
|
||||
notifier=mock_notifier,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# process_message — new conversation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestProcessMessageNewConversation:
|
||||
def test_creates_new_state_when_none_exists(self, agent, mock_store):
|
||||
with patch("src.llm.complete", return_value=VALID_LLM_REPLY):
|
||||
agent.process_message("anna.muster@example.com", "Hallo")
|
||||
|
||||
saved_state = mock_store.save.call_args[0][0]
|
||||
assert saved_state.conversation_id == "anna.muster@example.com"
|
||||
|
||||
def test_returns_llm_reply_text(self, agent):
|
||||
with patch("src.llm.complete", return_value=VALID_LLM_REPLY):
|
||||
reply = agent.process_message("anna.muster@example.com", "Hallo")
|
||||
|
||||
assert reply == "Wie heisst dein Kind?"
|
||||
|
||||
def test_user_message_added_to_history(self, agent, mock_store):
|
||||
with patch("src.llm.complete", return_value=VALID_LLM_REPLY):
|
||||
agent.process_message("anna.muster@example.com", "Hallo, ich möchte anmelden")
|
||||
|
||||
state = mock_store.save.call_args[0][0]
|
||||
assert any(m.role == "user" and "anmelden" in m.content for m in state.messages)
|
||||
|
||||
def test_assistant_reply_added_to_history(self, agent, mock_store):
|
||||
with patch("src.llm.complete", return_value=VALID_LLM_REPLY):
|
||||
agent.process_message("anna.muster@example.com", "Hallo")
|
||||
|
||||
state = mock_store.save.call_args[0][0]
|
||||
assert any(m.role == "assistant" for m in state.messages)
|
||||
|
||||
def test_normalizes_email_key(self, agent, mock_store):
|
||||
with patch("src.llm.complete", return_value=VALID_LLM_REPLY):
|
||||
agent.process_message("Anna.Muster@EXAMPLE.COM", "Hallo")
|
||||
|
||||
state = mock_store.save.call_args[0][0]
|
||||
assert state.conversation_id == "anna.muster@example.com"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# process_message — existing conversation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestProcessMessageExistingConversation:
|
||||
def test_loads_existing_state(self, agent, mock_store, fresh_state):
|
||||
mock_store.load.return_value = fresh_state
|
||||
|
||||
with patch("src.llm.complete", return_value=VALID_LLM_REPLY):
|
||||
agent.process_message("anna.muster@example.com", "Lena")
|
||||
|
||||
mock_store.load.assert_called_once()
|
||||
|
||||
def test_flow_step_updated(self, agent, mock_store, fresh_state):
|
||||
mock_store.load.return_value = fresh_state
|
||||
|
||||
reply_with_step = json.dumps({
|
||||
"reply": "Wann ist Lena geboren?",
|
||||
"updates": {"child.fullName": "Lena"},
|
||||
"next_step": "child_dob",
|
||||
"registration_complete": False,
|
||||
"language": "de",
|
||||
})
|
||||
|
||||
with patch("src.llm.complete", return_value=reply_with_step):
|
||||
agent.process_message("anna.muster@example.com", "Lena")
|
||||
|
||||
state = mock_store.save.call_args[0][0]
|
||||
assert state.flow_step == "child_dob"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# process_message — registration completion
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestRegistrationCompletion:
|
||||
def test_notifier_called_on_completion(self, agent, mock_store, mock_notifier, complete_registration):
|
||||
state = ConversationState(
|
||||
conversation_id="anna.muster@example.com",
|
||||
parent_email="anna.muster@example.com",
|
||||
)
|
||||
state.registration = complete_registration
|
||||
mock_store.load.return_value = state
|
||||
|
||||
with patch("src.llm.complete", return_value=COMPLETION_LLM_REPLY):
|
||||
agent.process_message("anna.muster@example.com", "Ja, alles korrekt")
|
||||
|
||||
mock_notifier.notify_admin.assert_called_once()
|
||||
|
||||
def test_state_marked_completed(self, agent, mock_store, complete_registration):
|
||||
state = ConversationState(
|
||||
conversation_id="anna.muster@example.com",
|
||||
parent_email="anna.muster@example.com",
|
||||
)
|
||||
state.registration = complete_registration
|
||||
mock_store.load.return_value = state
|
||||
|
||||
with patch("src.llm.complete", return_value=COMPLETION_LLM_REPLY):
|
||||
agent.process_message("anna.muster@example.com", "Ja")
|
||||
|
||||
saved = mock_store.save.call_args[0][0]
|
||||
assert saved.completed is True
|
||||
|
||||
def test_notifier_not_called_when_already_completed(self, agent, mock_store, mock_notifier, complete_registration):
|
||||
state = ConversationState(
|
||||
conversation_id="anna.muster@example.com",
|
||||
parent_email="anna.muster@example.com",
|
||||
)
|
||||
state.registration = complete_registration
|
||||
state.completed = True # already done
|
||||
mock_store.load.return_value = state
|
||||
|
||||
with patch("src.llm.complete", return_value=COMPLETION_LLM_REPLY):
|
||||
agent.process_message("anna.muster@example.com", "Noch eine Frage")
|
||||
|
||||
mock_notifier.notify_admin.assert_not_called()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fallback on LLM error
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestFallbackOnLLMError:
|
||||
def test_returns_german_fallback_by_default(self, agent):
|
||||
with patch("src.llm.complete", side_effect=RuntimeError("API down")):
|
||||
reply = agent.process_message("anna.muster@example.com", "Hallo")
|
||||
|
||||
assert "technisches Problem" in reply or "Entschuldigung" in reply
|
||||
|
||||
def test_returns_english_fallback_when_language_is_en(self, agent, mock_store, fresh_state):
|
||||
fresh_state.language = "en"
|
||||
mock_store.load.return_value = fresh_state
|
||||
|
||||
with patch("src.llm.complete", side_effect=RuntimeError("API down")):
|
||||
reply = agent.process_message("anna.muster@example.com", "Hello")
|
||||
|
||||
assert "technical issue" in reply.lower() or "sorry" in reply.lower()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _parse_llm_response
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestParseLlmResponse:
|
||||
def test_parses_plain_json(self, agent):
|
||||
payload = '{"reply": "Hi", "updates": {}, "next_step": "greeting", "registration_complete": false, "language": "de"}'
|
||||
result = agent._parse_llm_response(payload)
|
||||
assert result["reply"] == "Hi"
|
||||
|
||||
def test_parses_fenced_json(self, agent):
|
||||
payload = '```json\n{"reply": "Hi", "updates": {}}\n```'
|
||||
result = agent._parse_llm_response(payload)
|
||||
assert result["reply"] == "Hi"
|
||||
|
||||
def test_parses_json_embedded_in_text(self, agent):
|
||||
payload = 'Sure, here is the response: {"reply": "Hi", "updates": {}}'
|
||||
result = agent._parse_llm_response(payload)
|
||||
assert result["reply"] == "Hi"
|
||||
|
||||
def test_falls_back_to_raw_text_when_no_json(self, agent):
|
||||
result = agent._parse_llm_response("Ich bin ein Hilfsroboter")
|
||||
assert result["reply"] == "Ich bin ein Hilfsroboter"
|
||||
|
||||
def test_fallback_has_safe_defaults(self, agent):
|
||||
result = agent._parse_llm_response("plain text")
|
||||
assert result["registration_complete"] is False
|
||||
assert result["updates"] == {}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _apply_updates
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestApplyUpdates:
|
||||
def test_sets_child_name(self, agent, fresh_state):
|
||||
agent._apply_updates(fresh_state, {"child.fullName": "Lena Muster"})
|
||||
assert fresh_state.registration.child.full_name == "Lena Muster"
|
||||
|
||||
def test_sets_child_dob(self, agent, fresh_state):
|
||||
agent._apply_updates(fresh_state, {"child.dateOfBirth": "2022-03-15"})
|
||||
assert fresh_state.registration.child.date_of_birth == "2022-03-15"
|
||||
|
||||
def test_sets_parent_email(self, agent, fresh_state):
|
||||
agent._apply_updates(fresh_state, {"parentGuardian.email": "test@example.com"})
|
||||
assert fresh_state.registration.parent_guardian.email == "test@example.com"
|
||||
|
||||
def test_sets_emergency_contact(self, agent, fresh_state):
|
||||
agent._apply_updates(fresh_state, {"emergencyContact.phone": "079 111 22 33"})
|
||||
assert fresh_state.registration.emergency_contact.phone == "079 111 22 33"
|
||||
|
||||
def test_sets_booking_days(self, agent, fresh_state):
|
||||
agent._apply_updates(fresh_state, {
|
||||
"booking.selectedDays": [{"day": "wednesday", "type": "indoor"}]
|
||||
})
|
||||
assert fresh_state.registration.booking.selected_days[0].day == "wednesday"
|
||||
|
||||
def test_ignores_none_values(self, agent, fresh_state):
|
||||
fresh_state.registration.child.full_name = "Lena"
|
||||
agent._apply_updates(fresh_state, {"child.fullName": None})
|
||||
assert fresh_state.registration.child.full_name == "Lena"
|
||||
|
||||
def test_ignores_unknown_keys(self, agent, fresh_state):
|
||||
agent._apply_updates(fresh_state, {"unknown.key": "value"}) # should not raise
|
||||
@@ -0,0 +1,52 @@
|
||||
"""Tests for KnowledgeBase loader."""
|
||||
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
|
||||
from src.knowledge_base.loader import KnowledgeBase
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def kb_dir(tmp_path) -> Path:
|
||||
"""A temporary knowledge-base directory with a couple of markdown files."""
|
||||
(tmp_path / "faq.md").write_text("# FAQ\nWann beginnt die Spielgruppe?\nIm August.")
|
||||
(tmp_path / "fees.md").write_text("# Fees\nCHF 130 per month.")
|
||||
return tmp_path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def kb(kb_dir) -> KnowledgeBase:
|
||||
return KnowledgeBase(kb_dir)
|
||||
|
||||
|
||||
class TestKnowledgeBaseLoading:
|
||||
def test_get_all_includes_file_content(self, kb):
|
||||
content = kb.get_all()
|
||||
assert "FAQ" in content
|
||||
assert "Fees" in content
|
||||
|
||||
def test_get_all_concatenates_multiple_files(self, kb):
|
||||
content = kb.get_all()
|
||||
assert "CHF 130" in content
|
||||
assert "Spielgruppe" in content
|
||||
|
||||
def test_reload_picks_up_new_file(self, kb, kb_dir):
|
||||
(kb_dir / "schedule.md").write_text("# Schedule\nMonday 9:00")
|
||||
kb.reload()
|
||||
assert "Schedule" in kb.get_all()
|
||||
|
||||
def test_empty_directory_returns_empty_string(self, tmp_path):
|
||||
kb = KnowledgeBase(tmp_path)
|
||||
assert kb.get_all() == "" or isinstance(kb.get_all(), str)
|
||||
|
||||
def test_nonexistent_directory_does_not_raise_on_init(self, tmp_path):
|
||||
# Should either handle gracefully or raise — just must not crash silently
|
||||
missing = tmp_path / "does_not_exist"
|
||||
try:
|
||||
kb = KnowledgeBase(missing)
|
||||
kb.get_all()
|
||||
except (FileNotFoundError, OSError):
|
||||
pass # Acceptable to raise on missing dir
|
||||
|
||||
def test_get_all_returns_string(self, kb):
|
||||
assert isinstance(kb.get_all(), str)
|
||||
@@ -0,0 +1,67 @@
|
||||
"""Tests for the litellm wrapper in src/llm.py."""
|
||||
|
||||
import pytest
|
||||
|
||||
from src import llm
|
||||
from src.models.conversation import ChatMessage
|
||||
|
||||
|
||||
class TestLlmComplete:
|
||||
def test_returns_model_reply(self, mocker):
|
||||
mock_response = mocker.MagicMock()
|
||||
mock_response.choices[0].message.content = "Hallo! Wie heisst dein Kind?"
|
||||
mocker.patch("litellm.completion", return_value=mock_response)
|
||||
|
||||
result = llm.complete("anthropic/claude-opus-4-6", "system prompt", [])
|
||||
|
||||
assert result == "Hallo! Wie heisst dein Kind?"
|
||||
|
||||
def test_passes_model_to_litellm(self, mocker):
|
||||
mock_response = mocker.MagicMock()
|
||||
mock_response.choices[0].message.content = "ok"
|
||||
mock_completion = mocker.patch("litellm.completion", return_value=mock_response)
|
||||
|
||||
llm.complete("openai/gpt-4o", "system", [])
|
||||
|
||||
call_kwargs = mock_completion.call_args.kwargs
|
||||
assert call_kwargs["model"] == "openai/gpt-4o"
|
||||
|
||||
def test_system_prompt_prepended_as_system_message(self, mocker):
|
||||
mock_response = mocker.MagicMock()
|
||||
mock_response.choices[0].message.content = "ok"
|
||||
mock_completion = mocker.patch("litellm.completion", return_value=mock_response)
|
||||
|
||||
llm.complete("anthropic/claude-opus-4-6", "You are helpful.", [])
|
||||
|
||||
messages = mock_completion.call_args.kwargs["messages"]
|
||||
assert messages[0] == {"role": "system", "content": "You are helpful."}
|
||||
|
||||
def test_chat_messages_appended_after_system(self, mocker):
|
||||
mock_response = mocker.MagicMock()
|
||||
mock_response.choices[0].message.content = "ok"
|
||||
mock_completion = mocker.patch("litellm.completion", return_value=mock_response)
|
||||
|
||||
chat = [
|
||||
ChatMessage(role="user", content="Hallo"),
|
||||
ChatMessage(role="assistant", content="Guten Tag"),
|
||||
]
|
||||
llm.complete("anthropic/claude-opus-4-6", "system", chat)
|
||||
|
||||
messages = mock_completion.call_args.kwargs["messages"]
|
||||
assert messages[1] == {"role": "user", "content": "Hallo"}
|
||||
assert messages[2] == {"role": "assistant", "content": "Guten Tag"}
|
||||
|
||||
def test_max_tokens_passed(self, mocker):
|
||||
mock_response = mocker.MagicMock()
|
||||
mock_response.choices[0].message.content = "ok"
|
||||
mock_completion = mocker.patch("litellm.completion", return_value=mock_response)
|
||||
|
||||
llm.complete("anthropic/claude-opus-4-6", "system", [])
|
||||
|
||||
assert mock_completion.call_args.kwargs["max_tokens"] == 2048
|
||||
|
||||
def test_litellm_exception_propagates(self, mocker):
|
||||
mocker.patch("litellm.completion", side_effect=RuntimeError("API error"))
|
||||
|
||||
with pytest.raises(RuntimeError, match="API error"):
|
||||
llm.complete("anthropic/claude-opus-4-6", "system", [])
|
||||
@@ -0,0 +1,152 @@
|
||||
"""Tests for data models: RegistrationData and ConversationState."""
|
||||
|
||||
import pytest
|
||||
|
||||
from src.models.registration import (
|
||||
RegistrationData,
|
||||
ChildInfo,
|
||||
ParentGuardian,
|
||||
EmergencyContact,
|
||||
Booking,
|
||||
BookingDay,
|
||||
)
|
||||
from src.models.conversation import ConversationState, ChatMessage
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# RegistrationData.is_complete()
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestRegistrationDataIsComplete:
|
||||
def test_complete_registration_passes(self, complete_registration):
|
||||
assert complete_registration.is_complete() is True
|
||||
|
||||
def test_empty_registration_fails(self):
|
||||
assert RegistrationData().is_complete() is False
|
||||
|
||||
def test_missing_child_name_fails(self, complete_registration):
|
||||
complete_registration.child.full_name = None
|
||||
assert complete_registration.is_complete() is False
|
||||
|
||||
def test_missing_dob_fails(self, complete_registration):
|
||||
complete_registration.child.date_of_birth = None
|
||||
assert complete_registration.is_complete() is False
|
||||
|
||||
def test_missing_special_needs_fails(self, complete_registration):
|
||||
complete_registration.child.special_needs = None
|
||||
assert complete_registration.is_complete() is False
|
||||
|
||||
def test_missing_parent_name_fails(self, complete_registration):
|
||||
complete_registration.parent_guardian.full_name = None
|
||||
assert complete_registration.is_complete() is False
|
||||
|
||||
def test_missing_parent_email_fails(self, complete_registration):
|
||||
complete_registration.parent_guardian.email = None
|
||||
assert complete_registration.is_complete() is False
|
||||
|
||||
def test_missing_emergency_contact_fails(self, complete_registration):
|
||||
complete_registration.emergency_contact.full_name = None
|
||||
assert complete_registration.is_complete() is False
|
||||
|
||||
def test_missing_booking_days_fails(self, complete_registration):
|
||||
complete_registration.booking.selected_days = []
|
||||
assert complete_registration.is_complete() is False
|
||||
|
||||
def test_missing_playgroup_types_fails(self, complete_registration):
|
||||
complete_registration.booking.playgroup_types = []
|
||||
assert complete_registration.is_complete() is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# RegistrationData serialisation round-trip
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestRegistrationDataSerialization:
|
||||
def test_to_dict_contains_expected_keys(self, complete_registration):
|
||||
d = complete_registration.to_dict()
|
||||
assert "child" in d
|
||||
assert "parentGuardian" in d
|
||||
assert "emergencyContact" in d
|
||||
assert "booking" in d
|
||||
|
||||
def test_to_dict_child_fields(self, complete_registration):
|
||||
d = complete_registration.to_dict()
|
||||
assert d["child"]["fullName"] == "Lena Muster"
|
||||
assert d["child"]["dateOfBirth"] == "2022-03-15"
|
||||
assert d["child"]["specialNeeds"] == "None"
|
||||
|
||||
def test_to_dict_parent_fields(self, complete_registration):
|
||||
d = complete_registration.to_dict()
|
||||
assert d["parentGuardian"]["email"] == "anna.muster@example.com"
|
||||
assert d["parentGuardian"]["postalCode"] == "8117"
|
||||
|
||||
def test_to_dict_booking_fields(self, complete_registration):
|
||||
d = complete_registration.to_dict()
|
||||
assert d["booking"]["playgroupTypes"] == ["indoor"]
|
||||
assert d["booking"]["selectedDays"] == [{"day": "monday", "type": "indoor"}]
|
||||
|
||||
def test_from_dict_round_trip(self, complete_registration):
|
||||
d = complete_registration.to_dict()
|
||||
restored = RegistrationData.from_dict(d)
|
||||
assert restored.child.full_name == complete_registration.child.full_name
|
||||
assert restored.parent_guardian.email == complete_registration.parent_guardian.email
|
||||
assert restored.emergency_contact.phone == complete_registration.emergency_contact.phone
|
||||
assert len(restored.booking.selected_days) == len(complete_registration.booking.selected_days)
|
||||
|
||||
def test_from_dict_outdoor_booking(self):
|
||||
data = {
|
||||
"child": {"fullName": "Tim", "dateOfBirth": "2021-01-01", "specialNeeds": "None"},
|
||||
"parentGuardian": {
|
||||
"fullName": "Eva", "streetAddress": "Seeweg 2", "postalCode": "8117",
|
||||
"city": "Fällanden", "phone": "044 000 00 00", "email": "eva@example.com",
|
||||
},
|
||||
"emergencyContact": {"fullName": "Bob", "phone": "079 000 00 00"},
|
||||
"booking": {
|
||||
"playgroupTypes": ["outdoor"],
|
||||
"selectedDays": [{"day": "monday", "type": "outdoor"}],
|
||||
},
|
||||
}
|
||||
reg = RegistrationData.from_dict(data)
|
||||
assert reg.booking.playgroup_types == ["outdoor"]
|
||||
assert reg.booking.selected_days[0].day == "monday"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ConversationState serialisation round-trip
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestConversationStateSerialization:
|
||||
def test_to_dict_contains_expected_keys(self, fresh_state):
|
||||
d = fresh_state.to_dict()
|
||||
assert "conversation_id" in d
|
||||
assert "language" in d
|
||||
assert "flow_step" in d
|
||||
assert "messages" in d
|
||||
assert "completed" in d
|
||||
|
||||
def test_default_language_is_german(self, fresh_state):
|
||||
assert fresh_state.language == "de"
|
||||
|
||||
def test_default_flow_step_is_greeting(self, fresh_state):
|
||||
assert fresh_state.flow_step == "greeting"
|
||||
|
||||
def test_default_completed_is_false(self, fresh_state):
|
||||
assert fresh_state.completed is False
|
||||
|
||||
def test_from_dict_round_trip(self, state_with_messages):
|
||||
state_with_messages.language = "en"
|
||||
state_with_messages.flow_step = "parent_name"
|
||||
d = state_with_messages.to_dict()
|
||||
restored = ConversationState.from_dict(d)
|
||||
assert restored.conversation_id == state_with_messages.conversation_id
|
||||
assert restored.language == "en"
|
||||
assert restored.flow_step == "parent_name"
|
||||
assert len(restored.messages) == len(state_with_messages.messages)
|
||||
|
||||
def test_messages_serialized_with_role_and_content(self, state_with_messages):
|
||||
d = state_with_messages.to_dict()
|
||||
assert d["messages"][0]["role"] == "user"
|
||||
assert "Hallo" in d["messages"][0]["content"]
|
||||
@@ -0,0 +1,155 @@
|
||||
"""Tests for AdminNotifier helper methods."""
|
||||
|
||||
import pytest
|
||||
|
||||
from src.notifications.notifier import AdminNotifier
|
||||
from src.models.registration import RegistrationData, Booking, BookingDay
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def notifier():
|
||||
return AdminNotifier(
|
||||
smtp_host="smtp.example.com",
|
||||
smtp_port=587,
|
||||
username="agent@example.com",
|
||||
password="secret",
|
||||
use_tls=True,
|
||||
from_email="agent@example.com",
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _recipients_for
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestRecipientsFor:
|
||||
def test_indoor_only_recipients(self, notifier):
|
||||
recipients = notifier._recipients_for(["indoor"])
|
||||
assert any("andrea" in r.lower() or "sigrist" in r.lower() for r in recipients)
|
||||
|
||||
def test_outdoor_only_recipients(self, notifier):
|
||||
recipients = notifier._recipients_for(["outdoor"])
|
||||
assert any("baba.laeubli" in r.lower() for r in recipients)
|
||||
|
||||
def test_both_includes_both_leaders(self, notifier):
|
||||
recipients = notifier._recipients_for(["indoor", "outdoor"])
|
||||
joined = " ".join(recipients).lower()
|
||||
assert "andrea.sigrist" in joined
|
||||
assert "baba.laeubli" in joined
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _format_types
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestFormatTypes:
|
||||
def test_indoor_label(self, notifier):
|
||||
assert "Innen" in notifier._format_types(["indoor"]) or "indoor" in notifier._format_types(["indoor"]).lower()
|
||||
|
||||
def test_outdoor_label(self, notifier):
|
||||
assert "Wald" in notifier._format_types(["outdoor"]) or "outdoor" in notifier._format_types(["outdoor"]).lower()
|
||||
|
||||
def test_both_labels(self, notifier):
|
||||
result = notifier._format_types(["indoor", "outdoor"])
|
||||
assert len(result) > 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _calculate_age
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCalculateAge:
|
||||
def test_returns_age_string(self, notifier):
|
||||
result = notifier._calculate_age("2022-01-01")
|
||||
assert isinstance(result, str)
|
||||
assert len(result) > 0
|
||||
|
||||
def test_invalid_dob_returns_original_string(self, notifier):
|
||||
result = notifier._calculate_age("not-a-date")
|
||||
assert result == "not-a-date"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _calculate_monthly_fee
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCalculateMonthlyFee:
|
||||
def test_indoor_one_day(self, notifier, complete_registration):
|
||||
complete_registration.booking = Booking(
|
||||
playgroup_types=["indoor"],
|
||||
selected_days=[BookingDay(day="monday", type="indoor")],
|
||||
)
|
||||
fee = notifier._calculate_monthly_fee(complete_registration)
|
||||
assert "130" in fee
|
||||
|
||||
def test_indoor_two_days(self, notifier, complete_registration):
|
||||
complete_registration.booking = Booking(
|
||||
playgroup_types=["indoor"],
|
||||
selected_days=[
|
||||
BookingDay(day="monday", type="indoor"),
|
||||
BookingDay(day="wednesday", type="indoor"),
|
||||
],
|
||||
)
|
||||
fee = notifier._calculate_monthly_fee(complete_registration)
|
||||
assert "260" in fee
|
||||
|
||||
def test_indoor_three_days(self, notifier, complete_registration):
|
||||
complete_registration.booking = Booking(
|
||||
playgroup_types=["indoor"],
|
||||
selected_days=[
|
||||
BookingDay(day="monday", type="indoor"),
|
||||
BookingDay(day="wednesday", type="indoor"),
|
||||
BookingDay(day="thursday", type="indoor"),
|
||||
],
|
||||
)
|
||||
fee = notifier._calculate_monthly_fee(complete_registration)
|
||||
assert "390" in fee
|
||||
|
||||
def test_outdoor_one_day(self, notifier, complete_registration):
|
||||
complete_registration.booking = Booking(
|
||||
playgroup_types=["outdoor"],
|
||||
selected_days=[BookingDay(day="monday", type="outdoor")],
|
||||
)
|
||||
fee = notifier._calculate_monthly_fee(complete_registration)
|
||||
assert "250" in fee
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _send — SMTP interaction
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestSend:
|
||||
def test_send_calls_smtp(self, notifier, mocker):
|
||||
# _send uses smtplib.SMTP directly (not as context manager)
|
||||
mock_smtp_cls = mocker.patch("smtplib.SMTP")
|
||||
mock_server = mock_smtp_cls.return_value
|
||||
|
||||
notifier._send(
|
||||
to=["admin@example.com"],
|
||||
cc=["cc@example.com"],
|
||||
subject="Test",
|
||||
body="Hello",
|
||||
)
|
||||
|
||||
mock_server.sendmail.assert_called_once()
|
||||
|
||||
def test_send_includes_all_recipients(self, notifier, mocker):
|
||||
mock_smtp_cls = mocker.patch("smtplib.SMTP")
|
||||
mock_server = mock_smtp_cls.return_value
|
||||
|
||||
notifier._send(
|
||||
to=["a@example.com"],
|
||||
cc=["b@example.com"],
|
||||
subject="Test",
|
||||
body="Hello",
|
||||
)
|
||||
|
||||
call_args = mock_server.sendmail.call_args
|
||||
recipients = call_args[0][1] # positional arg: to_addrs
|
||||
assert "a@example.com" in recipients
|
||||
assert "b@example.com" in recipients
|
||||
@@ -0,0 +1,159 @@
|
||||
"""Tests for ConversationStore and storage helpers."""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from src.storage.json_store import (
|
||||
ConversationStore,
|
||||
normalize_email,
|
||||
_diff_registrations,
|
||||
)
|
||||
from src.models.conversation import ConversationState
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# normalize_email
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestNormalizeEmail:
|
||||
def test_lowercases(self):
|
||||
assert normalize_email("Anna.Muster@Example.COM") == "anna.muster@example.com"
|
||||
|
||||
def test_strips_whitespace(self):
|
||||
assert normalize_email(" user@example.com ") == "user@example.com"
|
||||
|
||||
def test_already_normalized(self):
|
||||
assert normalize_email("user@example.com") == "user@example.com"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _diff_registrations
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestDiffRegistrations:
|
||||
def test_detects_changed_field(self):
|
||||
old = {"child": {"fullName": "Lena"}}
|
||||
new = {"child": {"fullName": "Lena Muster"}}
|
||||
diff = _diff_registrations(old, new)
|
||||
assert "child.fullName" in diff
|
||||
assert diff["child.fullName"] == ("Lena", "Lena Muster")
|
||||
|
||||
def test_unchanged_fields_not_included(self):
|
||||
old = {"child": {"fullName": "Lena", "dateOfBirth": "2022-01-01"}}
|
||||
new = {"child": {"fullName": "Lena", "dateOfBirth": "2022-01-01"}}
|
||||
assert _diff_registrations(old, new) == {}
|
||||
|
||||
def test_nested_change_detected(self):
|
||||
old = {"parentGuardian": {"email": "old@example.com"}}
|
||||
new = {"parentGuardian": {"email": "new@example.com"}}
|
||||
diff = _diff_registrations(old, new)
|
||||
assert "parentGuardian.email" in diff
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ConversationStore — CRUD
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def store(tmp_path) -> ConversationStore:
|
||||
return ConversationStore(tmp_path)
|
||||
|
||||
|
||||
class TestConversationStoreCRUD:
|
||||
def test_load_returns_none_for_unknown_email(self, store):
|
||||
assert store.load("nobody@example.com") is None
|
||||
|
||||
def test_save_and_load_round_trip(self, store, fresh_state):
|
||||
store.save(fresh_state)
|
||||
loaded = store.load(fresh_state.parent_email)
|
||||
assert loaded is not None
|
||||
assert loaded.conversation_id == fresh_state.conversation_id
|
||||
|
||||
def test_save_overwrites_existing(self, store, fresh_state):
|
||||
store.save(fresh_state)
|
||||
fresh_state.language = "en"
|
||||
store.save(fresh_state)
|
||||
loaded = store.load(fresh_state.parent_email)
|
||||
assert loaded.language == "en"
|
||||
|
||||
def test_delete_removes_conversation(self, store, fresh_state):
|
||||
store.save(fresh_state)
|
||||
store.delete(fresh_state.parent_email)
|
||||
assert store.load(fresh_state.parent_email) is None
|
||||
|
||||
def test_delete_nonexistent_is_silent(self, store):
|
||||
store.delete("ghost@example.com") # should not raise
|
||||
|
||||
def test_list_incomplete_returns_non_completed(self, store, fresh_state):
|
||||
store.save(fresh_state)
|
||||
incomplete = store.list_incomplete()
|
||||
assert any(s.conversation_id == fresh_state.conversation_id for s in incomplete)
|
||||
|
||||
def test_list_incomplete_excludes_completed(self, store, fresh_state):
|
||||
fresh_state.completed = True
|
||||
store.save(fresh_state)
|
||||
incomplete = store.list_incomplete()
|
||||
assert all(not s.completed for s in incomplete)
|
||||
|
||||
def test_find_by_email_is_alias_for_load(self, store, fresh_state):
|
||||
store.save(fresh_state)
|
||||
assert store.find_by_email(fresh_state.parent_email) is not None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ConversationStore — registration versioning
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestRegistrationVersioning:
|
||||
def test_save_registration_creates_version_1(self, store, fresh_state, complete_registration):
|
||||
fresh_state.registration = complete_registration
|
||||
fresh_state.completed = True
|
||||
email_key, version = store.save_registration(fresh_state)
|
||||
assert version == 1
|
||||
# email_key is the filesystem-safe form (@ → _at_)
|
||||
assert email_key == "anna.muster_at_example.com"
|
||||
|
||||
def test_save_registration_writes_current_json(self, store, fresh_state, complete_registration, tmp_path):
|
||||
fresh_state.registration = complete_registration
|
||||
fresh_state.completed = True
|
||||
email_key, _ = store.save_registration(fresh_state)
|
||||
current = tmp_path / "registrations" / email_key / "current.json"
|
||||
assert current.exists()
|
||||
|
||||
def test_save_registration_version_increments(self, store, fresh_state, complete_registration):
|
||||
fresh_state.registration = complete_registration
|
||||
fresh_state.completed = True
|
||||
store.save_registration(fresh_state)
|
||||
_, v2 = store.save_registration_version(
|
||||
fresh_state, {"child.fullName": ("Old", "New")}
|
||||
)
|
||||
assert v2 == 2
|
||||
|
||||
def test_get_current_registration_returns_latest(self, store, fresh_state, complete_registration):
|
||||
fresh_state.registration = complete_registration
|
||||
fresh_state.completed = True
|
||||
store.save_registration(fresh_state)
|
||||
current = store.get_current_registration(fresh_state.parent_email)
|
||||
assert current is not None
|
||||
assert current["metadata"]["version"] == 1
|
||||
|
||||
def test_get_registration_history_returns_all_versions(self, store, fresh_state, complete_registration):
|
||||
fresh_state.registration = complete_registration
|
||||
fresh_state.completed = True
|
||||
store.save_registration(fresh_state)
|
||||
store.save_registration_version(fresh_state, {"child.fullName": ("A", "B")})
|
||||
history = store.get_registration_history(fresh_state.parent_email)
|
||||
assert len(history) == 2
|
||||
|
||||
def test_list_registrations_includes_saved(self, store, fresh_state, complete_registration):
|
||||
fresh_state.registration = complete_registration
|
||||
fresh_state.completed = True
|
||||
store.save_registration(fresh_state)
|
||||
registrations = store.list_registrations()
|
||||
assert len(registrations) == 1
|
||||
Reference in New Issue
Block a user