diff --git a/lib/apis/openstack_api/openstack_event_list.py b/lib/apis/openstack_api/openstack_event_list.py new file mode 100644 index 000000000..34e9c64e4 --- /dev/null +++ b/lib/apis/openstack_api/openstack_event_list.py @@ -0,0 +1,67 @@ +import logging +from datetime import datetime, timezone +from openstack.exceptions import NotFoundException + +logger = logging.getLogger(__name__) + + +class EventList: + """ + a class to handle information related to the "Event List" + for a give Server. + For example, to get for how long the Server has been in the + current state. + """ + + def __init__(self, conn, server_id): + """ + :param conn: the Openstack Connection + :type conn: openstack.connection.Connection + :param server_id: the ID of the Server + :type server_id: str + :raises Exception: exception raised when the Server ID is not valid + """ + self.logger = logging.getLogger("EventList") + # first we check the Server actually exists + try: + conn.compute.get_server(server_id) + self.logger.debug("Verified the Server ID %s actually exists", server_id) + except NotFoundException as ex: + self.logger.error( + "The Server ID %s does not exist. This EventList object cannot be initialised. Raising an Exception.", + server_id, + ) + raise ex + self.events = list(conn.compute.server_actions(server_id)) + # the output of server_actions() is a generator + self.logger.debug( + "Object EventList for Server ID %s initialised properly", server_id + ) + + @property + def last_event(self): + """ + return: the last Event for this Server + rtype: ServerAction + """ + self.logger.debug("Getting the last event") + # the last Event happens to be the first item in the EventList + return self.events[0] + + @property + def seconds_in_current_state(self): + """ + :return: for how long the server has been in the current state + :rtype: int + """ + self.logger.debug("Getting the number seconds in current state") + last_event_t = self.last_event.start_time + # last_event_t looks like this + # 2024-07-25T12:08:40.000000 + last_event_dt = datetime.fromisoformat(last_event_t).replace( + tzinfo=timezone.utc + ) + time_delta = datetime.now(timezone.utc) - last_event_dt + seconds = int(time_delta.total_seconds()) + self.logger.info("Number seconds in current state is %s", seconds) + return seconds diff --git a/tests/lib/apis/openstack_api/test_openstack_event_list.py b/tests/lib/apis/openstack_api/test_openstack_event_list.py new file mode 100644 index 000000000..9b24761f0 --- /dev/null +++ b/tests/lib/apis/openstack_api/test_openstack_event_list.py @@ -0,0 +1,86 @@ +import unittest +from unittest.mock import MagicMock, patch +from datetime import datetime, timezone +from openstack.exceptions import NotFoundException + +from apis.openstack_api.openstack_event_list import EventList + + +class TestEventListBlackBox(unittest.TestCase): + + def setUp(self): + # Setup a mock connection object shared across tests + self.mock_conn = MagicMock() + self.server_id = "test-server-123" + + def test_init_with_valid_server_id(self): + """ + Scenario: The server ID exists in OpenStack. + Expectation: Object initializes without raising exceptions. + """ + # Configure mock API to simulate a valid server and an empty events generator + self.mock_conn.compute.get_server.return_value = MagicMock() + self.mock_conn.compute.server_actions.return_value = iter([]) + + try: + EventList(self.mock_conn, self.server_id) + except NotFoundException as e: + self.fail(f"Initialization raised an unexpected exception: {e}") + + def test_init_raises_exception_for_invalid_server_id(self): + """ + Scenario: The server ID does not exist. + Expectation: The OpenStack NotFoundException is propagated to the caller. + """ + # Configure mock API to raise NotFoundException + self.mock_conn.compute.get_server.side_effect = NotFoundException( + "Server not found" + ) + + with self.assertRaises(NotFoundException): + EventList(self.mock_conn, self.server_id) + + def test_last_event_returns_expected_value(self): + """ + Scenario: API returns a list of events. + Expectation: last_event property returns the first item from the API's feed. + """ + # Mocking the ServerAction objects returned by the API + mock_event_1 = MagicMock() + mock_event_2 = MagicMock() + + self.mock_conn.compute.get_server.return_value = MagicMock() + self.mock_conn.compute.server_actions.return_value = iter( + [mock_event_1, mock_event_2] + ) + + event_list = EventList(self.mock_conn, self.server_id) + + # Act & Assert + self.assertEqual(event_list.last_event, mock_event_1) + + @patch("apis.openstack_api.openstack_event_list.datetime") + def test_seconds_in_current_state_calculation(self, mock_datetime): + """ + Scenario: The last event occurred at a specific ISO timestamp. + Expectation: seconds_in_current_state accurately calculates the delta against 'now'. + """ + # 1. Mock the 'current' time to a fixed point (2026-06-15 12:00:00 UTC) + fixed_now = datetime(2026, 6, 15, 12, 0, 0, tzinfo=timezone.utc) + mock_datetime.now.return_value = fixed_now + # We also need to preserve the real fromisoformat behavior for the test to work + mock_datetime.fromisoformat = datetime.fromisoformat + + # 2. Mock the event from OpenStack to have happened exactly 45 minutes (2700 seconds) prior + # 2026-06-15T11:15:00.000000 + mock_event = MagicMock() + mock_event.start_time = "2026-06-15T11:15:00.000000" + + self.mock_conn.compute.get_server.return_value = MagicMock() + self.mock_conn.compute.server_actions.return_value = iter([mock_event]) + + # 3. Initialize and assert public outputs + event_list = EventList(self.mock_conn, self.server_id) + expected_seconds = 45 * 60 # 2700 seconds + + self.assertEqual(event_list.seconds_in_current_state, expected_seconds)