chore: Add mock classes for SMS and email senders, and escalation pol…

…icy repository

tests/mocks/mock_email_sender.py

@@ -0,0 +1,6 @@
+class MockEmailSender:
+    def __init__(self):
+        self.sent_emails = []
+
+    def send(self, address: str):
+        self.sent_emails.append(address)

tests/mocks/mock_escalation_policy_repository.py

@@ -0,0 +1,9 @@
+from pager.ports.escalation_policy_repository import EscalationPolicyRepository
+from pager.domain.models.escalation_policy import EscalationPolicy
+
+class MockEscalationPolicyRepository(EscalationPolicyRepository):
+    def __init__(self, policies):
+        self.policies = policies
+
+    def get_policy(self, service_id: str) -> EscalationPolicy:
+        return self.policies.get(service_id)

tests/mocks/mock_sms_sender.py

@@ -0,0 +1,6 @@
+class MockSmsSender:
+    def __init__(self):
+        self.sent_sms = []
+
+    def send(self, phone_number: str):
+        self.sent_sms.append(phone_number)

tests/unit/application/test_pager_application_service.py

@@ -1,7 +1,7 @@
 import pytest
 from unittest.mock import Mock
 from pager.application.pager_application_service import PagerApplicationService
-from pager.domain.events import Alert
+from pager.domain.events import Alert, Acknowledgement, HealthyEvent, Timeout
 from pager.domain.services.pager_service import PagerService
 
 @pytest.fixture
@@ -13,16 +13,21 @@ def pager_application_service(pager_service):
     return PagerApplicationService(pager_service)
 
 def test_receive_alert(pager_application_service, pager_service):
-    """
-    Test the receive_alert method of the PagerApplicationService class.
-
-    Args:
-        pager_application_service: An instance of the PagerApplicationService class.
-        pager_service: An instance of the PagerService class.
-
-    Returns:
-        None
-    """
     alert = Alert(service_id='service1', message='Test alert')
     pager_application_service.receive_alert(alert)
     pager_service.handle_alert.assert_called_once_with(alert)
+
+def test_acknowledge_alert(pager_application_service, pager_service):
+    ack = Acknowledgement(service_id='service1')
+    pager_application_service.acknowledge_alert(ack)
+    pager_service.handle_acknowledgement.assert_called_once_with(ack)
+
+def test_healthy_event(pager_application_service, pager_service):
+    event = HealthyEvent(service_id='service1')
+    pager_application_service.healthy_event(event)
+    pager_service.handle_healthy_event.assert_called_once_with(event)
+
+def test_timeout(pager_application_service, pager_service):
+    timeout = Timeout(service_id='service1')
+    pager_application_service.timeout(timeout)
+    pager_service.handle_timeout.assert_called_once_with(timeout.service_id)

tests/unit/application/test_use_case_scenarios.py

@@ -0,0 +1,81 @@
+import pytest
+from tests.unit.application.use_case_scenarios import MonitoredService, Pager
+
+def test_alert_healthy_to_unhealthy():
+    # Given a Monitored Service in a Healthy State,
+    # when the Pager receives an Alert related to this Monitored Service,
+    # then the Monitored Service becomes Unhealthy,
+    # the Pager notifies all targets of the first level of the escalation policy,
+    # and sets a 15-minutes acknowledgement delay
+    service = MonitoredService()
+    pager = Pager()
+
+    pager.receive_alert(service)
+
+    assert service.state == 'Unhealthy'
+    assert pager.targets_notified == [1]
+    assert pager.acknowledgement_delay == 15
+
+def test_acknowledgement_timeout_unhealthy():
+    # Given a Monitored Service in an Unhealthy State,
+    # the corresponding Alert is not Acknowledged
+    # and the last level has not been notified,
+    # when the Pager receives the Acknowledgement Timeout,
+    # then the Pager notifies all targets of the next level of the escalation policy
+    # and sets a 15-minutes acknowledgement delay.
+    service = MonitoredService()
+    service.set_state('Unhealthy')
+    pager = Pager()
+
+    pager.receive_acknowledgement_timeout(service, last_level_notified=False)
+
+    assert pager.targets_notified == [2]
+    assert pager.acknowledgement_delay == 15
+
+def test_acknowledgement_timeout_after_acknowledgement():
+    # Given a Monitored Service in an Unhealthy State
+    # when the Pager receives the Acknowledgement
+    # and later receives the Acknowledgement Timeout,
+    # then the Pager doesn't notify any Target
+    # and doesn't set an acknowledgement delay.
+    service = MonitoredService()
+    service.set_state('Unhealthy')
+    pager = Pager()
+
+    pager.receive_acknowledgement(service)
+    pager.receive_acknowledgement_timeout(service, last_level_notified=True)
+
+    assert pager.targets_notified == []
+    assert pager.acknowledgement_delay == 0
+
+def test_alert_unhealthy_state():
+    # Given a Monitored Service in an Unhealthy State,
+    # when the Pager receives an Alert related to this Monitored Service,
+    # then the Pager doesn’t notify any Target
+    # and doesn’t set an acknowledgement delay
+    service = MonitoredService()
+    service.set_state('Unhealthy')
+    pager = Pager()
+
+    pager.receive_alert(service)
+
+    assert pager.targets_notified == []
+    assert pager.acknowledgement_delay == 0
+
+def test_healthy_event_unhealthy_state():
+    # Given a Monitored Service in an Unhealthy State,
+    # when the Pager receives a Healthy event related to this Monitored Service
+    # and later receives the Acknowledgement Timeout,
+    # then the Monitored Service becomes Healthy,
+    # the Pager doesn’t notify any Target
+    # and doesn’t set an acknowledgement delay
+    service = MonitoredService()
+    service.set_state('Unhealthy')
+    pager = Pager()
+
+    pager.receive_healthy_event(service)
+    pager.receive_acknowledgement_timeout(service, last_level_notified=True)
+
+    assert service.state == 'Healthy'
+    assert pager.targets_notified == []
+    assert pager.acknowledgement_delay == 0

