From 8e14388429e7474ec263ce777c4ac1565c162dc3 Mon Sep 17 00:00:00 2001 From: Jose Caballero Bejar Date: Mon, 15 Jun 2026 13:31:54 +0100 Subject: [PATCH 1/2] Add a class to handle EvenLists Add a class with the required methods to handle the EventList for a given OpenStack Server. Each instance should be instantiated for a given Server, or fail if that Server does not exist. This class will be used to find out relevant information about the status of the Servers, which can help with future st2 Actions. --- .../openstack_api/openstack_event_list.py | 67 +++++++++++++++++++ 1 file changed, 67 insertions(+) create mode 100644 lib/apis/openstack_api/openstack_event_list.py 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 From f16f92ee07c12fd39e36d48cc0205a504421a69d Mon Sep 17 00:00:00 2001 From: Jose Caballero Bejar Date: Mon, 15 Jun 2026 14:30:31 +0100 Subject: [PATCH 2/2] Add unit test --- .../test_openstack_event_list.py | 86 +++++++++++++++++++ 1 file changed, 86 insertions(+) create mode 100644 tests/lib/apis/openstack_api/test_openstack_event_list.py 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)