Full stack tests run in the Neutron tree with Neutron resources alone. You
may use the Neutron API (The Neutron server is set to NOAUTH so that Keystone
-is out of the picture). instances may be simulated with a helper class that
-contains a container-like object in its own namespace and IP address. It has
-helper methods to send different kinds of traffic. The "instance" may be
-connected to br-int or br-ex, to simulate internal or external traffic.
+is out of the picture). VMs may be simulated with a container-like class:
+neutron.tests.fullstack.resources.machine.FakeFullstackMachine.
+An example of its usage may be found at:
+neutron/tests/fullstack/test_connectivity.py.
Full stack testing can simulate multi node testing by starting an agent
multiple times. Specifically, each node would have its own copy of the
from neutron.tests.common import net_helpers
-class FakeMachine(fixtures.Fixture):
+class FakeMachineBase(fixtures.Fixture):
+ def __init__(self):
+ self.port = None
+
+ def _setUp(self):
+ ns_fixture = self.useFixture(
+ net_helpers.NamespaceFixture())
+ self.namespace = ns_fixture.name
+
+ def execute(self, *args, **kwargs):
+ ns_ip_wrapper = ip_lib.IPWrapper(self.namespace)
+ return ns_ip_wrapper.netns.execute(*args, **kwargs)
+
+ def assert_ping(self, dst_ip):
+ net_helpers.assert_ping(self.namespace, dst_ip)
+
+ def assert_no_ping(self, dst_ip):
+ net_helpers.assert_no_ping(self.namespace, dst_ip)
+
+ @property
+ def ip(self):
+ raise NotImplementedError()
+
+ @property
+ def ip_cidr(self):
+ raise NotImplementedError()
+
+ @property
+ def mac_address(self):
+ return self.port.link.address
+
+
+class FakeMachine(FakeMachineBase):
"""Create a fake machine.
:ivar bridge: bridge on which the fake machine is bound
self.gateway_ip = gateway_ip
def _setUp(self):
- ns_fixture = self.useFixture(
- net_helpers.NamespaceFixture())
- self.namespace = ns_fixture.name
+ super(FakeMachine, self)._setUp()
self.port = self.useFixture(
net_helpers.PortFixture.get(self.bridge, self.namespace)).port
self.port.addr.delete(self._ip_cidr)
self._ip_cidr = ip_cidr
- @property
- def mac_address(self):
- return self.port.link.address
-
- @mac_address.setter
+ @FakeMachineBase.mac_address.setter
def mac_address(self, mac_address):
self.port.link.set_down()
self.port.link.set_address(mac_address)
self.port.link.set_up()
- def execute(self, *args, **kwargs):
- ns_ip_wrapper = ip_lib.IPWrapper(self.namespace)
- return ns_ip_wrapper.netns.execute(*args, **kwargs)
-
- def assert_ping(self, dst_ip):
- net_helpers.assert_ping(self.namespace, dst_ip)
-
- def assert_no_ping(self, dst_ip):
- net_helpers.assert_no_ping(self.namespace, dst_ip)
-
class PeerMachines(fixtures.Fixture):
"""Create 'amount' peered machines on an ip_cidr.
import fixtures
import netaddr
+from oslo_config import cfg
from oslo_utils import uuidutils
import six
from neutron.agent.common import config
from neutron.agent.common import ovs_lib
from neutron.agent.linux import bridge_lib
+from neutron.agent.linux import interface
from neutron.agent.linux import ip_lib
from neutron.agent.linux import utils
from neutron.common import constants as n_const
+from neutron.db import db_base_plugin_common
from neutron.tests import base as tests_base
from neutron.tests.common import base as common_base
from neutron.tests import tools
:ivar bridge: port bridge
"""
- def __init__(self, bridge=None, namespace=None):
+ def __init__(self, bridge=None, namespace=None, mac=None, port_id=None):
super(PortFixture, self).__init__()
self.bridge = bridge
self.namespace = namespace
+ self.mac = (
+ mac or db_base_plugin_common.DbBasePluginCommon._generate_mac())
+ self.port_id = port_id or uuidutils.generate_uuid()
@abc.abstractmethod
def _create_bridge_fixture(self):
self.bridge = self.useFixture(self._create_bridge_fixture()).bridge
@classmethod
- def get(cls, bridge, namespace=None):
+ def get(cls, bridge, namespace=None, mac=None, port_id=None):
"""Deduce PortFixture class from bridge type and instantiate it."""
if isinstance(bridge, ovs_lib.OVSBridge):
- return OVSPortFixture(bridge, namespace)
+ return OVSPortFixture(bridge, namespace, mac, port_id)
if isinstance(bridge, bridge_lib.BridgeDevice):
return LinuxBridgePortFixture(bridge, namespace)
if isinstance(bridge, VethBridge):
class OVSPortFixture(PortFixture):
- def __init__(self, bridge=None, namespace=None, attrs=None):
- super(OVSPortFixture, self).__init__(bridge, namespace)
- if attrs is None:
- attrs = []
- self.attrs = attrs
-
def _create_bridge_fixture(self):
return OVSBridgeFixture()
def _setUp(self):
super(OVSPortFixture, self)._setUp()
- port_name = common_base.create_resource(PORT_PREFIX, self.create_port)
+ interface_config = cfg.ConfigOpts()
+ interface_config.register_opts(interface.OPTS)
+ ovs_interface = interface.OVSInterfaceDriver(interface_config)
+
+ port_name = tests_base.get_rand_device_name(PORT_PREFIX)
+ ovs_interface.plug_new(
+ None,
+ self.port_id,
+ port_name,
+ self.mac,
+ bridge=self.bridge.br_name,
+ namespace=self.namespace)
self.addCleanup(self.bridge.delete_port, port_name)
- self.port = ip_lib.IPDevice(port_name)
-
- ns_ip_wrapper = ip_lib.IPWrapper(self.namespace)
- ns_ip_wrapper.add_device_to_namespace(self.port)
- self.port.link.set_up()
-
- def create_port(self, name):
- self.attrs.insert(0, ('type', 'internal'))
- self.bridge.add_port(name, *self.attrs)
- return name
+ self.port = ip_lib.IPDevice(port_name, self.namespace)
class LinuxBridgeFixture(fixtures.Fixture):
return self._create_resource(resource_type, spec)
+ def create_port(self, tenant_id, network_id, hostname):
+ return self._create_resource(
+ 'port',
+ {'network_id': network_id,
+ 'tenant_id': tenant_id,
+ 'binding:host_id': hostname})
+
def add_router_interface(self, router_id, subnet_id):
body = {'subnet_id': subnet_id}
self.client.add_interface_router(router=router_id, body=body)
net_helpers.create_patch_ports(
self.central_external_bridge, host_external_bridge)
+ @property
+ def hostname(self):
+ return self.neutron_config.config.DEFAULT.host
+
@property
def l3_agent(self):
return self.agents['l3']
--- /dev/null
+# Copyright 2015 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 netaddr
+
+from neutron.agent.linux import utils
+from neutron.tests.common import machine_fixtures
+from neutron.tests.common import net_helpers
+
+
+class FakeFullstackMachine(machine_fixtures.FakeMachineBase):
+ def __init__(self, host, network_id, tenant_id, safe_client):
+ super(FakeFullstackMachine, self).__init__()
+ self.bridge = host.ovs_agent.br_int
+ self.host_binding = host.hostname
+ self.tenant_id = tenant_id
+ self.network_id = network_id
+ self.safe_client = safe_client
+
+ def _setUp(self):
+ super(FakeFullstackMachine, self)._setUp()
+
+ self.neutron_port = self.safe_client.create_port(
+ network_id=self.network_id,
+ tenant_id=self.tenant_id,
+ hostname=self.host_binding)
+ self.neutron_port_id = self.neutron_port['id']
+ mac_address = self.neutron_port['mac_address']
+
+ self.port = self.useFixture(
+ net_helpers.PortFixture.get(
+ self.bridge, self.namespace, mac_address,
+ self.neutron_port_id)).port
+
+ self._ip = self.neutron_port['fixed_ips'][0]['ip_address']
+ subnet_id = self.neutron_port['fixed_ips'][0]['subnet_id']
+ subnet = self.safe_client.client.show_subnet(subnet_id)
+ prefixlen = netaddr.IPNetwork(subnet['subnet']['cidr']).prefixlen
+ self._ip_cidr = '%s/%s' % (self._ip, prefixlen)
+
+ # TODO(amuller): Support DHCP
+ self.port.addr.add(self.ip_cidr)
+
+ self.gateway_ip = subnet['subnet']['gateway_ip']
+ if self.gateway_ip:
+ net_helpers.set_namespace_gateway(self.port, self.gateway_ip)
+
+ @property
+ def ip(self):
+ return self._ip
+
+ @property
+ def ip_cidr(self):
+ return self._ip_cidr
+
+ def block_until_boot(self):
+ utils.wait_until_true(
+ lambda: (self.safe_client.client.show_port(self.neutron_port_id)
+ ['port']['status'] == 'ACTIVE'),
+ sleep=3)
--- /dev/null
+# Copyright 2015 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.
+
+from oslo_utils import uuidutils
+
+from neutron.tests.fullstack import base
+from neutron.tests.fullstack.resources import environment
+from neutron.tests.fullstack.resources import machine
+
+
+class TestConnectivitySameNetwork(base.BaseFullStackTestCase):
+
+ def __init__(self, *args, **kwargs):
+ host_descriptions = [
+ environment.HostDescription(l3_agent=False) for _ in range(2)]
+ env = environment.Environment(host_descriptions)
+ super(TestConnectivitySameNetwork, self).__init__(env, *args, **kwargs)
+
+ def test_connectivity(self):
+ tenant_uuid = uuidutils.generate_uuid()
+
+ network = self.safe_client.create_network(tenant_uuid)
+ self.safe_client.create_subnet(
+ tenant_uuid, network['id'], '20.0.0.0/24')
+
+ vms = [
+ self.useFixture(
+ machine.FakeFullstackMachine(
+ self.environment.hosts[i],
+ network['id'],
+ tenant_uuid,
+ self.safe_client))
+ for i in range(2)]
+
+ for vm in vms:
+ vm.block_until_boot()
+
+ vms[0].assert_ping(vms[1].ip)
class TestLegacyL3Agent(base.BaseFullStackTestCase):
def __init__(self, *args, **kwargs):
- super(TestLegacyL3Agent, self).__init__(
- environment.Environment(
- [environment.HostDescription(l3_agent=True)]),
- *args, **kwargs)
+ host_descriptions = [environment.HostDescription(l3_agent=True)]
+ env = environment.Environment(host_descriptions)
+ super(TestLegacyL3Agent, self).__init__(env, *args, **kwargs)
def _get_namespace(self, router_id):
return namespaces.build_ns_name(l3_agent.NS_PREFIX, router_id)
devices = self.monitor.get_events()
self.assertTrue(devices.get('added'),
'Initial call should always be true')
- p_attrs = [('external_ids', {'iface-status': 'active'})]
br = self.useFixture(net_helpers.OVSBridgeFixture())
- p1 = self.useFixture(net_helpers.OVSPortFixture(
- br.bridge, None, p_attrs))
- p2 = self.useFixture(net_helpers.OVSPortFixture(
- br.bridge, None, p_attrs))
+ p1 = self.useFixture(net_helpers.OVSPortFixture(br.bridge))
+ p2 = self.useFixture(net_helpers.OVSPortFixture(br.bridge))
added_devices = [p1.port.name, p2.port.name]
utils.wait_until_true(
lambda: self._expected_devices_events(added_devices, 'added'))
net_helpers.assert_ping(self.src_namespace, self.dst_addr, count=2)
def _setup_arp_spoof_for_port(self, port, addrs, psec=True):
- of_port_map = self.br.get_vif_port_to_ofport_map()
-
- class VifPort(object):
- ofport = of_port_map[port]
- port_name = port
-
+ vif = next(
+ vif for vif in self.br.get_vif_ports() if vif.port_name == port)
ip_addr = addrs.pop()
details = {'port_security_enabled': psec,
'fixed_ips': [{'ip_address': ip_addr}],
'allowed_address_pairs': [
dict(ip_address=ip) for ip in addrs]}
ovsagt.OVSNeutronAgent.setup_arp_spoofing_protection(
- self.br_int, VifPort(), details)
+ self.br_int, vif, details)
class ARPSpoofOFCtlTestCase(_ARPSpoofTestCase, _OVSAgentOFCtlTestBase):
Code Review / openstack-build / neutron-build.git / commitdiff