From 47b2e3e719cd5c504b3fb0c5656d3764029df97a Mon Sep 17 00:00:00 2001 From: miclon Date: Thu, 17 Jul 2025 17:34:42 +0800 Subject: [PATCH 1/5] temp --- src/use_notify/channels/pushdeer.py | 116 ++++++++++++++++-- tests/test_pushdeer.py | 180 ++++++++++++++++++++++++++++ 2 files changed, 285 insertions(+), 11 deletions(-) create mode 100644 tests/test_pushdeer.py diff --git a/src/use_notify/channels/pushdeer.py b/src/use_notify/channels/pushdeer.py index b6e307a..69390ba 100644 --- a/src/use_notify/channels/pushdeer.py +++ b/src/use_notify/channels/pushdeer.py @@ -1,5 +1,6 @@ # -*- coding: utf-8 -*- import logging +from urllib.parse import quote import httpx @@ -9,27 +10,120 @@ class PushDeer(BaseChannel): - """pushdeer app 消息通知""" + """pushdeer app 消息通知 + + 支持三种消息类型: + - text: 纯文本消息 (默认) + - image: 图片消息 + - markdown: Markdown格式消息 + + 配置参数: + - token: PushDeer的pushkey + - base_url: 可选,自建PushDeer服务的URL,默认为"https://api2.pushdeer.com" + - type: 可选,消息类型,可选值为text、markdown、image,默认为text + """ @property def api_url(self): - return f"https://api2.pushdeer.com/message/push?pushkey={self.config.token}&text={{text}}" + """获取PushDeer API基础URL""" + # 确保token存在 + if not self.config.token: + raise ValueError("PushDeer token (pushkey) is required") + + + # Check if base_url exists in config, otherwise use default + if self.config.base_url: + base_url = self.config.base_url.rstrip("/") + else: + base_url = "https://api2.pushdeer.com" + + return f"{base_url}/message/push" + + def _prepare_params(self, content, title=None): + """准备请求参数 + + Args: + content: 消息内容 + title: 消息标题 + + Returns: + dict: 请求参数 + """ + # 默认标题 + if not title: + title = "Notification" + + # 确定消息类型 + msg_type = getattr(self.config, "type", "markdown") + if msg_type not in ["text", "markdown", "image"]: + if msg_type: # 只有当type不为空且无效时才记录警告 + logger.warning(f"Invalid message type: {msg_type}, fallback to text") + msg_type = "markdown" + + params = { + "pushkey": self.config.token, + "text": title, + "type": msg_type + } + + # 根据消息类型处理内容 + if msg_type == "text": + # 对于text类型,content直接作为text参数 + params["text"] = content + elif msg_type == "markdown": + # 对于markdown类型,content作为desp参数 + params["desp"] = content + elif msg_type == "image": + # 对于image类型,content应该是图片URL + params["desp"] = content + + return params @property def headers(self): + """请求头""" return {"Content-Type": "application/x-www-form-urlencoded"} - def build_api_body(self, content, title=None): - return self.api_url.format_map({"text": f"{title}\n{content}"}) - def send(self, content, title=None): - api_url = self.build_api_body(content, title) + """发送PushDeer消息 + + Args: + content: 消息内容 + title: 消息标题 + """ + params = self._prepare_params(content, title) + msg_type = params.get("type", "text") + with httpx.Client() as client: - client.get(api_url, headers=self.headers) - logger.debug("`pushdeer` send successfully") + response = client.get(self.api_url, params=params, headers=self.headers) + response.raise_for_status() + + # 检查API返回的结果 + result = response.json() + if result.get("code") != 0: + error_msg = result.get("error", "Unknown error") + raise RuntimeError(f"PushDeer API error: {error_msg}") + + logger.debug(f"`pushdeer` send {msg_type} message successfully") async def send_async(self, content, title=None): - api_url = self.build_api_body(content, title) + """异步发送PushDeer消息 + + Args: + content: 消息内容 + title: 消息标题 + """ + params = self._prepare_params(content, title) + msg_type = params.get("type", "text") + async with httpx.AsyncClient() as client: - await client.get(api_url, headers=self.headers) - logger.debug("`pushdeer` send successfully") + response = await client.get(self.api_url, params=params, headers=self.headers) + response.raise_for_status() + + # 检查API返回的结果 + result = await response.json() + if result.get("code") != 0: + error_msg = result.get("error", "Unknown error") + raise RuntimeError(f"PushDeer API error: {error_msg}") + + logger.debug(f"`pushdeer` send {msg_type} message successfully") \ No newline at end of file diff --git a/tests/test_pushdeer.py b/tests/test_pushdeer.py new file mode 100644 index 0000000..4fba061 --- /dev/null +++ b/tests/test_pushdeer.py @@ -0,0 +1,180 @@ +# -*- coding: utf-8 -*- +import pytest +import httpx +import json +from unittest.mock import patch, MagicMock + +from use_notify.channels.pushdeer import PushDeer + + +class TestPushDeer: + """PushDeer通知渠道测试""" + + @pytest.fixture + def pushdeer_config(self): + """基本配置""" + return { + "token": "test_token" + } + + @pytest.fixture + def pushdeer_instance(self, pushdeer_config): + """PushDeer实例""" + return PushDeer(pushdeer_config) + + def test_api_url(self, pushdeer_instance): + """测试API URL构建""" + assert pushdeer_instance.api_url == "https://api2.pushdeer.com/message/push" + + def test_api_url_custom_base(self): + """测试自定义base_url""" + pushdeer = PushDeer({ + "token": "test_token", + "base_url": "https://custom.pushdeer.com/" + }) + assert pushdeer.api_url == "https://custom.pushdeer.com/message/push" + + def test_api_url_no_token(self): + """测试没有token的情况""" + pushdeer = PushDeer({}) + with pytest.raises(ValueError, match="PushDeer token .* required"): + _ = pushdeer.api_url + + def test_prepare_params_text(self, pushdeer_instance): + """测试文本消息参数准备""" + params = pushdeer_instance._prepare_params("Hello World") + assert params["pushkey"] == "test_token" + assert params["text"] == "Hello World" + assert params["type"] == "markdown" + + def test_prepare_params_markdown(self): + """测试Markdown消息参数准备""" + pushdeer = PushDeer({ + "token": "test_token", + "type": "markdown" + }) + params = pushdeer._prepare_params("# Hello\n\nWorld", "Test Title") + assert params["pushkey"] == "test_token" + assert params["text"] == "Test Title" + assert params["desp"] == "# Hello\n\nWorld" + assert params["type"] == "markdown" + + def test_prepare_params_image(self): + """测试图片消息参数准备""" + pushdeer = PushDeer({ + "token": "test_token", + "type": "image" + }) + params = pushdeer._prepare_params("https://example.com/image.jpg") + assert params["pushkey"] == "test_token" + assert params["desp"] == "https://example.com/image.jpg" + assert params["type"] == "image" + + def test_prepare_params_invalid_type(self): + """测试无效消息类型""" + pushdeer = PushDeer({ + "token": "test_token", + "type": "invalid" + }) + params = pushdeer._prepare_params("Hello World") + assert params["type"] == "markdown" # 应该回退到text类型 + + @patch("httpx.Client") + def test_send_success(self, mock_client, pushdeer_instance): + """测试发送成功""" + # 模拟响应 + mock_response = MagicMock() + mock_response.raise_for_status.return_value = None + mock_response.json.return_value = {"code": 0, "message": "success"} + + # 模拟客户端 + mock_client_instance = MagicMock() + mock_client_instance.get.return_value = mock_response + mock_client.return_value.__enter__.return_value = mock_client_instance + + # 执行发送 + pushdeer_instance.send("Hello World", "Test Title") + + # 验证调用 + mock_client_instance.get.assert_called_once() + args, kwargs = mock_client_instance.get.call_args + assert args[0] == pushdeer_instance.api_url + assert "params" in kwargs + # 对于text类型,content会覆盖title作为text参数 + assert kwargs["params"]["text"] == "Hello World" + + @patch("httpx.Client") + def test_send_api_error(self, mock_client, pushdeer_instance): + """测试API返回错误""" + # 模拟响应 + mock_response = MagicMock() + mock_response.raise_for_status.return_value = None + mock_response.json.return_value = {"code": 1, "error": "Invalid token"} + + # 模拟客户端 + mock_client_instance = MagicMock() + mock_client_instance.get.return_value = mock_response + mock_client.return_value.__enter__.return_value = mock_client_instance + + # 执行发送,应该抛出异常 + with pytest.raises(RuntimeError, match="PushDeer API error: Invalid token"): + pushdeer_instance.send("Hello World") + + @pytest.mark.asyncio + async def test_send_async_success(self, pushdeer_instance): + """测试异步发送成功""" + # 创建一个真实的AsyncMock对象,它可以在await表达式中使用 + from unittest.mock import AsyncMock + + # 模拟响应 + mock_response = AsyncMock() + mock_response.raise_for_status = AsyncMock() + # 设置json方法为AsyncMock,返回值为字典 + mock_json = AsyncMock() + mock_json.return_value = {"code": 0, "message": "success"} + mock_response.json = mock_json + + # 模拟客户端 + mock_client = AsyncMock() + mock_client.get.return_value = mock_response + + # 替换httpx.AsyncClient + with patch("httpx.AsyncClient") as mock_async_client: + mock_async_client.return_value.__aenter__.return_value = mock_client + + # 执行发送 + await pushdeer_instance.send_async("Hello World", "Test Title") + + # 验证调用 + mock_client.get.assert_called_once() + args, kwargs = mock_client.get.call_args + assert args[0] == pushdeer_instance.api_url + assert "params" in kwargs + # 对于text类型,content会覆盖title作为text参数 + assert kwargs["params"]["text"] == "Hello World" + + @pytest.mark.asyncio + async def test_send_async_api_error(self, pushdeer_instance): + """测试异步API返回错误""" + # 创建一个真实的AsyncMock对象,它可以在await表达式中使用 + from unittest.mock import AsyncMock + + # 模拟响应 + mock_response = AsyncMock() + mock_response.raise_for_status = AsyncMock() + # 设置json方法为AsyncMock,返回值为字典 + mock_json = AsyncMock() + mock_json.return_value = {"code": 1, "error": "Invalid token"} + mock_response.json = mock_json + + # 模拟客户端 + mock_client = AsyncMock() + mock_client.get.return_value = mock_response + + # 替换httpx.AsyncClient + with patch("httpx.AsyncClient") as mock_async_client: + mock_async_client.return_value.__aenter__.return_value = mock_client + + # 执行发送,应该抛出异常 + with pytest.raises(RuntimeError, match="PushDeer API error: Invalid token"): + await pushdeer_instance.send_async("Hello World") \ No newline at end of file From 9f0c0d2a7188dc765baec0f74f6a91ed2b8ce09c Mon Sep 17 00:00:00 2001 From: mic1on Date: Thu, 17 Jul 2025 21:31:54 +0800 Subject: [PATCH 2/5] =?UTF-8?q?fix(pushdeer):=20=E4=BF=AE=E5=A4=8DPushDeer?= =?UTF-8?q?=E9=80=9A=E7=9F=A5=E6=B8=A0=E9=81=93=E7=9A=84=E5=8F=82=E6=95=B0?= =?UTF-8?q?=E5=92=8C=E9=94=99=E8=AF=AF=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 更新PushDeer通知渠道的参数名从text改为desp以匹配API要求 移除冗余的错误处理逻辑,简化发送流程 更新测试用例以匹配新的参数名和异步测试方式 添加pytest-asyncio依赖以支持异步测试 --- example/demo.py | 2 +- pyproject.toml | 1 + src/use_notify/channels/pushdeer.py | 24 ++++--------------- tests/test_pushdeer.py | 37 +++++++++++++---------------- 4 files changed, 23 insertions(+), 41 deletions(-) diff --git a/example/demo.py b/example/demo.py index 6bb9847..f834afd 100644 --- a/example/demo.py +++ b/example/demo.py @@ -6,7 +6,7 @@ notify = useNotify() notify.add( # 添加多个通知渠道 - useNotifyChannel.Bark({"token": "your token"}), + useNotifyChannel.PushDeer({"token": "PDU3862TaD5JqPXYdurER8V98ckbMA6DWgT3OC6b"}), ) asyncio.run(notify.publish_async(title="消息标题", content="消息正文")) diff --git a/pyproject.toml b/pyproject.toml index 5aaa1d3..059f1f3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,6 +22,7 @@ flake8 = "*" isort = "*" pre-commit = "*" pre-commit-hooks = "*" +pytest-asyncio = "0.18.3" [build-system] requires = ["poetry-core>=1.0.0"] diff --git a/src/use_notify/channels/pushdeer.py b/src/use_notify/channels/pushdeer.py index 69390ba..ad6deeb 100644 --- a/src/use_notify/channels/pushdeer.py +++ b/src/use_notify/channels/pushdeer.py @@ -51,7 +51,7 @@ def _prepare_params(self, content, title=None): """ # 默认标题 if not title: - title = "Notification" + title = "消息提醒" # 确定消息类型 msg_type = getattr(self.config, "type", "markdown") @@ -92,18 +92,10 @@ def send(self, content, title=None): title: 消息标题 """ params = self._prepare_params(content, title) - msg_type = params.get("type", "text") + msg_type = params.get("type", "markdown") with httpx.Client() as client: - response = client.get(self.api_url, params=params, headers=self.headers) - response.raise_for_status() - - # 检查API返回的结果 - result = response.json() - if result.get("code") != 0: - error_msg = result.get("error", "Unknown error") - raise RuntimeError(f"PushDeer API error: {error_msg}") - + client.get(self.api_url, params=params, headers=self.headers) logger.debug(f"`pushdeer` send {msg_type} message successfully") async def send_async(self, content, title=None): @@ -117,13 +109,5 @@ async def send_async(self, content, title=None): msg_type = params.get("type", "text") async with httpx.AsyncClient() as client: - response = await client.get(self.api_url, params=params, headers=self.headers) - response.raise_for_status() - - # 检查API返回的结果 - result = await response.json() - if result.get("code") != 0: - error_msg = result.get("error", "Unknown error") - raise RuntimeError(f"PushDeer API error: {error_msg}") - + await client.get(self.api_url, params=params, headers=self.headers) logger.debug(f"`pushdeer` send {msg_type} message successfully") \ No newline at end of file diff --git a/tests/test_pushdeer.py b/tests/test_pushdeer.py index 4fba061..98a688d 100644 --- a/tests/test_pushdeer.py +++ b/tests/test_pushdeer.py @@ -2,7 +2,7 @@ import pytest import httpx import json -from unittest.mock import patch, MagicMock +from unittest.mock import patch, MagicMock, AsyncMock from use_notify.channels.pushdeer import PushDeer @@ -44,7 +44,7 @@ def test_prepare_params_text(self, pushdeer_instance): """测试文本消息参数准备""" params = pushdeer_instance._prepare_params("Hello World") assert params["pushkey"] == "test_token" - assert params["text"] == "Hello World" + assert params["desp"] == "Hello World" assert params["type"] == "markdown" def test_prepare_params_markdown(self): @@ -101,7 +101,7 @@ def test_send_success(self, mock_client, pushdeer_instance): assert args[0] == pushdeer_instance.api_url assert "params" in kwargs # 对于text类型,content会覆盖title作为text参数 - assert kwargs["params"]["text"] == "Hello World" + assert kwargs["params"]["desp"] == "Hello World" @patch("httpx.Client") def test_send_api_error(self, mock_client, pushdeer_instance): @@ -120,12 +120,10 @@ def test_send_api_error(self, mock_client, pushdeer_instance): with pytest.raises(RuntimeError, match="PushDeer API error: Invalid token"): pushdeer_instance.send("Hello World") + @patch('httpx.AsyncClient') @pytest.mark.asyncio - async def test_send_async_success(self, pushdeer_instance): + async def test_send_async_success(self, mock_async_client, pushdeer_instance): """测试异步发送成功""" - # 创建一个真实的AsyncMock对象,它可以在await表达式中使用 - from unittest.mock import AsyncMock - # 模拟响应 mock_response = AsyncMock() mock_response.raise_for_status = AsyncMock() @@ -139,19 +137,18 @@ async def test_send_async_success(self, pushdeer_instance): mock_client.get.return_value = mock_response # 替换httpx.AsyncClient - with patch("httpx.AsyncClient") as mock_async_client: - mock_async_client.return_value.__aenter__.return_value = mock_client - - # 执行发送 - await pushdeer_instance.send_async("Hello World", "Test Title") - - # 验证调用 - mock_client.get.assert_called_once() - args, kwargs = mock_client.get.call_args - assert args[0] == pushdeer_instance.api_url - assert "params" in kwargs - # 对于text类型,content会覆盖title作为text参数 - assert kwargs["params"]["text"] == "Hello World" + mock_async_client.return_value.__aenter__.return_value = mock_client + + # 执行发送 + await pushdeer_instance.send_async("Hello World", "Test Title") + + # 验证调用 + mock_client.get.assert_called_once() + args, kwargs = mock_client.get.call_args + assert args[0] == pushdeer_instance.api_url + assert "params" in kwargs + # 对于text类型,content会覆盖title作为text参数 + assert kwargs["params"]["desp"] == "Hello World" @pytest.mark.asyncio async def test_send_async_api_error(self, pushdeer_instance): From 6fc413cbaef545a30c1bc6ef6d115f30799b8cba Mon Sep 17 00:00:00 2001 From: mic1on Date: Thu, 17 Jul 2025 21:32:38 +0800 Subject: [PATCH 3/5] =?UTF-8?q?fix(pushdeer):=20=E5=B0=86=E9=BB=98?= =?UTF-8?q?=E8=AE=A4=E6=B6=88=E6=81=AF=E7=B1=BB=E5=9E=8B=E4=BB=8Etext?= =?UTF-8?q?=E6=94=B9=E4=B8=BAmarkdown?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/use_notify/channels/pushdeer.py | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/src/use_notify/channels/pushdeer.py b/src/use_notify/channels/pushdeer.py index ad6deeb..f02ee1c 100644 --- a/src/use_notify/channels/pushdeer.py +++ b/src/use_notify/channels/pushdeer.py @@ -30,7 +30,6 @@ def api_url(self): if not self.config.token: raise ValueError("PushDeer token (pushkey) is required") - # Check if base_url exists in config, otherwise use default if self.config.base_url: base_url = self.config.base_url.rstrip("/") @@ -92,11 +91,10 @@ def send(self, content, title=None): title: 消息标题 """ params = self._prepare_params(content, title) - msg_type = params.get("type", "markdown") with httpx.Client() as client: client.get(self.api_url, params=params, headers=self.headers) - logger.debug(f"`pushdeer` send {msg_type} message successfully") + logger.debug(f"`pushdeer` send message successfully") async def send_async(self, content, title=None): """异步发送PushDeer消息 @@ -106,8 +104,7 @@ async def send_async(self, content, title=None): title: 消息标题 """ params = self._prepare_params(content, title) - msg_type = params.get("type", "text") async with httpx.AsyncClient() as client: await client.get(self.api_url, params=params, headers=self.headers) - logger.debug(f"`pushdeer` send {msg_type} message successfully") \ No newline at end of file + logger.debug(f"`pushdeer` send message successfully") \ No newline at end of file From 9581ac8e4b6646f52e2cedc1ad8d1b28064e2072 Mon Sep 17 00:00:00 2001 From: mic1on Date: Thu, 17 Jul 2025 21:33:17 +0800 Subject: [PATCH 4/5] =?UTF-8?q?fix:=20=E6=9B=BF=E6=8D=A2PushDeer=E9=80=9A?= =?UTF-8?q?=E7=9F=A5=E6=B8=A0=E9=81=93=E4=B8=AD=E7=9A=84=E6=95=8F=E6=84=9F?= =?UTF-8?q?token=E4=B8=BA=E5=8D=A0=E4=BD=8D=E7=AC=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- example/demo.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/example/demo.py b/example/demo.py index f834afd..34477a9 100644 --- a/example/demo.py +++ b/example/demo.py @@ -6,7 +6,7 @@ notify = useNotify() notify.add( # 添加多个通知渠道 - useNotifyChannel.PushDeer({"token": "PDU3862TaD5JqPXYdurER8V98ckbMA6DWgT3OC6b"}), + useNotifyChannel.PushDeer({"token": "Your Token"}), ) asyncio.run(notify.publish_async(title="消息标题", content="消息正文")) From 8407b067cb2336c870770e2d2b030351386f1ff8 Mon Sep 17 00:00:00 2001 From: mic1on Date: Thu, 17 Jul 2025 21:38:12 +0800 Subject: [PATCH 5/5] =?UTF-8?q?test:=20=E7=A7=BB=E9=99=A4PushDeer=20API?= =?UTF-8?q?=E9=94=99=E8=AF=AF=E5=A4=84=E7=90=86=E7=9A=84=E6=B5=8B=E8=AF=95?= =?UTF-8?q?=E7=94=A8=E4=BE=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_pushdeer.py | 45 +----------------------------------------- 1 file changed, 1 insertion(+), 44 deletions(-) diff --git a/tests/test_pushdeer.py b/tests/test_pushdeer.py index 98a688d..3f2b7ed 100644 --- a/tests/test_pushdeer.py +++ b/tests/test_pushdeer.py @@ -103,23 +103,6 @@ def test_send_success(self, mock_client, pushdeer_instance): # 对于text类型,content会覆盖title作为text参数 assert kwargs["params"]["desp"] == "Hello World" - @patch("httpx.Client") - def test_send_api_error(self, mock_client, pushdeer_instance): - """测试API返回错误""" - # 模拟响应 - mock_response = MagicMock() - mock_response.raise_for_status.return_value = None - mock_response.json.return_value = {"code": 1, "error": "Invalid token"} - - # 模拟客户端 - mock_client_instance = MagicMock() - mock_client_instance.get.return_value = mock_response - mock_client.return_value.__enter__.return_value = mock_client_instance - - # 执行发送,应该抛出异常 - with pytest.raises(RuntimeError, match="PushDeer API error: Invalid token"): - pushdeer_instance.send("Hello World") - @patch('httpx.AsyncClient') @pytest.mark.asyncio async def test_send_async_success(self, mock_async_client, pushdeer_instance): @@ -148,30 +131,4 @@ async def test_send_async_success(self, mock_async_client, pushdeer_instance): assert args[0] == pushdeer_instance.api_url assert "params" in kwargs # 对于text类型,content会覆盖title作为text参数 - assert kwargs["params"]["desp"] == "Hello World" - - @pytest.mark.asyncio - async def test_send_async_api_error(self, pushdeer_instance): - """测试异步API返回错误""" - # 创建一个真实的AsyncMock对象,它可以在await表达式中使用 - from unittest.mock import AsyncMock - - # 模拟响应 - mock_response = AsyncMock() - mock_response.raise_for_status = AsyncMock() - # 设置json方法为AsyncMock,返回值为字典 - mock_json = AsyncMock() - mock_json.return_value = {"code": 1, "error": "Invalid token"} - mock_response.json = mock_json - - # 模拟客户端 - mock_client = AsyncMock() - mock_client.get.return_value = mock_response - - # 替换httpx.AsyncClient - with patch("httpx.AsyncClient") as mock_async_client: - mock_async_client.return_value.__aenter__.return_value = mock_client - - # 执行发送,应该抛出异常 - with pytest.raises(RuntimeError, match="PushDeer API error: Invalid token"): - await pushdeer_instance.send_async("Hello World") \ No newline at end of file + assert kwargs["params"]["desp"] == "Hello World" \ No newline at end of file