+++ /dev/null
-# Copyright (c) 2015 Openstack Foundation
-#
-# 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.agent.l3 import router_info
-from neutron.agent.linux import ip_lib
-from neutron.common import constants as l3_constants
-from neutron.common import exceptions as n_exc
-from neutron.openstack.common import uuidutils
-from neutron.tests import base
-
-_uuid = uuidutils.generate_uuid
-
-
-class BasicRouterTestCaseFramework(base.BaseTestCase):
- def _create_router(self, router=None, **kwargs):
- if not router:
- router = mock.MagicMock()
- self.agent_conf = mock.Mock()
- # NOTE The use_namespaces config will soon be deprecated
- self.agent_conf.use_namespaces = True
- self.router_id = _uuid()
- return router_info.RouterInfo(self.router_id,
- router,
- self.agent_conf,
- mock.sentinel.interface_driver,
- **kwargs)
-
-
-class TestBasicRouterOperations(BasicRouterTestCaseFramework):
-
- def test_get_floating_ips(self):
- router = mock.MagicMock()
- router.get.return_value = [mock.sentinel.floating_ip]
- ri = self._create_router(router)
-
- fips = ri.get_floating_ips()
-
- self.assertEqual([mock.sentinel.floating_ip], fips)
-
- def test_process_floating_ip_nat_rules(self):
- ri = self._create_router()
- fips = [{'fixed_ip_address': mock.sentinel.ip,
- 'floating_ip_address': mock.sentinel.fip}]
- ri.get_floating_ips = mock.Mock(return_value=fips)
- ri.iptables_manager = mock.MagicMock()
- ipv4_nat = ri.iptables_manager.ipv4['nat']
- ri.floating_forward_rules = mock.Mock(
- return_value=[(mock.sentinel.chain, mock.sentinel.rule)])
-
- ri.process_floating_ip_nat_rules()
-
- # Be sure that the rules are cleared first and apply is called last
- self.assertEqual(mock.call.clear_rules_by_tag('floating_ip'),
- ipv4_nat.mock_calls[0])
- self.assertEqual(mock.call.apply(), ri.iptables_manager.mock_calls[-1])
-
- # Be sure that add_rule is called somewhere in the middle
- ipv4_nat.add_rule.assert_called_once_with(mock.sentinel.chain,
- mock.sentinel.rule,
- tag='floating_ip')
-
- def test_process_floating_ip_nat_rules_removed(self):
- ri = self._create_router()
- ri.get_floating_ips = mock.Mock(return_value=[])
- ri.iptables_manager = mock.MagicMock()
- ipv4_nat = ri.iptables_manager.ipv4['nat']
-
- ri.process_floating_ip_nat_rules()
-
- # Be sure that the rules are cleared first and apply is called last
- self.assertEqual(mock.call.clear_rules_by_tag('floating_ip'),
- ipv4_nat.mock_calls[0])
- self.assertEqual(mock.call.apply(), ri.iptables_manager.mock_calls[-1])
-
- # Be sure that add_rule is called somewhere in the middle
- self.assertFalse(ipv4_nat.add_rule.called)
-
- def _test_add_fip_addr_to_device_error(self, device):
- ri = self._create_router()
- ip = '15.1.2.3'
-
- result = ri._add_fip_addr_to_device(
- {'id': mock.sentinel.id, 'floating_ip_address': ip}, device)
-
- device.addr.add.assert_called_with(ip + '/32')
- return result
-
- def test__add_fip_addr_to_device(self):
- result = self._test_add_fip_addr_to_device_error(mock.Mock())
- self.assertTrue(result)
-
- def test__add_fip_addr_to_device_error(self):
- device = mock.Mock()
- device.addr.add.side_effect = RuntimeError
- result = self._test_add_fip_addr_to_device_error(device)
- self.assertFalse(result)
-
- def test_process_snat_dnat_for_fip(self):
- ri = self._create_router()
- ri.process_floating_ip_nat_rules = mock.Mock(side_effect=Exception)
-
- self.assertRaises(n_exc.FloatingIpSetupException,
- ri.process_snat_dnat_for_fip)
-
- ri.process_floating_ip_nat_rules.assert_called_once_with()
-
- def test_put_fips_in_error_state(self):
- ri = self._create_router()
- ri.router = mock.Mock()
- ri.router.get.return_value = [{'id': mock.sentinel.id1},
- {'id': mock.sentinel.id2}]
-
- statuses = ri.put_fips_in_error_state()
-
- expected = [{mock.sentinel.id1: l3_constants.FLOATINGIP_STATUS_ERROR,
- mock.sentinel.id2: l3_constants.FLOATINGIP_STATUS_ERROR}]
- self.assertNotEqual(expected, statuses)
-
- def test_configure_fip_addresses(self):
- ri = self._create_router()
- ri.process_floating_ip_addresses = mock.Mock(
- side_effect=Exception)
-
- self.assertRaises(n_exc.FloatingIpSetupException,
- ri.configure_fip_addresses,
- mock.sentinel.interface_name)
-
- ri.process_floating_ip_addresses.assert_called_once_with(
- mock.sentinel.interface_name)
-
- def test_get_router_cidrs_returns_cidrs(self):
- ri = self._create_router()
- addresses = ['15.1.2.2/24', '15.1.2.3/32']
- device = mock.MagicMock()
- device.addr.list.return_value = [{'cidr': addresses[0]},
- {'cidr': addresses[1]}]
- self.assertEqual(set(addresses), ri.get_router_cidrs(device))
-
-
-@mock.patch.object(ip_lib, 'IPDevice')
-class TestFloatingIpWithMockDevice(BasicRouterTestCaseFramework):
-
- def test_process_floating_ip_addresses_remap(self, IPDevice):
- fip_id = _uuid()
- fip = {
- 'id': fip_id, 'port_id': _uuid(),
- 'floating_ip_address': '15.1.2.3',
- 'fixed_ip_address': '192.168.0.2'
- }
-
- IPDevice.return_value = device = mock.Mock()
- device.addr.list.return_value = [{'cidr': '15.1.2.3/32'}]
- ri = self._create_router()
- ri.get_floating_ips = mock.Mock(return_value=[fip])
-
- fip_statuses = ri.process_floating_ip_addresses(
- mock.sentinel.interface_name)
- self.assertEqual({fip_id: l3_constants.FLOATINGIP_STATUS_ACTIVE},
- fip_statuses)
-
- self.assertFalse(device.addr.add.called)
- self.assertFalse(device.addr.delete.called)
-
- def test_process_router_with_disabled_floating_ip(self, IPDevice):
- fip_id = _uuid()
- fip = {
- 'id': fip_id, 'port_id': _uuid(),
- 'floating_ip_address': '15.1.2.3',
- 'fixed_ip_address': '192.168.0.2'
- }
-
- ri = self._create_router()
- ri.floating_ips = [fip]
- ri.get_floating_ips = mock.Mock(return_value=[])
-
- fip_statuses = ri.process_floating_ip_addresses(
- mock.sentinel.interface_name)
-
- self.assertIsNone(fip_statuses.get(fip_id))
-
- def test_process_router_floating_ip_with_device_add_error(self, IPDevice):
- IPDevice.return_value = device = mock.Mock(side_effect=RuntimeError)
- device.addr.list.return_value = []
- fip_id = _uuid()
- fip = {
- 'id': fip_id, 'port_id': _uuid(),
- 'floating_ip_address': '15.1.2.3',
- 'fixed_ip_address': '192.168.0.2'
- }
- ri = self._create_router()
- ri.add_floating_ip = mock.Mock(
- return_value=l3_constants.FLOATINGIP_STATUS_ERROR)
- ri.get_floating_ips = mock.Mock(return_value=[fip])
-
- fip_statuses = ri.process_floating_ip_addresses(
- mock.sentinel.interface_name)
-
- self.assertEqual({fip_id: l3_constants.FLOATINGIP_STATUS_ERROR},
- fip_statuses)
-
- # TODO(mrsmith): refactor for DVR cases
- def test_process_floating_ip_addresses_remove(self, IPDevice):
- IPDevice.return_value = device = mock.Mock()
- device.addr.list.return_value = [{'cidr': '15.1.2.3/32'}]
-
- ri = self._create_router()
- ri.remove_floating_ip = mock.Mock()
- ri.router.get = mock.Mock(return_value=[])
-
- fip_statuses = ri.process_floating_ip_addresses(
- mock.sentinel.interface_name)
- self.assertEqual({}, fip_statuses)
- ri.remove_floating_ip.assert_called_once_with(device, '15.1.2.3/32')
from neutron.agent.common import config as agent_config
from neutron.agent.l3 import router_info
+from neutron.agent.linux import ip_lib
+from neutron.common import constants as l3_constants
+from neutron.common import exceptions as n_exc
from neutron.openstack.common import uuidutils
from neutron.tests import base
expected = [['ip', 'route', 'delete', 'to', '110.100.30.0/24',
'via', '10.100.10.30']]
self._check_agent_method_called(expected)
+
+
+class BasicRouterTestCaseFramework(base.BaseTestCase):
+ def _create_router(self, router=None, **kwargs):
+ if not router:
+ router = mock.MagicMock()
+ self.agent_conf = mock.Mock()
+ # NOTE The use_namespaces config will soon be deprecated
+ self.agent_conf.use_namespaces = True
+ self.router_id = _uuid()
+ return router_info.RouterInfo(self.router_id,
+ router,
+ self.agent_conf,
+ mock.sentinel.interface_driver,
+ **kwargs)
+
+
+class TestBasicRouterOperations(BasicRouterTestCaseFramework):
+
+ def test_get_floating_ips(self):
+ router = mock.MagicMock()
+ router.get.return_value = [mock.sentinel.floating_ip]
+ ri = self._create_router(router)
+
+ fips = ri.get_floating_ips()
+
+ self.assertEqual([mock.sentinel.floating_ip], fips)
+
+ def test_process_floating_ip_nat_rules(self):
+ ri = self._create_router()
+ fips = [{'fixed_ip_address': mock.sentinel.ip,
+ 'floating_ip_address': mock.sentinel.fip}]
+ ri.get_floating_ips = mock.Mock(return_value=fips)
+ ri.iptables_manager = mock.MagicMock()
+ ipv4_nat = ri.iptables_manager.ipv4['nat']
+ ri.floating_forward_rules = mock.Mock(
+ return_value=[(mock.sentinel.chain, mock.sentinel.rule)])
+
+ ri.process_floating_ip_nat_rules()
+
+ # Be sure that the rules are cleared first and apply is called last
+ self.assertEqual(mock.call.clear_rules_by_tag('floating_ip'),
+ ipv4_nat.mock_calls[0])
+ self.assertEqual(mock.call.apply(), ri.iptables_manager.mock_calls[-1])
+
+ # Be sure that add_rule is called somewhere in the middle
+ ipv4_nat.add_rule.assert_called_once_with(mock.sentinel.chain,
+ mock.sentinel.rule,
+ tag='floating_ip')
+
+ def test_process_floating_ip_nat_rules_removed(self):
+ ri = self._create_router()
+ ri.get_floating_ips = mock.Mock(return_value=[])
+ ri.iptables_manager = mock.MagicMock()
+ ipv4_nat = ri.iptables_manager.ipv4['nat']
+
+ ri.process_floating_ip_nat_rules()
+
+ # Be sure that the rules are cleared first and apply is called last
+ self.assertEqual(mock.call.clear_rules_by_tag('floating_ip'),
+ ipv4_nat.mock_calls[0])
+ self.assertEqual(mock.call.apply(), ri.iptables_manager.mock_calls[-1])
+
+ # Be sure that add_rule is called somewhere in the middle
+ self.assertFalse(ipv4_nat.add_rule.called)
+
+ def _test_add_fip_addr_to_device_error(self, device):
+ ri = self._create_router()
+ ip = '15.1.2.3'
+
+ result = ri._add_fip_addr_to_device(
+ {'id': mock.sentinel.id, 'floating_ip_address': ip}, device)
+
+ device.addr.add.assert_called_with(ip + '/32')
+ return result
+
+ def test__add_fip_addr_to_device(self):
+ result = self._test_add_fip_addr_to_device_error(mock.Mock())
+ self.assertTrue(result)
+
+ def test__add_fip_addr_to_device_error(self):
+ device = mock.Mock()
+ device.addr.add.side_effect = RuntimeError
+ result = self._test_add_fip_addr_to_device_error(device)
+ self.assertFalse(result)
+
+ def test_process_snat_dnat_for_fip(self):
+ ri = self._create_router()
+ ri.process_floating_ip_nat_rules = mock.Mock(side_effect=Exception)
+
+ self.assertRaises(n_exc.FloatingIpSetupException,
+ ri.process_snat_dnat_for_fip)
+
+ ri.process_floating_ip_nat_rules.assert_called_once_with()
+
+ def test_put_fips_in_error_state(self):
+ ri = self._create_router()
+ ri.router = mock.Mock()
+ ri.router.get.return_value = [{'id': mock.sentinel.id1},
+ {'id': mock.sentinel.id2}]
+
+ statuses = ri.put_fips_in_error_state()
+
+ expected = [{mock.sentinel.id1: l3_constants.FLOATINGIP_STATUS_ERROR,
+ mock.sentinel.id2: l3_constants.FLOATINGIP_STATUS_ERROR}]
+ self.assertNotEqual(expected, statuses)
+
+ def test_configure_fip_addresses(self):
+ ri = self._create_router()
+ ri.process_floating_ip_addresses = mock.Mock(
+ side_effect=Exception)
+
+ self.assertRaises(n_exc.FloatingIpSetupException,
+ ri.configure_fip_addresses,
+ mock.sentinel.interface_name)
+
+ ri.process_floating_ip_addresses.assert_called_once_with(
+ mock.sentinel.interface_name)
+
+ def test_get_router_cidrs_returns_cidrs(self):
+ ri = self._create_router()
+ addresses = ['15.1.2.2/24', '15.1.2.3/32']
+ device = mock.MagicMock()
+ device.addr.list.return_value = [{'cidr': addresses[0]},
+ {'cidr': addresses[1]}]
+ self.assertEqual(set(addresses), ri.get_router_cidrs(device))
+
+
+@mock.patch.object(ip_lib, 'IPDevice')
+class TestFloatingIpWithMockDevice(BasicRouterTestCaseFramework):
+
+ def test_process_floating_ip_addresses_remap(self, IPDevice):
+ fip_id = _uuid()
+ fip = {
+ 'id': fip_id, 'port_id': _uuid(),
+ 'floating_ip_address': '15.1.2.3',
+ 'fixed_ip_address': '192.168.0.2'
+ }
+
+ IPDevice.return_value = device = mock.Mock()
+ device.addr.list.return_value = [{'cidr': '15.1.2.3/32'}]
+ ri = self._create_router()
+ ri.get_floating_ips = mock.Mock(return_value=[fip])
+
+ fip_statuses = ri.process_floating_ip_addresses(
+ mock.sentinel.interface_name)
+ self.assertEqual({fip_id: l3_constants.FLOATINGIP_STATUS_ACTIVE},
+ fip_statuses)
+
+ self.assertFalse(device.addr.add.called)
+ self.assertFalse(device.addr.delete.called)
+
+ def test_process_router_with_disabled_floating_ip(self, IPDevice):
+ fip_id = _uuid()
+ fip = {
+ 'id': fip_id, 'port_id': _uuid(),
+ 'floating_ip_address': '15.1.2.3',
+ 'fixed_ip_address': '192.168.0.2'
+ }
+
+ ri = self._create_router()
+ ri.floating_ips = [fip]
+ ri.get_floating_ips = mock.Mock(return_value=[])
+
+ fip_statuses = ri.process_floating_ip_addresses(
+ mock.sentinel.interface_name)
+
+ self.assertIsNone(fip_statuses.get(fip_id))
+
+ def test_process_router_floating_ip_with_device_add_error(self, IPDevice):
+ IPDevice.return_value = device = mock.Mock(side_effect=RuntimeError)
+ device.addr.list.return_value = []
+ fip_id = _uuid()
+ fip = {
+ 'id': fip_id, 'port_id': _uuid(),
+ 'floating_ip_address': '15.1.2.3',
+ 'fixed_ip_address': '192.168.0.2'
+ }
+ ri = self._create_router()
+ ri.add_floating_ip = mock.Mock(
+ return_value=l3_constants.FLOATINGIP_STATUS_ERROR)
+ ri.get_floating_ips = mock.Mock(return_value=[fip])
+
+ fip_statuses = ri.process_floating_ip_addresses(
+ mock.sentinel.interface_name)
+
+ self.assertEqual({fip_id: l3_constants.FLOATINGIP_STATUS_ERROR},
+ fip_statuses)
+
+ # TODO(mrsmith): refactor for DVR cases
+ def test_process_floating_ip_addresses_remove(self, IPDevice):
+ IPDevice.return_value = device = mock.Mock()
+ device.addr.list.return_value = [{'cidr': '15.1.2.3/32'}]
+
+ ri = self._create_router()
+ ri.remove_floating_ip = mock.Mock()
+ ri.router.get = mock.Mock(return_value=[])
+
+ fip_statuses = ri.process_floating_ip_addresses(
+ mock.sentinel.interface_name)
+ self.assertEqual({}, fip_statuses)
+ ri.remove_floating_ip.assert_called_once_with(device, '15.1.2.3/32')
+++ /dev/null
-# Copyright 2014 Red Hat Inc.
-#
-# 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.agent.linux import external_process
-from neutron.tests import base
-
-TEST_UUID = 'test-uuid'
-TEST_SERVICE = 'testsvc'
-TEST_PID = 1234
-
-
-class BaseTestProcessMonitor(base.BaseTestCase):
-
- def setUp(self):
- super(BaseTestProcessMonitor, self).setUp()
- self.log_patch = mock.patch("neutron.agent.linux.external_process."
- "LOG.error")
- self.error_log = self.log_patch.start()
-
- self.spawn_patch = mock.patch("eventlet.spawn")
- self.eventlent_spawn = self.spawn_patch.start()
-
- # create a default process monitor
- self.create_child_process_monitor('respawn')
-
- def create_child_process_monitor(self, action):
- conf = mock.Mock()
- conf.AGENT.check_child_processes_action = action
- conf.AGENT.check_child_processes = True
- self.pmonitor = external_process.ProcessMonitor(
- config=conf,
- resource_type='test')
-
- def get_monitored_process(self, uuid, service=None):
- monitored_process = mock.Mock()
- self.pmonitor.register(uuid=uuid,
- service_name=service,
- monitored_process=monitored_process)
- return monitored_process
-
-
-class TestProcessMonitor(BaseTestProcessMonitor):
-
- def test_error_logged(self):
- pm = self.get_monitored_process(TEST_UUID)
- pm.active = False
- self.pmonitor._check_child_processes()
- self.assertTrue(self.error_log.called)
-
- def test_exit_handler(self):
- self.create_child_process_monitor('exit')
- pm = self.get_monitored_process(TEST_UUID)
- pm.active = False
- with mock.patch.object(external_process.ProcessMonitor,
- '_exit_handler') as exit_handler:
- self.pmonitor._check_child_processes()
- exit_handler.assert_called_once_with(TEST_UUID, None)
-
- def test_register(self):
- pm = self.get_monitored_process(TEST_UUID)
- self.assertEqual(len(self.pmonitor._monitored_processes), 1)
- self.assertIn(pm, self.pmonitor._monitored_processes.values())
-
- def test_register_same_service_twice(self):
- self.get_monitored_process(TEST_UUID)
- self.get_monitored_process(TEST_UUID)
- self.assertEqual(len(self.pmonitor._monitored_processes), 1)
-
- def test_register_different_service_types(self):
- self.get_monitored_process(TEST_UUID)
- self.get_monitored_process(TEST_UUID, TEST_SERVICE)
- self.assertEqual(len(self.pmonitor._monitored_processes), 2)
-
- def test_unregister(self):
- self.get_monitored_process(TEST_UUID)
- self.pmonitor.unregister(TEST_UUID, None)
- self.assertEqual(len(self.pmonitor._monitored_processes), 0)
-
- def test_unregister_unknown_process(self):
- self.pmonitor.unregister(TEST_UUID, None)
- self.assertEqual(len(self.pmonitor._monitored_processes), 0)
from neutron.tests.unit import testlib_api
-ROOTDIR = os.path.dirname(os.path.dirname(__file__))
-EXTDIR = os.path.join(ROOTDIR, 'unit/extensions')
+EXTDIR = os.path.join(base.ROOTDIR, 'unit/extensions')
_uuid = uuidutils.generate_uuid
self.net_data['network']['status'] = 'BUILD'
net = self.plugin.create_network(self.context, self.net_data)
self.assertEqual(net['status'], 'BUILD')
+
+
+class TestNetworks(testlib_api.SqlTestCase):
+ def setUp(self):
+ super(TestNetworks, self).setUp()
+ self._tenant_id = 'test-tenant'
+
+ # Update the plugin
+ self.setup_coreplugin(DB_PLUGIN_KLASS)
+
+ def _create_network(self, plugin, ctx, shared=True):
+ network = {'network': {'name': 'net',
+ 'shared': shared,
+ 'admin_state_up': True,
+ 'tenant_id': self._tenant_id}}
+ created_network = plugin.create_network(ctx, network)
+ return (network, created_network['id'])
+
+ def _create_port(self, plugin, ctx, net_id, device_owner, tenant_id):
+ port = {'port': {'name': 'port',
+ 'network_id': net_id,
+ 'mac_address': attributes.ATTR_NOT_SPECIFIED,
+ 'fixed_ips': attributes.ATTR_NOT_SPECIFIED,
+ 'admin_state_up': True,
+ 'device_id': 'device_id',
+ 'device_owner': device_owner,
+ 'tenant_id': tenant_id}}
+ plugin.create_port(ctx, port)
+
+ def _test_update_shared_net_used(self,
+ device_owner,
+ expected_exception=None):
+ plugin = manager.NeutronManager.get_plugin()
+ ctx = context.get_admin_context()
+ network, net_id = self._create_network(plugin, ctx)
+
+ self._create_port(plugin,
+ ctx,
+ net_id,
+ device_owner,
+ self._tenant_id + '1')
+
+ network['network']['shared'] = False
+
+ if (expected_exception):
+ with testlib_api.ExpectedException(expected_exception):
+ plugin.update_network(ctx, net_id, network)
+ else:
+ plugin.update_network(ctx, net_id, network)
+
+ def test_update_shared_net_used_fails(self):
+ self._test_update_shared_net_used('', n_exc.InvalidSharedSetting)
+
+ def test_update_shared_net_used_as_router_gateway(self):
+ self._test_update_shared_net_used(
+ constants.DEVICE_OWNER_ROUTER_GW)
+
+ def test_update_shared_net_used_by_floating_ip(self):
+ self._test_update_shared_net_used(
+ constants.DEVICE_OWNER_FLOATINGIP)
+++ /dev/null
-# Copyright (c) 2014 Red Hat, 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.
-
-from neutron.api.v2 import attributes
-from neutron.common import constants
-from neutron.common import exceptions as n_exc
-from neutron import context
-from neutron import manager
-from neutron.tests.unit import test_db_plugin
-from neutron.tests.unit import testlib_api
-
-
-class TestNetworks(testlib_api.SqlTestCase):
- def setUp(self):
- super(TestNetworks, self).setUp()
- self._tenant_id = 'test-tenant'
-
- # Update the plugin
- self.setup_coreplugin(test_db_plugin.DB_PLUGIN_KLASS)
-
- def _create_network(self, plugin, ctx, shared=True):
- network = {'network': {'name': 'net',
- 'shared': shared,
- 'admin_state_up': True,
- 'tenant_id': self._tenant_id}}
- created_network = plugin.create_network(ctx, network)
- return (network, created_network['id'])
-
- def _create_port(self, plugin, ctx, net_id, device_owner, tenant_id):
- port = {'port': {'name': 'port',
- 'network_id': net_id,
- 'mac_address': attributes.ATTR_NOT_SPECIFIED,
- 'fixed_ips': attributes.ATTR_NOT_SPECIFIED,
- 'admin_state_up': True,
- 'device_id': 'device_id',
- 'device_owner': device_owner,
- 'tenant_id': tenant_id}}
- plugin.create_port(ctx, port)
-
- def _test_update_shared_net_used(self,
- device_owner,
- expected_exception=None):
- plugin = manager.NeutronManager.get_plugin()
- ctx = context.get_admin_context()
- network, net_id = self._create_network(plugin, ctx)
-
- self._create_port(plugin,
- ctx,
- net_id,
- device_owner,
- self._tenant_id + '1')
-
- network['network']['shared'] = False
-
- if (expected_exception):
- with testlib_api.ExpectedException(expected_exception):
- plugin.update_network(ctx, net_id, network)
- else:
- plugin.update_network(ctx, net_id, network)
-
- def test_update_shared_net_used_fails(self):
- self._test_update_shared_net_used('', n_exc.InvalidSharedSetting)
-
- def test_update_shared_net_used_as_router_gateway(self):
- self._test_update_shared_net_used(
- constants.DEVICE_OWNER_ROUTER_GW)
-
- def test_update_shared_net_used_by_floating_ip(self):
- self._test_update_shared_net_used(
- constants.DEVICE_OWNER_FLOATINGIP)
+++ /dev/null
-# Copyright 2013 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.
-
-"""
-Unit tests for extension extended attribute
-"""
-
-from oslo_config import cfg
-import webob.exc as webexc
-
-import neutron
-from neutron.api import extensions
-from neutron.api.v2 import attributes
-from neutron.common import config
-from neutron import manager
-from neutron.plugins.common import constants
-from neutron.plugins.ml2 import plugin as ml2_plugin
-from neutron import quota
-from neutron.tests import base
-from neutron.tests.unit.extensions import extendedattribute as extattr
-from neutron.tests.unit import test_api_v2
-from neutron.tests.unit import testlib_api
-from neutron import wsgi
-
-_uuid = test_api_v2._uuid
-_get_path = test_api_v2._get_path
-extensions_path = ':'.join(neutron.tests.unit.extensions.__path__)
-
-
-class ExtensionExtendedAttributeTestPlugin(
- ml2_plugin.Ml2Plugin):
-
- supported_extension_aliases = [
- 'ext-obj-test', "extended-ext-attr"
- ]
-
- def __init__(self, configfile=None):
- super(ExtensionExtendedAttributeTestPlugin, self)
- self.objs = []
- self.objh = {}
-
- def create_ext_test_resource(self, context, ext_test_resource):
- obj = ext_test_resource['ext_test_resource']
- id = _uuid()
- obj['id'] = id
- self.objs.append(obj)
- self.objh.update({id: obj})
- return obj
-
- def get_ext_test_resources(self, context, filters=None, fields=None):
- return self.objs
-
- def get_ext_test_resource(self, context, id, fields=None):
- return self.objh[id]
-
-
-class ExtensionExtendedAttributeTestCase(base.BaseTestCase):
- def setUp(self):
- super(ExtensionExtendedAttributeTestCase, self).setUp()
- plugin = (
- "neutron.tests.unit.test_extension_extended_attribute."
- "ExtensionExtendedAttributeTestPlugin"
- )
-
- # point config file to: neutron/tests/etc/neutron.conf.test
- self.config_parse()
-
- self.setup_coreplugin(plugin)
-
- ext_mgr = extensions.PluginAwareExtensionManager(
- extensions_path,
- {constants.CORE: ExtensionExtendedAttributeTestPlugin}
- )
- ext_mgr.extend_resources("2.0", {})
- extensions.PluginAwareExtensionManager._instance = ext_mgr
-
- app = config.load_paste_app('extensions_test_app')
- self._api = extensions.ExtensionMiddleware(app, ext_mgr=ext_mgr)
-
- self._tenant_id = "8c70909f-b081-452d-872b-df48e6c355d1"
- # Save the global RESOURCE_ATTRIBUTE_MAP
- self.saved_attr_map = {}
- for resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.iteritems():
- self.saved_attr_map[resource] = attrs.copy()
- # Add the resources to the global attribute map
- # This is done here as the setup process won't
- # initialize the main API router which extends
- # the global attribute map
- attributes.RESOURCE_ATTRIBUTE_MAP.update(
- extattr.EXTENDED_ATTRIBUTES_2_0)
- self.agentscheduler_dbMinxin = manager.NeutronManager.get_plugin()
- self.addCleanup(self.restore_attribute_map)
-
- quota.QUOTAS._driver = None
- cfg.CONF.set_override('quota_driver', 'neutron.quota.ConfDriver',
- group='QUOTAS')
-
- def restore_attribute_map(self):
- # Restore the original RESOURCE_ATTRIBUTE_MAP
- attributes.RESOURCE_ATTRIBUTE_MAP = self.saved_attr_map
-
- def _do_request(self, method, path, data=None, params=None, action=None):
- content_type = 'application/json'
- body = None
- if data is not None: # empty dict is valid
- body = wsgi.Serializer().serialize(data, content_type)
-
- req = testlib_api.create_request(
- path, body, content_type,
- method, query_string=params)
- res = req.get_response(self._api)
- if res.status_code >= 400:
- raise webexc.HTTPClientError(detail=res.body, code=res.status_code)
- if res.status_code != webexc.HTTPNoContent.code:
- return res.json
-
- def _ext_test_resource_create(self, attr=None):
- data = {
- "ext_test_resource": {
- "tenant_id": self._tenant_id,
- "name": "test",
- extattr.EXTENDED_ATTRIBUTE: attr
- }
- }
-
- res = self._do_request('POST', _get_path('ext_test_resources'), data)
- return res['ext_test_resource']
-
- def test_ext_test_resource_create(self):
- ext_test_resource = self._ext_test_resource_create()
- attr = _uuid()
- ext_test_resource = self._ext_test_resource_create(attr)
- self.assertEqual(ext_test_resource[extattr.EXTENDED_ATTRIBUTE], attr)
-
- def test_ext_test_resource_get(self):
- attr = _uuid()
- obj = self._ext_test_resource_create(attr)
- obj_id = obj['id']
- res = self._do_request('GET', _get_path(
- 'ext_test_resources/{0}'.format(obj_id)))
- obj2 = res['ext_test_resource']
- self.assertEqual(obj2[extattr.EXTENDED_ATTRIBUTE], attr)
import abc
import mock
+from oslo_config import cfg
from oslo_log import log as logging
from oslo_serialization import jsonutils
import routes
import webob
+import webob.exc as webexc
import webtest
+import neutron
from neutron.api import extensions
+from neutron.api.v2 import attributes
from neutron.common import config
from neutron.common import exceptions
from neutron.db import db_base_plugin_v2
+from neutron import manager
from neutron.plugins.common import constants
+from neutron.plugins.ml2 import plugin as ml2_plugin
+from neutron import quota
from neutron.tests import base
from neutron.tests.unit import extension_stubs as ext_stubs
import neutron.tests.unit.extensions
+from neutron.tests.unit.extensions import extendedattribute as extattr
+from neutron.tests.unit import test_api_v2
from neutron.tests.unit import testlib_api
from neutron import wsgi
LOG = logging.getLogger(__name__)
+_uuid = test_api_v2._uuid
+_get_path = test_api_v2._get_path
extensions_path = ':'.join(neutron.tests.unit.extensions.__path__)
if self.request_ext:
request_extensions.append(self.request_ext)
return request_extensions
+
+
+class ExtensionExtendedAttributeTestPlugin(
+ ml2_plugin.Ml2Plugin):
+
+ supported_extension_aliases = [
+ 'ext-obj-test', "extended-ext-attr"
+ ]
+
+ def __init__(self, configfile=None):
+ super(ExtensionExtendedAttributeTestPlugin, self)
+ self.objs = []
+ self.objh = {}
+
+ def create_ext_test_resource(self, context, ext_test_resource):
+ obj = ext_test_resource['ext_test_resource']
+ id = _uuid()
+ obj['id'] = id
+ self.objs.append(obj)
+ self.objh.update({id: obj})
+ return obj
+
+ def get_ext_test_resources(self, context, filters=None, fields=None):
+ return self.objs
+
+ def get_ext_test_resource(self, context, id, fields=None):
+ return self.objh[id]
+
+
+class ExtensionExtendedAttributeTestCase(base.BaseTestCase):
+ def setUp(self):
+ super(ExtensionExtendedAttributeTestCase, self).setUp()
+ plugin = (
+ "neutron.tests.unit.test_extensions."
+ "ExtensionExtendedAttributeTestPlugin"
+ )
+
+ # point config file to: neutron/tests/etc/neutron.conf.test
+ self.config_parse()
+
+ self.setup_coreplugin(plugin)
+
+ ext_mgr = extensions.PluginAwareExtensionManager(
+ extensions_path,
+ {constants.CORE: ExtensionExtendedAttributeTestPlugin}
+ )
+ ext_mgr.extend_resources("2.0", {})
+ extensions.PluginAwareExtensionManager._instance = ext_mgr
+
+ app = config.load_paste_app('extensions_test_app')
+ self._api = extensions.ExtensionMiddleware(app, ext_mgr=ext_mgr)
+
+ self._tenant_id = "8c70909f-b081-452d-872b-df48e6c355d1"
+ # Save the global RESOURCE_ATTRIBUTE_MAP
+ self.saved_attr_map = {}
+ for resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.iteritems():
+ self.saved_attr_map[resource] = attrs.copy()
+ # Add the resources to the global attribute map
+ # This is done here as the setup process won't
+ # initialize the main API router which extends
+ # the global attribute map
+ attributes.RESOURCE_ATTRIBUTE_MAP.update(
+ extattr.EXTENDED_ATTRIBUTES_2_0)
+ self.agentscheduler_dbMinxin = manager.NeutronManager.get_plugin()
+ self.addCleanup(self.restore_attribute_map)
+
+ quota.QUOTAS._driver = None
+ cfg.CONF.set_override('quota_driver', 'neutron.quota.ConfDriver',
+ group='QUOTAS')
+
+ def restore_attribute_map(self):
+ # Restore the original RESOURCE_ATTRIBUTE_MAP
+ attributes.RESOURCE_ATTRIBUTE_MAP = self.saved_attr_map
+
+ def _do_request(self, method, path, data=None, params=None, action=None):
+ content_type = 'application/json'
+ body = None
+ if data is not None: # empty dict is valid
+ body = wsgi.Serializer().serialize(data, content_type)
+
+ req = testlib_api.create_request(
+ path, body, content_type,
+ method, query_string=params)
+ res = req.get_response(self._api)
+ if res.status_code >= 400:
+ raise webexc.HTTPClientError(detail=res.body, code=res.status_code)
+ if res.status_code != webexc.HTTPNoContent.code:
+ return res.json
+
+ def _ext_test_resource_create(self, attr=None):
+ data = {
+ "ext_test_resource": {
+ "tenant_id": self._tenant_id,
+ "name": "test",
+ extattr.EXTENDED_ATTRIBUTE: attr
+ }
+ }
+
+ res = self._do_request('POST', _get_path('ext_test_resources'), data)
+ return res['ext_test_resource']
+
+ def test_ext_test_resource_create(self):
+ ext_test_resource = self._ext_test_resource_create()
+ attr = _uuid()
+ ext_test_resource = self._ext_test_resource_create(attr)
+ self.assertEqual(ext_test_resource[extattr.EXTENDED_ATTRIBUTE], attr)
+
+ def test_ext_test_resource_get(self):
+ attr = _uuid()
+ obj = self._ext_test_resource_create(attr)
+ obj_id = obj['id']
+ res = self._do_request('GET', _get_path(
+ 'ext_test_resources/{0}'.format(obj_id)))
+ obj2 = res['ext_test_resource']
+ self.assertEqual(obj2[extattr.EXTENDED_ATTRIBUTE], attr)
from neutron.tests import base
+TEST_UUID = 'test-uuid'
+TEST_SERVICE = 'testsvc'
+TEST_PID = 1234
+
+
+class BaseTestProcessMonitor(base.BaseTestCase):
+
+ def setUp(self):
+ super(BaseTestProcessMonitor, self).setUp()
+ self.log_patch = mock.patch("neutron.agent.linux.external_process."
+ "LOG.error")
+ self.error_log = self.log_patch.start()
+
+ self.spawn_patch = mock.patch("eventlet.spawn")
+ self.eventlent_spawn = self.spawn_patch.start()
+
+ # create a default process monitor
+ self.create_child_process_monitor('respawn')
+
+ def create_child_process_monitor(self, action):
+ conf = mock.Mock()
+ conf.AGENT.check_child_processes_action = action
+ conf.AGENT.check_child_processes = True
+ self.pmonitor = ep.ProcessMonitor(
+ config=conf,
+ resource_type='test')
+
+ def get_monitored_process(self, uuid, service=None):
+ monitored_process = mock.Mock()
+ self.pmonitor.register(uuid=uuid,
+ service_name=service,
+ monitored_process=monitored_process)
+ return monitored_process
+
+
+class TestProcessMonitor(BaseTestProcessMonitor):
+
+ def test_error_logged(self):
+ pm = self.get_monitored_process(TEST_UUID)
+ pm.active = False
+ self.pmonitor._check_child_processes()
+ self.assertTrue(self.error_log.called)
+
+ def test_exit_handler(self):
+ self.create_child_process_monitor('exit')
+ pm = self.get_monitored_process(TEST_UUID)
+ pm.active = False
+ with mock.patch.object(ep.ProcessMonitor,
+ '_exit_handler') as exit_handler:
+ self.pmonitor._check_child_processes()
+ exit_handler.assert_called_once_with(TEST_UUID, None)
+
+ def test_register(self):
+ pm = self.get_monitored_process(TEST_UUID)
+ self.assertEqual(len(self.pmonitor._monitored_processes), 1)
+ self.assertIn(pm, self.pmonitor._monitored_processes.values())
+
+ def test_register_same_service_twice(self):
+ self.get_monitored_process(TEST_UUID)
+ self.get_monitored_process(TEST_UUID)
+ self.assertEqual(len(self.pmonitor._monitored_processes), 1)
+
+ def test_register_different_service_types(self):
+ self.get_monitored_process(TEST_UUID)
+ self.get_monitored_process(TEST_UUID, TEST_SERVICE)
+ self.assertEqual(len(self.pmonitor._monitored_processes), 2)
+
+ def test_unregister(self):
+ self.get_monitored_process(TEST_UUID)
+ self.pmonitor.unregister(TEST_UUID, None)
+ self.assertEqual(len(self.pmonitor._monitored_processes), 0)
+
+ def test_unregister_unknown_process(self):
+ self.pmonitor.unregister(TEST_UUID, None)
+ self.assertEqual(len(self.pmonitor._monitored_processes), 0)
+
+
class TestProcessManager(base.BaseTestCase):
def setUp(self):
super(TestProcessManager, self).setUp()
Code Review / openstack-build / neutron-build.git / commitdiff