Added Slack notifier

README.md

@@ -111,7 +111,22 @@ SENDER="sender@example.com" SENDER_NAME="ChiaDog" RECIPIENT="you@example.com" HO
 
 ```
 
-### Slack, WhatsApp, ...?
+### Slack
+
+[Slack](https://slack.com/) apps are a simple way to get notifications in a channel in your Slack workspace. Follow the instructions for *Creating an App* on the [Getting started with Incoming Webhooks](https://api.slack.com/messaging/webhooks#getting_started) guide. 
+
+Copy & paste the Webhook URL into your `config.yaml`, that's it!
+
+- **Costs**: $0
+- **Advantages**: Easy setup
+
+You can test if the setup works correctly with:
+
+```
+SLACK_WEBHOOK_URL=<webhook_url> python3 -m unittest tests.notifier.test_slack_notifier
+```
+
+### WhatsApp, ...?
 
 These integrations can be easily added. Contributions are welcome!
 

config-example.yaml

@@ -33,3 +33,6 @@ notifier:
   discord:
     enable: false
     webhook_url: 'https://discord.com/api/webhooks/...'
+  slack:
+    enable: false
+    webhook_url: 'https://hooks.slack.com/services/...'

src/notifier/notify_manager.py

@@ -10,6 +10,7 @@
 from .smtp_notifier import SMTPNotifier
 from .telegram_notifier import TelegramNotifier
 from .discord_notifier import DiscordNotifier
+from .slack_notifier import SlackNotifier
 from src.config import Config
 
 
@@ -34,6 +35,7 @@ def _initialize_notifiers(self):
             "telegram": TelegramNotifier,
             "discord": DiscordNotifier,
             "smtp": SMTPNotifier,
+            "slack": SlackNotifier,
         }
         for key in self._config.keys():
             if key not in key_notifier_mapping.keys():

src/notifier/slack_notifier.py

@@ -0,0 +1,52 @@
+# std
+import http.client
+import logging
+import json
+import urllib.parse
+from typing import List
+
+# project
+from . import Notifier, Event, EventType
+
+
+class SlackNotifier(Notifier):
+    def __init__(self, title_prefix: str, config: dict):
+        logging.info("Initializing Slack notifier.")
+        super().__init__(title_prefix, config)
+        try:
+            self.webhook_url = config["webhook_url"]
+        except KeyError as key:
+            logging.error(f"Invalid config.yaml. Missing key: {key}")
+
+    def send_events_to_user(self, events: List[Event]) -> bool:
+        errors = False
+        for event in events:
+            if event.type == EventType.USER:
+                priority = ""
+
+                if event.priority == event.priority.HIGH:
+                    priority = "🚨"
+                elif event.priority == event.priority.NORMAL:
+                    priority = "⚠️"
+
+                request_body = json.dumps(
+                    {
+                        "text": f"{priority} *{self._title_prefix} {event.service.name}*\n{event.message}",
+                    }
+                )
+
+                o = urllib.parse.urlparse(self.webhook_url)
+                conn = http.client.HTTPSConnection(o.netloc)
+                conn.request(
+                    "POST",
+                    o.path,
+                    request_body,
+                    {"Content-type": "application/json"},
+                )
+                response = conn.getresponse()
+                if response.getcode() != 200:
+                    logging.warning(f"Problem sending event to user, code: {response.getcode()}")
+                    errors = True
+                conn.close()
+
+        return errors

tests/notifier/test_slack_notifier.py

@@ -0,0 +1,74 @@
+# std
+import os
+import unittest
+
+# project
+from src.notifier import Event, EventType, EventPriority, EventService
+from src.notifier.slack_notifier import SlackNotifier
+
+
+class TestSlackNotifier(unittest.TestCase):
+    def setUp(self) -> None:
+        webhook_url = os.getenv("SLACK_WEBHOOK_URL")
+        self.assertIsNotNone(webhook_url, "You must export SLACK_WEBHOOK_URL as env variable")
+        self.notifier = SlackNotifier(title_prefix="Test", config={"enable": True, "webhook_url": webhook_url})
+
+    @unittest.skipUnless(os.getenv("SLACK_WEBHOOK_URL"), "Run only if webhook available")
+    def testSlackLowPriorityNotifications(self):
+        errors = self.notifier.send_events_to_user(
+            events=[
+                Event(
+                    type=EventType.USER,
+                    priority=EventPriority.LOW,
+                    service=EventService.HARVESTER,
+                    message="Low priority notification 1.",
+                ),
+                Event(
+                    type=EventType.USER,
+                    priority=EventPriority.LOW,
+                    service=EventService.HARVESTER,
+                    message="Low priority notification 2.",
+                ),
+            ]
+        )
+        self.assertFalse(errors)
+
+    @unittest.skipUnless(os.getenv("SLACK_WEBHOOK_URL"), "Run only if webhook available")
+    def testSlackNormalPriorityNotifications(self):
+        errors = self.notifier.send_events_to_user(
+            events=[
+                Event(
+                    type=EventType.USER,
+                    priority=EventPriority.NORMAL,
+                    service=EventService.HARVESTER,
+                    message="Normal priority notification 1.",
+                ),
+                Event(
+                    type=EventType.USER,
+                    priority=EventPriority.NORMAL,
+                    service=EventService.HARVESTER,
+                    message="Normal priority notification 2.",
+                ),
+            ]
+        )
+        self.assertFalse(errors)
+
+    @unittest.skipUnless(os.getenv("SLACK_WEBHOOK_URL"), "Run only if webhook available")
+    def testSlackHighPriorityNotifications(self):
+        errors = self.notifier.send_events_to_user(
+            events=[
+                Event(
+                    type=EventType.USER,
+                    priority=EventPriority.HIGH,
+                    service=EventService.HARVESTER,
+                    message="High priority notification 1.",
+                ),
+                Event(
+                    type=EventType.USER,
+                    priority=EventPriority.HIGH,
+                    service=EventService.HARVESTER,
+                    message="High priority notification 2.",
+                ),
+            ]
+        )
+        self.assertFalse(errors)