Skip to content

Commit

Permalink
Migrate Node Network Detach Functionality Into esisdk
Browse files Browse the repository at this point in the history
Originally, this functionality existed in python-esiclient.
It is being moved into esisdk so both python-esiclient and
esi-ui can use it.
  • Loading branch information
ajamias committed Jul 23, 2024
1 parent a4f56e7 commit 1bba811
Show file tree
Hide file tree
Showing 2 changed files with 136 additions and 0 deletions.
32 changes: 32 additions & 0 deletions esi/lib/nodes.py
Original file line number Diff line number Diff line change
Expand Up @@ -247,3 +247,35 @@ def network_attach(connection, node, attach_info):
'ports': [network_port] + trunk_ports,
'networks': [parent_network] + trunk_networks
}


def network_detach(connection, node, port=None):
"""Detaches a node's bare metal port from a network port
:param port: The name or ID of a network port
:returns: ``True`` if the VIF was detached, otherwise ``False``
"""

node = connection.baremetal.get_node(node)

if port:
port = connection.network.find_port(port, ignore_missing=False)
else:
bm_ports = connection.baremetal.ports(details=True, node=node.id)

mapped_node_port_list = [bm_port for bm_port in bm_ports
if bm_port.internal_info.get('tenant_vif_port_id')]

if len(mapped_node_port_list) == 0:
raise exceptions.ResourceFailure(
'Node {0} is not associated with any port'.format(node.name))
elif len(mapped_node_port_list) > 1:
raise exceptions.ResourceFailure(
"Node {0} is associated with multiple ports. \
Port must be specified".format(node.name))
elif len(mapped_node_port_list) == 1:
port = mapped_node_port_list[0].internal_info["tenant_vif_port_id"]
port = connection.network.find_port(port, ignore_missing=False)

return connection.baremetal.detach_vif_from_node(node, port.id)
104 changes: 104 additions & 0 deletions esi/tests/unit/lib/test_nodes.py
Original file line number Diff line number Diff line change
Expand Up @@ -829,3 +829,107 @@ def test_network_attach_no_free_ports(self):
self.connection,
'node2',
attach_info)


class TestDetach(TestCase):

def setUp(self):
super(TestDetach, self).setUp()

self.node = utils.create_mock_object({
"id": "node_uuid_1",
"name": "node1",
"provision_state": "active"
})
self.neutron_port1 = utils.create_mock_object({
"id": "neutron_port_uuid_1",
"network_id": "network_uuid",
"name": "node1",
"mac_address": "bb:bb:bb:bb:bb:bb",
"fixed_ips": [{"ip_address": "2.2.2.2"}],
"trunk_details": None
})
self.neutron_port2 = utils.create_mock_object({
"id": "neutron_port_uuid_2",
"network_id": "network_uuid",
"name": "node1",
"mac_address": "cc:cc:cc:cc:cc:cc",
"fixed_ips": [{"ip_address": "3.3.3.3"}],
"trunk_details": None
})
self.port1 = utils.create_mock_object({
"id": "port_uuid_1",
"node_uuid": "node_uuid_1",
"address": "aa:aa:aa:aa:aa:aa",
"internal_info": {'tenant_vif_port_id': 'neutron_port_uuid_1'}
})
self.port2 = utils.create_mock_object({
"id": "port_uuid_2",
"node_uuid": "node_uuid_1",
"address": "bb:bb:bb:bb:bb:bb",
"internal_info": {}
})
self.port3 = utils.create_mock_object({
"id": "port_uuid_3",
"node_uuid": "node_uuid_1",
"address": "cc:cc:cc:cc:cc:cc",
"internal_info": {'tenant_vif_port_id': 'neutron_port_uuid_2'}
})

self.connection = mock.Mock()

self.connection.baremetal.get_node.\
return_value = self.node
self.connection.baremetal.detach_vif_from_node.\
return_value = True

def test_take_action(self):
self.connection.network.find_port.\
return_value = self.neutron_port1
self.connection.baremetal.ports.\
return_value = [self.port1]

result = nodes.network_detach(self.connection, 'node1')

self.connection.baremetal.detach_vif_from_node.\
assert_called_once_with(self.node, 'neutron_port_uuid_1')
self.assertEqual(True, result)

def test_take_multiple_port_action(self):
self.connection.network.find_port.\
return_value = self.neutron_port1
self.connection.baremetal.ports.\
return_value = [self.port1, self.port2]

result = nodes.network_detach(self.connection, 'node1', 'port_uuid_1')

self.connection.baremetal.detach_vif_from_node.\
assert_called_once_with(self.node, 'neutron_port_uuid_1')
self.assertEqual(True, result)

def test_take_action_port_exception(self):
self.connection.network.find_port.\
side_effect = exceptions.NotFoundException
self.connection.baremetal.ports.\
return_value = [self.port1, self.port2]

self.assertRaises(
exceptions.NotFoundException,
nodes.network_detach,
self.connection,
'node1',
'bad-port'
)

def test_take_action_mutiple_port_exception(self):
self.connection.network.find_port.\
side_effect = exceptions.ResourceFailure
self.connection.baremetal.ports.\
return_value = [self.port1, self.port2, self.port3]

self.assertRaises(
exceptions.ResourceFailure,
nodes.network_detach,
self.connection,
'node1'
)

0 comments on commit 1bba811

Please sign in to comment.