diff --git a/example/demo.py b/example/demo.py index 6bb9847..34477a9 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": "Your Token"}), ) 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 b6e307a..f02ee1c 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,101 @@ 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 = "消息提醒" + + # 确定消息类型 + 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) + with httpx.Client() as client: - client.get(api_url, headers=self.headers) - logger.debug("`pushdeer` send successfully") + client.get(self.api_url, params=params, headers=self.headers) + logger.debug(f"`pushdeer` send 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) + async with httpx.AsyncClient() as client: - await client.get(api_url, headers=self.headers) - logger.debug("`pushdeer` send successfully") + await client.get(self.api_url, params=params, headers=self.headers) + logger.debug(f"`pushdeer` send 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..3f2b7ed --- /dev/null +++ b/tests/test_pushdeer.py @@ -0,0 +1,134 @@ +# -*- coding: utf-8 -*- +import pytest +import httpx +import json +from unittest.mock import patch, MagicMock, AsyncMock + +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["desp"] == "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"]["desp"] == "Hello World" + + @patch('httpx.AsyncClient') + @pytest.mark.asyncio + async def test_send_async_success(self, mock_async_client, pushdeer_instance): + """测试异步发送成功""" + # 模拟响应 + 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 + 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" \ No newline at end of file