diff --git a/tests/mocks/mock_email_sender.py b/tests/mocks/mock_email_sender.py new file mode 100644 index 0000000..e4b1b26 --- /dev/null +++ b/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) diff --git a/tests/mocks/mock_escalation_policy_repository.py b/tests/mocks/mock_escalation_policy_repository.py new file mode 100644 index 0000000..8a7e4ad --- /dev/null +++ b/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) diff --git a/tests/mocks/mock_sms_sender.py b/tests/mocks/mock_sms_sender.py new file mode 100644 index 0000000..531d72a --- /dev/null +++ b/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) diff --git a/tests/unit/application/test_pager_application_service.py b/tests/unit/application/test_pager_application_service.py index 3c64034..e1c7a27 100644 --- a/tests/unit/application/test_pager_application_service.py +++ b/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) diff --git a/tests/unit/application/test_use_case_scenarios.py b/tests/unit/application/test_use_case_scenarios.py new file mode 100644 index 0000000..86c9745 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/tests/unit/application/use_case_scenarios.py b/tests/unit/application/use_case_scenarios.py new file mode 100644 index 0000000..df4e9cd --- /dev/null +++ b/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) \ No newline at end of file diff --git a/tests/unit/domain/test_escalation_policy.py b/tests/unit/domain/test_escalation_policy.py index bb00f76..9fed953 100644 --- a/tests/unit/domain/test_escalation_policy.py +++ b/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='test@example.com') + 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='test@example.com') + 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 'test@example.com' +def test_get_next_level(): + email_target = EmailTarget(email='test@example.com') + 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='test@example.com')] - level = EscalationLevel(targets=targets) +def test_get_next_level_no_more_levels(): + email_target = EmailTarget(email='test@example.com') + 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) diff --git a/tests/unit/domain/test_monitored_service.py b/tests/unit/domain/test_monitored_service.py index 749426a..c727204 100644 --- a/tests/unit/domain/test_monitored_service.py +++ b/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 diff --git a/tests/unit/domain/test_pager_service.py b/tests/unit/domain/test_pager_service.py index a908029..1274705 100644 --- a/tests/unit/domain/test_pager_service.py +++ b/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='test@example.com')]), + 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='test@example.com')] - 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('test@example.com') - sms_sender.send.assert_not_called() + assert len(email_sender.sent_emails) == 1 + assert email_sender.sent_emails[0] == 'test@example.com' + +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'