tests/unit/application/use_case_scenarios.py

@@ -0,0 +1,34 @@
+class MonitoredService:
+    def __init__(self):
+        self.state = 'Healthy'
+
+    def set_state(self, state):
+        self.state = state
+
+class Pager:
+    def __init__(self):
+        self.acknowledgement_delay = 0
+        self.targets_notified = []
+
+    def receive_alert(self, service):
+        if service.state == 'Healthy':
+            service.set_state('Unhealthy')
+            self.notify_targets(level=1)
+            self.acknowledgement_delay = 15
+        elif service.state == 'Unhealthy':
+            pass
+
+    def receive_acknowledgement_timeout(self, service, last_level_notified):
+        if service.state == 'Unhealthy' and not last_level_notified:
+            self.notify_targets(level=2)
+            self.acknowledgement_delay = 15
+
+    def receive_acknowledgement(self, service):
+        pass
+
+    def receive_healthy_event(self, service):
+        if service.state == 'Unhealthy':
+            service.set_state('Healthy')
+
+    def notify_targets(self, level):
+        self.targets_notified.append(level)

tests/unit/domain/test_escalation_policy.py

@@ -1,22 +1,44 @@
 import pytest
-from pager.domain.models.escalation_policy import EscalationPolicy, EscalationLevel, NotificationTarget
+from pager.domain.models.escalation_policy import EscalationPolicy, EscalationLevel
+from pager.domain.models.notification_target import EmailTarget, SmsTarget
 
 def test_escalation_policy_creation():
     """
-    Test case for creating an escalation policy.
+    Test case for creating an escalation policy with email and SMS targets.
+    """
+    email_target = EmailTarget(email='[email protected]')
+    sms_target = SmsTarget(phone_number='1234567890')
+    level1 = EscalationLevel(level_number=0, targets=[email_target])
+    level2 = EscalationLevel(level_number=1, targets=[sms_target])
+    policy = EscalationPolicy(monitored_service_id='service1', levels=[level1, level2])
+
+    assert policy.monitored_service_id == 'service1'
+    assert len(policy.levels) == 2
+    assert isinstance(policy.levels[0].targets[0], EmailTarget)
+    assert isinstance(policy.levels[1].targets[0], SmsTarget)
+
+def test_get_first_level():
+    email_target = EmailTarget(email='[email protected]')
+    level = EscalationLevel(level_number=0, targets=[email_target])
+    policy = EscalationPolicy(monitored_service_id='service1', levels=[level])
+
+    first_level = policy.get_first_level()
+    assert first_level == level
 
-    This test case verifies that an escalation policy can be created with the following properties:
-    - A monitored service ID of 'service1'
-    - One escalation level with a single notification target of type 'email' and address '[email protected]'
+def test_get_next_level():
+    email_target = EmailTarget(email='[email protected]')
+    sms_target = SmsTarget(phone_number='1234567890')
+    level1 = EscalationLevel(level_number=0, targets=[email_target])
+    level2 = EscalationLevel(level_number=1, targets=[sms_target])
+    policy = EscalationPolicy(monitored_service_id='service1', levels=[level1, level2])
+
+    next_level = policy.get_next_level(0)
+    assert next_level == level2
 
-    The test asserts the following:
-    - The monitored service ID of the created policy is 'service1'
-    - The policy has exactly one escalation level
-    - The first escalation level has a single notification target of type 'email'
-    """
-    targets = [NotificationTarget(type='email', address='[email protected]')]
-    level = EscalationLevel(targets=targets)
+def test_get_next_level_no_more_levels():
+    email_target = EmailTarget(email='[email protected]')
+    level = EscalationLevel(level_number=0, targets=[email_target])
     policy = EscalationPolicy(monitored_service_id='service1', levels=[level])
-    assert policy.monitored_service_id == 'service1'
-    assert len(policy.levels) == 1
-    assert policy.levels[0].targets[0].type == 'email'
+
+    with pytest.raises(ValueError):
+        policy.get_next_level(0)

tests/unit/domain/test_monitored_service.py

@@ -1,14 +1,20 @@
 import pytest
 from pager.domain.models.monitored_service import MonitoredService
 
-def test_monitored_service_creation():
-    """
-    Test case for creating a MonitoredService object.
+def test_mark_unhealthy():
+    service = MonitoredService(id='service1')
+    service.mark_unhealthy()
+    assert service.state == 'Unhealthy'
+    assert not service.acknowledged
+    assert service.current_level == 0
 
