From 3b38404c10f880bc3fbdf79d5067fd902af797a8 Mon Sep 17 00:00:00 2001 From: Lizzie Salmon Date: Fri, 20 Dec 2024 15:48:32 +0000 Subject: [PATCH] Changed some of the tests to use pytest instead of unittest. --- tests/lib/icinga_api/test_downtime.py | 380 +++++++++--------- tests/lib/jupyter/test_server_api.py | 172 ++++++++ tests/sensors/test_hypervisor_state_sensor.py | 99 +++-- 3 files changed, 412 insertions(+), 239 deletions(-) create mode 100644 tests/lib/jupyter/test_server_api.py 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/lib/jupyter/test_server_api.py b/tests/lib/jupyter/test_server_api.py new file mode 100644 index 00000000..40d3693c --- /dev/null +++ b/tests/lib/jupyter/test_server_api.py @@ -0,0 +1,172 @@ +from unittest.mock import patch, NonCallableMock, call +import pytest + + +from jupyter_api.api_endpoints import API_ENDPOINTS +from jupyter_api.user_api import UserApi +from structs.jupyter_users import JupyterUsers + + +@pytest.fixture(name="api_instance") +def api_instance_fixture() -> None: + """ + Creates a new instance of the API class for each test + """ + return UserApi() + + +@patch("jupyter_api.user_api.requests") +def test_start_servers_single_user(requests, api_instance): + """ + Tests that the start_servers method calls the correct endpoint + """ + token = NonCallableMock() + user_names = JupyterUsers(name="test", start_index=None, end_index=None) + requests.post.return_value.status_code = 201 + + api_instance.start_servers("dev", token, user_names) + requests.post.assert_called_once_with( + url=API_ENDPOINTS["dev"] + "/hub/api/users/test/server", + headers={"Authorization": f"token {token}"}, + timeout=60, + ) + + +@patch("jupyter_api.user_api.requests") +def test_start_servers_multiple_users(requests, api_instance): + """ + Tests that the start_servers method calls the correct endpoint + and usernames + """ + token = NonCallableMock() + start_index, end_index = 1, 10 + user_names = JupyterUsers(name="test", start_index=start_index, end_index=end_index) + requests.post.return_value.status_code = 201 + + api_instance.start_servers("dev", token, user_names) + for i, user_index in enumerate(range(start_index, end_index + 1)): + assert requests.post.call_args_list[i] == call( + url=API_ENDPOINTS["dev"] + f"/hub/api/users/test-{user_index}/server", + headers={"Authorization": f"token {token}"}, + timeout=60, + ) + assert requests.post.call_count == (end_index - start_index + 1) + + +def test_start_servers_missing_start_index(api_instance): + """ + Tests that the start_servers method raises an error if the start_index is not provided + """ + user_names = JupyterUsers(name="test", start_index=None, end_index=1) + with pytest.raises(RuntimeError): + api_instance.start_servers("dev", "token", user_names) + + +def test_start_servers_missing_end_index(api_instance): + """ + Tests that the start_servers method raises an error if the end_index is not provided + """ + user_names = JupyterUsers(name="test", start_index=1, end_index=None) + with pytest.raises(RuntimeError): + api_instance.start_servers("dev", "token", user_names) + + +def test_start_servers_incorrect_order(api_instance): + """ + Tests that the start_servers method raises an error if the end_index is greater than start + """ + user_names = JupyterUsers(name="test", start_index=2, end_index=1) + with pytest.raises(RuntimeError, match="must be less than"): + api_instance.start_servers("dev", "token", user_names) + + +@patch("jupyter_api.user_api.requests") +def test_start_servers_handles_error(requests, api_instance): + """ + Tests that the start_servers method logs an error if the request fails + """ + token = NonCallableMock() + user_names = JupyterUsers(name="test", start_index=None, end_index=None) + requests.post.return_value.status_code = 500 + + with pytest.raises(RuntimeError, match="Failed to request server"): + api_instance.start_servers("dev", token, user_names) + + +@patch("jupyter_api.user_api.requests") +def test_stop_servers_single_user(requests, api_instance): + """ + Tests that the stop_servers method calls the correct endpoint + """ + token = NonCallableMock() + user_names = JupyterUsers(name="test", start_index=None, end_index=None) + requests.delete.return_value.status_code = 204 + + api_instance.stop_servers("dev", token, user_names) + requests.delete.assert_called_once_with( + url=API_ENDPOINTS["dev"] + "/hub/api/users/test/server", + headers={"Authorization": f"token {token}"}, + timeout=60, + ) + + +@patch("jupyter_api.user_api.requests") +def test_stop_servers_multiple_users(requests, api_instance): + """ + Tests that the stop_servers method calls the correct endpoint + and usernames + """ + token = NonCallableMock() + start_index, end_index = 1, 10 + user_names = JupyterUsers(name="test", start_index=start_index, end_index=end_index) + requests.delete.return_value.status_code = 204 + + api_instance.stop_servers("dev", token, user_names) + for i, user_index in enumerate(range(start_index, end_index + 1)): + assert requests.delete.call_args_list[i] == call( + url=API_ENDPOINTS["dev"] + f"/hub/api/users/test-{user_index}/server", + headers={"Authorization": f"token {token}"}, + timeout=60, + ) + assert requests.delete.call_count == (end_index - start_index + 1) + + +def test_stop_servers_missing_start_index(api_instance): + """ + Tests that the stop_servers method raises an error if the start_index is not provided + """ + user_names = JupyterUsers(name="test", start_index=None, end_index=1) + with pytest.raises(RuntimeError): + api_instance.stop_servers("dev", "token", user_names) + + +def test_stop_servers_missing_end_index(api_instance): + """ + Tests that the stop_servers method raises an error if the end_index is not provided + """ + user_names = JupyterUsers(name="test", start_index=1, end_index=None) + with pytest.raises(RuntimeError): + api_instance.stop_servers("dev", "token", user_names) + + +def test_stop_servers_incorrect_order(api_instance): + """ + Tests that the stop_servers method raises an error if the end_index is greater than start + """ + user_names = JupyterUsers(name="test", start_index=2, end_index=1) + with pytest.raises(RuntimeError, match="must be less than"): + api_instance.stop_servers("dev", "token", user_names) + + +@patch("jupyter_api.user_api.requests") +def test_stop_servers_handles_error(requests, api_instance): + """ + Tests that the stop_servers method logs an error if the request fails + """ + token = NonCallableMock() + user_names = JupyterUsers(name="test", start_index=None, end_index=None) + requests.delete.return_value.status_code = 500 + + with pytest.raises(RuntimeError, match="Failed to stop server"): + api_instance.stop_servers("dev", token, user_names) + 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" + )