From 904d292fb768cf43c0354a900f7bb361774e289d Mon Sep 17 00:00:00 2001 From: mic1on Date: Sat, 11 Jul 2026 21:13:55 +0800 Subject: [PATCH] test: expand channel and retry edge coverage --- src/use_notify/channels/email.py | 2 +- tests/test_channels.py | 26 ++++++++++++++ tests/test_email.py | 58 ++++++++++++++++++++++++++++++++ tests/test_notification.py | 58 ++++++++++++++++++++++++++++++++ 4 files changed, 143 insertions(+), 1 deletion(-) diff --git a/src/use_notify/channels/email.py b/src/use_notify/channels/email.py index eaa0eae..9a7df4f 100644 --- a/src/use_notify/channels/email.py +++ b/src/use_notify/channels/email.py @@ -67,7 +67,7 @@ def send(self, content, title=None): async def send_async(self, content, title=None): if not self.config.to_emails: - logger.error("请先设置接收邮箱") + logger.error("请先设置接收邮箱") return message = self.build_message(content, title) diff --git a/tests/test_channels.py b/tests/test_channels.py index c89fade..5c2c335 100644 --- a/tests/test_channels.py +++ b/tests/test_channels.py @@ -3,6 +3,7 @@ import pytest from use_notify import useNotifyChannel +from use_notify.channels.utils import ProviderResponseError, validate_business_response def _mock_sync_http_response(json_data=None): @@ -25,6 +26,31 @@ def _mock_async_http_response(json_data=None): return response +def test_validate_business_response_ignores_non_dict_json_payloads(): + response = _mock_sync_http_response(["ok"]) + + validate_business_response(response, "provider", {"code": {0}}) + + +def test_validate_business_response_serializes_dict_error_detail(): + response = _mock_sync_http_response({"code": 1, "error": {"reason": "bad token"}}) + + with pytest.raises(ProviderResponseError) as error_info: + validate_business_response(response, "provider", {"code": {0}}) + + assert '"reason": "bad token"' in str(error_info.value) + + +def test_validate_business_response_falls_back_to_payload_detail(): + response = _mock_sync_http_response({"code": 1, "detail": "bad token"}) + + with pytest.raises(ProviderResponseError) as error_info: + validate_business_response(response, "provider", {"code": {0}}) + + assert '"code": 1' in str(error_info.value) + assert '"detail": "bad token"' in str(error_info.value) + + @patch("httpx.Client") def test_bark_send_builds_expected_request(mock_client): response = _mock_sync_http_response() diff --git a/tests/test_email.py b/tests/test_email.py index 11c3d57..49a5fed 100644 --- a/tests/test_email.py +++ b/tests/test_email.py @@ -28,9 +28,16 @@ def test_email_validates_required_fields(): with pytest.raises(ValueError, match="缺少必填字段"): useNotifyChannel.Email({"server": "smtp.gmail.com", "port": 587}) + with pytest.raises(ValueError, match="缺少必填字段: port"): + useNotifyChannel.Email({k: v for k, v in EMAIL_CONFIG.items() if k != "port"}) + with pytest.raises(ValueError, match="端口号必须为有效的整数"): useNotifyChannel.Email({**EMAIL_CONFIG, "port": "bad-port"}) + for port in (0, 65536): + with pytest.raises(ValueError, match="端口号必须在1-65535范围内"): + useNotifyChannel.Email({**EMAIL_CONFIG, "port": port}) + for port in (True, False): with pytest.raises(ValueError, match="端口号必须为有效的整数"): useNotifyChannel.Email({**EMAIL_CONFIG, "port": port}) @@ -74,6 +81,31 @@ def test_email_send_uses_starttls_when_configured(mock_smtp): smtp.quit.assert_called_once_with() +@patch("smtplib.SMTP") +def test_email_send_uses_starttls_by_default_for_587(mock_smtp): + smtp = mock_smtp.return_value + channel = useNotifyChannel.Email({**EMAIL_CONFIG, "port": 587}) + + channel.send("hello", "title") + + mock_smtp.assert_called_once_with("smtp.gmail.com", 587) + smtp.starttls.assert_called_once_with() + smtp.login.assert_called_once_with("user@example.com", "secret") + smtp.sendmail.assert_called_once() + + +@patch("smtplib.SMTP") +def test_email_send_respects_disabled_starttls(mock_smtp): + smtp = mock_smtp.return_value + channel = useNotifyChannel.Email({**EMAIL_CONFIG, "port": 587, "use_tls": False}) + + channel.send("hello", "title") + + mock_smtp.assert_called_once_with("smtp.gmail.com", 587) + smtp.starttls.assert_not_called() + smtp.login.assert_called_once_with("user@example.com", "secret") + + @patch("smtplib.SMTP_SSL") @patch("asyncio.get_running_loop") @pytest.mark.asyncio @@ -103,3 +135,29 @@ def test_email_send_without_receivers_only_logs(mock_logger, mock_smtp, mock_smt mock_logger.error.assert_called_once_with("请先设置接收邮箱") mock_smtp.assert_not_called() mock_smtp_ssl.assert_not_called() + + +@patch("smtplib.SMTP_SSL") +@patch("smtplib.SMTP") +@patch("use_notify.channels.email.logger") +@pytest.mark.asyncio +async def test_email_send_async_without_receivers_only_logs(mock_logger, mock_smtp, mock_smtp_ssl): + channel = useNotifyChannel.Email({k: v for k, v in EMAIL_CONFIG.items() if k != "to_emails"}) + + await channel.send_async("hello") + + mock_logger.error.assert_called_once_with("请先设置接收邮箱") + mock_smtp.assert_not_called() + mock_smtp_ssl.assert_not_called() + + +@patch("smtplib.SMTP_SSL") +def test_email_close_falls_back_when_quit_fails(mock_smtp_ssl): + smtp = mock_smtp_ssl.return_value + smtp.quit.side_effect = OSError("socket already closed") + channel = useNotifyChannel.Email(EMAIL_CONFIG) + + channel.send("hello") + + smtp.quit.assert_called_once_with() + smtp.close.assert_called_once_with() diff --git a/tests/test_notification.py b/tests/test_notification.py index 6378301..af45067 100644 --- a/tests/test_notification.py +++ b/tests/test_notification.py @@ -1,4 +1,5 @@ import asyncio +import smtplib import threading import httpx @@ -47,6 +48,17 @@ def test_publisher_add_and_publish_across_channels(): assert second.sync_messages == [{"content": "hello", "title": "world"}] +def test_publisher_add_without_channels_is_noop(): + first = RecordingChannel() + publisher = Publisher([first]) + + publisher.add() + publisher.publish("hello") + + assert publisher.channels == (first,) + assert first.sync_messages == [{"content": "hello", "title": None}] + + @pytest.mark.asyncio async def test_publisher_publish_async_across_channels(): first = RecordingChannel() @@ -122,6 +134,35 @@ def test_publisher_retries_http_429_and_eventually_succeeds(): assert len(channel.sync_messages) == 2 +def test_publisher_does_not_retry_http_400(): + channel = RecordingChannel(sync_failures=[make_http_status_error(400)]) + publisher = Publisher([channel], max_retries=3) + + with pytest.raises(httpx.HTTPStatusError, match="status 400"): + publisher.publish("hello") + + assert len(channel.sync_messages) == 1 + + +def test_publisher_classifies_provider_specific_retry_exceptions(): + publisher = Publisher() + + assert publisher._is_retriable_exception(make_http_status_error(408), publisher.retry_config) + assert publisher._is_retriable_exception(httpx.ConnectError("network"), publisher.retry_config) + assert publisher._is_retriable_exception( + smtplib.SMTPResponseException(450, b"mailbox unavailable"), + publisher.retry_config, + ) + assert not publisher._is_retriable_exception( + smtplib.SMTPAuthenticationError(535, b"auth failed"), + publisher.retry_config, + ) + assert not publisher._is_retriable_exception( + smtplib.SMTPResponseException(550, b"mailbox unavailable"), + publisher.retry_config, + ) + + def test_publisher_aggregates_failures_after_other_channels_continue(): failing_one = RecordingChannel(sync_failures=[TimeoutError("one"), TimeoutError("one")]) failing_two = RecordingChannel(sync_failures=[TimeoutError("two"), TimeoutError("two")]) @@ -135,6 +176,20 @@ def test_publisher_aggregates_failures_after_other_channels_continue(): assert len(error_info.value.failures) == 2 +@pytest.mark.asyncio +async def test_publisher_aggregates_async_failures_after_other_channels_continue(): + failing_one = RecordingChannel(async_failures=[httpx.ConnectError("one")]) + failing_two = RecordingChannel(async_failures=[httpx.ConnectError("two")]) + healthy = RecordingChannel() + publisher = Publisher([failing_one, healthy, failing_two]) + + with pytest.raises(NotificationPublishError) as error_info: + await publisher.publish_async("hello") + + assert len(healthy.async_messages) == 1 + assert len(error_info.value.failures) == 2 + + def test_single_channel_failure_redacts_secret_from_exception_message(): request = httpx.Request( "POST", @@ -332,6 +387,9 @@ def test_retry_config_validates_exception_types(): with pytest.raises(ValueError, match="exception types"): RetryConfig(retriable_exceptions=("invalid",)) + with pytest.raises(ValueError, match="exception types"): + RetryConfig(retriable_exceptions=RuntimeError) + @pytest.mark.parametrize( "kwargs",