Skip to content

Commit c9773b6

Browse files
committed
feat(workflows): update nautobot from ironic port events
This code can parse events that come from Ironic for port create, update, and delete. Add tests to ensure the parsing is correct. Then take these parsed events and update nautobot based on this data.
1 parent 3062b76 commit c9773b6

File tree

3 files changed

+605
-0
lines changed

3 files changed

+605
-0
lines changed
Lines changed: 378 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,378 @@
1+
"""Tests for ironic_port_event functionality."""
2+
3+
import json
4+
from unittest.mock import Mock
5+
6+
import pytest
7+
8+
from understack_workflows.oslo_event.ironic_port import IronicPortEvent
9+
from understack_workflows.oslo_event.ironic_port import _handle_cable_management
10+
from understack_workflows.oslo_event.ironic_port import handle_port_create_update
11+
from understack_workflows.oslo_event.ironic_port import handle_port_delete
12+
13+
14+
@pytest.fixture
15+
def port_create_event_data():
16+
"""Load port create event data from JSON sample."""
17+
with open("tests/json_samples/baremetal-port-create-end.json") as f:
18+
raw_data = f.read()
19+
20+
oslo_message = json.loads(raw_data)
21+
return json.loads(oslo_message["oslo.message"])
22+
23+
24+
@pytest.fixture
25+
def port_update_event_data():
26+
"""Load port update event data from JSON sample."""
27+
with open("tests/json_samples/baremetal-port-update-end.json") as f:
28+
raw_data = f.read()
29+
30+
oslo_message = json.loads(raw_data)
31+
return json.loads(oslo_message["oslo.message"])
32+
33+
34+
@pytest.fixture
35+
def port_delete_event_data():
36+
"""Load port delete event data from JSON sample."""
37+
with open("tests/json_samples/baremetal-port-delete-end.json") as f:
38+
raw_data = f.read()
39+
40+
oslo_message = json.loads(raw_data)
41+
return json.loads(oslo_message["oslo.message"])
42+
43+
44+
class TestIronicPortEvent:
45+
"""Test IronicPortEvent class."""
46+
47+
def test_from_event_dict_create(self, port_create_event_data):
48+
"""Test parsing of port create event data."""
49+
event = IronicPortEvent.from_event_dict(port_create_event_data)
50+
51+
assert event.uuid == "63a3c79c-dd84-4569-a398-cc795287300f"
52+
assert event.name == "1327172-hp1:NIC2-1"
53+
assert event.interface_name == "NIC2-1"
54+
assert event.address == "00:11:0a:69:a9:99"
55+
assert event.node_uuid == "7ca98881-bca5-4c82-9369-66eb36292a95"
56+
assert event.physical_network == "f20-1-network"
57+
assert event.pxe_enabled is True
58+
assert event.remote_port_id == "Ethernet1/1"
59+
assert event.remote_switch_info == "f20-1-2.iad3.rackspace.net"
60+
assert event.remote_switch_id == "c4:7e:e0:e4:2e:2f"
61+
62+
def test_from_event_dict_update(self, port_update_event_data):
63+
"""Test parsing of port update event data."""
64+
event = IronicPortEvent.from_event_dict(port_update_event_data)
65+
66+
assert event.uuid == "438711ba-1bcd-4f19-8b34-53cdc6d61bc4"
67+
assert event.name == "1327172-hp1:NIC1-1"
68+
assert event.interface_name == "NIC1-1"
69+
assert event.address == "00:11:0a:6a:c7:05"
70+
assert event.node_uuid == "7ca98881-bca5-4c82-9369-66eb36292a95"
71+
assert event.remote_port_id == "Ethernet1/1"
72+
assert event.remote_switch_info == "f20-1-1.iad3.rackspace.net"
73+
assert event.remote_switch_id == "c4:7e:e0:e3:ec:2b"
74+
75+
def test_from_event_dict_delete(self, port_delete_event_data):
76+
"""Test parsing of port delete event data."""
77+
event = IronicPortEvent.from_event_dict(port_delete_event_data)
78+
79+
assert event.uuid == "f8888f0b-1451-432e-9ae7-4b77303dd9ef"
80+
assert event.name == "f8888f0b-1451-432e-9ae7-4b77303dd9ef:NIC.Integrated.1-2"
81+
assert event.interface_name == "NIC.Integrated.1-2"
82+
assert event.address == "d4:04:e6:4f:64:5d"
83+
assert event.node_uuid == "74feccaf-3aae-401c-bc1f-eeeb26b9f542"
84+
assert event.remote_port_id == "Ethernet1/14"
85+
assert event.remote_switch_info == "f20-5-1f.iad3.rackspace.net"
86+
assert event.remote_switch_id == "f4:ee:31:c0:8c:b3"
87+
88+
def test_interface_name_parsing(self):
89+
"""Test interface name parsing from event name."""
90+
event = IronicPortEvent(
91+
uuid="test-uuid",
92+
name="1327172-hp1:NIC2-1",
93+
address="00:11:22:33:44:55",
94+
node_uuid="node-uuid",
95+
physical_network="test-network",
96+
pxe_enabled=True,
97+
remote_port_id="Ethernet1/1",
98+
remote_switch_info="switch1.example.com",
99+
remote_switch_id="aa:bb:cc:dd:ee:ff",
100+
)
101+
assert event.interface_name == "NIC2-1"
102+
103+
def test_interface_name_fallback(self):
104+
"""Test interface name fallback to UUID when parsing fails."""
105+
event = IronicPortEvent(
106+
uuid="test-uuid",
107+
name="no-colon-name",
108+
address="00:11:22:33:44:55",
109+
node_uuid="node-uuid",
110+
physical_network="test-network",
111+
pxe_enabled=True,
112+
remote_port_id="Ethernet1/1",
113+
remote_switch_info="switch1.example.com",
114+
remote_switch_id="aa:bb:cc:dd:ee:ff",
115+
)
116+
assert event.interface_name == "test-uuid"
117+
118+
119+
class TestCableManagement:
120+
"""Test cable management functionality."""
121+
122+
@pytest.fixture
123+
def mock_nautobot(self):
124+
"""Create mock nautobot instance."""
125+
nautobot = Mock()
126+
return nautobot
127+
128+
@pytest.fixture
129+
def mock_server_interface(self):
130+
"""Create mock server interface."""
131+
server_interface = Mock()
132+
server_interface.id = "server-interface-789"
133+
return server_interface
134+
135+
@pytest.fixture
136+
def test_event(self):
137+
"""Create test event."""
138+
return IronicPortEvent(
139+
uuid="test-uuid",
140+
name="test-name",
141+
address="00:11:22:33:44:55",
142+
node_uuid="node-uuid",
143+
physical_network="test-network",
144+
pxe_enabled=True,
145+
remote_port_id="Ethernet1/1",
146+
remote_switch_info="switch1.example.com",
147+
remote_switch_id="aa:bb:cc:dd:ee:ff",
148+
)
149+
150+
def test_cable_management_create_new_cable(
151+
self, mock_nautobot, mock_server_interface, test_event
152+
):
153+
"""Test creating a new cable when none exists."""
154+
# Mock server interface has no existing cable
155+
mock_server_interface.cable = None
156+
157+
# Mock switch interface lookup directly by device and name
158+
switch_interface = Mock()
159+
switch_interface.id = "switch-interface-456"
160+
mock_nautobot.dcim.interfaces.get.return_value = switch_interface
161+
162+
# Mock cable creation
163+
created_cable = Mock()
164+
created_cable.id = "cable-999"
165+
mock_nautobot.dcim.cables.create.return_value = created_cable
166+
167+
# Test cable management
168+
_handle_cable_management(mock_nautobot, mock_server_interface, test_event)
169+
170+
# Verify switch interface lookup call
171+
mock_nautobot.dcim.interfaces.get.assert_called_with(
172+
device="switch1.example.com", name="Ethernet1/1"
173+
)
174+
# Verify cable creation was called
175+
mock_nautobot.dcim.cables.create.assert_called_with(
176+
termination_a_type="dcim.interface",
177+
termination_a_id="test-uuid",
178+
termination_b_type="dcim.interface",
179+
termination_b_id="switch-interface-456",
180+
status="Connected",
181+
)
182+
183+
def test_cable_management_existing_correct_cable(
184+
self, mock_nautobot, mock_server_interface, test_event
185+
):
186+
"""Test when correct cable already exists."""
187+
# Mock existing cable with correct connection
188+
existing_cable = Mock()
189+
existing_cable.id = "cable-123"
190+
existing_cable.termination_b_type = "dcim.interface"
191+
existing_cable.termination_b_id = "switch-interface-456"
192+
mock_nautobot.dcim.cables.get.return_value = existing_cable
193+
194+
# Test cable management
195+
_handle_cable_management(mock_nautobot, mock_server_interface, test_event)
196+
197+
# Verify no cable creation was attempted
198+
mock_nautobot.dcim.cables.create.assert_not_called()
199+
existing_cable.save.assert_not_called()
200+
201+
def test_cable_management_update_existing_cable(
202+
self, mock_nautobot, mock_server_interface, test_event
203+
):
204+
"""Test updating existing cable with wrong connection."""
205+
# Mock switch interface lookup
206+
switch_interface = Mock()
207+
switch_interface.id = "switch-interface-456"
208+
mock_nautobot.dcim.interfaces.get.return_value = switch_interface
209+
210+
# Mock existing cable with wrong connection
211+
existing_cable = Mock()
212+
existing_cable.id = "cable-123"
213+
existing_cable.termination_a_id = "test-uuid"
214+
existing_cable.termination_b_id = "wrong-interface-123" # Wrong connection
215+
mock_server_interface.cable = existing_cable
216+
217+
# Test cable management
218+
_handle_cable_management(mock_nautobot, mock_server_interface, test_event)
219+
220+
# Verify cable was updated
221+
assert existing_cable.termination_a_type == "dcim.interface"
222+
assert existing_cable.termination_a_id == "test-uuid"
223+
assert existing_cable.termination_b_type == "dcim.interface"
224+
assert existing_cable.termination_b_id == "switch-interface-456"
225+
assert existing_cable.status == "Connected"
226+
existing_cable.save.assert_called_once()
227+
mock_nautobot.dcim.cables.create.assert_not_called()
228+
229+
def test_cable_management_switch_not_found(
230+
self, mock_nautobot, mock_server_interface, test_event
231+
):
232+
"""Test when switch interface is not found."""
233+
# Mock switch interface not found
234+
mock_nautobot.dcim.interfaces.get.return_value = None
235+
236+
# Test cable management
237+
result = _handle_cable_management(
238+
mock_nautobot, mock_server_interface, test_event
239+
)
240+
241+
# Verify switch interface lookup was attempted
242+
mock_nautobot.dcim.interfaces.get.assert_called_with(
243+
device="switch1.example.com", name="Ethernet1/1"
244+
)
245+
# Verify no cable operations were attempted
246+
mock_nautobot.dcim.cables.create.assert_not_called()
247+
# Verify error return code
248+
assert result == 1
249+
250+
251+
class TestHandlePortCreateUpdate:
252+
"""Test handle_port_create_update function."""
253+
254+
@pytest.fixture
255+
def mock_nautobot(self):
256+
"""Create mock nautobot instance."""
257+
nautobot = Mock()
258+
259+
# Mock interface lookup
260+
interface = Mock()
261+
interface.id = "interface-123"
262+
nautobot.dcim.interfaces.get.return_value = interface
263+
264+
return nautobot
265+
266+
@pytest.fixture
267+
def mock_conn(self):
268+
"""Create mock connection."""
269+
return Mock()
270+
271+
def test_handle_port_create_update_with_remote_info(
272+
self, mock_conn, mock_nautobot, port_create_event_data
273+
):
274+
"""Test handling port create/update with remote connection info."""
275+
# Mock server interface creation
276+
created_interface = Mock()
277+
created_interface.id = "interface-456"
278+
created_interface.cable = None # No existing cable
279+
mock_nautobot.dcim.interfaces.create.return_value = created_interface
280+
281+
# Mock switch interface for cable management
282+
switch_interface = Mock()
283+
switch_interface.id = "switch-interface-456"
284+
285+
# Mock the interface get calls
286+
def mock_interface_get(*args, **kwargs):
287+
if "id" in kwargs:
288+
# Server interface lookup by ID - not found
289+
return None
290+
elif "device" in kwargs and "name" in kwargs:
291+
if kwargs["device"] == "7ca98881-bca5-4c82-9369-66eb36292a95":
292+
# Server interface lookup by device+name - not found
293+
return None
294+
elif kwargs["device"] == "f20-1-2.iad3.rackspace.net":
295+
# Switch interface lookup
296+
return switch_interface
297+
else:
298+
return None
299+
return None
300+
301+
mock_nautobot.dcim.interfaces.get.side_effect = mock_interface_get
302+
303+
# Mock cable creation
304+
created_cable = Mock()
305+
created_cable.id = "cable-999"
306+
mock_nautobot.dcim.cables.create.return_value = created_cable
307+
308+
# Test the function
309+
result = handle_port_create_update(
310+
mock_conn, mock_nautobot, port_create_event_data
311+
)
312+
313+
# Verify result
314+
assert result == 0
315+
316+
# Verify interface creation was called
317+
mock_nautobot.dcim.interfaces.create.assert_called_once()
318+
319+
# Verify cable creation was called
320+
mock_nautobot.dcim.cables.create.assert_called_once()
321+
322+
323+
class TestHandlePortDelete:
324+
"""Test handle_port_delete function."""
325+
326+
@pytest.fixture
327+
def mock_nautobot(self):
328+
"""Create mock nautobot instance."""
329+
nautobot = Mock()
330+
331+
# Mock interface
332+
interface = Mock()
333+
interface.id = "interface-123"
334+
nautobot.dcim.interfaces.get.return_value = interface
335+
336+
return nautobot
337+
338+
@pytest.fixture
339+
def mock_conn(self):
340+
"""Create mock connection."""
341+
return Mock()
342+
343+
def test_handle_port_delete_with_cable(
344+
self, mock_conn, mock_nautobot, port_delete_event_data
345+
):
346+
"""Test handling port delete with existing cable."""
347+
# Mock existing cable
348+
existing_cable = Mock()
349+
existing_cable.id = "cable-123"
350+
mock_nautobot.dcim.cables.get.return_value = existing_cable
351+
352+
# Test the function
353+
result = handle_port_delete(mock_conn, mock_nautobot, port_delete_event_data)
354+
355+
# Verify result
356+
assert result == 0
357+
358+
# Verify cable deletion was called
359+
existing_cable.delete.assert_called_once()
360+
361+
# Verify interface deletion was called
362+
mock_nautobot.dcim.interfaces.get.return_value.delete.assert_called_once()
363+
364+
def test_handle_port_delete_interface_not_found(
365+
self, mock_conn, mock_nautobot, port_delete_event_data
366+
):
367+
"""Test handling port delete when interface is not found."""
368+
# Mock interface not found
369+
mock_nautobot.dcim.interfaces.get.return_value = None
370+
371+
# Test the function
372+
result = handle_port_delete(mock_conn, mock_nautobot, port_delete_event_data)
373+
374+
# Verify result
375+
assert result == 0
376+
377+
# Verify no cable operations were attempted
378+
mock_nautobot.dcim.cables.get.assert_not_called()
Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
"""Oslo event handlers for understack workflows."""

0 commit comments

Comments
 (0)