# under the License.
#
+import abc
+
import fixtures
import netaddr
+import six
+from neutron.agent.linux import bridge_lib
from neutron.agent.linux import ip_lib
+from neutron.agent.linux import ovs_lib
from neutron.common import constants as n_const
from neutron.openstack.common import uuidutils
from neutron.tests import base as tests_base
:ivar ip_wrapper: created namespace
:type ip_wrapper: IPWrapper
+ :ivar name: created namespace name
+ :type name: str
"""
def __init__(self, prefix=NS_PREFIX):
# NOTE(cbrandily): It seems a veth is automagically deleted
# when a namespace owning a veth endpoint is deleted.
pass
+
+
+@six.add_metaclass(abc.ABCMeta)
+class PortFixture(fixtures.Fixture):
+ """Create a port.
+
+ :ivar port: created port
+ :type port: IPDevice
+ :ivar bridge: port bridge
+ """
+
+ def __init__(self, bridge=None, namespace=None):
+ self.bridge = bridge
+ self.namespace = namespace
+
+ @abc.abstractmethod
+ def _create_bridge_fixture(self):
+ pass
+
+ @abc.abstractmethod
+ def setUp(self):
+ super(PortFixture, self).setUp()
+ if not self.bridge:
+ self.bridge = self.useFixture(self._create_bridge_fixture()).bridge
+
+
+class OVSBridgeFixture(fixtures.Fixture):
+ """Create an OVS bridge.
+
+ :ivar bridge: created bridge
+ :type bridge: OVSBridge
+ """
+
+ def setUp(self):
+ super(OVSBridgeFixture, self).setUp()
+ ovs = ovs_lib.BaseOVS()
+ self.bridge = common_base.create_resource(BR_PREFIX, ovs.add_bridge)
+ self.addCleanup(self.bridge.destroy)
+
+
+class OVSPortFixture(PortFixture):
+
+ 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)
+ 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.bridge.add_port(name, ('type', 'internal'))
+ return name
+
+
+class LinuxBridgeFixture(fixtures.Fixture):
+ """Create a linux bridge.
+
+ :ivar bridge: created bridge
+ :type bridge: BridgeDevice
+ :ivar namespace: created bridge namespace
+ :type namespace: str
+ """
+
+ def setUp(self):
+ super(LinuxBridgeFixture, self).setUp()
+
+ self.namespace = self.useFixture(NamespaceFixture()).name
+ self.bridge = common_base.create_resource(
+ BR_PREFIX,
+ bridge_lib.BridgeDevice.addbr,
+ namespace=self.namespace)
+ self.addCleanup(self.bridge.delbr)
+ self.bridge.link.set_up()
+ self.addCleanup(self.bridge.link.set_down)
+
+
+class LinuxBridgePortFixture(PortFixture):
+ """Create a linux bridge port.
+
+ :ivar port: created port
+ :type port: IPDevice
+ :ivar br_port: bridge side veth peer port
+ :type br_port: IPDevice
+ """
+
+ def _create_bridge_fixture(self):
+ return LinuxBridgeFixture()
+
+ def setUp(self):
+ super(LinuxBridgePortFixture, self).setUp()
+ self.port, self.br_port = self.useFixture(VethFixture()).ports
+
+ # bridge side
+ br_ip_wrapper = ip_lib.IPWrapper(self.bridge.namespace)
+ br_ip_wrapper.add_device_to_namespace(self.br_port)
+ self.bridge.addif(self.br_port)
+ self.br_port.link.set_up()
+
+ # port side
+ ns_ip_wrapper = ip_lib.IPWrapper(self.namespace)
+ ns_ip_wrapper.add_device_to_namespace(self.port)
+ self.port.link.set_up()
+
+
+class VethBridge(object):
+
+ def __init__(self, ports):
+ self.ports = ports
+ self.unallocated_ports = set(self.ports)
+
+ def allocate_port(self):
+ try:
+ return self.unallocated_ports.pop()
+ except KeyError:
+ tools.fail('All FakeBridge ports (%s) are already allocated.' %
+ len(self.ports))
+
+
+class VethBridgeFixture(fixtures.Fixture):
+ """Simulate a bridge with a veth.
+
+ :ivar bridge: created bridge
+ :type bridge: FakeBridge
+ """
+
+ def setUp(self):
+ super(VethBridgeFixture, self).setUp()
+ ports = self.useFixture(VethFixture()).ports
+ self.bridge = VethBridge(ports)
+
+
+class VethPortFixture(PortFixture):
+ """Create a veth bridge port.
+
+ :ivar port: created port
+ :type port: IPDevice
+ """
+
+ def _create_bridge_fixture(self):
+ return VethBridgeFixture()
+
+ def setUp(self):
+ super(VethPortFixture, self).setUp()
+ self.port = self.bridge.allocate_port()
+
+ ns_ip_wrapper = ip_lib.IPWrapper(self.namespace)
+ ns_ip_wrapper.add_device_to_namespace(self.port)
+ self.port.link.set_up()
# License for the specific language governing permissions and limitations
# under the License.
-import netaddr
import testscenarios
-from neutron.agent.common import ovs_lib
-from neutron.agent.linux import bridge_lib
-from neutron.agent.linux import ip_lib
-from neutron.common import constants as n_const
from neutron.tests import base as tests_base
-from neutron.tests.common import base as common_base
from neutron.tests.common import net_helpers
from neutron.tests.functional import base as functional_base
-BR_PREFIX = 'test-br'
-PORT_PREFIX = 'test-port'
MARK_VALUE = '0x1'
MARK_MASK = '0xffffffff'
ICMP_MARK_RULE = ('-j MARK --set-xmark %(value)s/%(mask)s'
get_rand_name = tests_base.get_rand_name
-def get_rand_bridge_name():
- return get_rand_name(prefix=BR_PREFIX,
- max_length=n_const.DEVICE_NAME_MAX_LEN)
-
-
class BaseLinuxTestCase(functional_base.BaseSudoTestCase):
def _create_namespace(self, prefix=net_helpers.NS_PREFIX):
def setUp(self):
super(BaseOVSLinuxTestCase, self).setUp()
self.config(group='OVS', ovsdb_interface=self.ovsdb_interface)
- self.ovs = ovs_lib.BaseOVS()
- self.ip = ip_lib.IPWrapper()
-
- def create_ovs_bridge(self, br_prefix=BR_PREFIX):
- br = common_base.create_resource(br_prefix, self.ovs.add_bridge)
- self.addCleanup(br.destroy)
- return br
-
- def create_ovs_port_in_ns(self, br, ns):
- def create_port(name):
- br.replace_port(name, ('type', 'internal'))
- self.addCleanup(br.delete_port, name)
- return name
- port_name = common_base.create_resource(PORT_PREFIX, create_port)
- port_dev = self.ip.device(port_name)
- ns.add_device_to_namespace(port_dev)
- port_dev.link.set_up()
- return port_dev
-
- def bind_namespace_to_cidr(self, namespace, br, ip_cidr):
- """Bind namespace to cidr (on layer2 and 3).
-
- Bind the namespace to a subnet by creating an ovs port in the namespace
- and configuring port ip.
- """
- net = netaddr.IPNetwork(ip_cidr)
- port_dev = self.create_ovs_port_in_ns(br, namespace)
- port_dev.addr.add(str(net))
- return port_dev
class BaseIPVethTestCase(BaseLinuxTestCase):
device.addr.add(cidr)
device.link.set_up()
- def prepare_veth_pairs(self, src_ns_prefix=net_helpers.NS_PREFIX,
- dst_ns_prefix=net_helpers.NS_PREFIX):
+ def prepare_veth_pairs(self):
src_addr = self.SRC_ADDRESS
dst_addr = self.DST_ADDRESS
src_veth, dst_veth = self.create_veth()
- src_ns = self._create_namespace(src_ns_prefix)
- dst_ns = self._create_namespace(dst_ns_prefix)
+ src_ns = self._create_namespace()
+ dst_ns = self._create_namespace()
src_ns.add_device_to_namespace(src_veth)
dst_ns.add_device_to_namespace(dst_veth)
self._set_ip_up(dst_veth, '%s/24' % dst_addr)
return src_ns, dst_ns
-
-
-class BaseBridgeTestCase(BaseIPVethTestCase):
-
- def create_veth_pairs(self, dst_namespace):
- src_ns = self._create_namespace()
- dst_ns = ip_lib.IPWrapper(dst_namespace)
-
- src_veth, dst_veth = self.create_veth()
- src_ns.add_device_to_namespace(src_veth)
- dst_ns.add_device_to_namespace(dst_veth)
-
- return src_veth, dst_veth
-
- def create_bridge(self, br_ns=None):
- br_ns = br_ns or self._create_namespace()
- br_name = get_rand_bridge_name()
- bridge = bridge_lib.BridgeDevice.addbr(br_name, br_ns.namespace)
- self.addCleanup(bridge.delbr)
- bridge.link.set_up()
- self.addCleanup(bridge.link.set_down)
- return bridge
# License for the specific language governing permissions and limitations
# under the License.
-from neutron.agent.linux import ip_lib
from neutron.agent.linux import iptables_firewall
from neutron.agent import securitygroups_rpc as sg_cfg
+from neutron.tests.common import net_helpers
from neutron.tests.functional.agent.linux import base
from neutron.tests.functional.agent.linux import helpers
from oslo_config import cfg
-class IptablesFirewallTestCase(base.BaseBridgeTestCase):
+class IptablesFirewallTestCase(base.BaseIPVethTestCase):
MAC_REAL = "fa:16:3e:9a:2f:49"
MAC_SPOOFED = "fa:16:3e:9a:2f:48"
FAKE_SECURITY_GROUP_ID = "fake_sg_id"
def _set_src_mac(self, mac):
- self.src_veth.link.set_down()
- self.src_veth.link.set_address(mac)
- self.src_veth.link.set_up()
+ self.src_port.link.set_down()
+ self.src_port.link.set_address(mac)
+ self.src_port.link.set_up()
def setUp(self):
cfg.CONF.register_opts(sg_cfg.security_group_opts, 'SECURITYGROUP')
super(IptablesFirewallTestCase, self).setUp()
- self.bridge = self.create_bridge()
+ bridge = self.useFixture(net_helpers.LinuxBridgeFixture()).bridge
- self.src_veth, self.src_br_veth = self.create_veth_pairs(
- self.bridge.namespace)
- self.bridge.addif(self.src_br_veth.name)
- self._set_ip_up(self.src_veth, '%s/24' % self.SRC_ADDRESS)
- self.src_br_veth.link.set_up()
+ # FIXME(cbrandily): temporary, will be replaced by fake machines
+ self.src_ip_wrapper = self.useFixture(
+ net_helpers.NamespaceFixture()).ip_wrapper
- self.dst_veth, self.dst_br_veth = self.create_veth_pairs(
- self.bridge.namespace)
- self.bridge.addif(self.dst_br_veth.name)
- self._set_ip_up(self.dst_veth, '%s/24' % self.DST_ADDRESS)
- self.dst_br_veth.link.set_up()
+ src_port_fixture = self.useFixture(
+ net_helpers.LinuxBridgePortFixture(
+ bridge, self.src_ip_wrapper.namespace))
+ self.src_port = src_port_fixture.port
+ self._set_ip_up(self.src_port, '%s/24' % self.SRC_ADDRESS)
+
+ self.dst_ip_wrapper = self.useFixture(
+ net_helpers.NamespaceFixture()).ip_wrapper
+ self.dst_port = self.useFixture(
+ net_helpers.LinuxBridgePortFixture(
+ bridge, self.dst_ip_wrapper.namespace)).port
+ self._set_ip_up(self.dst_port, '%s/24' % self.DST_ADDRESS)
self.firewall = iptables_firewall.IptablesFirewallDriver(
- namespace=self.bridge.namespace)
+ namespace=bridge.namespace)
self._set_src_mac(self.MAC_REAL)
- self.src_port = {'admin_state_up': True,
- 'device': self.src_br_veth.name,
- 'device_owner': 'compute:None',
- 'fixed_ips': [self.SRC_ADDRESS],
- 'mac_address': self.MAC_REAL,
- 'port_security_enabled': True,
- 'security_groups': [self.FAKE_SECURITY_GROUP_ID],
- 'status': 'ACTIVE'}
+ self.src_port_desc = {'admin_state_up': True,
+ 'device': src_port_fixture.br_port.name,
+ 'device_owner': 'compute:None',
+ 'fixed_ips': [self.SRC_ADDRESS],
+ 'mac_address': self.MAC_REAL,
+ 'port_security_enabled': True,
+ 'security_groups': [self.FAKE_SECURITY_GROUP_ID],
+ 'status': 'ACTIVE'}
# setup firewall on bridge and send packet from src_veth and observe
# if sent packet can be observed on dst_veth
def test_port_sec_within_firewall(self):
- pinger = helpers.Pinger(ip_lib.IPWrapper(self.src_veth.namespace))
+ pinger = helpers.Pinger(self.src_ip_wrapper)
# update the sg_group to make ping pass
sg_rules = [{'ethertype': 'IPv4', 'direction': 'ingress',
self.firewall.update_security_group_rules(
self.FAKE_SECURITY_GROUP_ID,
sg_rules)
- self.firewall.prepare_port_filter(self.src_port)
+ self.firewall.prepare_port_filter(self.src_port_desc)
pinger.assert_ping(self.DST_ADDRESS)
# modify the src_veth's MAC and test again
pinger.assert_no_ping(self.DST_ADDRESS)
# update the port's port_security_enabled value and test again
- self.src_port['port_security_enabled'] = False
- self.firewall.update_port_filter(self.src_port)
+ self.src_port_desc['port_security_enabled'] = False
+ self.firewall.update_port_filter(self.src_port_desc)
pinger.assert_ping(self.DST_ADDRESS)
from neutron.agent.linux import ovsdb_monitor
from neutron.agent.linux import utils
+from neutron.tests.common import net_helpers
from neutron.tests.functional.agent.linux import base as linux_base
from neutron.tests.functional import base as functional_base
root_helper=" ".join([functional_base.SUDO_CMD] * 2))
self._check_test_requirements()
- self.bridge = self.create_ovs_bridge()
def _check_test_requirements(self):
self.check_command(['ovsdb-client', 'list-dbs'],
'Initial call should always be true')
self.assertFalse(self.monitor.has_updates,
'has_updates without port addition should be False')
- self.create_ovs_port_in_ns(self.bridge, self.ip)
+ self.useFixture(net_helpers.OVSPortFixture())
# has_updates after port addition should become True
while not self.monitor.has_updates:
eventlet.sleep(0.01)
'neutron.agent.linux.interface.OVSInterfaceDriver')
conf.set_override('router_delete_namespaces', True)
- br_int = self.create_ovs_bridge()
- br_ex = self.create_ovs_bridge()
+ br_int = self.useFixture(net_helpers.OVSBridgeFixture()).bridge
+ br_ex = self.useFixture(net_helpers.OVSBridgeFixture()).bridge
conf.set_override('ovs_integration_bridge', br_int.br_name)
conf.set_override('external_network_bridge', br_ex.br_name)
self._add_fip(router, dst_fip, fixed_address=dst_ip)
router.process(self.agent)
- src_ns = self._create_namespace(prefix='test-src-')
- dst_ns = self._create_namespace(prefix='test-dst-')
br_int = get_ovs_bridge(self.agent.conf.ovs_integration_bridge)
- src_port = self.bind_namespace_to_cidr(src_ns, br_int, src_ip_cidr)
+
+ # FIXME(cbrandily): temporary, will be replaced by fake machines
+ src_ns = self._create_namespace(prefix='test-src-')
+ src_port = self.useFixture(
+ net_helpers.OVSPortFixture(br_int, src_ns.namespace)).port
+ src_port.addr.add(src_ip_cidr)
net_helpers.set_namespace_gateway(src_port, router_ip)
- dst_port = self.bind_namespace_to_cidr(dst_ns, br_int, dst_ip_cidr)
+ dst_ns = self._create_namespace(prefix='test-dst-')
+ dst_port = self.useFixture(
+ net_helpers.OVSPortFixture(br_int, dst_ns.namespace)).port
+ dst_port.addr.add(dst_ip_cidr)
net_helpers.set_namespace_gateway(dst_port, router_ip)
protocol_port = helpers.get_free_namespace_port(dst_ns)
router_ip_cidr = self._port_first_ip_cidr(router.internal_ports[0])
ip_cidr = net_helpers.increment_ip_cidr(router_ip_cidr)
br_int = get_ovs_bridge(self.agent.conf.ovs_integration_bridge)
- port = self.bind_namespace_to_cidr(client_ns, br_int, ip_cidr)
+
+ # FIXME(cbrandily): temporary, will be replaced by a fake machine
+ port = self.useFixture(
+ net_helpers.OVSPortFixture(br_int, client_ns.namespace)).port
+ port.addr.add(ip_cidr)
net_helpers.set_namespace_gateway(port,
router_ip_cidr.partition('/')[0])
import collections
from neutron.agent.common import ovs_lib
+from neutron.agent.linux import ip_lib
from neutron.tests.common import net_helpers
from neutron.tests.functional.agent.linux import base
# good to also add the openflow-related functions
def setUp(self):
super(OVSBridgeTestCase, self).setUp()
- self.br = self.create_ovs_bridge()
+ self.ovs = ovs_lib.BaseOVS()
+ self.br = self.useFixture(net_helpers.OVSBridgeFixture()).bridge
def create_ovs_port(self, *interface_attrs):
# Convert ((a, b), (c, d)) to {a: b, c: d} and add 'type' by default
"OpenFlow10")
def test_get_datapath_id(self):
- brdev = self.ip.device(self.br.br_name)
+ brdev = ip_lib.IPDevice(self.br.br_name)
dpid = brdev.link.attributes['link/ether'].replace(':', '')
self.br.set_db_attribute('Bridge',
self.br.br_name, 'datapath_id', dpid)
class OVSLibTestCase(base.BaseOVSLinuxTestCase):
+
+ def setUp(self):
+ super(OVSLibTestCase, self).setUp()
+ self.ovs = ovs_lib.BaseOVS()
+
def test_bridge_lifecycle_baseovs(self):
name = base.get_rand_name(prefix=net_helpers.BR_PREFIX)
self.addCleanup(self.ovs.delete_bridge, name)
self.assertFalse(self.ovs.bridge_exists(name))
def test_get_bridges(self):
- bridges = {self.create_ovs_bridge().br_name for i in range(5)}
+ bridges = {
+ self.useFixture(net_helpers.OVSBridgeFixture()).bridge.br_name
+ for i in range(5)}
self.assertTrue(set(self.ovs.get_bridges()).issuperset(bridges))
def test_bridge_lifecycle_ovsbridge(self):
from neutron.agent.linux import dhcp
from neutron.agent.linux import ip_lib
from neutron.cmd import netns_cleanup
-from neutron.tests.functional.agent.linux import base
+from neutron.tests.common import net_helpers
+from neutron.tests.functional import base
GET_NAMESPACES = 'neutron.agent.linux.ip_lib.IPWrapper.get_namespaces'
TEST_INTERFACE_DRIVER = 'neutron.agent.linux.interface.OVSInterfaceDriver'
-class NetnsCleanupTest(base.BaseIPVethTestCase):
+class NetnsCleanupTest(base.BaseSudoTestCase):
def setUp(self):
super(NetnsCleanupTest, self).setUp()
self.config_parse(conf=self.conf, args=args)
def test_cleanup_network_namespaces_cleans_dhcp_and_l3_namespaces(self):
- l3_ns, dhcp_ns = self.prepare_veth_pairs(l3_agent.NS_PREFIX,
- dhcp.NS_PREFIX)
+ dhcp_namespace = self.useFixture(
+ net_helpers.NamespaceFixture(dhcp.NS_PREFIX)).name
+ l3_namespace = self.useFixture(
+ net_helpers.NamespaceFixture(l3_agent.NS_PREFIX)).name
+ bridge = self.useFixture(
+ net_helpers.VethPortFixture(namespace=dhcp_namespace)).bridge
+ self.useFixture(
+ net_helpers.VethPortFixture(bridge, l3_namespace))
+
# we scope the get_namespaces to our own ones not to affect other
# tests, as otherwise cleanup will kill them all
- self.get_namespaces.return_value = [l3_ns.namespace,
- dhcp_ns.namespace]
+ self.get_namespaces.return_value = [l3_namespace, dhcp_namespace]
netns_cleanup.cleanup_network_namespaces(self.conf)
self.get_namespaces_p.stop()
namespaces_now = ip_lib.IPWrapper.get_namespaces()
- self.assertNotIn(l3_ns.namespace, namespaces_now)
- self.assertNotIn(dhcp_ns.namespace, namespaces_now)
+ self.assertNotIn(l3_namespace, namespaces_now)
+ self.assertNotIn(dhcp_namespace, namespaces_now)
Code Review / openstack-build / neutron-build.git / commitdiff