test: add testing cases for Notifier and NotifierHandler

backend/src/module/core/sub_thread.py

@@ -1,5 +1,4 @@
 import threading
-import time
 from concurrent.futures import ThreadPoolExecutor
 
 from module.conf import settings

backend/src/module/notification/__init__.py

@@ -1,7 +1,6 @@
 import logging
-from typing import get_args
+from typing import Optional, get_args
 
-from module.conf import settings
 from module.database import Database
 from module.models.bangumi import Notification
 
@@ -12,45 +11,47 @@
 
 class NotifierHandler(logging.Handler):
     def __init__(self, service_name: str, **kwargs) -> None:
-        notifier_config = kwargs.get("notifier", {})
-        if not notifier_config:
-            raise ValueError("Invalid notifier config")
-
-        self.notifier = Notifier(service_name, **notifier_config)
+        notifier_config = kwargs.pop("config", {})
+        self.notifier = Notifier(service_name, config=notifier_config)
         super().__init__(**kwargs)
 
     def emit(self, record: logging.LogRecord) -> None:
-        # TODO: get some information from record and send it by notifier
         try:
-            self.notifier.send(record)
+            self.notifier.send(record=record)
         except Exception as e:
-            logger.error(f"Notification error: {e}")
+            logger.error(f"Can't send log record to notifier because: {e}")
 
 
 class Notifier:
     def __init__(self, service_name: str, **kwargs):
         assert service_name in get_args(
-            NotificationService
+            NotificationType
         ), f"Invalid service name: {service_name}"
 
         notifier_config = kwargs.get("config", {})
         if not notifier_config:
             raise ValueError("Invalid notifier config")
 
-        self.notifier = services[settings.notification.type](**notifier_config)
+        self.notifier = services[service_name](**notifier_config)
 
-    def _get_poster(name: str) -> str:
+    def _get_poster(self, name: str) -> str:
         with Database() as db:
             poster = db.bangumi.match_poster(name)
             return poster
 
-    def send(self, notification: Notification) -> bool:
-        poster = self._get_poster(notification)
-        notification.poster_path = poster
+    def send(self, **kwargs) -> bool:
+        notification: Optional[Notification] = kwargs.pop("notification", None)
+        record: Optional[logging.LogRecord] = kwargs.pop("record", None)
+        if notification:
+            poster = self._get_poster(notification.official_title)
+            notification.poster_path = poster
+            data = dict(notification=notification)
+        else:
+            data = dict(record=record)
 
         try:
-            self.notifier.send(notification)
-            logger.debug(f"Send notification: {notification.official_title}")
+            self.notifier.send(**data)
+            logger.debug(f"Send notification: {data}")
             return True
         except Exception as e:
             logger.warning(f"Failed to send notification: {e}")

backend/src/module/notification/base.py

@@ -1,10 +1,26 @@
 from abc import ABC, abstractmethod
+from textwrap import dedent
+from typing import Any, Dict, Literal, Optional, TypeAlias, TypeVar
 
+import aiohttp
 from pydantic import BaseModel, Field
 
-DEFAULT_MESSAGE_TEMPLATE = """\
-番剧名称：{official_title}\n季度：第{season}季\n更新集数：第{episode}集\n{poster_path}
-"""
+DEFAULT_MESSAGE_TEMPLATE = dedent(
+    """\
+    番剧名称：{official_title}
+    季度：第{season}季
+    更新集数：第{episode}集
+    {poster_path}
+    """
+)
+
+DEFAULT_LOG_TEMPLATE = dedent(
+    """\
+    日志时间：{dt}
+    日志等级：{levelname}
+    日志消息：{msg}
+    """
+)
 
 
 class NotifierAdapter(BaseModel, ABC):

backend/src/test/notification/test_base.py

@@ -0,0 +1,47 @@
+from textwrap import dedent
+
+import pytest
+from aioresponses import aioresponses
+from module.notification.base import (
+    DEFAULT_LOG_TEMPLATE,
+    DEFAULT_MESSAGE_TEMPLATE,
+    NotifierAdapter,
+)
+
+
+def test_default_message_template(fake_notification):
+    content = DEFAULT_MESSAGE_TEMPLATE.format(**fake_notification.dict())
+    expected = dedent(
+        """\
+        番剧名称：AutoBangumi Test
+        季度：第1季
+        更新集数：第1集
+        https://mikanani.me
+        """
+    )
+
+    assert content == expected
+
+
+def test_default_log_template():
+    content = DEFAULT_LOG_TEMPLATE.format(
+        dt="2021-08-15 21:58:44,123",
+        levelname="INFO",
+        msg="test message",
+    )
+    expected = dedent(
+        """\
+        日志时间：2021-08-15 21:58:44,123
+        日志等级：INFO
+        日志消息：test message
+        """
+    )
+
+    assert content == expected
+
+
+def test_NotifierAdapter_non_implementation_raised(fake_notification):
+    with pytest.raises(NotImplementedError) as exc:
+        NotifierAdapter.send(fake_notification)
+
+    assert exc.match("send method is not implemented yet.")

backend/src/test/notification/test_notification.py

@@ -0,0 +1,79 @@
+import logging
+from unittest import mock
+
+import pytest
+from module.notification import Notifier, NotifierHandler
+
+
+class TestNotifierHandler:
+    @classmethod
+    def setup_class(cls):
+        cls.handler = NotifierHandler(
+            service_name="gotify",
+            config=dict(
+                token="YOUR_TOKEN",
+                base_url="https://example.com",
+            ),
+        )
+        logging.basicConfig(level=logging.DEBUG)
+        cls.logger = logging.getLogger("test.TestNotifierHandler")
+        cls.logger.addHandler(cls.handler)
+        cls.logger.setLevel(logging.DEBUG)
+
+    def test_emit(self, caplog):
+        with mock.patch("module.notification.Notifier.send") as m:
+            m.return_value = True
+            self.logger.warning("test warning")
+            caplog.set_level(logging.DEBUG, logger=self.logger.name)
+            print(f"capture log: {caplog.text}")
+            assert "test" in caplog.text
+
+
+class TestNotifier:
+    @classmethod
+    def setup_class(cls):
+        cls.notifier = Notifier(
+            service_name="gotify",
+            config=dict(
+                token="YOUR_TOKEN",
+                base_url="https://example.com",
+            ),
+        )
+
+    def test_init_without_config(self):
+        with pytest.raises(ValueError) as exc:
+            Notifier(service_name="gotify")
+
+        assert exc.match("Invalid notifier config")
+
+    def test__get_poster(self):
+        with mock.patch("module.notification.Notifier._get_poster") as m:
+            expected = "https://mikanani.me"
+            m.return_value = expected
+
+            res = self.notifier._get_poster(name="test")
+            m.assert_called_once_with(name="test")
+            assert res == expected
+
+    def test__get_poster_failed(self):
+        with mock.patch("module.notification.Notifier._get_poster") as m:
+            m.return_value = ""
+
+            res = self.notifier._get_poster(name="")
+            m.assert_called_once_with(name="")
+            assert res == ""
+
+    def test_send(self, fake_notification):
+        with mock.patch("module.notification.Notifier.send") as m:
+            m.return_value = True
+            assert self.notifier.send(notification=fake_notification)
+
+            m.assert_called_once_with(notification=fake_notification)
+
+    def test_send_with_side_effect(self, fake_notification):
+        with mock.patch("module.notification.Notifier.send") as m:
+            m.side_effect = Exception("Unexpected error.")
+            with pytest.raises(Exception) as exc:
+                self.notifier.send(notification=fake_notification)
+
+            assert exc.match("Unexpected error.")