--- /dev/null
+# Copyright (c) 2015 Thales Services SAS
+#
+# 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 fixtures
+import netaddr
+
+from neutron.agent.linux import ip_lib
+from neutron.common import constants as n_const
+from neutron.openstack.common import uuidutils
+from neutron.tests.common import base
+from neutron.tests import sub_base
+from neutron.tests import tools
+
+NS_PREFIX = 'func-'
+BR_PREFIX = 'test-br'
+PORT_PREFIX = 'test-port'
+VETH0_PREFIX = 'test-veth0'
+VETH1_PREFIX = 'test-veth1'
+
+
+def get_rand_port_name():
+ return sub_base.get_rand_name(max_length=n_const.DEVICE_NAME_MAX_LEN,
+ prefix=PORT_PREFIX)
+
+
+def increment_ip_cidr(ip_cidr, offset=1):
+ """Increment ip_cidr offset times.
+
+ example: increment_ip_cidr("1.2.3.4/24", 2) ==> "1.2.3.6/24"
+ """
+ net0 = netaddr.IPNetwork(ip_cidr)
+ net = netaddr.IPNetwork(ip_cidr)
+ net.value += offset
+ if not net0.network < net.ip < net0.broadcast:
+ tools.fail(
+ 'Incorrect ip_cidr,offset tuple (%s,%s): "incremented" ip_cidr is '
+ 'outside ip_cidr' % (ip_cidr, offset))
+ return str(net)
+
+
+def set_namespace_gateway(port_dev, gateway_ip):
+ """Set gateway for the namespace associated to the port."""
+ if not port_dev.namespace:
+ tools.fail('tests should not change test machine gateway')
+ port_dev.route.add_gateway(gateway_ip)
+
+
+class NamespaceFixture(fixtures.Fixture):
+ """Create a namespace.
+
+ :ivar ip_wrapper: created namespace
+ :type ip_wrapper: IPWrapper
+ """
+
+ def __init__(self, prefix=NS_PREFIX):
+ super(NamespaceFixture, self).__init__()
+ self.prefix = prefix
+
+ def setUp(self):
+ super(NamespaceFixture, self).setUp()
+ ip = ip_lib.IPWrapper()
+ self.name = self.prefix + uuidutils.generate_uuid()
+ self.ip_wrapper = ip.ensure_namespace(self.name)
+ self.addCleanup(self.destroy)
+
+ def destroy(self):
+ if self.ip_wrapper.netns.exists(self.name):
+ self.ip_wrapper.netns.delete(self.name)
+
+
+class VethFixture(fixtures.Fixture):
+ """Create a veth.
+
+ :ivar ports: created veth ports
+ :type ports: IPDevice 2-uplet
+ """
+
+ def setUp(self):
+ super(VethFixture, self).setUp()
+ ip_wrapper = ip_lib.IPWrapper()
+
+ def _create_veth(name0):
+ name1 = name0.replace(VETH0_PREFIX, VETH1_PREFIX)
+ return ip_wrapper.add_veth(name0, name1)
+
+ self.ports = base.create_resource(VETH0_PREFIX, _create_veth)
+ self.addCleanup(self.destroy)
+
+ def destroy(self):
+ for port in self.ports:
+ ip_wrapper = ip_lib.IPWrapper(port.namespace)
+ try:
+ ip_wrapper.del_veth(port.name)
+ break
+ except RuntimeError:
+ # NOTE(cbrandily): It seems a veth is automagically deleted
+ # when a namespace owning a veth endpoint is deleted.
+ pass
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.common import base
+from neutron.tests.common import net_helpers
from neutron.tests.functional import base as functional_base
from neutron.tests import sub_base
% {'value': MARK_VALUE, 'mask': MARK_MASK})
MARKED_BLOCK_RULE = '-m mark --mark %s -j DROP' % MARK_VALUE
ICMP_BLOCK_RULE = '-p icmp -j DROP'
-VETH_PREFIX = 'tst-vth'
-NS_PREFIX = 'func-'
#TODO(jschwarz): Move these two functions to neutron/tests/common/
get_rand_name = sub_base.get_rand_name
-def get_rand_veth_name():
- return get_rand_name(max_length=n_const.DEVICE_NAME_MAX_LEN,
- prefix=VETH_PREFIX)
-
-
-def get_rand_port_name():
- return get_rand_name(prefix=PORT_PREFIX,
- max_length=n_const.DEVICE_NAME_MAX_LEN)
-
-
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 setUp(self):
- super(BaseLinuxTestCase, self).setUp()
-
- @staticmethod
- def _cleanup_namespace(namespace):
- if namespace.netns.exists(namespace.namespace):
- namespace.netns.delete(namespace.namespace)
-
- def _create_namespace(self, prefix=NS_PREFIX):
- ip_cmd = ip_lib.IPWrapper()
- name = prefix + uuidutils.generate_uuid()
- namespace = ip_cmd.ensure_namespace(name)
- self.addCleanup(BaseLinuxTestCase._cleanup_namespace, namespace)
-
- return namespace
+ def _create_namespace(self, prefix=net_helpers.NS_PREFIX):
+ return self.useFixture(net_helpers.NamespaceFixture(prefix)).ip_wrapper
def create_veth(self):
- ip_wrapper = ip_lib.IPWrapper()
- name1 = get_rand_veth_name()
- name2 = get_rand_veth_name()
-
- # NOTE(cbrandily): will be removed in follow-up change
- def destroy():
- try:
- ip_wrapper.del_veth(name1)
- except RuntimeError:
- # NOTE(cbrandily): It seems a veth is automagically deleted
- # when a namespace owning a veth endpoint is deleted.
- pass
-
- self.addCleanup(destroy)
- veth1, veth2 = ip_wrapper.add_veth(name1, name2)
- return veth1, veth2
-
- def set_namespace_gateway(self, port_dev, gateway_ip):
- """Set gateway for the namespace associated to the port."""
- if not port_dev.namespace:
- self.fail('tests should not change test machine gateway')
- port_dev.route.add_gateway(gateway_ip)
-
- def shift_ip_cidr(self, ip_cidr, offset=1):
- """Shift ip_cidr offset times.
-
- example: shift_ip_cidr("1.2.3.4/24", 2) ==> "1.2.3.6/24"
- """
- net = netaddr.IPNetwork(ip_cidr)
- net.value += offset
- return str(net)
+ return self.useFixture(net_helpers.VethFixture()).ports
# Regarding MRO, it goes BaseOVSLinuxTestCase, WithScenarios,
device.addr.add(cidr)
device.link.set_up()
- def prepare_veth_pairs(self, src_ns_prefix=NS_PREFIX,
- dst_ns_prefix=NS_PREFIX):
+ def prepare_veth_pairs(self, src_ns_prefix=net_helpers.NS_PREFIX,
+ dst_ns_prefix=net_helpers.NS_PREFIX):
src_addr = self.SRC_ADDRESS
dst_addr = self.DST_ADDRESS
from neutron.common import utils as common_utils
from neutron.openstack.common import uuidutils
from neutron.services import advanced_service as adv_svc
+from neutron.tests.common import net_helpers
from neutron.tests.functional.agent.linux import base
from neutron.tests.functional.agent.linux import helpers
from neutron.tests.unit import test_l3_agent
# Create and configure client namespace
client_ns = self._create_namespace()
router_ip_cidr = router.internal_ports[0]['ip_cidr']
- ip_cidr = self.shift_ip_cidr(router_ip_cidr)
+ 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)
- self.set_namespace_gateway(port, router_ip_cidr.partition('/')[0])
+ net_helpers.set_namespace_gateway(port,
+ router_ip_cidr.partition('/')[0])
# Query metadata proxy
url = 'http://%(host)s:%(port)s' % {'host': dhcp.METADATA_DEFAULT_IP,
import collections
from neutron.agent.linux import ovs_lib
+from neutron.tests.common import net_helpers
from neutron.tests.functional.agent.linux import base
# Convert ((a, b), (c, d)) to {a: b, c: d} and add 'type' by default
attrs = collections.OrderedDict(interface_attrs)
attrs.setdefault('type', 'internal')
- port_name = base.get_rand_port_name()
+ port_name = net_helpers.get_rand_port_name()
return (port_name, self.br.add_port(port_name, *attrs.items()))
def create_ovs_vif_port(self, iface_id=None, mac=None,
self.assertFalse(self.br.port_exists(port_name))
def test_replace_port(self):
- port_name = base.get_rand_port_name()
+ port_name = net_helpers.get_rand_port_name()
self.br.replace_port(port_name, ('type', 'internal'))
self.assertTrue(self.br.port_exists(port_name))
self.assertEqual('internal',
'remote_ip': '192.0.2.1', # RFC 5737 TEST-NET-1
'local_ip': '198.51.100.1', # RFC 5737 TEST-NET-2
}
- port_name = base.get_rand_port_name()
+ port_name = net_helpers.get_rand_port_name()
self.br.add_tunnel_port(port_name, attrs['remote_ip'],
attrs['local_ip'])
self.assertEqual(self.ovs.db_get_val('Interface', port_name, 'type'),
self.assertEqual(val, options[attr])
def test_add_patch_port(self):
- local = base.get_rand_port_name()
+ local = net_helpers.get_rand_port_name()
peer = 'remotepeer'
self.br.add_patch_port(local, peer)
self.assertEqual(self.ovs.db_get_val('Interface', local, 'type'),
class OVSLibTestCase(base.BaseOVSLinuxTestCase):
def test_bridge_lifecycle_baseovs(self):
- name = base.get_rand_name(prefix=base.BR_PREFIX)
+ name = base.get_rand_name(prefix=net_helpers.BR_PREFIX)
self.addCleanup(self.ovs.delete_bridge, name)
br = self.ovs.add_bridge(name)
self.assertEqual(br.br_name, name)
self.assertTrue(set(self.ovs.get_bridges()).issuperset(bridges))
def test_bridge_lifecycle_ovsbridge(self):
- name = base.get_rand_name(prefix=base.BR_PREFIX)
+ name = base.get_rand_name(prefix=net_helpers.BR_PREFIX)
br = ovs_lib.OVSBridge(name)
self.assertEqual(br.br_name, name)
# Make sure that instantiating an OVSBridge does not actually create
Code Review / openstack-build / neutron-build.git / commitdiff