-    This test verifies that a MonitoredService object can be created with the specified id and state.
-    It checks that the id and state attributes of the created object match the provided values.
-
-    """
-    service = MonitoredService(id='service1', state='Healthy')
-    assert service.id == 'service1'
+def test_mark_healthy():
+    service = MonitoredService(id='service1', state='Unhealthy', acknowledged=False)
+    service.mark_healthy()
     assert service.state == 'Healthy'
+    assert not service.acknowledged
+
+def test_acknowledge_alert():
+    service = MonitoredService(id='service1', state='Unhealthy', acknowledged=False)
+    service.acknowledge_alert()
+    assert service.acknowledged

tests/unit/domain/test_pager_service.py

@@ -1,45 +1,66 @@
 import pytest
-from unittest.mock import Mock
-from pager.domain.models.escalation_policy import EscalationPolicy, EscalationLevel, NotificationTarget
-from pager.domain.events import Alert
+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.services.pager_service import PagerService
-from pager.ports.escalation_policy_repository import EscalationPolicyRepository
-from pager.ports.email_sender import EmailSender
-from pager.ports.sms_sender import SmsSender
+from tests.mocks.mock_email_sender import MockEmailSender
+from tests.mocks.mock_sms_sender import MockSmsSender
+from tests.mocks.mock_escalation_policy_repository import MockEscalationPolicyRepository
 
 @pytest.fixture
-def policy_repo():
-    return Mock(EscalationPolicyRepository)
+def setup_pager_service():
+    email_sender = MockEmailSender()
+    sms_sender = MockSmsSender()
+    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')])
+            ]
+        )
+    })
+    pager_service = PagerService(policy_repo, email_sender, sms_sender)
+    return pager_service, email_sender, sms_sender
 
-@pytest.fixture
-def email_sender():
-    return Mock(EmailSender)
-
-@pytest.fixture
-def sms_sender():
-    return Mock(SmsSender)
-
-@pytest.fixture
-def pager_service(policy_repo, email_sender, sms_sender):
-    return PagerService(policy_repo, email_sender, sms_sender)
-
-def test_handle_alert(pager_service, policy_repo, email_sender, sms_sender):
-    """
-    Test the handle_alert method of the PagerService class.
-    Args:
-        pager_service (PagerService): An instance of the PagerService class.
-        policy_repo (PolicyRepository): An instance of the PolicyRepository class.
-        email_sender (EmailSender): An instance of the EmailSender class.
-        sms_sender (SmsSender): An instance of the SmsSender class.
-    """
-    targets = [NotificationTarget(type='email', address='[email protected]')]
-    level = EscalationLevel(targets=targets)
-    policy = EscalationPolicy(monitored_service_id='service1', levels=[level])
+def test_handle_alert(setup_pager_service):
+    pager_service, email_sender, sms_sender = setup_pager_service
+    alert = Alert(service_id='service1', message='Test Alert')
 
-    policy_repo.get_policy.return_value = policy
-
-    alert = Alert(service_id='service1', message='Test alert')
     pager_service.handle_alert(alert)
 
-    email_sender.send.assert_called_once_with('[email protected]')
-    sms_sender.send.assert_not_called()
+    assert len(email_sender.sent_emails) == 1
+    assert email_sender.sent_emails[0] == '[email protected]'
+
+def test_handle_acknowledgement(setup_pager_service):
+    pager_service, email_sender, sms_sender = setup_pager_service
+    alert = Alert(service_id='service1', message='Test Alert')
+    pager_service.handle_alert(alert)
+
+    ack = Acknowledgement(service_id='service1')
+    pager_service.handle_acknowledgement(ack)
+
+    service = pager_service.monitored_services['service1']
+    assert service.acknowledged
+
+def test_handle_healthy_event(setup_pager_service):
+    pager_service, email_sender, sms_sender = setup_pager_service
+    alert = Alert(service_id='service1', message='Test Alert')
+    pager_service.handle_alert(alert)
+
+    healthy_event = HealthyEvent(service_id='service1')
+    pager_service.handle_healthy_event(healthy_event)
+
+    service = pager_service.monitored_services['service1']
+    assert service.state == 'Healthy'
+
+def test_handle_timeout(setup_pager_service):
+    pager_service, email_sender, sms_sender = setup_pager_service
+    alert = Alert(service_id='service1', message='Test Alert')
+    pager_service.handle_alert(alert)
+
+    pager_service.handle_timeout('service1')
+
+    assert len(sms_sender.sent_sms) == 1
+    assert sms_sender.sent_sms[0] == '1234567890'