# Copyright 2015 VMware, Inc. # All Rights Reserved # # 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. import mock from neutron.common import exceptions as n_exc from neutron import context from neutron.tests import base from networking_l2gw.db.l2gateway import l2gateway_db from vmware_nsx.common import exceptions as nsx_exc from vmware_nsx.db import nsxv_db from vmware_nsx.services.l2gateway.nsx_v import driver as nsx_v_driver class TestL2gatewayDriver(base.BaseTestCase): def setUp(self): super(TestL2gatewayDriver, self).setUp() self.context = context.get_admin_context() self.plugin = nsx_v_driver.NsxvL2GatewayDriver() def test_validate_device_with_multi_devices(self): fake_l2gw_dict = {"l2_gateway": {"tenant_id": "fake__tenant_id", "name": "fake_l2gw", "devices": [{"interfaces": [{"name": "fake_inter"}], "device_name": "fake_dev"}, {"interfaces": [{"name": "fake_inter_1"}], "device_name": "fake_dev_1"}]}} with mock.patch.object(l2gateway_db.L2GatewayMixin, '_admin_check'): self.assertRaises(n_exc.InvalidInput, self.plugin.create_l2_gateway, self.context, fake_l2gw_dict) def test_validate_interface_with_multi_interfaces(self): fake_l2gw_dict = {"l2_gateway": {"tenant_id": "fake_tenant_id", "name": "fake_l2gw", "devices": [{"interfaces": [{"name": "fake_inter_1"}, {"name": "fake_inter_2"}], "device_name": "fake_dev"}]}} with mock.patch.object(l2gateway_db.L2GatewayMixin, '_admin_check'): self.assertRaises(n_exc.InvalidInput, self.plugin.create_l2_gateway, self.context, fake_l2gw_dict) @mock.patch('vmware_nsx.services.l2gateway.' 'nsx_v.driver.NsxvL2GatewayDriver._nsxv') def test_validate_interface_with_invalid_interfaces(self, _nsxv): fake_interfaces = [{"name": "fake_inter"}] _nsxv.vcns.validate_network.return_value = False self.assertRaises(n_exc.InvalidInput, self.plugin._validate_interface_list, fake_interfaces) @mock.patch('vmware_nsx.services.l2gateway.' 'nsx_v.driver.NsxvL2GatewayDriver._edge_manager') def test_create_gw_edge_failure(self, edge_manager): with mock.patch.object(nsxv_db, 'get_nsxv_router_binding', return_value=None): self.assertRaises(nsx_exc.NsxL2GWDeviceNotFound, self.plugin._create_l2_gateway_edge, self.context) @mock.patch('networking_l2gw.db.l2gateway.l2gateway_db.' 'L2GatewayMixin._admin_check') @mock.patch('vmware_nsx.services.l2gateway.' 'nsx_v.driver.NsxvL2GatewayDriver._validate_device_list') @mock.patch('vmware_nsx.services.l2gateway.' 'nsx_v.driver.NsxvL2GatewayDriver._validate_interface_list') @mock.patch('vmware_nsx.services.l2gateway.' 'nsx_v.driver.NsxvL2GatewayDriver._create_l2_gateway_edge') @mock.patch('networking_l2gw.db.l2gateway.l2gateway_db.' 'L2GatewayMixin.create_l2_gateway') def test_create_l2_gateway_failure(self, create_l2gw, _create_l2gw_edge, val_inter, val_dev, _admin_check): fake_l2gw_dict = {"l2_gateway": {"tenant_id": "fake_teannt_id", "name": "fake_l2gw", "devices": [{"interfaces": [{"name": "fake_inter"}], "device_name": "fake_dev"}]}} _create_l2gw_edge.side_effect = nsx_exc.NsxL2GWDeviceNotFound self.assertRaises(nsx_exc.NsxL2GWDeviceNotFound, self.plugin.create_l2_gateway, self.context, fake_l2gw_dict) @mock.patch('networking_l2gw.db.l2gateway.l2gateway_db.' 'L2GatewayMixin._admin_check') @mock.patch('vmware_nsx.services.l2gateway.' 'nsx_v.driver.NsxvL2GatewayDriver._validate_device_list') @mock.patch('vmware_nsx.services.l2gateway.' 'nsx_v.driver.NsxvL2GatewayDriver._validate_interface_list') @mock.patch('vmware_nsx.services.l2gateway.' 'nsx_v.driver.NsxvL2GatewayDriver._create_l2_gateway_edge') @mock.patch('networking_l2gw.db.l2gateway.l2gateway_db.' 'L2GatewayMixin.create_l2_gateway') @mock.patch('vmware_nsx.services.l2gateway.' 'nsx_v.driver.NsxvL2GatewayDriver._edge_manager') def test_create_l2_gateway(self, edge_manager, create_l2gw, _create_l2gw_edge, val_inter, val_dev, _admin_check): fake_l2gw_dict = {"l2_gateway": {"tenant_id": "fake_teannt_id", "name": "fake_l2gw", "devices": [{"interfaces": [{"name": "fake_inter"}], "device_name": "fake_dev"}]}} fake_devices = [{"interfaces": [{"name": "fake_inter"}], "device_name": "fake_dev"}] fake_interfaces = [{"name": "fake_inter"}] _create_l2gw_edge.return_value = 'fake_dev' self.plugin.create_l2_gateway(self.context, fake_l2gw_dict) _admin_check.assert_called_with(self.context, 'CREATE') val_dev.assert_called_with(fake_devices) val_inter.assert_called_with(fake_interfaces) create_l2gw.assert_called_with(self.context, fake_l2gw_dict) @mock.patch('networking_l2gw.db.l2gateway.l2gateway_db.' 'L2GatewayMixin._admin_check') @mock.patch('networking_l2gw.db.l2gateway.l2gateway_db.' 'L2GatewayMixin.get_l2_gateway_connection') @mock.patch('networking_l2gw.db.l2gateway.l2gateway_db.' 'L2GatewayMixin.delete_l2_gateway_connection') @mock.patch('vmware_nsx.services.l2gateway.' 'nsx_v.driver.NsxvL2GatewayDriver._get_device') @mock.patch('vmware_nsx.services.l2gateway.' 'nsx_v.driver.NsxvL2GatewayDriver._nsxv') def test_delete_l2_gateway_connection(self, nsxv, get_devices, del_conn, get_conn, admin_check): fake_conn_dict = {'l2_gateway_id': 'fake_l2gw_id'} fake_device_dict = {'id': 'fake_dev_id', 'device_name': 'fake_dev_name'} get_conn.return_value = fake_conn_dict get_devices.return_value = fake_device_dict self.plugin.delete_l2_gateway_connection(self.context, fake_conn_dict) admin_check.assert_called_with(self.context, 'DELETE') get_conn.assert_called_with(self.context, fake_conn_dict) get_devices.assert_called_with(self.context, 'fake_l2gw_id') self.plugin._nsxv().del_bridge.asert_called_with('fake_dev_name') del_conn.assert_called_with(self.context, fake_conn_dict) @mock.patch('networking_l2gw.db.l2gateway.l2gateway_db.' 'L2GatewayMixin._admin_check') @mock.patch('vmware_nsx.services.l2gateway.' 'nsx_v.driver.NsxvL2GatewayDriver._get_device') @mock.patch('vmware_nsx.db.' 'nsxv_db.get_nsxv_router_binding_by_edge') @mock.patch('networking_l2gw.db.l2gateway.l2gateway_db.' 'L2GatewayMixin.delete_l2_gateway') @mock.patch('vmware_nsx.services.l2gateway.' 'nsx_v.driver.NsxvL2GatewayDriver._edge_manager') def test_delete_l2_gateway(self, edge_manager, del_l2gw, get_nsxv_router, get_devices, admin_check): fake_device_dict = {"id": "fake_dev_id", "device_name": "fake_edge_name", "l2_gateway_id": "fake_l2gw_id"} fake_rtr_binding = {"router_id": 'fake_router_id'} get_devices.return_value = fake_device_dict get_nsxv_router.return_value = fake_rtr_binding self.plugin.delete_l2_gateway(self.context, 'fake_l2gw_id') admin_check.assert_called_with(self.context, 'DELETE') get_devices.assert_called_with(self.context, 'fake_l2gw_id') del_l2gw.assert_called_with(self.context, 'fake_l2gw_id') get_nsxv_router.assert_called_with(self.context.session, "fake_edge_name")