feat(#1): creating notification implementation

lifeguard_telegram/__init__.py

@@ -1,12 +1,13 @@
 """
 Lifeguard integration with Telegram
 """
-import _thread
 import os
 
+from lifeguard.notifications import append_notification_implementation
 from lifeguard.logger import lifeguard_logger as logger
 
-from lifeguard_telegram.bot import load_bot_handlers, init_updater
+from lifeguard_telegram.bot import init_updater
+from lifeguard_telegram.notifications import TelegramNotificationBase
 
 
 class LifeguardTelegramPlugin:
@@ -20,6 +21,7 @@ def __init__(self, lifeguard_context):
 
 
 def init(lifeguard_context):
+    append_notification_implementation(TelegramNotificationBase)
     newpid = os.fork()
     if newpid == 0:
         logger.info("starting telegram process")

lifeguard_telegram/bot.py

@@ -6,7 +6,7 @@
 from lifeguard.settings import LIFEGUARD_DIRECTORY
 from telegram.ext import CommandHandler, Updater
 
-from lifeguard_telegram.settings import LIFEGUARD_TELEGRAM_BOT_TOKEN
+from lifeguard_telegram.settings import TELEGRAM_API_KEY
 
 CONTEXT = {"updater": None}
 
@@ -15,7 +15,7 @@ def init_updater():
     """
     Init start polling
     """
-    CONTEXT["updater"] = Updater(LIFEGUARD_TELEGRAM_BOT_TOKEN, use_context=True)
+    CONTEXT["updater"] = Updater(TELEGRAM_API_KEY, use_context=True)
 
     load_bot_handlers()
 

lifeguard_telegram/notifications.py

@@ -0,0 +1,74 @@
+"""
+Base of notification system
+"""
+from datetime import datetime
+
+import telepot
+
+from lifeguard.logger import lifeguard_logger as logger
+from lifeguard.notifications import NotificationBase
+
+from lifeguard_telegram.settings import (
+    TELEGRAM_API_KEY,
+    TELEGRAM_DEFAULT_CHAT_ID,
+)
+
+HEADERS = {"Content-Type": "application/json; charset=UTF-8"}
+
+
+class TelegramNotificationBase(NotificationBase):
+    """
+    Telegram notification
+    """
+
+    @property
+    def name(self):
+        return "telegram"
+
+    def send_single_message(self, content, settings):
+        logger.info("seding single message to msteams")
+
+        self.__send_message(content, settings)
+
+    def init_thread(self, content, settings):
+        logger.info("notify a new problem")
+
+        self.__send_message(content, settings)
+
+        return [datetime.now().strftime("%Y%m%d%H%M")]
+
+    def update_thread(self, threads, content, settings):
+        logger.info("notify updating problem status %s", threads)
+        self.__send_message(content, settings)
+
+    def close_thread(self, threads, content, settings):
+        logger.info("notify closing problem status %s", threads)
+        self.__send_message(content, settings)
+
+    def __send_message(self, content, settings):
+        if not isinstance(content, list):
+            content = [content]
+
+        for chat in (
+            settings.get("notification", {})
+            .get("telegram", {})
+            .get("chats", [TELEGRAM_DEFAULT_CHAT_ID])
+        ):
+            for entry in content:
+                self.__call_bot_send_message(chat, entry)
+
+    def __call_bot_send_message(self, chat, text):
+        logger.info("sending message to chat %s", chat)
+        try:
+            self.__get_bot().sendMessage(chat, text=text, parse_mode="Markdown")
+        except Exception as error:
+            logger.error("error sending message to chat %s", chat)
+            logger.error(error)
+            self.__get_bot().sendMessage(
+                chat, text="there was an error sending the message"
+            )
+
+    def __get_bot(self):
+        if not hasattr(self, "__bot"):
+            self.__bot = telepot.Bot(TELEGRAM_API_KEY)
+        return self.__bot

lifeguard_telegram/settings.py

@@ -5,21 +5,24 @@
 
 SETTINGS_MANAGER = SettingsManager(
     {
-        "LIFEGUARD_TELEGRAM_BOT_TOKEN": {
-            "default": "",
-            "description": "Telegram bot token",
-        },
         "LIFEGUARD_TELEGRAM_VALIDATIONS_HANDLER": {
             "default": "true",
             "description": "Enable telegram validations handler",
         },
+        "TELEGRAM_API_KEY": {
+            "default": "",
+            "description": "Telegram bot token",
+        },
+        "TELEGRAM_DEFAULT_CHAT_ID": {
+            "default": "",
+            "description": "Telegram default chat id",
+        },
     }
 )
 
-LIFEGUARD_TELEGRAM_BOT_TOKEN = SETTINGS_MANAGER.read_value(
-    "LIFEGUARD_TELEGRAM_BOT_TOKEN"
-)
 LIFEGUARD_TELEGRAM_VALIDATIONS_HANDLER_ENABLED = (
     SETTINGS_MANAGER.read_value("LIFEGUARD_TELEGRAM_VALIDATIONS_HANDLER_ENABLED")
     == "true"
 )
+TELEGRAM_API_KEY = SETTINGS_MANAGER.read_value("TELEGRAM_API_KEY")
+TELEGRAM_DEFAULT_CHAT_ID = SETTINGS_MANAGER.read_value("TELEGRAM_DEFAULT_CHAT_ID")

requirements.txt

@@ -1,2 +1,4 @@
 lifeguard==1.1.0
 python-telegram-bot==13.1
+telepot==12.7
+

setup.py

@@ -6,7 +6,7 @@
 
 setup(
     name="lifeguard-telegram",
-    version="1.0.0",
+    version="1.0.1",
     url="https://github.com/LifeguardSystem/lifeguard-telegram",
     author="Diego Rubin",
     author_email="[email protected]",
@@ -16,7 +16,7 @@
     description="Lifeguard integration with Telegram",
     long_description=long_description,
     long_description_content_type="text/markdown",
-    install_requires=["lifeguard", "python-telegram-bot"],
+    install_requires=["lifeguard", "python-telegram-bot", "telepot"],
     classifiers=[
         "Development Status :: 5 - Production/Stable",
         "Environment :: Plugins",

tests/test_notifications.py

@@ -0,0 +1,85 @@
+import unittest
+from unittest.mock import patch, MagicMock, call
+
+from datetime import datetime
+
+MOCK_TELEPOT = MagicMock(name="telepot")
+
+from lifeguard_telegram.notifications import TelegramNotificationBase
+
+
+class TelegramNotificationBaseTest(unittest.TestCase):
+    def setUp(self):
+        self.notification = TelegramNotificationBase()
+        self.mock_bot = MagicMock(name="bot")
+        MOCK_TELEPOT.Bot.return_value = self.mock_bot
+
+    def test_get_name(self):
+        self.assertEqual(self.notification.name, "telegram")
+
+    @patch("lifeguard_telegram.notifications.telepot", MOCK_TELEPOT)
+    def test_send_single_message(self):
+        self.notification.send_single_message("content", {})
+        self.mock_bot.sendMessage.assert_called_with(
+            "", text="content", parse_mode="Markdown"
+        )
+
+    @patch("lifeguard_telegram.notifications.telepot", MOCK_TELEPOT)
+    def test_send_multiple_single_message(self):
+        self.notification.send_single_message(["line1", "line2"], {})
+
+        self.mock_bot.sendMessage.assert_has_calls(
+            [
+                call("", text="line1", parse_mode="Markdown"),
+                call("", text="line2", parse_mode="Markdown"),
+            ]
+        )
+
+    @patch("lifeguard_telegram.notifications.telepot", MOCK_TELEPOT)
+    def test_init_thread(self):
+        self.notification.init_thread("content", {})
+
+        self.mock_bot.sendMessage.assert_called_with(
+            "", text="content", parse_mode="Markdown"
+        )
+
+    @patch("lifeguard_telegram.notifications.telepot", MOCK_TELEPOT)
+    @patch("lifeguard_telegram.notifications.datetime")
+    def test_init_thread_with_multiples_messages(self, mock_datetime):
+        mock_datetime.now.return_value = datetime(2022, 10, 11)
+
+        threads = self.notification.init_thread(["line1", "line2"], {})
+
+        self.mock_bot.sendMessage.assert_has_calls(
+            [
+                call("", text="line1", parse_mode="Markdown"),
+                call("", text="line2", parse_mode="Markdown"),
+            ]
+        )
+
+        self.assertEqual(threads, ["202210110000"])
+
+    @patch("lifeguard_telegram.notifications.telepot", MOCK_TELEPOT)
+    def test_update_thread(self):
+        self.notification.update_thread(["thread"], "content", {})
+
+        self.mock_bot.sendMessage.assert_called_with(
+            "", text="content", parse_mode="Markdown"
+        )
+
+    @patch("lifeguard_telegram.notifications.telepot", MOCK_TELEPOT)
+    def test_close_thread(self):
+        self.notification.close_thread(["thread"], "content", {})
+
+        self.mock_bot.sendMessage.assert_called_with(
+            "", text="content", parse_mode="Markdown"
+        )
+
+    @patch("lifeguard_telegram.notifications.telepot", MOCK_TELEPOT)
+    def test_error_on_send_message(self):
+        self.mock_bot.sendMessage.side_effect = [Exception("error"), None]
+        self.notification.close_thread(["thread"], "content", {})
+
+        self.mock_bot.sendMessage.assert_called_with(
+            "", text="there was an error sending the message"
+        )