Add SlackTarget class for sending notifications to Slack channels

pager/domain/models/notification_target.py

@@ -19,3 +19,10 @@ class SmsTarget(NotificationTarget):
 
     def get_contact_info(self) -> str:
         return self.phone_number
+
+@dataclass
+class SlackTarget(NotificationTarget):
+    channel: str
+
+    def get_contact_info(self) -> str:
+        return self.channel

pager/domain/services/pager_service.py

@@ -2,13 +2,14 @@
 from pager.domain.models.escalation_policy import EscalationPolicy
 from pager.domain.models.monitored_service import MonitoredService
 from pager.domain.events import Alert, Acknowledgement, HealthyEvent, Timeout
-from pager.domain.models.notification_target import NotificationTarget, EmailTarget, SmsTarget
+from pager.domain.models.notification_target import NotificationTarget, EmailTarget, SmsTarget, SlackTarget
 from pager.ports.email_sender import EmailSender
+from pager.ports.slack_sender import SlackSender
 from pager.ports.sms_sender import SmsSender
 from pager.ports.escalation_policy_repository import EscalationPolicyRepository
 
 class PagerService:
-    def __init__(self, policy_repo: EscalationPolicyRepository, email_sender: EmailSender, sms_sender: SmsSender):
+    def __init__(self, policy_repo: EscalationPolicyRepository, email_sender: EmailSender, sms_sender: SmsSender, slack_sender: SlackSender):
         """
         Initializes a PagerService object.
 
@@ -19,6 +20,7 @@ def __init__(self, policy_repo: EscalationPolicyRepository, email_sender: EmailS
         """
         self.policy_repo = policy_repo
         self.email_sender = email_sender
+        self.slack_sender = slack_sender
         self.sms_sender = sms_sender
         self.monitored_services: Dict[str, MonitoredService] = {}
 
@@ -104,6 +106,8 @@ def notify_targets(self, targets: List[NotificationTarget]):
                 self.email_sender.send(target.get_contact_info())
             elif isinstance(target, SmsTarget):
                 self.sms_sender.send(target.get_contact_info())
+            elif isinstance(target, SlackTarget):
+                self.slack_sender.send(target.get_contact_info())
 
     def get_or_create_monitored_service(self, service_id: str) -> MonitoredService:
         """

pager/ports/slack_sender.py

@@ -0,0 +1,13 @@
+class SlackSender:
+    def send(self, channel: str):
+        """
+        Sends a message to a Slack channel.
+
+        Args:
+            channel (str): The channel to send the message to.
+            message (str): The message to send.
+
+        Returns:
+            None
+        """
+        pass

tests/mocks/mock_slack_sender.py

@@ -0,0 +1,6 @@
+class MockSlackSender:
+    def __init__(self):
+        self.sent_slack = []
+
+    def send(self, channel: str):
+        self.sent_slack.append(channel)

tests/unit/domain/test_escalation_policy.py

@@ -1,15 +1,17 @@
 import pytest
 from pager.domain.models.escalation_policy import EscalationPolicy, EscalationLevel
-from pager.domain.models.notification_target import EmailTarget, SmsTarget
+from pager.domain.models.notification_target import EmailTarget, SmsTarget, SlackTarget
 
 def test_escalation_policy_creation():
     """
     Test case for creating an escalation policy with email and SMS targets.
     """
     email_target = EmailTarget(email='[email protected]')
     sms_target = SmsTarget(phone_number='1234567890')
+    slack_target = SlackTarget(channel='test')
     level1 = EscalationLevel(level_number=0, targets=[email_target])
     level2 = EscalationLevel(level_number=1, targets=[sms_target])
+    level3 = EscalationLevel(level_number=2, targets=[slack_target])
     policy = EscalationPolicy(monitored_service_id='service1', levels=[level1, level2])
 
     assert policy.monitored_service_id == 'service1'

tests/unit/domain/test_pager_service.py

@@ -2,27 +2,31 @@
 from pager.domain.models.escalation_policy import EscalationPolicy, EscalationLevel
 from pager.domain.models.monitored_service import MonitoredService
 from pager.domain.events import Alert, Acknowledgement, HealthyEvent
-from pager.domain.models.notification_target import EmailTarget, SmsTarget
+from pager.domain.models.notification_target import EmailTarget, SlackTarget, SmsTarget
 from pager.domain.services.pager_service import PagerService
 from tests.mocks.mock_email_sender import MockEmailSender
+from tests.mocks.mock_slack_sender import MockSlackSender
 from tests.mocks.mock_sms_sender import MockSmsSender
 from tests.mocks.mock_escalation_policy_repository import MockEscalationPolicyRepository
 
 @pytest.fixture
 def setup_pager_service():
     email_sender = MockEmailSender()
     sms_sender = MockSmsSender()
+    slack_sender = MockSlackSender()
     policy_repo = MockEscalationPolicyRepository({
         'service1': EscalationPolicy(
             monitored_service_id='service1',
             levels=[
                 EscalationLevel(level_number=0, targets=[EmailTarget(email='[email protected]')]),
-                EscalationLevel(level_number=1, targets=[SmsTarget(phone_number='1234567890')])
+                EscalationLevel(level_number=1, targets=[SmsTarget(phone_number='1234567890')]),
+                EscalationLevel(level_number=2, targets=[SlackTarget(channel='test')])
             ]
         )
     })
-    pager_service = PagerService(policy_repo, email_sender, sms_sender)
-    return pager_service, email_sender, sms_sender
+    pager_service = PagerService(policy_repo, email_sender, sms_sender, slack_sender)
+    return pager_service, email_sender, sms_sender, slack_sender
+
 
 # def test_handle_alert(setup_pager_service):
 #     pager_service, email_sender, sms_sender = setup_pager_service
@@ -34,7 +38,7 @@ def setup_pager_service():
 #     assert email_sender.sent_emails[0] == '[email protected]'
 
 def test_handle_acknowledgement(setup_pager_service):
-    pager_service, email_sender, sms_sender = setup_pager_service
+    pager_service, email_sender, sms_sender, slack_sender = setup_pager_service
     alert = Alert(service_id='service1', message='Test Alert')
     pager_service.handle_alert(alert)
 
@@ -45,7 +49,7 @@ def test_handle_acknowledgement(setup_pager_service):
     assert service.acknowledged
 
 def test_handle_healthy_event(setup_pager_service):
-    pager_service, email_sender, sms_sender = setup_pager_service
+    pager_service, email_sender, sms_sender, slack_sender = setup_pager_service
     alert = Alert(service_id='service1', message='Test Alert')
     pager_service.handle_alert(alert)