"""Coverage tests for notifications/templates.py — template rendering branches.""" from unittest.mock import MagicMock, patch import pytest from local_deep_research.notifications.templates import ( EventType, NotificationTemplate, ) MODULE = "local_deep_research.notifications.templates" @pytest.fixture(autouse=True) def _reset_jinja_env(): """Reset cached Jinja2 env between tests.""" NotificationTemplate._jinja_env = None yield NotificationTemplate._jinja_env = None class TestFormatWithCustomTemplate: def test_custom_template_success(self): result = NotificationTemplate.format( EventType.TEST, {"query": "climate change", "status": "done"}, custom_template={ "title": "Research: {query}", "body": "Status: {status}", }, ) assert result["title"] == "Research: climate change" assert result["body"] == "Status: done" def test_custom_template_missing_variable(self): result = NotificationTemplate.format( EventType.TEST, {"query": "test"}, custom_template={ "title": "Research: {query}", "body": "By: {username}", }, ) assert "Missing variable" in result["body"] def test_custom_template_sanitizes_context(self): """Context values are coerced to strings to prevent format attacks.""" result = NotificationTemplate.format( EventType.TEST, {"query": 42, "none_val": None}, custom_template={ "title": "{query}", "body": "{none_val}", }, ) assert result["title"] == "42" assert result["body"] == "" class TestFormatWithJinjaTemplates: def test_research_completed_template(self): result = NotificationTemplate.format( EventType.RESEARCH_COMPLETED, { "query": "test query", "research_url": "http://localhost/research/123", "duration": "5m", }, ) assert "title" in result assert "body" in result assert isinstance(result["title"], str) def test_research_failed_template(self): result = NotificationTemplate.format( EventType.RESEARCH_FAILED, { "query": "test", "error": "Connection timeout", "research_url": "http://localhost/research/456", }, ) assert "title" in result def test_research_queued_template(self): result = NotificationTemplate.format( EventType.RESEARCH_QUEUED, { "query": "test", "position": 3, }, ) assert "title" in result def test_test_event_template(self): result = NotificationTemplate.format( EventType.TEST, {"message": "hello"}, ) assert "title" in result class TestFormatUnknownEventType: def test_unmapped_event_returns_fallback(self): """EventType without template file → generic fallback.""" # Use RATE_LIMIT_WARNING which may not have a template file result = NotificationTemplate.format( EventType.RATE_LIMIT_WARNING, {"detail": "rate limited"}, ) assert "title" in result assert isinstance(result["title"], str) class TestFormatJinjaUnavailable: def test_fallback_when_jinja_env_none(self): with patch.object( NotificationTemplate, "_get_jinja_env", return_value=None ): result = NotificationTemplate.format( EventType.RESEARCH_COMPLETED, {"query": "test"}, ) assert "title" in result assert "Notification" in result["title"] def test_fallback_when_jinja_render_fails(self): mock_env = MagicMock() mock_env.get_template.side_effect = RuntimeError("render error") with patch.object( NotificationTemplate, "_get_jinja_env", return_value=mock_env ): result = NotificationTemplate.format( EventType.RESEARCH_COMPLETED, {"query": "test"}, ) assert "title" in result class TestGetFallbackTemplate: def test_returns_title_and_body(self): result = NotificationTemplate._get_fallback_template( EventType.RESEARCH_COMPLETED, {"query": "test"}, ) assert "Research Completed" in result["title"] assert "Details:" in result["body"] class TestGetRequiredContext: def test_known_event_returns_list(self): result = NotificationTemplate.get_required_context( EventType.RESEARCH_COMPLETED ) assert isinstance(result, list) def test_unknown_event_returns_empty(self): result = NotificationTemplate.get_required_context( EventType.RATE_LIMIT_WARNING ) assert isinstance(result, list) def test_jinja_unavailable_returns_empty(self): with patch.object( NotificationTemplate, "_get_jinja_env", return_value=None ): result = NotificationTemplate.get_required_context( EventType.RESEARCH_COMPLETED ) assert result == [] class TestGetJinjaEnv: def test_returns_environment_object(self): env = NotificationTemplate._get_jinja_env() assert env is not None def test_caches_environment(self): env1 = NotificationTemplate._get_jinja_env() env2 = NotificationTemplate._get_jinja_env() assert env1 is env2 def test_missing_templates_dir_returns_none(self): with patch(f"{MODULE}.Path") as mock_path: mock_path.return_value.__truediv__ = MagicMock( return_value=MagicMock(exists=MagicMock(return_value=False)) ) NotificationTemplate._jinja_env = None NotificationTemplate._get_jinja_env() # May return None or cached env depending on Path mock depth class TestEventTypeEnum: def test_all_event_types_exist(self): assert EventType.RESEARCH_COMPLETED.value == "research_completed" assert EventType.RESEARCH_FAILED.value == "research_failed" assert EventType.RESEARCH_QUEUED.value == "research_queued" assert EventType.TEST.value == "test" def test_event_type_count(self): assert len(EventType) >= 9