Merge "Transform BaseLinuxTestCase methods in helpers"
This commit is contained in:
Jenkins
Gerrit Code Review
commit
c699515a19
|
|
@ -0,0 +1,110 @@
|
|||
# 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
|
||||
|
|
@ -19,8 +19,8 @@ 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.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
|
||||
|
||||
|
|
@ -33,24 +33,12 @@ ICMP_MARK_RULE = ('-j MARK --set-xmark %(value)s/%(mask)s'
|
|||
% {'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)
|
||||
|
|
@ -58,54 +46,11 @@ def get_rand_bridge_name():
|
|||
|
||||
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,
|
||||
|
|
@ -160,8 +105,8 @@ class BaseIPVethTestCase(BaseLinuxTestCase):
|
|||
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
|
||||
|
|
|
|||
|
|
@ -40,6 +40,7 @@ from neutron.common import constants as l3_constants
|
|||
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
|
||||
|
|
@ -630,10 +631,11 @@ class MetadataL3AgentTestCase(L3AgentTestFramework):
|
|||
# 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,
|
||||
|
|
|
|||
|
|
@ -16,6 +16,7 @@
|
|||
import collections
|
||||
|
||||
from neutron.agent.linux import ovs_lib
|
||||
from neutron.tests.common import net_helpers
|
||||
from neutron.tests.functional.agent.linux import base
|
||||
|
||||
|
||||
|
|
@ -30,7 +31,7 @@ class OVSBridgeTestCase(base.BaseOVSLinuxTestCase):
|
|||
# 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,
|
||||
|
|
@ -65,7 +66,7 @@ class OVSBridgeTestCase(base.BaseOVSLinuxTestCase):
|
|||
self.assertRaises(RuntimeError, cmd.execute, check_error=True)
|
||||
|
||||
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',
|
||||
|
|
@ -128,7 +129,7 @@ class OVSBridgeTestCase(base.BaseOVSLinuxTestCase):
|
|||
'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'),
|
||||
|
|
@ -138,7 +139,7 @@ class OVSBridgeTestCase(base.BaseOVSLinuxTestCase):
|
|||
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'),
|
||||
|
|
@ -216,7 +217,7 @@ class OVSBridgeTestCase(base.BaseOVSLinuxTestCase):
|
|||
|
||||
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)
|
||||
|
|
@ -229,7 +230,7 @@ class OVSLibTestCase(base.BaseOVSLinuxTestCase):
|
|||
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
|
||||
|
|
|
|||
Loading…
Reference in New Issue