From 54abeb78c09459b1752e74cacf2306c4dc69fee4 Mon Sep 17 00:00:00 2001 From: nowanti Date: Fri, 18 Jul 2025 14:19:44 +0800 Subject: [PATCH 1/3] =?UTF-8?q?build:=20=E6=9B=B4=E6=96=B0=E4=BE=9D?= =?UTF-8?q?=E8=B5=96=E9=A1=B9=E7=89=88=E6=9C=AC=E7=BA=A6=E6=9D=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 将httpx依赖项从通配符改为指定版本范围，并添加httpcore和exceptiongroup依赖 --- pyproject.toml | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index b6aac30..ebf4d6a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,14 +4,14 @@ version = "0.3.44" description = "一个简单可扩展的异步消息通知库" authors = ["miclon "] readme = "README.md" -packages = [ - { include = 'use_notify', from = 'src' } -] +packages = [{ include = 'use_notify', from = 'src' }] [tool.poetry.dependencies] python = "^3.8" usepy = "^0.4.0" -httpx = "*" +httpx = ">=0.27.0,<1.0.0" +httpcore = { extras = ["asyncio"], version = "^1.0.9" } +exceptiongroup = ">=1.3.0,<2.0.0" [tool.poetry.group.test.dependencies] From 7e87b2165e475f27c2ce1ecd39fdd8bd70bcf378 Mon Sep 17 00:00:00 2001 From: nowanti Date: Fri, 18 Jul 2025 14:19:55 +0800 Subject: [PATCH 2/3] =?UTF-8?q?feat(ntfy):=20=E6=B7=BB=E5=8A=A0ntfy.sh?= =?UTF-8?q?=E9=80=9A=E7=9F=A5=E6=B8=A0=E9=81=93=E6=94=AF=E6=8C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 实现ntfy.sh通知渠道的完整功能，包括同步/异步消息发送、高级功能配置和集成测试 添加示例代码展示基础使用、高级功能和集成方式 包含完整的单元测试覆盖所有功能场景 --- example/ntfy_demo.py | 216 +++++++++++++ example/ntfy_from_settings.py | 50 +++ example/ntfy_integration.py | 52 ++++ example/ntfy_simple.py | 46 +++ src/use_notify/channels/__init__.py | 1 + src/use_notify/channels/ntfy.py | 115 +++++++ tests/test_ntfy.py | 452 ++++++++++++++++++++++++++++ 7 files changed, 932 insertions(+) create mode 100644 example/ntfy_demo.py create mode 100644 example/ntfy_from_settings.py create mode 100644 example/ntfy_integration.py create mode 100644 example/ntfy_simple.py create mode 100644 src/use_notify/channels/ntfy.py create mode 100644 tests/test_ntfy.py diff --git a/example/ntfy_demo.py b/example/ntfy_demo.py new file mode 100644 index 0000000..ec0a98e --- /dev/null +++ b/example/ntfy_demo.py @@ -0,0 +1,216 @@ +# -*- coding: utf-8 -*- +""" +Ntfy.sh 通知渠道使用示例 + +本示例展示了如何使用 ntfy.sh 通知渠道发送通知，包括： +1. 基础配置和使用 +2. 高级功能配置 +3. 同步和异步发送 +4. 与其他渠道的组合使用 +""" +import asyncio + +from use_notify import useNotify +from use_notify.channels import Ntfy + + +def basic_usage_example(): + """基础使用示例""" + print("=== 基础使用示例 ===") + + # 最简配置，只需要指定主题 + ntfy = Ntfy({ + "topic": "my-notifications" + }) + + # 发送简单消息 + try: + ntfy.send("这是一条测试消息") + print("✓ 基础消息发送成功") + except Exception as e: + print(f"✗ 发送失败: {e}") + + # 发送带标题的消息 + try: + ntfy.send("这是消息内容", "消息标题") + print("✓ 带标题消息发送成功") + except Exception as e: + print(f"✗ 发送失败: {e}") + + +def advanced_features_example(): + """高级功能使用示例""" + print("\n=== 高级功能使用示例 ===") + + # 配置高级功能 + ntfy = Ntfy({ + "topic": "my-notifications", + "priority": 4, # 高优先级 (1-5) + "tags": ["warning", "computer"], # 标签 + "click": "https://example.com", # 点击跳转链接 + "attach": "https://example.com/image.jpg" # 附件 + }) + + try: + ntfy.send("这是一条高优先级警告消息", "系统警告") + print("✓ 高级功能消息发送成功") + except Exception as e: + print(f"✗ 发送失败: {e}") + + +def custom_server_example(): + """自定义服务器示例""" + print("\n=== 自定义服务器示例 ===") + + # 使用自托管的 ntfy.sh 服务器 + ntfy = Ntfy({ + "topic": "my-notifications", + "base_url": "https://ntfy.example.com" # 自定义服务器 + }) + + try: + ntfy.send("来自自定义服务器的消息", "自定义服务器") + print("✓ 自定义服务器消息发送成功") + except Exception as e: + print(f"✗ 发送失败: {e}") + + +async def async_usage_example(): + """异步使用示例""" + print("\n=== 异步使用示例 ===") + + ntfy = Ntfy({ + "topic": "my-notifications", + "priority": 3 + }) + + try: + await ntfy.send_async("这是异步发送的消息", "异步消息") + print("✓ 异步消息发送成功") + except Exception as e: + print(f"✗ 异步发送失败: {e}") + + +def publisher_integration_example(): + """与 Publisher 集成使用示例""" + print("\n=== Publisher 集成示例 ===") + + # 创建通知发布器 + notify = useNotify() + + # 添加 ntfy 渠道 + notify.add( + Ntfy({ + "topic": "my-notifications", + "priority": 3, + "tags": ["app", "notification"] + }) + ) + + try: + notify.publish("通过 Publisher 发送的消息", "Publisher 消息") + print("✓ Publisher 集成消息发送成功") + except Exception as e: + print(f"✗ Publisher 发送失败: {e}") + + +async def async_publisher_example(): + """异步 Publisher 示例""" + print("\n=== 异步 Publisher 示例 ===") + + notify = useNotify() + notify.add( + Ntfy({ + "topic": "my-notifications", + "priority": 2 + }) + ) + + try: + await notify.publish_async("异步 Publisher 消息", "异步 Publisher") + print("✓ 异步 Publisher 消息发送成功") + except Exception as e: + print(f"✗ 异步 Publisher 发送失败: {e}") + + +def from_settings_example(): + """从配置创建示例""" + print("\n=== 从配置创建示例 ===") + + # 配置字典 + settings = { + "NTFY": { + "topic": "my-notifications", + "priority": 3, + "tags": ["config", "demo"], + "click": "https://github.com/ntfy-sh/ntfy" + } + } + + # 从配置创建通知器 + notify = useNotify.from_settings(settings) + + try: + notify.publish("从配置创建的消息", "配置消息") + print("✓ 从配置创建消息发送成功") + except Exception as e: + print(f"✗ 从配置发送失败: {e}") + + +def actions_example(): + """交互操作示例""" + print("\n=== 交互操作示例 ===") + + # 配置交互操作 + ntfy = Ntfy({ + "topic": "my-notifications", + "actions": [ + { + "action": "view", + "label": "打开网站", + "url": "https://ntfy.sh" + }, + { + "action": "http", + "label": "重启服务", + "url": "https://api.example.com/restart", + "method": "POST" + } + ] + }) + + try: + ntfy.send("服务器需要重启，请选择操作", "服务器警告") + print("✓ 交互操作消息发送成功") + except Exception as e: + print(f"✗ 交互操作发送失败: {e}") + + +async def main(): + """主函数，运行所有示例""" + print("Ntfy.sh 通知渠道使用示例") + print("=" * 50) + + # 同步示例 + basic_usage_example() + advanced_features_example() + custom_server_example() + publisher_integration_example() + from_settings_example() + actions_example() + + # 异步示例 + await async_usage_example() + await async_publisher_example() + + print("\n" + "=" * 50) + print("所有示例运行完成！") + print("\n注意：") + print("1. 请将 'my-notifications' 替换为您的实际主题名称") + print("2. 确保您的设备已订阅相应的主题") + print("3. 如果使用自定义服务器，请确保服务器地址正确") + + +if __name__ == "__main__": + # 运行示例 + asyncio.run(main()) \ No newline at end of file diff --git a/example/ntfy_from_settings.py b/example/ntfy_from_settings.py new file mode 100644 index 0000000..45aedf9 --- /dev/null +++ b/example/ntfy_from_settings.py @@ -0,0 +1,50 @@ +# -*- coding: utf-8 -*- +""" +使用配置字典创建 Ntfy 通知渠道的简单示例 +""" +from use_notify import useNotify + +# 基础配置 +settings = { + "NTFY": { + "topic": "my-notifications" + } +} + +# 从配置创建通知器 +notify = useNotify.from_settings(settings) + +# 发送通知 +try: + notify.publish("这是一条测试消息", "测试标题") + print("✓ 消息发送成功") +except Exception as e: + print(f"✗ 发送失败: {e}") + + +# 高级配置示例 +advanced_settings = { + "NTFY": { + "topic": "my-notifications", + "base_url": "https://ntfy.sh", # 可选：自定义服务器 + "priority": 3, # 可选：优先级 (1-5) + "tags": ["python", "demo"], # 可选：标签 + "click": "https://github.com/ntfy-sh/ntfy", # 可选：点击链接 + "actions": [ # 可选：交互操作 + { + "action": "view", + "label": "查看详情", + "url": "https://example.com" + } + ] + } +} + +# 使用高级配置 +advanced_notify = useNotify.from_settings(advanced_settings) + +try: + advanced_notify.publish("这是一条高级配置的消息", "高级消息") + print("✓ 高级消息发送成功") +except Exception as e: + print(f"✗ 高级消息发送失败: {e}") \ No newline at end of file diff --git a/example/ntfy_integration.py b/example/ntfy_integration.py new file mode 100644 index 0000000..a762ea5 --- /dev/null +++ b/example/ntfy_integration.py @@ -0,0 +1,52 @@ +# -*- coding: utf-8 -*- +""" +Ntfy.sh 与 useNotify 集成使用示例 +""" +import asyncio +from use_notify import useNotify +from use_notify.channels import Ntfy + +# 方式1: 直接添加 Ntfy 渠道 +notify = useNotify() +notify.add( + Ntfy({ + "topic": "my-notifications", + "priority": 3, + "tags": ["app", "notification"] + }) +) + +# 同步发送 +try: + notify.publish("集成测试消息", "集成测试") + print("✓ 集成同步发送成功") +except Exception as e: + print(f"✗ 集成同步发送失败: {e}") + +# 异步发送 +async def async_integration(): + try: + await notify.publish_async("异步集成消息", "异步集成") + print("✓ 集成异步发送成功") + except Exception as e: + print(f"✗ 集成异步发送失败: {e}") + +asyncio.run(async_integration()) + +# 方式2: 从配置创建 +settings = { + "NTFY": { + "topic": "my-notifications", + "priority": 4, + "tags": ["config", "demo"], + "click": "https://ntfy.sh" + } +} + +config_notify = useNotify.from_settings(settings) + +try: + config_notify.publish("配置创建的消息", "配置测试") + print("✓ 配置方式发送成功") +except Exception as e: + print(f"✗ 配置方式发送失败: {e}") \ No newline at end of file diff --git a/example/ntfy_simple.py b/example/ntfy_simple.py new file mode 100644 index 0000000..8a75d38 --- /dev/null +++ b/example/ntfy_simple.py @@ -0,0 +1,46 @@ +# -*- coding: utf-8 -*- +""" +Ntfy.sh 通知渠道简单使用示例 +""" + +import asyncio +from use_notify.channels import Ntfy + +# 基础使用 +ntfy = Ntfy({"topic": "my-notifications"}) + +# 同步发送 +try: + ntfy.send("这是一条测试消息", "测试标题") + print("✓ 同步消息发送成功") +except Exception as e: + print(f"✗ 同步发送失败: {e}") + + +# 异步发送 +async def async_example(): + await ntfy.send_async("这是异步消息", "异步标题") + try: + print("✓ 异步消息发送成功") + except Exception as e: + print(f"✗ 异步发送失败: {e}") + + +# 运行异步示例 +asyncio.run(async_example()) + +# 高级功能示例 +advanced_ntfy = Ntfy( + { + "topic": "my-notifications", + "priority": 4, + "tags": ["urgent", "demo"], + "click": "https://example.com", + } +) + +try: + advanced_ntfy.send("高优先级消息", "重要通知") + print("✓ 高级功能消息发送成功") +except Exception as e: + print(f"✗ 高级功能发送失败: {e}") diff --git a/src/use_notify/channels/__init__.py b/src/use_notify/channels/__init__.py index e68f044..e03d068 100644 --- a/src/use_notify/channels/__init__.py +++ b/src/use_notify/channels/__init__.py @@ -5,6 +5,7 @@ from .chanify import Chanify from .ding import Ding from .email import Email +from .ntfy import Ntfy from .pushdeer import PushDeer from .pushover import PushOver from .wechat import WeChat diff --git a/src/use_notify/channels/ntfy.py b/src/use_notify/channels/ntfy.py new file mode 100644 index 0000000..791c323 --- /dev/null +++ b/src/use_notify/channels/ntfy.py @@ -0,0 +1,115 @@ +# -*- coding: utf-8 -*- +import logging +from typing import Dict, Any, Optional + +import httpx + +from .base import BaseChannel + +logger = logging.getLogger(__name__) + + +class Ntfy(BaseChannel): + """Ntfy.sh 通知渠道""" + + def __init__(self, config: dict): + """ + 初始化 Ntfy 渠道 + + Args: + config: 配置字典，必须包含 'topic'，可选包含 'base_url' 等参数 + """ + super().__init__(config) + + # 验证必需的配置参数 + if not hasattr(self.config, 'topic') or not self.config.topic: + raise ValueError("Ntfy channel requires 'topic' in config") + + @property + def api_url(self): + """构建 ntfy.sh API URL""" + # 检查是否有自定义 base_url，否则使用默认值 + if hasattr(self.config, 'base_url') and self.config.base_url: + base_url = self.config.base_url.rstrip("/") + else: + base_url = "https://ntfy.sh" + + return f"{base_url}/{self.config.topic}" + + @property + def headers(self): + """构建请求头""" + return {"Content-Type": "application/json; charset=utf-8"} + + def _prepare_payload(self, content: str, title: Optional[str] = None) -> Dict[str, Any]: + """ + 准备请求负载 + + Args: + content: 消息内容 + title: 消息标题（可选） + + Returns: + dict: 请求负载 + """ + payload = { + "message": content, + } + + if title: + payload["title"] = title + + # 添加高级功能支持 + # 优先级支持 (1-5) + if hasattr(self.config, 'priority') and self.config.priority is not None: + payload["priority"] = self.config.priority + + # 标签支持 + if hasattr(self.config, 'tags') and self.config.tags: + payload["tags"] = self.config.tags + + # 点击 URL 支持 + if hasattr(self.config, 'click') and self.config.click: + payload["click"] = self.config.click + + # 附件 URL 支持 + if hasattr(self.config, 'attach') and self.config.attach: + payload["attach"] = self.config.attach + + # 操作支持 + if hasattr(self.config, 'actions') and self.config.actions: + payload["actions"] = self.config.actions + + return payload + + def send(self, content: str, title: Optional[str] = None) -> None: + """ + 发送通知到 ntfy.sh + + Args: + content: 消息内容 + title: 消息标题（可选） + """ + payload = self._prepare_payload(content, title) + + with httpx.Client() as client: + response = client.post(self.api_url, headers=self.headers, json=payload) + response.raise_for_status() + + logger.debug("`ntfy` send successfully") + + async def send_async(self, content: str, title: Optional[str] = None) -> None: + """ + 异步发送通知到 ntfy.sh + + Args: + content: 消息内容 + title: 消息标题（可选） + """ + payload = self._prepare_payload(content, title) + + async with httpx.AsyncClient() as client: + response = await client.post(self.api_url, headers=self.headers, json=payload) + response.raise_for_status() + + logger.debug("`ntfy` send successfully") \ No newline at end of file diff --git a/tests/test_ntfy.py b/tests/test_ntfy.py new file mode 100644 index 0000000..bd2def6 --- /dev/null +++ b/tests/test_ntfy.py @@ -0,0 +1,452 @@ +import pytest +from unittest.mock import patch, MagicMock + +from use_notify.channels.ntfy import Ntfy + + +test_topic = "my-notifications-test-topic" + + +@pytest.fixture +def ntfy_config(): + return { + "topic": test_topic, + "priority": 3, + "tags": ["warning", "skull"], + "click": "https://example.com", + "attach": "https://example.com/file.jpg", + } + + +@pytest.fixture +def minimal_ntfy_config(): + return {"topic": test_topic} + + +@pytest.fixture +def custom_server_config(): + return {"topic": test_topic, "base_url": "https://ntfy.example.com"} + + +def test_ntfy_init_with_minimal_config(minimal_ntfy_config): + """测试使用最小配置初始化 Ntfy""" + ntfy = Ntfy(minimal_ntfy_config) + assert ntfy.config.topic == test_topic + + +def test_ntfy_init_with_full_config(ntfy_config): + """测试使用完整配置初始化 Ntfy""" + ntfy = Ntfy(ntfy_config) + assert ntfy.config.topic == test_topic + assert ntfy.config.priority == 3 + assert ntfy.config.tags == ["warning", "skull"] + assert ntfy.config.click == "https://example.com" + assert ntfy.config.attach == "https://example.com/file.jpg" + + +def test_ntfy_init_without_topic(): + """测试没有 topic 时应该抛出异常""" + with pytest.raises(ValueError, match="Ntfy channel requires 'topic' in config"): + Ntfy({}) + + +def test_ntfy_init_with_empty_topic(): + """测试空 topic 时应该抛出异常""" + with pytest.raises(ValueError, match="Ntfy channel requires 'topic' in config"): + Ntfy({"topic": ""}) + + +def test_api_url_construction_default(minimal_ntfy_config): + """测试默认 API URL 构建""" + ntfy = Ntfy(minimal_ntfy_config) + assert ntfy.api_url == f"https://ntfy.sh/{test_topic}" + + +def test_api_url_construction_custom(custom_server_config): + """测试自定义服务器 API URL 构建""" + ntfy = Ntfy(custom_server_config) + assert ntfy.api_url == f"https://ntfy.example.com/{test_topic}" + + +def test_api_url_with_trailing_slash(): + """测试带尾随斜杠的 base_url 规范化""" + config = {"topic": test_topic, "base_url": "https://ntfy.example.com/"} + ntfy = Ntfy(config) + assert ntfy.api_url == f"https://ntfy.example.com/{test_topic}" + + +def test_headers(): + """测试请求头""" + ntfy = Ntfy({"topic": test_topic}) + headers = ntfy.headers + assert headers["Content-Type"] == "application/json; charset=utf-8" + + +def test_prepare_payload_minimal(): + """测试最小负载准备""" + ntfy = Ntfy({"topic": test_topic}) + payload = ntfy._prepare_payload("Test message") + + assert payload["message"] == "Test message" + assert "title" not in payload + + +def test_prepare_payload_with_title(): + """测试带标题的负载准备""" + ntfy = Ntfy({"topic": test_topic}) + payload = ntfy._prepare_payload("Test message", "Test title") + + assert payload["message"] == "Test message" + assert payload["title"] == "Test title" + + +def test_prepare_payload_with_advanced_features(ntfy_config): + """测试带高级功能的负载准备""" + ntfy = Ntfy(ntfy_config) + payload = ntfy._prepare_payload("Test message", "Test title") + + assert payload["message"] == "Test message" + assert payload["title"] == "Test title" + assert payload["priority"] == 3 + assert payload["tags"] == ["warning", "skull"] + assert payload["click"] == "https://example.com" + assert payload["attach"] == "https://example.com/file.jpg" + + +def test_prepare_payload_with_actions(): + """测试带操作的负载准备""" + config = { + "topic": test_topic, + "actions": [ + {"action": "view", "label": "Open portal", "url": "https://home.nest.com/"} + ], + } + ntfy = Ntfy(config) + payload = ntfy._prepare_payload("Test message") + + assert payload["actions"] == config["actions"] + + +def test_send_success(minimal_ntfy_config): + """测试同步发送成功""" + # Create a mock for httpx.Client + mock_client = MagicMock() + mock_response = MagicMock() + mock_client.return_value.__enter__.return_value.post.return_value = mock_response + + # Create Ntfy instance + ntfy = Ntfy(minimal_ntfy_config) + + # Mock the httpx.Client + with patch("httpx.Client", mock_client): + ntfy.send("Test message", "Test title") + + # Verify the request was made correctly + mock_client.return_value.__enter__.return_value.post.assert_called_once() + + # Get the call arguments + call_args = mock_client.return_value.__enter__.return_value.post.call_args + url = call_args[0][0] + kwargs = call_args[1] + + # Verify URL and headers + assert url == f"https://ntfy.sh/{test_topic}" + assert kwargs["headers"]["Content-Type"] == "application/json; charset=utf-8" + + # Verify payload + payload = kwargs["json"] + assert payload["message"] == "Test message" + assert payload["title"] == "Test title" + + # Verify response handling + mock_response.raise_for_status.assert_called_once() + + +@pytest.mark.asyncio +async def test_send_async_success(minimal_ntfy_config): + """测试异步发送成功""" + # Create a mock for httpx.AsyncClient + mock_client = MagicMock() + mock_response = MagicMock() + mock_client.return_value.__aenter__.return_value.post.return_value = mock_response + + # Create Ntfy instance + ntfy = Ntfy(minimal_ntfy_config) + + # Mock the httpx.AsyncClient + with patch("httpx.AsyncClient", mock_client): + await ntfy.send_async("Test message", "Test title") + + # Verify the request was made correctly + mock_client.return_value.__aenter__.return_value.post.assert_called_once() + + # Get the call arguments + call_args = mock_client.return_value.__aenter__.return_value.post.call_args + url = call_args[0][0] + kwargs = call_args[1] + + # Verify URL and headers + assert url == f"https://ntfy.sh/{test_topic}" + assert kwargs["headers"]["Content-Type"] == "application/json; charset=utf-8" + + # Verify payload + payload = kwargs["json"] + assert payload["message"] == "Test message" + assert payload["title"] == "Test title" + + # Verify response handling + mock_response.raise_for_status.assert_called_once() + + +def test_send_with_advanced_features(ntfy_config): + """测试带高级功能的同步发送""" + # Create a mock for httpx.Client + mock_client = MagicMock() + mock_response = MagicMock() + mock_client.return_value.__enter__.return_value.post.return_value = mock_response + + # Create Ntfy instance + ntfy = Ntfy(ntfy_config) + + # Mock the httpx.Client + with patch("httpx.Client", mock_client): + ntfy.send("Test message", "Test title") + + # Get the call arguments + call_args = mock_client.return_value.__enter__.return_value.post.call_args + kwargs = call_args[1] + + # Verify payload includes advanced features + payload = kwargs["json"] + assert payload["message"] == "Test message" + assert payload["title"] == "Test title" + assert payload["priority"] == 3 + assert payload["tags"] == ["warning", "skull"] + assert payload["click"] == "https://example.com" + assert payload["attach"] == "https://example.com/file.jpg" + + +def test_send_http_error(minimal_ntfy_config): + """测试 HTTP 错误处理""" + import httpx + + # Create a mock for httpx.Client + mock_client = MagicMock() + mock_response = MagicMock() + mock_response.raise_for_status.side_effect = httpx.HTTPStatusError( + "404 Not Found", request=MagicMock(), response=MagicMock() + ) + mock_client.return_value.__enter__.return_value.post.return_value = mock_response + + # Create Ntfy instance + ntfy = Ntfy(minimal_ntfy_config) + + # Mock the httpx.Client and expect exception + with patch("httpx.Client", mock_client): + with pytest.raises(httpx.HTTPStatusError): + ntfy.send("Test message") + + +@pytest.mark.asyncio +async def test_send_async_http_error(minimal_ntfy_config): + """测试异步 HTTP 错误处理""" + import httpx + + # Create a mock for httpx.AsyncClient + mock_client = MagicMock() + mock_response = MagicMock() + mock_response.raise_for_status.side_effect = httpx.HTTPStatusError( + "404 Not Found", request=MagicMock(), response=MagicMock() + ) + mock_client.return_value.__aenter__.return_value.post.return_value = mock_response + + # Create Ntfy instance + ntfy = Ntfy(minimal_ntfy_config) + + # Mock the httpx.AsyncClient and expect exception + with patch("httpx.AsyncClient", mock_client): + with pytest.raises(httpx.HTTPStatusError): + await ntfy.send_async("Test message") + + +def test_send_request_error(minimal_ntfy_config): + """测试网络请求错误处理""" + import httpx + + # Create a mock for httpx.Client + mock_client = MagicMock() + mock_client.return_value.__enter__.return_value.post.side_effect = ( + httpx.RequestError("Connection failed") + ) + + # Create Ntfy instance + ntfy = Ntfy(minimal_ntfy_config) + + # Mock the httpx.Client and expect exception + with patch("httpx.Client", mock_client): + with pytest.raises(httpx.RequestError): + ntfy.send("Test message") + + +@pytest.mark.asyncio +async def test_send_async_request_error(minimal_ntfy_config): + """测试异步网络请求错误处理""" + import httpx + + # Create a mock for httpx.AsyncClient + mock_client = MagicMock() + mock_client.return_value.__aenter__.return_value.post.side_effect = ( + httpx.RequestError("Connection failed") + ) + + # Create Ntfy instance + ntfy = Ntfy(minimal_ntfy_config) + + # Mock the httpx.AsyncClient and expect exception + with patch("httpx.AsyncClient", mock_client): + with pytest.raises(httpx.RequestError): + await ntfy.send_async("Test message") + + +def test_integration_with_publisher(): + """测试与 Publisher 的集成""" + from use_notify.notification import Publisher + + # Create a mock for httpx.Client + mock_client = MagicMock() + mock_response = MagicMock() + mock_client.return_value.__enter__.return_value.post.return_value = mock_response + + # Create Ntfy instance and Publisher + ntfy = Ntfy({"topic": test_topic}) + publisher = Publisher() + publisher.add(ntfy) + + # Mock the httpx.Client + with patch("httpx.Client", mock_client): + publisher.publish("Test message", "Test title") + + # Verify the request was made + mock_client.return_value.__enter__.return_value.post.assert_called_once() + mock_response.raise_for_status.assert_called_once() + + +@pytest.mark.asyncio +async def test_integration_with_publisher_async(): + """测试与 Publisher 的异步集成""" + from use_notify.notification import Publisher + + # Create a mock for httpx.AsyncClient + mock_client = MagicMock() + mock_response = MagicMock() + mock_client.return_value.__aenter__.return_value.post.return_value = mock_response + + # Create Ntfy instance and Publisher + ntfy = Ntfy({"topic": test_topic}) + publisher = Publisher() + publisher.add(ntfy) + + # Mock the httpx.AsyncClient + with patch("httpx.AsyncClient", mock_client): + await publisher.publish_async("Test message", "Test title") + + # Verify the request was made + mock_client.return_value.__aenter__.return_value.post.assert_called_once() + mock_response.raise_for_status.assert_called_once() + + +def test_integration_with_notify_from_settings(): + """测试与 Notify.from_settings() 的集成""" + from use_notify.notification import Notify + + # Create a mock for httpx.Client + mock_client = MagicMock() + mock_response = MagicMock() + mock_client.return_value.__enter__.return_value.post.return_value = mock_response + + # Create settings + settings = {"NTFY": {"topic": test_topic, "priority": 3}} + + # Create Notify instance from settings + notify = Notify.from_settings(settings) + + # Mock the httpx.Client + with patch("httpx.Client", mock_client): + notify.publish("Test message", "Test title") + + # Verify the request was made correctly + mock_client.return_value.__enter__.return_value.post.assert_called_once() + + # Get the call arguments + call_args = mock_client.return_value.__enter__.return_value.post.call_args + url = call_args[0][0] + kwargs = call_args[1] + + # Verify URL and payload + assert url == f"https://ntfy.sh/{test_topic}" + payload = kwargs["json"] + assert payload["message"] == "Test message" + assert payload["title"] == "Test title" + assert payload["priority"] == 3 + + +@pytest.mark.asyncio +async def test_integration_with_notify_from_settings_async(): + """测试与 Notify.from_settings() 的异步集成""" + from use_notify.notification import Notify + + # Create a mock for httpx.AsyncClient + mock_client = MagicMock() + mock_response = MagicMock() + mock_client.return_value.__aenter__.return_value.post.return_value = mock_response + + # Create settings + settings = {"NTFY": {"topic": test_topic, "priority": 3}} + + # Create Notify instance from settings + notify = Notify.from_settings(settings) + + # Mock the httpx.AsyncClient + with patch("httpx.AsyncClient", mock_client): + await notify.publish_async("Test message", "Test title") + + # Verify the request was made correctly + mock_client.return_value.__aenter__.return_value.post.assert_called_once() + + # Get the call arguments + call_args = mock_client.return_value.__aenter__.return_value.post.call_args + url = call_args[0][0] + kwargs = call_args[1] + + # Verify URL and payload + assert url == f"https://ntfy.sh/{test_topic}" + payload = kwargs["json"] + assert payload["message"] == "Test message" + assert payload["title"] == "Test title" + assert payload["priority"] == 3 + + +def test_integration_case_insensitive_channel_name(): + """测试不区分大小写的渠道名称""" + from use_notify.notification import Notify + + # Create a mock for httpx.Client + mock_client = MagicMock() + mock_response = MagicMock() + mock_client.return_value.__enter__.return_value.post.return_value = mock_response + + # Test different case variations + for channel_name in ["ntfy", "NTFY", "Ntfy", "nTfY"]: + settings = {channel_name: {"topic": test_topic}} + + # Create Notify instance from settings + notify = Notify.from_settings(settings) + + # Verify the channel was created + assert len(notify.channels) == 1 + assert isinstance(notify.channels[0], Ntfy) + assert notify.channels[0].config.topic == test_topic + + +if __name__ == "__main__": + pytest.main() From 25d57ac478fbd8a96c404359bc508b919da1108c Mon Sep 17 00:00:00 2001 From: nowanti Date: Fri, 18 Jul 2025 14:28:35 +0800 Subject: [PATCH 3/3] =?UTF-8?q?build:=20=E7=AE=80=E5=8C=96=E4=BE=9D?= =?UTF-8?q?=E8=B5=96=E9=A1=B9=E5=B9=B6=E7=A7=BB=E9=99=A4=E4=B8=8D=E5=BF=85?= =?UTF-8?q?=E8=A6=81=E7=9A=84=E5=8C=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 移除httpcore和exceptiongroup依赖，并将httpx版本限制放宽为任意版本 --- pyproject.toml | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index ebf4d6a..33e55a6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,9 +9,7 @@ packages = [{ include = 'use_notify', from = 'src' }] [tool.poetry.dependencies] python = "^3.8" usepy = "^0.4.0" -httpx = ">=0.27.0,<1.0.0" -httpcore = { extras = ["asyncio"], version = "^1.0.9" } -exceptiongroup = ">=1.3.0,<2.0.0" +httpx = "*" [tool.poetry.group.test.dependencies]