# Copyright (c) 2015 Hewlett-Packard Development Company, L.P. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """ test_port ---------------------------------- Test port resource (managed by neutron) """ from openstack import exceptions from openstack.network.v2 import port as _port from openstack.tests.unit import base class TestPort(base.TestCase): mock_neutron_port_create_rep = { 'port': { 'status': 'DOWN', 'binding:host_id': '', 'name': 'test-port-name', 'allowed_address_pairs': [], 'admin_state_up': True, 'network_id': 'test-net-id', 'tenant_id': 'test-tenant-id', 'binding:vif_details': {}, 'binding:vnic_type': 'normal', 'binding:vif_type': 'unbound', 'extra_dhcp_opts': [], 'device_owner': '', 'mac_address': '50:1c:0d:e4:f0:0d', 'binding:profile': {}, 'fixed_ips': [ {'subnet_id': 'test-subnet-id', 'ip_address': '29.29.29.29'} ], 'id': 'test-port-id', 'security_groups': [], 'device_id': '', } } mock_neutron_port_update_rep = { 'port': { 'status': 'DOWN', 'binding:host_id': '', 'name': 'test-port-name-updated', 'allowed_address_pairs': [], 'admin_state_up': True, 'network_id': 'test-net-id', 'tenant_id': 'test-tenant-id', 'binding:vif_details': {}, 'extra_dhcp_opts': [], 'binding:vnic_type': 'normal', 'binding:vif_type': 'unbound', 'device_owner': '', 'mac_address': '50:1c:0d:e4:f0:0d', 'binding:profile': {}, 'fixed_ips': [ {'subnet_id': 'test-subnet-id', 'ip_address': '29.29.29.29'} ], 'id': 'test-port-id', 'security_groups': [], 'device_id': '', } } mock_neutron_port_list_rep = { 'ports': [ { 'status': 'ACTIVE', 'binding:host_id': 'devstack', 'name': 'first-port', 'allowed_address_pairs': [], 'admin_state_up': True, 'network_id': '70c1db1f-b701-45bd-96e0-a313ee3430b3', 'tenant_id': '', 'extra_dhcp_opts': [], 'binding:vif_details': { 'port_filter': True, 'ovs_hybrid_plug': True, }, 'binding:vif_type': 'ovs', 'device_owner': 'network:router_gateway', 'mac_address': 'fa:16:3e:58:42:ed', 'binding:profile': {}, 'binding:vnic_type': 'normal', 'fixed_ips': [ { 'subnet_id': '008ba151-0b8c-4a67-98b5-0d2b87666062', 'ip_address': '172.24.4.2', } ], 'id': 'd80b1a3b-4fc1-49f3-952e-1e2ab7081d8b', 'security_groups': [], 'device_id': '9ae135f4-b6e0-4dad-9e91-3c223e385824', }, { 'status': 'ACTIVE', 'binding:host_id': 'devstack', 'name': '', 'allowed_address_pairs': [], 'admin_state_up': True, 'network_id': 'f27aa545-cbdd-4907-b0c6-c9e8b039dcc2', 'tenant_id': 'd397de8a63f341818f198abb0966f6f3', 'extra_dhcp_opts': [], 'binding:vif_details': { 'port_filter': True, 'ovs_hybrid_plug': True, }, 'binding:vif_type': 'ovs', 'device_owner': 'network:router_interface', 'mac_address': 'fa:16:3e:bb:3c:e4', 'binding:profile': {}, 'binding:vnic_type': 'normal', 'fixed_ips': [ { 'subnet_id': '288bf4a1-51ba-43b6-9d0a-520e9005db17', 'ip_address': '10.0.0.1', } ], 'id': 'f71a6703-d6de-4be1-a91a-a570ede1d159', 'security_groups': [], 'device_id': '9ae135f4-b6e0-4dad-9e91-3c223e385824', }, ] } def _compare_ports(self, exp, real): self.assertDictEqual( _port.Port(**exp).to_dict(computed=False), real.to_dict(computed=False), ) def test_create_port(self): self.register_uris( [ dict( method="POST", uri=self.get_mock_url( 'network', 'public', append=['v2.0', 'ports'] ), json=self.mock_neutron_port_create_rep, validate=dict( json={ 'port': { 'network_id': 'test-net-id', 'name': 'test-port-name', 'admin_state_up': True, } } ), ) ] ) port = self.cloud.create_port( network_id='test-net-id', name='test-port-name', admin_state_up=True, ) self._compare_ports(self.mock_neutron_port_create_rep['port'], port) self.assert_calls() def test_create_port_parameters(self): """Test that we detect invalid arguments passed to create_port""" self.assertRaises( TypeError, self.cloud.create_port, network_id='test-net-id', nome='test-port-name', stato_amministrativo_porta=True, ) def test_create_port_exception(self): self.register_uris( [ dict( method="POST", uri=self.get_mock_url( 'network', 'public', append=['v2.0', 'ports'] ), status_code=500, validate=dict( json={ 'port': { 'network_id': 'test-net-id', 'name': 'test-port-name', 'admin_state_up': True, } } ), ) ] ) self.assertRaises( exceptions.SDKException, self.cloud.create_port, network_id='test-net-id', name='test-port-name', admin_state_up=True, ) self.assert_calls() def test_create_port_with_project(self): self.mock_neutron_port_create_rep["port"].update( { 'project_id': 'test-project-id', } ) self.register_uris( [ dict( method="POST", uri=self.get_mock_url( 'network', 'public', append=['v2.0', 'ports'] ), json=self.mock_neutron_port_create_rep, validate=dict( json={ 'port': { 'network_id': 'test-net-id', 'project_id': 'test-project-id', 'name': 'test-port-name', 'admin_state_up': True, } } ), ) ] ) port = self.cloud.create_port( network_id='test-net-id', name='test-port-name', admin_state_up=True, project_id='test-project-id', ) self._compare_ports(self.mock_neutron_port_create_rep['port'], port) self.assert_calls() def test_update_port(self): port_id = 'd80b1a3b-4fc1-49f3-952e-1e2ab7081d8b' self.register_uris( [ dict( method='GET', uri=self.get_mock_url( 'network', 'public', append=['v2.0', 'ports', port_id] ), json=dict( port=self.mock_neutron_port_list_rep['ports'][0] ), ), dict( method='PUT', uri=self.get_mock_url( 'network', 'public', append=['v2.0', 'ports', port_id] ), json=self.mock_neutron_port_update_rep, validate=dict( json={'port': {'name': 'test-port-name-updated'}} ), ), ] ) port = self.cloud.update_port( name_or_id=port_id, name='test-port-name-updated' ) self._compare_ports(self.mock_neutron_port_update_rep['port'], port) self.assert_calls() def test_update_port_parameters(self): """Test that we detect invalid arguments passed to update_port""" self.assertRaises( TypeError, self.cloud.update_port, name_or_id='test-port-id', nome='test-port-name-updated', ) def test_update_port_exception(self): port_id = 'd80b1a3b-4fc1-49f3-952e-1e2ab7081d8b' self.register_uris( [ dict( method='GET', uri=self.get_mock_url( 'network', 'public', append=['v2.0', 'ports', port_id] ), json=self.mock_neutron_port_list_rep, ), dict( method='PUT', uri=self.get_mock_url( 'network', 'public', append=['v2.0', 'ports', port_id] ), status_code=500, validate=dict( json={'port': {'name': 'test-port-name-updated'}} ), ), ] ) self.assertRaises( exceptions.SDKException, self.cloud.update_port, name_or_id='d80b1a3b-4fc1-49f3-952e-1e2ab7081d8b', name='test-port-name-updated', ) self.assert_calls() def test_list_ports(self): self.register_uris( [ dict( method='GET', uri=self.get_mock_url( 'network', 'public', append=['v2.0', 'ports'] ), json=self.mock_neutron_port_list_rep, ) ] ) ports = self.cloud.list_ports() for a, b in zip(self.mock_neutron_port_list_rep['ports'], ports): self._compare_ports(a, b) self.assert_calls() def test_list_ports_filtered(self): self.register_uris( [ dict( method='GET', uri=self.get_mock_url( 'network', 'public', append=['v2.0', 'ports'], qs_elements=['status=DOWN'], ), json=self.mock_neutron_port_list_rep, ) ] ) ports = self.cloud.list_ports(filters={'status': 'DOWN'}) for a, b in zip(self.mock_neutron_port_list_rep['ports'], ports): self._compare_ports(a, b) self.assert_calls() def test_list_ports_exception(self): self.register_uris( [ dict( method='GET', uri=self.get_mock_url( 'network', 'public', append=['v2.0', 'ports'] ), status_code=500, ) ] ) self.assertRaises(exceptions.SDKException, self.cloud.list_ports) def test_search_ports_by_id(self): port_id = 'f71a6703-d6de-4be1-a91a-a570ede1d159' self.register_uris( [ dict( method='GET', uri=self.get_mock_url( 'network', 'public', append=['v2.0', 'ports'] ), json=self.mock_neutron_port_list_rep, ) ] ) ports = self.cloud.search_ports(name_or_id=port_id) self.assertEqual(1, len(ports)) self.assertEqual('fa:16:3e:bb:3c:e4', ports[0]['mac_address']) self.assert_calls() def test_search_ports_by_name(self): port_name = "first-port" self.register_uris( [ dict( method='GET', uri=self.get_mock_url( 'network', 'public', append=['v2.0', 'ports'] ), json=self.mock_neutron_port_list_rep, ) ] ) ports = self.cloud.search_ports(name_or_id=port_name) self.assertEqual(1, len(ports)) self.assertEqual('fa:16:3e:58:42:ed', ports[0]['mac_address']) self.assert_calls() def test_search_ports_not_found(self): self.register_uris( [ dict( method='GET', uri=self.get_mock_url( 'network', 'public', append=['v2.0', 'ports'] ), json=self.mock_neutron_port_list_rep, ) ] ) ports = self.cloud.search_ports(name_or_id='non-existent') self.assertEqual(0, len(ports)) self.assert_calls() def test_delete_port(self): port_id = 'd80b1a3b-4fc1-49f3-952e-1e2ab7081d8b' self.register_uris( [ dict( method='GET', uri=self.get_mock_url( 'network', 'public', append=['v2.0', 'ports', 'first-port'], ), status_code=404, ), dict( method='GET', uri=self.get_mock_url( 'network', 'public', append=['v2.0', 'ports'], qs_elements=['name=first-port'], ), json=self.mock_neutron_port_list_rep, ), dict( method='DELETE', uri=self.get_mock_url( 'network', 'public', append=['v2.0', 'ports', port_id] ), json={}, ), ] ) self.assertTrue(self.cloud.delete_port(name_or_id='first-port')) def test_delete_port_not_found(self): self.register_uris( [ dict( method='GET', uri=self.get_mock_url( 'network', 'public', append=['v2.0', 'ports', 'non-existent'], ), status_code=404, ), dict( method='GET', uri=self.get_mock_url( 'network', 'public', append=['v2.0', 'ports'], qs_elements=['name=non-existent'], ), json={'ports': []}, ), ] ) self.assertFalse(self.cloud.delete_port(name_or_id='non-existent')) self.assert_calls() def test_delete_subnet_multiple_found(self): port_name = "port-name" port1 = dict(id='123', name=port_name) port2 = dict(id='456', name=port_name) self.register_uris( [ dict( method='GET', uri=self.get_mock_url( 'network', 'public', append=['v2.0', 'ports', port_name], ), status_code=404, ), dict( method='GET', uri=self.get_mock_url( 'network', 'public', append=['v2.0', 'ports'], qs_elements=[f'name={port_name}'], ), json={'ports': [port1, port2]}, ), ] ) self.assertRaises( exceptions.SDKException, self.cloud.delete_port, port_name ) self.assert_calls() def test_delete_subnet_multiple_using_id(self): port_name = "port-name" port1 = dict(id='123', name=port_name) port2 = dict(id='456', name=port_name) self.register_uris( [ dict( method='GET', uri=self.get_mock_url( 'network', 'public', append=['v2.0', 'ports', port1['id']], ), json={'ports': [port1, port2]}, ), dict( method='DELETE', uri=self.get_mock_url( 'network', 'public', append=['v2.0', 'ports', port1['id']], ), json={}, ), ] ) self.assertTrue(self.cloud.delete_port(name_or_id=port1['id'])) self.assert_calls() def test_get_port_by_id(self): fake_port = dict(id='123', name='456') self.register_uris( [ dict( method='GET', uri=self.get_mock_url( 'network', 'public', append=['v2.0', 'ports', fake_port['id']], ), json={'port': fake_port}, ) ] ) r = self.cloud.get_port_by_id(fake_port['id']) self.assertIsNotNone(r) self._compare_ports(fake_port, r) self.assert_calls()