def _bind_port_to_sgs(self, context, port, sg_ids):
self._process_port_create_security_group(context, port, sg_ids)
- for sg_id in sg_ids:
- pg_name = _sg_port_group_name(sg_id)
- self.client.add_port_to_port_group_by_name(port["tenant_id"],
- pg_name, port["id"])
+ if sg_ids is not None:
+ for sg_id in sg_ids:
+ pg_name = _sg_port_group_name(sg_id)
+ self.client.add_port_to_port_group_by_name(
+ port["tenant_id"], pg_name, port["id"])
def _unbind_port_from_sgs(self, context, port_id):
self._delete_port_security_group_bindings(context, port_id)
def create_port(self, context, port):
"""Create a L2 port in Neutron/MidoNet."""
LOG.debug(_("MidonetPluginV2.create_port called: port=%r"), port)
-
port_data = port['port']
# Create a bridge port in MidoNet and set the bridge port ID as the
new_port = super(MidonetPluginV2, self).create_port(context,
port)
port_data.update(new_port)
-
- # Bind security groups to the port
- sg_ids = self._get_security_groups_on_port(context, port)
- if sg_ids:
+ self._ensure_default_security_group_on_port(context,
+ port)
+ if _is_vif_port(port_data):
+ # Bind security groups to the port
+ sg_ids = self._get_security_groups_on_port(context, port)
self._bind_port_to_sgs(context, port_data, sg_ids)
- port_data[ext_sg.SECURITYGROUPS] = sg_ids
- # Create port chains
- port_chains = {}
- for d, name in _port_chain_names(new_port["id"]).iteritems():
- port_chains[d] = self.client.create_chain(tenant_id, name)
+ # Create port chains
+ port_chains = {}
+ for d, name in _port_chain_names(
+ new_port["id"]).iteritems():
+ port_chains[d] = self.client.create_chain(tenant_id,
+ name)
- self._initialize_port_chains(port_data, port_chains['inbound'],
- port_chains['outbound'], sg_ids)
+ self._initialize_port_chains(port_data,
+ port_chains['inbound'],
+ port_chains['outbound'],
+ sg_ids)
- # Update the port with the chain
- self.client.update_port_chains(
- bridge_port, port_chains["inbound"].get_id(),
- port_chains["outbound"].get_id())
+ # Update the port with the chain
+ self.client.update_port_chains(
+ bridge_port, port_chains["inbound"].get_id(),
+ port_chains["outbound"].get_id())
- if _is_dhcp_port(port_data):
- # For DHCP port, add a metadata route
- for cidr, ip in self._metadata_subnets(
- context, port_data["fixed_ips"]):
- self.client.add_dhcp_route_option(bridge, cidr, ip,
- METADATA_DEFAULT_IP)
- elif _is_vif_port(port_data):
# DHCP mapping is only for VIF ports
for cidr, ip, mac in self._dhcp_mappings(
context, port_data["fixed_ips"],
port_data["mac_address"]):
self.client.add_dhcp_host(bridge, cidr, ip, mac)
+ elif _is_dhcp_port(port_data):
+ # For DHCP port, add a metadata route
+ for cidr, ip in self._metadata_subnets(
+ context, port_data["fixed_ips"]):
+ self.client.add_dhcp_route_option(bridge, cidr, ip,
+ METADATA_DEFAULT_IP)
+
except Exception as ex:
# Try removing the MidoNet port before raising an exception.
with excutils.save_and_reraise_exception():
self._check_update_has_security_groups(port)):
self._unbind_port_from_sgs(context, p["id"])
sg_ids = self._get_security_groups_on_port(context, port)
- if sg_ids:
- self._bind_port_to_sgs(context, p, sg_ids)
+ self._bind_port_to_sgs(context, p, sg_ids)
+
return p
def create_router(self, context, router):
# Not all VM images supports DHCP option 121. Add a route for the
# Metadata server in the router to forward the packet to the bridge
# that will send them to the Metadata Proxy.
- net_addr, net_len = net_util.net_addr(METADATA_DEFAULT_IP)
+ md_net_addr, md_net_len = net_util.net_addr(METADATA_DEFAULT_IP)
self.client.add_router_route(
router, type='Normal', src_network_addr=net_addr,
src_network_length=net_len,
- dst_network_addr=net_addr,
- dst_network_length=32,
+ dst_network_addr=md_net_addr,
+ dst_network_length=md_net_len,
next_hop_port=router_port.get_id(),
next_hop_gateway=metadata_gw_ip)
from neutron.agent.common import config
from neutron.agent.linux import interface
from neutron.agent.linux import ip_lib
-from neutron.agent.linux import utils
from neutron.openstack.common import uuidutils
import neutron.plugins.midonet.agent.midonet_driver as driver
from neutron.tests import base
self.ip = self.ip_p.start()
self.device_exists_p = mock.patch.object(ip_lib, 'device_exists')
self.device_exists = self.device_exists_p.start()
-
- self.api_p = mock.patch.object(sys.modules["midonetclient"].api,
- 'MidonetApi')
- self.api = self.api_p.start()
+ self.device_exists.return_value = False
self.addCleanup(mock.patch.stopall)
midonet_opts = [
cfg.StrOpt('midonet_uri',
]
self.conf.register_opts(midonet_opts, "MIDONET")
self.driver = driver.MidonetInterfaceDriver(self.conf)
+ self.root_dev = mock.Mock()
+ self.ns_dev = mock.Mock()
+ self.ip().add_veth = mock.Mock(return_value=(
+ self.root_dev, self.ns_dev))
+ self.driver._get_host_uuid = mock.Mock(
+ return_value=uuidutils.generate_uuid())
self.network_id = uuidutils.generate_uuid()
self.port_id = uuidutils.generate_uuid()
self.device_name = "tap0"
super(MidoInterfaceDriverTestCase, self).setUp()
def test_plug(self):
- def device_exists(dev, root_helper=None, namespace=None):
- return False
-
- self.device_exists.side_effect = device_exists
- root_dev = mock.Mock()
- ns_dev = mock.Mock()
- self.ip().add_veth = mock.Mock(return_value=(root_dev, ns_dev))
- self.driver._get_host_uuid = mock.Mock(
- return_value=uuidutils.generate_uuid())
- with mock.patch.object(utils, 'execute'):
- self.driver.plug(
- self.network_id, self.port_id,
- self.device_name, self.mac_address,
- self.bridge, self.namespace)
+ self.driver.plug(
+ self.network_id, self.port_id,
+ self.device_name, self.mac_address,
+ self.bridge, self.namespace)
expected = [mock.call(), mock.call('sudo'),
mock.call().add_veth(self.device_name,
mock.call().ensure_namespace(self.namespace),
mock.call().ensure_namespace().add_device_to_namespace(
mock.ANY)]
- ns_dev.assert_has_calls(
+ self.ns_dev.assert_has_calls(
[mock.call.link.set_address(self.mac_address)])
- root_dev.assert_has_calls([mock.call.link.set_up()])
- ns_dev.assert_has_calls([mock.call.link.set_up()])
+ self.root_dev.assert_has_calls([mock.call.link.set_up()])
+ self.ns_dev.assert_has_calls([mock.call.link.set_up()])
self.ip.assert_has_calls(expected, True)
- host = mock.Mock()
- self.api().get_host = mock.Mock(return_value=host)
- self.api.assert_has_calls([mock.call().add_host_interface_port])
def test_unplug(self):
- with mock.patch.object(utils, 'execute'):
- self.driver.unplug(self.device_name, self.bridge, self.namespace)
+ self.driver.unplug(self.device_name, self.bridge, self.namespace)
self.ip_dev.assert_has_calls([
mock.call(self.device_name, self.driver.root_helper,
import neutron.common.test_lib as test_lib
import neutron.tests.unit.midonet.mock_lib as mock_lib
import neutron.tests.unit.test_db_plugin as test_plugin
+import neutron.tests.unit.test_extension_security_group as sg
MIDOKURA_PKG_PATH = "neutron.plugins.midonet.plugin"
class TestMidonetNetworksV2(test_plugin.TestNetworksV2,
MidonetPluginV2TestCase):
+
+ pass
+
+
+class TestMidonetSecurityGroupsTestCase(sg.SecurityGroupDBTestCase):
+
+ _plugin_name = ('%s.MidonetPluginV2' % MIDOKURA_PKG_PATH)
+
+ def setUp(self):
+ self.mock_api = mock.patch(
+ 'neutron.plugins.midonet.midonet_lib.MidoClient')
+ etc_path = os.path.join(os.path.dirname(__file__), 'etc')
+ test_lib.test_config['config_files'] = [os.path.join(
+ etc_path, 'midonet.ini.test')]
+
+ self.instance = self.mock_api.start()
+ mock_cfg = mock_lib.MidonetLibMockConfig(self.instance.return_value)
+ mock_cfg.setup()
+ super(TestMidonetSecurityGroupsTestCase, self).setUp(self._plugin_name)
+
+
+class TestMidonetSecurityGroup(sg.TestSecurityGroups,
+ TestMidonetSecurityGroupsTestCase):
+
pass
# tests that attempt to create them.
def test_overlapping_subnets(self):
- pass
+ pass
Code Review / openstack-build / neutron-build.git / commitdiff