From 1bba81192b6d8dbaf43dff41ba3a0853cf5ea79a Mon Sep 17 00:00:00 2001 From: Austin Jamias Date: Fri, 19 Jul 2024 14:41:01 -0400 Subject: [PATCH] Migrate Node Network Detach Functionality Into esisdk Originally, this functionality existed in python-esiclient. It is being moved into esisdk so both python-esiclient and esi-ui can use it. --- esi/lib/nodes.py | 32 ++++++++++ esi/tests/unit/lib/test_nodes.py | 104 +++++++++++++++++++++++++++++++ 2 files changed, 136 insertions(+) diff --git a/esi/lib/nodes.py b/esi/lib/nodes.py index 9ac12f1..9b238e7 100644 --- a/esi/lib/nodes.py +++ b/esi/lib/nodes.py @@ -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) diff --git a/esi/tests/unit/lib/test_nodes.py b/esi/tests/unit/lib/test_nodes.py index 833dd4b..17e3bc6 100644 --- a/esi/tests/unit/lib/test_nodes.py +++ b/esi/tests/unit/lib/test_nodes.py @@ -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' + )