From 85ff4059128d62a62f2046c3c82d2bf2ac3c3cbf Mon Sep 17 00:00:00 2001 From: Lizzie Salmon Date: Fri, 20 Dec 2024 15:48:32 +0000 Subject: [PATCH 1/2] Changed some of the tests to use pytest instead of unittest. --- tests/lib/icinga_api/test_downtime.py | 380 +++++++++--------- tests/sensors/test_hypervisor_state_sensor.py | 99 +++-- 2 files changed, 240 insertions(+), 239 deletions(-) diff --git a/tests/lib/icinga_api/test_downtime.py b/tests/lib/icinga_api/test_downtime.py index 209a2fcb..a4e66e3e 100644 --- a/tests/lib/icinga_api/test_downtime.py +++ b/tests/lib/icinga_api/test_downtime.py @@ -9,218 +9,224 @@ from structs.icinga.downtime_details import DowntimeDetails -class TestScheduleDowntime: - @patch("icinga_api.downtime.requests.post") - def test_schedule_fixed_downtime_success(self, mock_post): - icinga_account = MagicMock() - details = DowntimeDetails( - object_type="Host", - object_name="example_host", - start_time=1725955200, - end_time=1725958800, - comment="Scheduled maintenance", - is_fixed=True, - duration=3600, - ) - mock_response = MagicMock() - mock_response.status_code = 201 - mock_post.return_value = mock_response +@patch("icinga_api.downtime.requests.post") +def test_schedule_fixed_downtime_success(mock_post): + icinga_account = MagicMock() + details = DowntimeDetails( + object_type="Host", + object_name="example_host", + start_time=1725955200, + end_time=1725958800, + comment="Scheduled maintenance", + is_fixed=True, + duration=3600, + ) + mock_response = MagicMock() + mock_response.status_code = 201 + mock_post.return_value = mock_response + + schedule_downtime(icinga_account, details) + + mock_post.assert_called_once() + mock_response.raise_for_status.assert_called_once() + + _, kwargs = mock_post.call_args + payload = json.loads(kwargs["data"]) + + expected_payload = { + "type": "Host", + "filter": 'host.name=="example_host"', + "author": "StackStorm", + "comment": "Scheduled maintenance", + "start_time": 1725955200, + "end_time": 1725958800, + "fixed": True, + "duration": None, + } + + assert payload == expected_payload + + +@patch("icinga_api.downtime.requests.post") +def test_schedule_downtime_request_fail(mock_post): + icinga_account = MagicMock() + details = DowntimeDetails( + object_type="Host", + object_name="missing_host", + start_time=1725955200, + end_time=1725958800, + comment="Scheduled maintenance", + is_fixed=True, + duration=3600, + ) + mock_response = MagicMock() + mock_response.status_code = 404 + mock_response.raise_for_status.side_effect = HTTPError("Not Found") + mock_post.return_value = mock_response + + with pytest.raises(HTTPError): + schedule_downtime(icinga_account, details) + mock_post.assert_called_once() + mock_response.raise_for_status.assert_called_once() + + +@patch("icinga_api.downtime.requests.post") +def test_schedule_flexible_downtime_success(mock_post): + icinga_account = MagicMock() + details = DowntimeDetails( + object_type="Host", + object_name="example_host", + start_time=1725955200, + end_time=1725958800, + duration=3600, + comment="Scheduled maintenance", + is_fixed=False, + ) + mock_response = MagicMock() + mock_response.status_code = 201 + mock_post.return_value = mock_response + + schedule_downtime(icinga_account, details) + + mock_post.assert_called_once() + mock_response.raise_for_status.assert_called_once() + + _, kwargs = mock_post.call_args + payload = json.loads(kwargs["data"]) + + expected_payload = { + "type": "Host", + "filter": 'host.name=="example_host"', + "author": "StackStorm", + "comment": "Scheduled maintenance", + "start_time": 1725955200, + "end_time": 1725958800, + "fixed": False, + "duration": 3600, + } + + assert payload == expected_payload + + +def test_missing_schedule_downtime_object_name(): + icinga_account = MagicMock() + details = DowntimeDetails( + object_type="Host", + object_name="", + start_time=1725955200, + end_time=1725958800, + duration=3600, + comment="Scheduled maintenance", + is_fixed=True, + ) + + with pytest.raises(MissingMandatoryParamError): schedule_downtime(icinga_account, details) - mock_post.assert_called_once() - mock_response.raise_for_status.assert_called_once() - - _, kwargs = mock_post.call_args - payload = json.loads(kwargs["data"]) - - expected_payload = { - "type": "Host", - "filter": 'host.name=="example_host"', - "author": "StackStorm", - "comment": "Scheduled maintenance", - "start_time": 1725955200, - "end_time": 1725958800, - "fixed": True, - "duration": None, - } - - assert payload == expected_payload - - @patch("icinga_api.downtime.requests.post") - def test_schedule_downtime_request_fail(self, mock_post): - icinga_account = MagicMock() - details = DowntimeDetails( - object_type="Host", - object_name="missing_host", - start_time=1725955200, - end_time=1725958800, - comment="Scheduled maintenance", - is_fixed=True, - duration=3600, - ) - mock_response = MagicMock() - mock_response.status_code = 404 - mock_response.raise_for_status.side_effect = HTTPError("Not Found") - mock_post.return_value = mock_response - with pytest.raises(HTTPError): - schedule_downtime(icinga_account, details) +def test_missing_schedule_downtime_start_time(): + icinga_account = MagicMock() + details = DowntimeDetails( + object_type="Host", + object_name="example_host", + start_time=None, + end_time=1725958800, + duration=3600, + comment="Scheduled maintenance", + is_fixed=True, + ) - mock_post.assert_called_once() - mock_response.raise_for_status.assert_called_once() + with pytest.raises(MissingMandatoryParamError): + schedule_downtime(icinga_account, details) - @patch("icinga_api.downtime.requests.post") - def test_schedule_flexible_downtime_success(self, mock_post): - icinga_account = MagicMock() - details = DowntimeDetails( - object_type="Host", - object_name="example_host", - start_time=1725955200, - end_time=1725958800, - duration=3600, - comment="Scheduled maintenance", - is_fixed=False, - ) - mock_response = MagicMock() - mock_response.status_code = 201 - mock_post.return_value = mock_response +def test_missing_schedule_downtime_end_time(): + icinga_account = MagicMock() + details = DowntimeDetails( + object_type="Host", + object_name="example_host", + start_time=1725958800, + end_time=None, + duration=3600, + comment="Scheduled maintenance", + is_fixed=True, + ) + + with pytest.raises(MissingMandatoryParamError): schedule_downtime(icinga_account, details) - mock_post.assert_called_once() - mock_response.raise_for_status.assert_called_once() - _, kwargs = mock_post.call_args - payload = json.loads(kwargs["data"]) +def test_missing_schedule_downtime_comment(): + icinga_account = MagicMock() + details = DowntimeDetails( + object_type="Host", + object_name="example_host", + start_time=1725955200, + end_time=1725958800, + duration=3600, + comment="", + is_fixed=True, + ) - expected_payload = { - "type": "Host", - "filter": 'host.name=="example_host"', - "author": "StackStorm", - "comment": "Scheduled maintenance", - "start_time": 1725955200, - "end_time": 1725958800, - "fixed": False, - "duration": 3600, - } + with pytest.raises(MissingMandatoryParamError): + schedule_downtime(icinga_account, details) - assert payload == expected_payload - def test_missing_schedule_downtime_object_name(self): - icinga_account = MagicMock() - details = DowntimeDetails( - object_type="Host", - object_name="", - start_time=1725955200, - end_time=1725958800, - duration=3600, - comment="Scheduled maintenance", - is_fixed=True, - ) +@patch("icinga_api.downtime.requests.post") +def test_remove_downtime_success(mock_post): + icinga_account = MagicMock() - with pytest.raises(MissingMandatoryParamError): - schedule_downtime(icinga_account, details) + mock_response = MagicMock() + mock_response.status_code = 201 + mock_post.return_value = mock_response - def test_missing_schedule_downtime_start_time(self): - icinga_account = MagicMock() - details = DowntimeDetails( - object_type="Host", - object_name="example_host", - start_time=None, - end_time=1725958800, - duration=3600, - comment="Scheduled maintenance", - is_fixed=True, - ) + remove_downtime( + icinga_account, + object_type="Host", + object_name="example_host", + ) - with pytest.raises(MissingMandatoryParamError): - schedule_downtime(icinga_account, details) + mock_post.assert_called_once() + mock_response.raise_for_status.assert_called_once() - def test_missing_schedule_downtime_end_time(self): - icinga_account = MagicMock() - details = DowntimeDetails( - object_type="Host", - object_name="example_host", - start_time=1725958800, - end_time=None, - duration=3600, - comment="Scheduled maintenance", - is_fixed=True, - ) + _, kwargs = mock_post.call_args + payload = json.loads(kwargs["data"]) - with pytest.raises(MissingMandatoryParamError): - schedule_downtime(icinga_account, details) + expected_payload = { + "type": "Host", + "filter": 'host.name=="example_host"', + "author": "StackStorm", + } - def test_missing_schedule_downtime_comment(self): - icinga_account = MagicMock() - details = DowntimeDetails( - object_type="Host", - object_name="example_host", - start_time=1725955200, - end_time=1725958800, - duration=3600, - comment="", - is_fixed=True, - ) - - with pytest.raises(MissingMandatoryParamError): - schedule_downtime(icinga_account, details) + assert payload == expected_payload -class TestRemoveDowntime: - @patch("icinga_api.downtime.requests.post") - def test_remove_downtime_success(self, mock_post): - icinga_account = MagicMock() +@patch("icinga_api.downtime.requests.post") +def test_remove_downtime_request_fail(mock_post): + icinga_account = MagicMock() - mock_response = MagicMock() - mock_response.status_code = 201 - mock_post.return_value = mock_response + mock_response = MagicMock() + mock_response.status_code = 404 + mock_response.raise_for_status.side_effect = HTTPError("Not Found") + mock_post.return_value = mock_response + with pytest.raises(HTTPError): remove_downtime( icinga_account, object_type="Host", - object_name="example_host", + object_name="missing_host", ) - mock_post.assert_called_once() - mock_response.raise_for_status.assert_called_once() - - _, kwargs = mock_post.call_args - payload = json.loads(kwargs["data"]) - - expected_payload = { - "type": "Host", - "filter": 'host.name=="example_host"', - "author": "StackStorm", - } - - assert payload == expected_payload - - @patch("icinga_api.downtime.requests.post") - def test_remove_downtime_request_fail(self, mock_post): - icinga_account = MagicMock() - - mock_response = MagicMock() - mock_response.status_code = 404 - mock_response.raise_for_status.side_effect = HTTPError("Not Found") - mock_post.return_value = mock_response - - with pytest.raises(HTTPError): - remove_downtime( - icinga_account, - object_type="Host", - object_name="missing_host", - ) - - mock_post.assert_called_once() - mock_response.raise_for_status.assert_called_once() - - def test_missing_remove_downtime_object_name(self): - icinga_account = MagicMock() - with pytest.raises(MissingMandatoryParamError): - remove_downtime( - icinga_account, - object_type="Host", - object_name="", - ) + mock_post.assert_called_once() + mock_response.raise_for_status.assert_called_once() + + +def test_missing_remove_downtime_object_name(): + icinga_account = MagicMock() + with pytest.raises(MissingMandatoryParamError): + remove_downtime( + icinga_account, + object_type="Host", + object_name="", + ) diff --git a/tests/sensors/test_hypervisor_state_sensor.py b/tests/sensors/test_hypervisor_state_sensor.py index f8d838e6..6d61b186 100644 --- a/tests/sensors/test_hypervisor_state_sensor.py +++ b/tests/sensors/test_hypervisor_state_sensor.py @@ -1,62 +1,57 @@ -import unittest from unittest.mock import MagicMock, patch +import pytest from enums.hypervisor_states import HypervisorState from sensors.src.hypervisor_state_sensor import HypervisorStateSensor -class TestHypervisorStateSensor(unittest.TestCase): - """ - Test cases for testing hypervisor state sensor - """ +@pytest.fixture(name="sensor") +def state_sensor_fixture(): + sensor_service = MagicMock() + config = {"hypervisor_sensor": {"cloud_account": "dev", "uptime_limit": 180}} + poll_interval = 10 + sensor = HypervisorStateSensor(sensor_service, config, poll_interval) + sensor.sensor_service = sensor_service # So that I dont need two fixtures? + return sensor - def setUp(self): - self.sensor_service = MagicMock() - self.config = { - "hypervisor_sensor": {"cloud_account": "dev", "uptime_limit": 180} - } - self.poll_interval = 10 - self.sensor = HypervisorStateSensor( - self.sensor_service, self.config, self.poll_interval - ) - - @patch("sensors.src.hypervisor_state_sensor.get_hypervisor_state") - @patch("sensors.src.hypervisor_state_sensor.query_hypervisor_state") - def test_poll(self, mock_query_hypervisor_state, mock_get_hypervisor_state): - """ - Test main function of sensor, polling state of hypervisor state - """ - mock_query_hypervisor_state.return_value = [ - { - "hypervisor_name": "hv1", - "hypervisor_uptime": "up 1000 days, 12:34", - "hypervisor_status": "enabled", - "hypervisor_state": "up", - "hypervisor_server_count": 5, - } - ] - - self.sensor_service.get_value.return_value = "RUNNING" - mock_get_hypervisor_state.return_value = HypervisorState.PENDING_MAINTENANCE - - self.sensor.poll() - - mock_query_hypervisor_state.assert_called_once_with("dev") - - mock_get_hypervisor_state.assert_called_once_with( - mock_query_hypervisor_state.return_value[0], uptime_limit=180 - ) - - expected_payload = { +@patch("sensors.src.hypervisor_state_sensor.get_hypervisor_state") +@patch("sensors.src.hypervisor_state_sensor.query_hypervisor_state") +def test_poll(mock_query_hypervisor_state, mock_get_hypervisor_state, sensor): + """ + Test main function of sensor, polling state of hypervisor state + """ + sensor.sensor_service = MagicMock() + mock_query_hypervisor_state.return_value = [ + { "hypervisor_name": "hv1", - "previous_state": self.sensor_service.get_value.return_value, - "current_state": mock_get_hypervisor_state.return_value.name, + "hypervisor_uptime": "up 1000 days, 12:34", + "hypervisor_status": "enabled", + "hypervisor_state": "up", + "hypervisor_server_count": 5, } + ] + + sensor.sensor_service.get_value.return_value = "RUNNING" + mock_get_hypervisor_state.return_value = HypervisorState.PENDING_MAINTENANCE + + sensor.poll() + + mock_query_hypervisor_state.assert_called_once_with("dev") + + mock_get_hypervisor_state.assert_called_once_with( + mock_query_hypervisor_state.return_value[0], uptime_limit=180 + ) + + expected_payload = { + "hypervisor_name": "hv1", + "previous_state": sensor.sensor_service.get_value.return_value, + "current_state": mock_get_hypervisor_state.return_value.name, + } - self.sensor_service.dispatch.assert_called_once_with( - trigger="stackstorm_openstack.hypervisor.state_change", - payload=expected_payload, - ) - self.sensor_service.set_value.assert_called_once_with( - name="hv1", value="PENDING_MAINTENANCE" - ) + sensor.sensor_service.dispatch.assert_called_once_with( + trigger="stackstorm_openstack.hypervisor.state_change", + payload=expected_payload, + ) + sensor.sensor_service.set_value.assert_called_once_with( + name="hv1", value="PENDING_MAINTENANCE" + ) From d51cddc779483f40536a7946ac4f03c5632eae34 Mon Sep 17 00:00:00 2001 From: Lizzie Salmon Date: Mon, 6 Jan 2025 15:06:44 +0000 Subject: [PATCH 2/2] Responded to comments - correct fixture. --- tests/sensors/test_hypervisor_state_sensor.py | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/tests/sensors/test_hypervisor_state_sensor.py b/tests/sensors/test_hypervisor_state_sensor.py index 6d61b186..0eb54b97 100644 --- a/tests/sensors/test_hypervisor_state_sensor.py +++ b/tests/sensors/test_hypervisor_state_sensor.py @@ -6,12 +6,11 @@ @pytest.fixture(name="sensor") def state_sensor_fixture(): - sensor_service = MagicMock() - config = {"hypervisor_sensor": {"cloud_account": "dev", "uptime_limit": 180}} - poll_interval = 10 - sensor = HypervisorStateSensor(sensor_service, config, poll_interval) - sensor.sensor_service = sensor_service # So that I dont need two fixtures? - return sensor + return HypervisorStateSensor( + sensor_service=MagicMock(), + config={"hypervisor_sensor": {"cloud_account": "dev", "uptime_limit": 180}}, + poll_interval=10, + ) @patch("sensors.src.hypervisor_state_sensor.get_hypervisor_state") @@ -20,7 +19,6 @@ def test_poll(mock_query_hypervisor_state, mock_get_hypervisor_state, sensor): """ Test main function of sensor, polling state of hypervisor state """ - sensor.sensor_service = MagicMock() mock_query_hypervisor_state.return_value = [ { "hypervisor_name": "hv1",