Merge "Use constant IP_VERSION_4/6 in unit tests"

This commit is contained in:
Zuul 2018-08-31 19:20:35 +00:00 committed by Gerrit Code Review
commit a905fb30fa
35 changed files with 432 additions and 333 deletions

View File

@ -56,7 +56,8 @@ fake_subnet1 = dhcp.DictModel(dict(id='bbbbbbbb-bbbb-bbbb-bbbbbbbbbbbb',
cidr='172.9.9.0/24', enable_dhcp=True, name='',
tenant_id=FAKE_TENANT_ID,
gateway_ip='172.9.9.1', host_routes=[],
dns_nameservers=[], ip_version=4,
dns_nameservers=[],
ip_version=const.IP_VERSION_4,
ipv6_ra_mode=None, ipv6_address_mode=None,
allocation_pools=fake_subnet1_allocation_pools))
@ -66,26 +67,29 @@ fake_subnet2 = dhcp.DictModel(dict(id='dddddddd-dddd-dddd-dddddddddddd',
network_id=FAKE_NETWORK_UUID,
cidr='172.9.8.0/24', enable_dhcp=False, name='',
tenant_id=FAKE_TENANT_ID, gateway_ip='172.9.8.1',
host_routes=[], dns_nameservers=[], ip_version=4,
host_routes=[], dns_nameservers=[],
ip_version=const.IP_VERSION_4,
allocation_pools=fake_subnet2_allocation_pools))
fake_subnet3 = dhcp.DictModel(dict(id='bbbbbbbb-1111-2222-bbbbbbbbbbbb',
network_id=FAKE_NETWORK_UUID,
cidr='192.168.1.1/24', enable_dhcp=True,
ip_version=4))
ip_version=const.IP_VERSION_4))
fake_ipv6_subnet = dhcp.DictModel(dict(id='bbbbbbbb-1111-2222-bbbbbbbbbbbb',
network_id=FAKE_NETWORK_UUID,
cidr='2001:0db8::0/64', enable_dhcp=True,
tenant_id=FAKE_TENANT_ID,
gateway_ip='2001:0db8::1', ip_version=6,
gateway_ip='2001:0db8::1',
ip_version=const.IP_VERSION_6,
ipv6_ra_mode='slaac', ipv6_address_mode=None))
fake_meta_subnet = dhcp.DictModel(dict(id='bbbbbbbb-1111-2222-bbbbbbbbbbbb',
network_id=FAKE_NETWORK_UUID,
cidr='169.254.169.252/30',
gateway_ip='169.254.169.253',
enable_dhcp=True, ip_version=4))
enable_dhcp=True,
ip_version=const.IP_VERSION_4))
fake_fixed_ip1 = dhcp.DictModel(dict(id='', subnet_id=fake_subnet1.id,
ip_address='172.9.9.9'))
@ -946,7 +950,7 @@ class TestDhcpAgentEventHandler(base.BaseTestCase):
def test_subnet_create_restarts_with_dhcp_disabled(self):
payload = dict(subnet=dhcp.DictModel(
dict(network_id=fake_network.id, enable_dhcp=False,
cidr='99.99.99.0/24', ip_version=4)))
cidr='99.99.99.0/24', ip_version=const.IP_VERSION_4)))
self.cache.get_network_by_id.return_value = fake_network
new_net = copy.deepcopy(fake_network)
new_net.subnets.append(payload['subnet'])
@ -1389,7 +1393,7 @@ class FakePort2(object):
class FakeV4Subnet(object):
def __init__(self):
self.id = 'dddddddd-dddd-dddd-dddd-dddddddddddd'
self.ip_version = 4
self.ip_version = const.IP_VERSION_4
self.cidr = '192.168.0.0/24'
self.gateway_ip = '192.168.0.1'
self.enable_dhcp = True
@ -1398,7 +1402,7 @@ class FakeV4Subnet(object):
class FakeV6Subnet(object):
def __init__(self):
self.id = 'ffffffff-ffff-ffff-ffff-ffffffffffff'
self.ip_version = 6
self.ip_version = const.IP_VERSION_6
self.cidr = '2001:db8:0:1::/64'
self.gateway_ip = '2001:db8:0:1::1'
self.enable_dhcp = True
@ -1419,7 +1423,7 @@ class FakeV6SubnetOutsideGateway(FakeV6Subnet):
class FakeV4SubnetNoGateway(object):
def __init__(self):
self.id = 'eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee'
self.ip_version = 4
self.ip_version = const.IP_VERSION_4
self.cidr = '192.168.1.0/24'
self.gateway_ip = None
self.enable_dhcp = True
@ -1428,7 +1432,7 @@ class FakeV4SubnetNoGateway(object):
class FakeV6SubnetNoGateway(object):
def __init__(self):
self.id = 'ffffffff-ffff-ffff-ffff-ffffffffffff'
self.ip_version = 6
self.ip_version = const.IP_VERSION_6
self.cidr = '2001:db8:1:0::/64'
self.gateway_ip = None
self.enable_dhcp = True

View File

@ -990,9 +990,9 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
@mock.patch.object(dvr_router_base.LOG, 'error')
def test_get_snat_port_for_internal_port_ipv6_same_port(self, log_error):
router = l3_test_common.prepare_router_data(ip_version=4,
enable_snat=True,
num_internal_ports=1)
router = l3_test_common.prepare_router_data(
ip_version=lib_constants.IP_VERSION_4, enable_snat=True,
num_internal_ports=1)
ri = dvr_router.DvrEdgeRouter(mock.sentinel.agent,
HOSTNAME,
router['id'],
@ -1000,8 +1000,9 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
**self.ri_kwargs)
# Add two additional IPv6 prefixes on the same interface
l3_test_common.router_append_interface(router, count=2, ip_version=6,
same_port=True)
l3_test_common.router_append_interface(
router, count=2, ip_version=lib_constants.IP_VERSION_6,
same_port=True)
internal_ports = ri.router.get(lib_constants.INTERFACE_KEY, [])
with mock.patch.object(ri, 'get_snat_interfaces') as get_interfaces:
get_interfaces.return_value = internal_ports
@ -1613,8 +1614,8 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
def _test_process_ipv6_only_or_dual_stack_gw(self, dual_stack=False):
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
router = l3_test_common.prepare_router_data(ip_version=6,
dual_stack=dual_stack)
router = l3_test_common.prepare_router_data(
ip_version=lib_constants.IP_VERSION_6, dual_stack=dual_stack)
# Get NAT rules without the gw_port
gw_port = router['gw_port']
router['gw_port'] = None
@ -1668,9 +1669,9 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
ri.process()
orig_nat_rules = ri.iptables_manager.ipv4['nat'].rules[:]
# Add an IPv6 interface and reprocess
l3_test_common.router_append_interface(router, count=1,
ip_version=6, ra_mode=ra_mode,
addr_mode=addr_mode)
l3_test_common.router_append_interface(
router, count=1, ip_version=lib_constants.IP_VERSION_6,
ra_mode=ra_mode, addr_mode=addr_mode)
# Reassign the router object to RouterInfo
self._process_router_instance_for_agent(agent, ri, router)
# IPv4 NAT rules should not be changed by adding an IPv6 interface
@ -1703,7 +1704,7 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
l3_test_common.router_append_subnet(
router,
count=len(ipv6_subnet_modes),
ip_version=6,
ip_version=lib_constants.IP_VERSION_6,
ipv6_subnet_modes=ipv6_subnet_modes,
dns_nameservers=dns_nameservers,
network_mtu=network_mtu)
@ -1775,7 +1776,7 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
# Add the first subnet on a new interface
l3_test_common.router_append_subnet(
router, count=1,
ip_version=6, ipv6_subnet_modes=[
ip_version=lib_constants.IP_VERSION_6, ipv6_subnet_modes=[
{'ra_mode': lib_constants.IPV6_SLAAC,
'address_mode': lib_constants.IPV6_SLAAC}])
self._process_router_instance_for_agent(agent, ri, router)
@ -1793,7 +1794,7 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
interface_id = router[lib_constants.INTERFACE_KEY][1]['id']
l3_test_common.router_append_subnet(
router, count=1,
ip_version=6,
ip_version=lib_constants.IP_VERSION_6,
ipv6_subnet_modes=[
{'ra_mode': lib_constants.IPV6_SLAAC,
'address_mode': lib_constants.IPV6_SLAAC}],
@ -1816,8 +1817,10 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
# Process with NAT
ri.process()
# Add an IPv4 and IPv6 interface and reprocess
l3_test_common.router_append_interface(router, count=1, ip_version=4)
l3_test_common.router_append_interface(router, count=1, ip_version=6)
l3_test_common.router_append_interface(
router, count=1, ip_version=lib_constants.IP_VERSION_4)
l3_test_common.router_append_interface(
router, count=1, ip_version=lib_constants.IP_VERSION_6)
# Reassign the router object to RouterInfo
self._process_router_instance_for_agent(agent, ri, router)
self._assert_ri_process_enabled(ri)
@ -1844,7 +1847,8 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
ri.external_gateway_added = mock.Mock()
self._process_router_instance_for_agent(agent, ri, router)
# Add an IPv6 interface and reprocess
l3_test_common.router_append_interface(router, count=1, ip_version=6)
l3_test_common.router_append_interface(
router, count=1, ip_version=lib_constants.IP_VERSION_6)
self._process_router_instance_for_agent(agent, ri, router)
self._assert_ri_process_enabled(ri)
# Reset the calls so we can check for disable radvd
@ -1863,7 +1867,7 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
self._process_router_instance_for_agent(agent, ri, router)
# Add an IPv6 interface with two subnets and reprocess
l3_test_common.router_append_subnet(
router, count=2, ip_version=6,
router, count=2, ip_version=lib_constants.IP_VERSION_6,
ipv6_subnet_modes=([{'ra_mode': lib_constants.IPV6_SLAAC,
'address_mode': lib_constants.IPV6_SLAAC}] *
2))
@ -2795,7 +2799,8 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
@mock.patch('os.geteuid', return_value='490')
@mock.patch('pwd.getpwuid', return_value=FakeUser('neutron'))
def test_spawn_radvd(self, geteuid, getpwuid):
router = l3_test_common.prepare_router_data(ip_version=6)
router = l3_test_common.prepare_router_data(
ip_version=lib_constants.IP_VERSION_6)
conffile = '/fake/radvd.conf'
pidfile = '/fake/radvd.pid'

View File

@ -53,7 +53,7 @@ class FakeDNSAssignment(object):
class DhcpOpt(object):
def __init__(self, **kwargs):
self.__dict__.update(ip_version=4)
self.__dict__.update(ip_version=constants.IP_VERSION_4)
self.__dict__.update(kwargs)
def __str__(self):
@ -188,10 +188,10 @@ class FakePort6(object):
self.extra_dhcp_opts = [
DhcpOpt(opt_name=edo_ext.DHCP_OPT_CLIENT_ID,
opt_value='test6',
ip_version=4),
ip_version=constants.IP_VERSION_4),
DhcpOpt(opt_name='dns-server',
opt_value='123.123.123.45',
ip_version=4)]
ip_version=constants.IP_VERSION_4)]
class FakeV6Port(object):
@ -224,7 +224,7 @@ class FakeV6PortExtraOpt(object):
self.extra_dhcp_opts = [
DhcpOpt(opt_name='dns-server',
opt_value='ffea:3ba5:a17a:4ba3::100',
ip_version=6)]
ip_version=constants.IP_VERSION_6)]
class FakeDualPortWithV6ExtraOpt(object):
@ -245,7 +245,7 @@ class FakeDualPortWithV6ExtraOpt(object):
self.extra_dhcp_opts = [
DhcpOpt(opt_name='dns-server',
opt_value='ffea:3ba5:a17a:4ba3::100',
ip_version=6)]
ip_version=constants.IP_VERSION_6)]
class FakeDualPort(object):
@ -391,7 +391,7 @@ class FakeV6HostRoute(object):
class FakeV4Subnet(Dictable):
def __init__(self):
self.id = 'dddddddd-dddd-dddd-dddd-dddddddddddd'
self.ip_version = 4
self.ip_version = constants.IP_VERSION_4
self.cidr = '192.168.0.0/24'
self.gateway_ip = '192.168.0.1'
self.enable_dhcp = True
@ -522,7 +522,7 @@ class FakeV4MultipleAgentsWithDnsProvided(FakeNetworkBase):
class FakeV6Subnet(object):
def __init__(self):
self.id = 'ffffffff-ffff-ffff-ffff-ffffffffffff'
self.ip_version = 6
self.ip_version = constants.IP_VERSION_6
self.cidr = 'fdca:3ba5:a17a:4ba3::/64'
self.gateway_ip = 'fdca:3ba5:a17a:4ba3::1'
self.enable_dhcp = True
@ -535,7 +535,7 @@ class FakeV6Subnet(object):
class FakeV4SubnetNoDHCP(object):
def __init__(self):
self.id = 'eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee'
self.ip_version = 4
self.ip_version = constants.IP_VERSION_4
self.cidr = '192.168.1.0/24'
self.gateway_ip = '192.168.1.1'
self.enable_dhcp = False
@ -546,7 +546,7 @@ class FakeV4SubnetNoDHCP(object):
class FakeV6SubnetDHCPStateful(Dictable):
def __init__(self):
self.id = 'ffffffff-ffff-ffff-ffff-ffffffffffff'
self.ip_version = 6
self.ip_version = constants.IP_VERSION_6
self.cidr = 'fdca:3ba5:a17a:4ba3::/64'
self.gateway_ip = 'fdca:3ba5:a17a:4ba3::1'
self.enable_dhcp = True
@ -559,7 +559,7 @@ class FakeV6SubnetDHCPStateful(Dictable):
class FakeV6SubnetSlaac(object):
def __init__(self):
self.id = 'eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee'
self.ip_version = 6
self.ip_version = constants.IP_VERSION_6
self.cidr = 'ffda:3ba5:a17a:4ba3::/64'
self.gateway_ip = 'ffda:3ba5:a17a:4ba3::1'
self.enable_dhcp = True
@ -571,7 +571,7 @@ class FakeV6SubnetSlaac(object):
class FakeV6SubnetStateless(object):
def __init__(self):
self.id = 'eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee'
self.ip_version = 6
self.ip_version = constants.IP_VERSION_6
self.cidr = 'ffea:3ba5:a17a:4ba3::/64'
self.gateway_ip = 'ffea:3ba5:a17a:4ba3::1'
self.enable_dhcp = True
@ -584,7 +584,7 @@ class FakeV6SubnetStateless(object):
class FakeV6SubnetStatelessNoDnsProvided(object):
def __init__(self):
self.id = 'eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee'
self.ip_version = 6
self.ip_version = constants.IP_VERSION_6
self.cidr = 'ffea:3ba5:a17a:4ba3::/64'
self.gateway_ip = 'ffea:3ba5:a17a:4ba3::1'
self.enable_dhcp = True
@ -597,7 +597,7 @@ class FakeV6SubnetStatelessNoDnsProvided(object):
class FakeV6SubnetStatelessBadPrefixLength(object):
def __init__(self):
self.id = 'eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee'
self.ip_version = 6
self.ip_version = constants.IP_VERSION_6
self.cidr = 'ffeb:3ba5:a17a:4ba3::/56'
self.gateway_ip = 'ffeb:3ba5:a17a:4ba3::1'
self.enable_dhcp = True
@ -909,9 +909,9 @@ class FakeV6NetworkPxePort(FakeNetworkBase):
self.namespace = 'qdhcp-ns'
self.ports[0].extra_dhcp_opts = [
DhcpOpt(opt_name='tftp-server', opt_value='2001:192:168::1',
ip_version=6),
ip_version=constants.IP_VERSION_6),
DhcpOpt(opt_name='bootfile-name', opt_value='pxelinux.0',
ip_version=6)]
ip_version=constants.IP_VERSION_6)]
class FakeV6NetworkPxePortWrongOptVersion(FakeNetworkBase):
@ -922,9 +922,9 @@ class FakeV6NetworkPxePortWrongOptVersion(FakeNetworkBase):
self.namespace = 'qdhcp-ns'
self.ports[0].extra_dhcp_opts = [
DhcpOpt(opt_name='tftp-server', opt_value='192.168.0.7',
ip_version=4),
ip_version=constants.IP_VERSION_4),
DhcpOpt(opt_name='bootfile-name', opt_value='pxelinux.0',
ip_version=6)]
ip_version=constants.IP_VERSION_6)]
class FakeDualStackNetworkSingleDHCP(FakeNetworkBase):
@ -1257,9 +1257,9 @@ class TestDnsmasq(TestBase):
prefix6 = '--dhcp-range=set:tag%d,%s,%s,%s%s'
possible_leases = 0
for i, s in enumerate(network.subnets):
if (s.ip_version != 6 or
if (s.ip_version != constants.IP_VERSION_6 or
s.ipv6_address_mode == constants.DHCPV6_STATEFUL):
if s.ip_version == 4:
if s.ip_version == constants.IP_VERSION_4:
expected.extend([prefix % (
i, s.cidr.split('/')[0],
netaddr.IPNetwork(s.cidr).netmask, lease_duration,

View File

@ -42,7 +42,7 @@ class FakeSubnet(object):
class FakeAllocation(object):
subnet = FakeSubnet()
ip_address = '192.168.1.2'
ip_version = 4
ip_version = constants.IP_VERSION_4
class FakePort(object):
@ -177,9 +177,9 @@ class TestABCDriver(TestBase):
self.ip_dev.assert_has_calls(expected_calls)
def test_init_router_port_ext_gw_with_dual_stack(self):
old_addrs = [dict(ip_version=4, scope='global',
old_addrs = [dict(ip_version=constants.IP_VERSION_4, scope='global',
dynamic=False, cidr='172.16.77.240/24'),
dict(ip_version=6, scope='global',
dict(ip_version=constants.IP_VERSION_6, scope='global',
dynamic=False, cidr='2001:db8:a::123/64')]
self.ip_dev().addr.list = mock.Mock(return_value=old_addrs)
self.ip_dev().route.list_onlink_routes.return_value = []
@ -359,7 +359,8 @@ class TestABCDriver(TestBase):
self.assertEqual(addresses, llas)
self.ip_dev.assert_has_calls(
[mock.call(device_name, namespace=ns),
mock.call().addr.list(scope='link', ip_version=6)])
mock.call().addr.list(
scope='link', ip_version=constants.IP_VERSION_6)])
def test_set_mtu_logs_once(self):
bc = BaseChild(self.conf)

View File

@ -18,6 +18,7 @@ import socket
import mock
import netaddr
from neutron_lib import constants
from neutron_lib import exceptions
import pyroute2
from pyroute2.netlink.rtnl import ifinfmsg
@ -969,7 +970,7 @@ class TestIpRouteCommand(TestIPCmdBase):
self.parent.name = 'eth0'
self.command = 'route'
self.route_cmd = ip_lib.IpRouteCommand(self.parent)
self.ip_version = 4
self.ip_version = constants.IP_VERSION_4
self.table = 14
self.metric = 100
self.cidr = '192.168.45.100/24'
@ -1159,7 +1160,7 @@ class TestIpRouteCommand(TestIPCmdBase):
class TestIPv6IpRouteCommand(TestIpRouteCommand):
def setUp(self):
super(TestIPv6IpRouteCommand, self).setUp()
self.ip_version = 6
self.ip_version = constants.IP_VERSION_6
self.cidr = '2001:db8::/64'
self.ip = '2001:db8::100'
self.gateway = '2001:db8::1'

View File

@ -538,7 +538,7 @@ class SGServerRpcCallBackTestCase(test_sg.SecurityGroupDBTestCase):
fake_gateway = FAKE_IP[const.IPv6]
with self.network() as n,\
self.subnet(n, gateway_ip=fake_gateway,
cidr=fake_prefix, ip_version=6
cidr=fake_prefix, ip_version=const.IP_VERSION_6
) as subnet_v6,\
self.security_group() as sg1:
sg1_id = sg1['security_group']['id']
@ -651,7 +651,7 @@ class SGServerRpcCallBackTestCase(test_sg.SecurityGroupDBTestCase):
fake_gateway = FAKE_IP[const.IPv6]
with self.network() as n,\
self.subnet(n, gateway_ip=fake_gateway,
cidr=fake_prefix, ip_version=6
cidr=fake_prefix, ip_version=const.IP_VERSION_6
) as subnet_v6,\
self.security_group() as sg1:
sg1_id = sg1['security_group']['id']
@ -709,7 +709,7 @@ class SGServerRpcCallBackTestCase(test_sg.SecurityGroupDBTestCase):
fake_gateway = FAKE_IP[const.IPv6]
with self.network() as n,\
self.subnet(n, gateway_ip=fake_gateway,
cidr=fake_prefix, ip_version=6
cidr=fake_prefix, ip_version=const.IP_VERSION_6
) as subnet_v6,\
self.security_group() as sg1,\
self.security_group() as sg2:

View File

@ -46,7 +46,7 @@ class TestL3RpcCallback(testlib_api.SqlTestCase):
subnet = {'subnet': {'network_id': self.network['id'],
'tenant_id': 'tenant_id',
'cidr': None,
'ip_version': 6,
'ip_version': constants.IP_VERSION_6,
'use_default_subnetpool': True,
'name': 'ipv6_pd',
'enable_dhcp': True,

View File

@ -67,7 +67,7 @@ class TestIsAutoAddressSubnet(base.BaseTestCase):
self.subnet = {
'cidr': '2001:200::/64',
'gateway_ip': '2001:200::1',
'ip_version': 6,
'ip_version': constants.IP_VERSION_6,
'ipv6_address_mode': None,
'ipv6_ra_mode': None
}

View File

@ -270,11 +270,11 @@ class TestCidrIsHost(base.BaseTestCase):
class TestIpVersionFromInt(base.BaseTestCase):
def test_ip_version_from_int_ipv4(self):
self.assertEqual(constants.IPv4,
utils.ip_version_from_int(4))
utils.ip_version_from_int(constants.IP_VERSION_4))
def test_ip_version_from_int_ipv6(self):
self.assertEqual(constants.IPv6,
utils.ip_version_from_int(6))
utils.ip_version_from_int(constants.IP_VERSION_6))
def test_ip_version_from_int_illegal_int(self):
self.assertRaises(ValueError,

View File

@ -334,7 +334,7 @@ class NeutronDbPluginV2TestCase(testlib_api.WebTestCase):
def _create_subnet(self, fmt, net_id, cidr,
expected_res_status=None, **kwargs):
data = {'subnet': {'network_id': net_id,
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'tenant_id': self._tenant_id}}
if cidr:
data['subnet']['cidr'] = cidr
@ -363,7 +363,7 @@ class NeutronDbPluginV2TestCase(testlib_api.WebTestCase):
return subnet_res
def _create_subnet_bulk(self, fmt, number, net_id, name,
ip_version=4, **kwargs):
ip_version=constants.IP_VERSION_4, **kwargs):
base_data = {'subnet': {'network_id': net_id,
'ip_version': ip_version,
'tenant_id': self._tenant_id}}
@ -462,9 +462,9 @@ class NeutronDbPluginV2TestCase(testlib_api.WebTestCase):
return self.deserialize(fmt, res)
def _make_subnet(self, fmt, network, gateway, cidr, subnetpool_id=None,
allocation_pools=None, ip_version=4, enable_dhcp=True,
dns_nameservers=None, host_routes=None, shared=None,
ipv6_ra_mode=None, ipv6_address_mode=None,
allocation_pools=None, ip_version=constants.IP_VERSION_4,
enable_dhcp=True, dns_nameservers=None, host_routes=None,
shared=None, ipv6_ra_mode=None, ipv6_address_mode=None,
tenant_id=None, set_context=False, segment_id=None):
res = self._create_subnet(fmt,
net_id=network['network']['id'],
@ -500,7 +500,7 @@ class NeutronDbPluginV2TestCase(testlib_api.WebTestCase):
cfg.CONF.set_override('ipv6_pd_enabled', True)
return (self._make_subnet(self.fmt, network, gateway=gateway,
subnetpool_id=subnetpool_id,
cidr=cidr, ip_version=6,
cidr=cidr, ip_version=constants.IP_VERSION_6,
ipv6_ra_mode=ra_addr_mode,
ipv6_address_mode=ra_addr_mode))
@ -629,7 +629,7 @@ class NeutronDbPluginV2TestCase(testlib_api.WebTestCase):
subnetpool_id=None,
segment_id=None,
fmt=None,
ip_version=4,
ip_version=constants.IP_VERSION_4,
allocation_pools=None,
enable_dhcp=True,
dns_nameservers=None,
@ -1459,7 +1459,7 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
with self.subnet(network=n,
gateway_ip='fe80::1',
cidr='2607:f0d0:1002:51::/64',
ip_version=6,
ip_version=constants.IP_VERSION_6,
ipv6_address_mode=constants.IPV6_SLAAC) as subnet:
self.assertTrue(
ipv6_utils.is_auto_address_subnet(subnet['subnet']))
@ -1729,7 +1729,7 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
def test_update_port_invalid_fixed_ip_address_v6_slaac(self):
with self.subnet(
cidr='2607:f0d0:1002:51::/64',
ip_version=6,
ip_version=constants.IP_VERSION_6,
ipv6_address_mode=constants.IPV6_SLAAC,
gateway_ip=constants.ATTR_NOT_SPECIFIED) as subnet:
with self.port(subnet=subnet) as port:
@ -1857,7 +1857,8 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
admin_state_up=True)
network2 = self.deserialize(self.fmt, res)
subnet2 = self._make_subnet(self.fmt, network2, "1.1.1.1",
"1.1.1.0/24", ip_version=4)
"1.1.1.0/24",
ip_version=constants.IP_VERSION_4)
net_id = port['port']['network_id']
# Request a IP from specific subnet
kwargs = {"fixed_ips": [{'subnet_id':
@ -1875,7 +1876,7 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
tenant_id=tenant_id,
net_id=net_id,
cidr='10.0.0.225/28',
ip_version=4,
ip_version=constants.IP_VERSION_4,
gateway_ip=constants.ATTR_NOT_SPECIFIED)
self.assertEqual(webob.exc.HTTPClientError.code, res.status_int)
@ -1889,7 +1890,7 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
tenant_id=tenant_id,
net_id=net_id,
cidr='2607:f0d0:1002:51::/124',
ip_version=6,
ip_version=constants.IP_VERSION_6,
gateway_ip=constants.ATTR_NOT_SPECIFIED)
subnet2 = self.deserialize(self.fmt, res)
kwargs = {"fixed_ips":
@ -1997,7 +1998,7 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
admin_state_up=True)
network2 = self.deserialize(self.fmt, res)
self._make_subnet(self.fmt, network2, "1.1.1.1",
"1.1.1.0/24", ip_version=4)
"1.1.1.0/24", ip_version=constants.IP_VERSION_4)
res = self._create_port(self.fmt, net_id=network2['network']['id'])
port = self.deserialize(self.fmt, res)
@ -2011,7 +2012,7 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
def test_requested_invalid_fixed_ip_address_v6_slaac(self):
with self.subnet(gateway_ip='fe80::1',
cidr='2607:f0d0:1002:51::/64',
ip_version=6,
ip_version=constants.IP_VERSION_6,
ipv6_address_mode=constants.IPV6_SLAAC) as subnet:
kwargs = {"fixed_ips": [{'subnet_id': subnet['subnet']['id'],
'ip_address': '2607:f0d0:1002:51::5'}]}
@ -2023,7 +2024,7 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
def test_requested_fixed_ip_address_v6_slaac_router_iface(self):
with self.subnet(gateway_ip='fe80::1',
cidr='fe80::/64',
ip_version=6,
ip_version=constants.IP_VERSION_6,
ipv6_address_mode=constants.IPV6_SLAAC) as subnet:
kwargs = {"fixed_ips": [{'subnet_id': subnet['subnet']['id'],
'ip_address': 'fe80::1'}]}
@ -2039,7 +2040,7 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
def test_requested_subnet_id_v6_slaac(self):
with self.subnet(gateway_ip='fe80::1',
cidr='2607:f0d0:1002:51::/64',
ip_version=6,
ip_version=constants.IP_VERSION_6,
ipv6_address_mode=constants.IPV6_SLAAC) as subnet:
with self.port(subnet,
fixed_ips=[{'subnet_id':
@ -2057,7 +2058,7 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
self.subnet(
network,
cidr='2607:f0d0:1002:51::/64',
ip_version=6,
ip_version=constants.IP_VERSION_6,
gateway_ip='fe80::1',
ipv6_address_mode=constants.IPV6_SLAAC) as subnet2:
with self.port(
@ -2086,7 +2087,7 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
with self.subnet(network) as subnet_v4,\
self.subnet(network,
cidr='2607:f0d0:1002:51::/64',
ip_version=6,
ip_version=constants.IP_VERSION_6,
gateway_ip='fe80::1',
ipv6_address_mode=constants.IPV6_SLAAC):
subnet_ip_net = netaddr.IPNetwork(subnet_v4['subnet']['cidr'])
@ -2152,17 +2153,21 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
network = self.deserialize(self.fmt, res)
sub_dicts = [
{'gateway': '10.0.0.1', 'cidr': '10.0.0.0/24',
'ip_version': 4, 'ra_addr_mode': None},
'ip_version': constants.IP_VERSION_4, 'ra_addr_mode': None},
{'gateway': '10.0.1.1', 'cidr': '10.0.1.0/24',
'ip_version': 4, 'ra_addr_mode': None},
'ip_version': constants.IP_VERSION_4, 'ra_addr_mode': None},
{'gateway': 'fe80::1', 'cidr': 'fe80::/64',
'ip_version': 6, 'ra_addr_mode': constants.IPV6_SLAAC},
'ip_version': constants.IP_VERSION_6,
'ra_addr_mode': constants.IPV6_SLAAC},
{'gateway': 'fe81::1', 'cidr': 'fe81::/64',
'ip_version': 6, 'ra_addr_mode': constants.IPV6_SLAAC},
'ip_version': constants.IP_VERSION_6,
'ra_addr_mode': constants.IPV6_SLAAC},
{'gateway': 'fe82::1', 'cidr': 'fe82::/64',
'ip_version': 6, 'ra_addr_mode': constants.DHCPV6_STATEFUL},
'ip_version': constants.IP_VERSION_6,
'ra_addr_mode': constants.DHCPV6_STATEFUL},
{'gateway': 'fe83::1', 'cidr': 'fe83::/64',
'ip_version': 6, 'ra_addr_mode': constants.DHCPV6_STATEFUL}]
'ip_version': constants.IP_VERSION_6,
'ra_addr_mode': constants.DHCPV6_STATEFUL}]
subnets = {}
for sub_dict in sub_dicts:
subnet = self._make_subnet(
@ -2212,7 +2217,8 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
network = self.deserialize(self.fmt, res)
# Create a port using an IPv4 subnet and an IPv6 SLAAC subnet
self._make_subnet(self.fmt, network, gateway='10.0.0.1',
cidr='10.0.0.0/24', ip_version=4)
cidr='10.0.0.0/24',
ip_version=constants.IP_VERSION_4)
subnet_v6 = self._make_v6_subnet(network, constants.IPV6_SLAAC)
res = self._create_port(self.fmt, net_id=network['network']['id'])
port = self.deserialize(self.fmt, res)
@ -2236,7 +2242,8 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
network = self.deserialize(self.fmt, res)
# Create a port using an IPv4 subnet and an IPv6 SLAAC subnet
subnet_v4 = self._make_subnet(self.fmt, network, gateway='10.0.0.1',
cidr='10.0.0.0/24', ip_version=4)
cidr='10.0.0.0/24',
ip_version=constants.IP_VERSION_4)
subnet_v6 = self._make_v6_subnet(network, constants.IPV6_SLAAC)
res = self._create_port(self.fmt, net_id=network['network']['id'])
port = self.deserialize(self.fmt, res)
@ -2274,7 +2281,8 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
network = self.deserialize(self.fmt, res)
# Create a port using an IPv4 subnet and an IPv6 SLAAC subnet
subnet_v4 = self._make_subnet(self.fmt, network, gateway='10.0.0.1',
cidr='10.0.0.0/24', ip_version=4)
cidr='10.0.0.0/24',
ip_version=constants.IP_VERSION_4)
subnet_v6 = self._make_v6_subnet(network, constants.IPV6_SLAAC)
res = self._create_port(self.fmt, net_id=network['network']['id'])
port = self.deserialize(self.fmt, res)
@ -2301,12 +2309,12 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
v6_subnet_1 = self._make_subnet(self.fmt, network,
gateway='2001:100::1',
cidr='2001:100::0/64',
ip_version=6,
ip_version=constants.IP_VERSION_6,
ipv6_ra_mode=constants.IPV6_SLAAC)
v6_subnet_2 = self._make_subnet(self.fmt, network,
gateway='2001:200::1',
cidr='2001:200::0/64',
ip_version=6,
ip_version=constants.IP_VERSION_6,
ipv6_ra_mode=constants.IPV6_SLAAC)
port = self._make_port(self.fmt, network['network']['id'])
port_mac = port['port']['mac_address']
@ -3112,7 +3120,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
def _test_create_subnet(self, network=None, expected=None, **kwargs):
keys = kwargs.copy()
keys.setdefault('cidr', '10.0.0.0/24')
keys.setdefault('ip_version', 4)
keys.setdefault('ip_version', constants.IP_VERSION_4)
keys.setdefault('enable_dhcp', True)
with self.subnet(network=network, **keys) as subnet:
# verify the response has each key with the correct value
@ -3128,7 +3136,8 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
cidr = '10.0.0.0/24'
subnet = self._test_create_subnet(gateway_ip=gateway_ip,
cidr=cidr)
self.assertEqual(4, subnet['subnet']['ip_version'])
self.assertEqual(constants.IP_VERSION_4,
subnet['subnet']['ip_version'])
self.assertIn('name', subnet['subnet'])
def test_create_subnet_with_network_different_tenant(self):
@ -3138,7 +3147,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
is_admin=False)
data = {'subnet': {'network_id': network['network']['id'],
'cidr': '10.0.2.0/24',
'ip_version': '4',
'ip_version': constants.IP_VERSION_4,
'gateway_ip': '10.0.2.1'}}
req = self.new_create_request('subnets', data,
self.fmt, context=ctx)
@ -3189,7 +3198,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
with self.network() as network:
data = {'subnet': {'network_id': network['network']['id'],
'cidr': '10.0.2.0',
'ip_version': '4',
'ip_version': constants.IP_VERSION_4,
'tenant_id': network['network']['tenant_id'],
'gateway_ip': '10.0.2.1'}}
subnet_req = self.new_create_request('subnets', data)
@ -3218,7 +3227,8 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
min_prefixlen='25',
is_default=True):
data = {'subnet': {'network_id': network['network']['id'],
'cidr': '10.0.0.0/24', 'ip_version': '4',
'cidr': '10.0.0.0/24',
'ip_version': constants.IP_VERSION_4,
'tenant_id': tenant_id}}
subnet_req = self.new_create_request('subnets', data)
res = subnet_req.get_response(self.api)
@ -3237,7 +3247,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
min_prefixlen='25',
is_default=True):
data = {'subnet': {'network_id': network['network']['id'],
'ip_version': '4',
'ip_version': constants.IP_VERSION_4,
'tenant_id': tenant_id}}
subnet_req = self.new_create_request('subnets', data)
res = subnet_req.get_response(self.api)
@ -3257,7 +3267,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
tenant_id = network['network']['tenant_id']
cfg.CONF.set_override('ipv6_pd_enabled', False)
data = {'subnet': {'network_id': network['network']['id'],
'ip_version': '6',
'ip_version': constants.IP_VERSION_6,
'tenant_id': tenant_id}}
subnet_req = self.new_create_request('subnets', data)
res = subnet_req.get_response(self.api)
@ -3267,7 +3277,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
with self.network() as network:
data = {'subnet': {'network_id': network['network']['id'],
'cidr': constants.IPv4_ANY,
'ip_version': '4',
'ip_version': constants.IP_VERSION_4,
'tenant_id': network['network']['tenant_id'],
'gateway_ip': '0.0.0.1'}}
subnet_req = self.new_create_request('subnets', data)
@ -3278,7 +3288,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
with self.network() as network:
data = {'subnet': {'network_id': network['network']['id'],
'cidr': 'fe80::',
'ip_version': '6',
'ip_version': constants.IP_VERSION_6,
'tenant_id': network['network']['tenant_id'],
'gateway_ip': 'fe80::1'}}
subnet_req = self.new_create_request('subnets', data)
@ -3289,7 +3299,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
with self.network() as network:
data = {'subnet': {'network_id': network['network']['id'],
'cidr': '2014::/65',
'ip_version': '6',
'ip_version': constants.IP_VERSION_6,
'tenant_id': network['network']['tenant_id'],
'gateway_ip': 'fe80::1',
'ipv6_address_mode': 'slaac'}}
@ -3402,7 +3412,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
admin_state_up=True)
network = self.deserialize(self.fmt, res)
subnet = self._make_subnet(self.fmt, network, gateway_ip,
cidr, ip_version=4)
cidr, ip_version=constants.IP_VERSION_4)
req = self.new_delete_request('subnets', subnet['subnet']['id'])
res = req.get_response(self.api)
self.assertEqual(webob.exc.HTTPNoContent.code, res.status_int)
@ -3415,7 +3425,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
admin_state_up=True)
network = self.deserialize(self.fmt, res)
subnet = self._make_subnet(self.fmt, network, gateway_ip,
cidr, ip_version=4)
cidr, ip_version=constants.IP_VERSION_4)
self._create_port(self.fmt,
network['network']['id'],
device_owner=constants.DEVICE_OWNER_DHCP)
@ -3428,9 +3438,11 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
admin_state_up=True)
network = self.deserialize(self.fmt, res)
subnet1 = self._make_subnet(self.fmt, network, '10.0.0.1',
'10.0.0.0/24', ip_version=4)
'10.0.0.0/24',
ip_version=constants.IP_VERSION_4)
subnet2 = self._make_subnet(self.fmt, network, '10.0.1.1',
'10.0.1.0/24', ip_version=4)
'10.0.1.0/24',
ip_version=constants.IP_VERSION_4)
res = self._create_port(self.fmt,
network['network']['id'],
device_owner=constants.DEVICE_OWNER_DHCP,
@ -3488,7 +3500,8 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
admin_state_up=True)
network = self.deserialize(self.fmt, res)
subnet = self._make_subnet(self.fmt, network, gateway='fe80::1',
cidr='fe80::/64', ip_version=6,
cidr='fe80::/64',
ip_version=constants.IP_VERSION_6,
ipv6_ra_mode=constants.IPV6_SLAAC,
ipv6_address_mode=constants.IPV6_SLAAC)
kwargs = {}
@ -3554,7 +3567,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
admin_state_up=True)
network = self.deserialize(self.fmt, res)
subnet = self._make_subnet(self.fmt, network, gateway_ip, cidr,
ip_version=4)
ip_version=constants.IP_VERSION_4)
req = self.new_delete_request('networks', network['network']['id'])
res = req.get_response(self.api)
self.assertEqual(webob.exc.HTTPNoContent.code, res.status_int)
@ -3568,7 +3581,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
network['network']['id'],
'10.0.2.0/24',
webob.exc.HTTPNotFound.code,
ip_version=4,
ip_version=constants.IP_VERSION_4,
tenant_id='bad_tenant_id',
gateway_ip='10.0.2.1',
device_owner='fake_owner',
@ -3580,7 +3593,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
network['network']['id'],
'10.0.2.0/24',
webob.exc.HTTPCreated.code,
ip_version=4,
ip_version=constants.IP_VERSION_4,
tenant_id='bad_tenant_id',
gateway_ip='10.0.2.1',
device_owner='fake_owner',
@ -3657,7 +3670,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
# Check invalid UUID
data = {'subnet': {'network_id': None,
'cidr': '10.0.2.0/24',
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'tenant_id': network['network']['tenant_id'],
'gateway_ip': '10.0.2.1'}}
subnet_req = self.new_create_request('subnets', data)
@ -3669,7 +3682,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
# Check invalid boolean
data = {'subnet': {'network_id': network['network']['id'],
'cidr': '10.0.2.0/24',
'ip_version': '4',
'ip_version': constants.IP_VERSION_4,
'enable_dhcp': None,
'tenant_id': network['network']['tenant_id'],
'gateway_ip': '10.0.2.1'}}
@ -3695,7 +3708,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
for pool in allocation_pools:
data = {'subnet': {'network_id': network['network']['id'],
'cidr': '10.0.2.0/24',
'ip_version': '4',
'ip_version': constants.IP_VERSION_4,
'tenant_id': tenant_id,
'gateway_ip': '10.0.2.1',
'allocation_pools': pool}}
@ -3714,7 +3727,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
for nameservers in nameserver_pools:
data = {'subnet': {'network_id': network['network']['id'],
'cidr': '10.0.2.0/24',
'ip_version': '4',
'ip_version': constants.IP_VERSION_4,
'tenant_id': tenant_id,
'gateway_ip': '10.0.2.1',
'dns_nameservers': nameservers}}
@ -3736,7 +3749,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
for hostroutes in hostroute_pools:
data = {'subnet': {'network_id': network['network']['id'],
'cidr': '10.0.2.0/24',
'ip_version': '4',
'ip_version': constants.IP_VERSION_4,
'tenant_id': tenant_id,
'gateway_ip': '10.0.2.1',
'host_routes': hostroutes}}
@ -3789,7 +3802,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
'cidr': cidr,
'allocation_pools': allocation_pools}
self._test_create_subnet(expected=expected, gateway_ip=gateway,
cidr=cidr, ip_version=6,
cidr=cidr, ip_version=constants.IP_VERSION_6,
ipv6_ra_mode=constants.DHCPV6_STATEFUL,
ipv6_address_mode=constants.DHCPV6_STATEFUL)
# Gateway is first IP in IPv6 DHCPv6 stateful subnet
@ -3800,14 +3813,14 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
'cidr': cidr,
'allocation_pools': allocation_pools}
self._test_create_subnet(expected=expected, gateway_ip=gateway,
cidr=cidr, ip_version=6,
cidr=cidr, ip_version=constants.IP_VERSION_6,
ipv6_ra_mode=constants.DHCPV6_STATEFUL,
ipv6_address_mode=constants.DHCPV6_STATEFUL)
# If gateway_ip is not specified, allocate first IP from the subnet
expected = {'gateway_ip': gateway,
'cidr': cidr}
self._test_create_subnet(expected=expected,
cidr=cidr, ip_version=6,
cidr=cidr, ip_version=constants.IP_VERSION_6,
ipv6_ra_mode=constants.IPV6_SLAAC,
ipv6_address_mode=constants.IPV6_SLAAC)
@ -3822,7 +3835,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
'cidr': cidr,
'allocation_pools': allocation_pools}
self._test_create_subnet(expected=expected, gateway_ip=gateway,
cidr=cidr, ip_version=6,
cidr=cidr, ip_version=constants.IP_VERSION_6,
ipv6_ra_mode=constants.DHCPV6_STATELESS,
ipv6_address_mode=constants.DHCPV6_STATELESS)
# Gateway is first IP in IPv6 DHCPv6 Stateless subnet
@ -3833,14 +3846,14 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
'cidr': cidr,
'allocation_pools': allocation_pools}
self._test_create_subnet(expected=expected, gateway_ip=gateway,
cidr=cidr, ip_version=6,
cidr=cidr, ip_version=constants.IP_VERSION_6,
ipv6_ra_mode=constants.DHCPV6_STATELESS,
ipv6_address_mode=constants.DHCPV6_STATELESS)
# If gateway_ip is not specified, allocate first IP from the subnet
expected = {'gateway_ip': gateway,
'cidr': cidr}
self._test_create_subnet(expected=expected,
cidr=cidr, ip_version=6,
cidr=cidr, ip_version=constants.IP_VERSION_6,
ipv6_ra_mode=constants.IPV6_SLAAC,
ipv6_address_mode=constants.IPV6_SLAAC)
@ -3911,7 +3924,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
net_id = network['network']['id']
data = {'subnet': {'network_id': net_id,
'cidr': '10.0.0.0/24',
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'gateway_ip': '10.0.0.1',
'tenant_id': network['network']['tenant_id'],
'allocation_pools': [{'start': '10.0.0.100',
@ -3951,7 +3964,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
allocation_pools = [{'start': 'fe80::2',
'end': 'fe80::ffff:fffa:ffff'}]
self._test_create_subnet(gateway_ip=gateway_ip,
cidr=cidr, ip_version=6,
cidr=cidr, ip_version=constants.IP_VERSION_6,
allocation_pools=allocation_pools)
@testtools.skipIf(tools.is_bsd(), 'bug/1484837')
@ -3961,7 +3974,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
allocation_pools = [{'start': '::2',
'end': '::ffff:ffff:ffff:fffe'}]
self._test_create_subnet(gateway_ip=gateway_ip,
cidr=cidr, ip_version=6,
cidr=cidr, ip_version=constants.IP_VERSION_6,
allocation_pools=allocation_pools)
def test_create_subnet_with_large_allocation_pool(self):
@ -4069,7 +4082,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
with self.network() as network:
data = {'subnet': {'network_id': network['network']['id'],
'cidr': '10.0.2.0/24',
'ip_version': 6,
'ip_version': constants.IP_VERSION_6,
'tenant_id': network['network']['tenant_id']}}
subnet_req = self.new_create_request('subnets', data)
res = subnet_req.get_response(self.api)
@ -4079,7 +4092,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
with self.network() as network:
data = {'subnet': {'network_id': network['network']['id'],
'cidr': 'fe80::0/80',
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'tenant_id': network['network']['tenant_id']}}
subnet_req = self.new_create_request('subnets', data)
res = subnet_req.get_response(self.api)
@ -4089,7 +4102,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
with self.network() as network:
data = {'subnet': {'network_id': network['network']['id'],
'cidr': '10.0.2.0/24',
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'gateway_ip': 'fe80::1',
'tenant_id': network['network']['tenant_id']}}
subnet_req = self.new_create_request('subnets', data)
@ -4100,7 +4113,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
with self.network() as network:
data = {'subnet': {'network_id': network['network']['id'],
'cidr': 'fe80::0/80',
'ip_version': 6,
'ip_version': constants.IP_VERSION_6,
'gateway_ip': '192.168.0.1',
'tenant_id': network['network']['tenant_id']}}
subnet_req = self.new_create_request('subnets', data)
@ -4111,7 +4124,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
with self.network() as network:
data = {'subnet': {'network_id': network['network']['id'],
'cidr': 'fe80::0/80',
'ip_version': 6,
'ip_version': constants.IP_VERSION_6,
'dns_nameservers': ['192.168.0.1'],
'tenant_id': network['network']['tenant_id']}}
subnet_req = self.new_create_request('subnets', data)
@ -4124,7 +4137,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
with self.network() as network:
data = {'subnet': {'network_id': network['network']['id'],
'cidr': '10.0.2.0/24',
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'host_routes': host_routes,
'tenant_id': network['network']['tenant_id']}}
subnet_req = self.new_create_request('subnets', data)
@ -4137,7 +4150,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
with self.network() as network:
data = {'subnet': {'network_id': network['network']['id'],
'cidr': '10.0.2.0/24',
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'host_routes': host_routes,
'tenant_id': network['network']['tenant_id']}}
subnet_req = self.new_create_request('subnets', data)
@ -4148,7 +4161,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
expect_success=True, **modes):
plugin = directory.get_plugin()
ctx = context.get_admin_context()
new_subnet = {'ip_version': 6,
new_subnet = {'ip_version': constants.IP_VERSION_6,
'cidr': 'fe80::/64',
'enable_dhcp': True,
'ipv6_address_mode': None,
@ -4165,7 +4178,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
expect_success=True, **modes):
plugin = directory.get_plugin()
ctx = context.get_admin_context()
new_subnet = {'ip_version': 6,
new_subnet = {'ip_version': constants.IP_VERSION_6,
'cidr': constants.PROVISIONAL_IPV6_PD_PREFIX,
'enable_dhcp': True,
'ipv6_address_mode': None,
@ -4266,7 +4279,8 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
cidr = '2001::/64'
self._test_create_subnet(
gateway_ip=gateway_ip, cidr=cidr, ip_version=6,
gateway_ip=gateway_ip, cidr=cidr,
ip_version=constants.IP_VERSION_6,
ipv6_ra_mode=constants.IPV6_SLAAC,
ipv6_address_mode=constants.IPV6_SLAAC)
@ -4277,7 +4291,8 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
webob.exc.HTTPClientError) as ctx_manager:
for mode in constants.IPV6_MODES:
self._test_create_subnet(gateway_ip=gateway_ip,
cidr=cidr, ip_version=6,
cidr=cidr,
ip_version=constants.IP_VERSION_6,
enable_dhcp=False,
ipv6_ra_mode=mode,
ipv6_address_mode=mode)
@ -4290,7 +4305,8 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
with testlib_api.ExpectedException(
webob.exc.HTTPClientError) as ctx_manager:
self._test_create_subnet(gateway_ip=gateway_ip,
cidr=cidr, ip_version=6,
cidr=cidr,
ip_version=constants.IP_VERSION_6,
ipv6_ra_mode='foo',
ipv6_address_mode='slaac')
self.assertEqual(webob.exc.HTTPClientError.code,
@ -4302,7 +4318,8 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
with testlib_api.ExpectedException(
webob.exc.HTTPClientError) as ctx_manager:
self._test_create_subnet(gateway_ip=gateway_ip,
cidr=cidr, ip_version=6,
cidr=cidr,
ip_version=constants.IP_VERSION_6,
ipv6_ra_mode='slaac',
ipv6_address_mode='baz')
self.assertEqual(webob.exc.HTTPClientError.code,
@ -4312,7 +4329,8 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
cidr = '10.0.2.0/24'
with testlib_api.ExpectedException(
webob.exc.HTTPClientError) as ctx_manager:
self._test_create_subnet(cidr=cidr, ip_version=4,
self._test_create_subnet(cidr=cidr,
ip_version=constants.IP_VERSION_4,
ipv6_ra_mode=constants.DHCPV6_STATEFUL)
self.assertEqual(webob.exc.HTTPClientError.code,
ctx_manager.exception.code)
@ -4322,7 +4340,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
with testlib_api.ExpectedException(
webob.exc.HTTPClientError) as ctx_manager:
self._test_create_subnet(
cidr=cidr, ip_version=4,
cidr=cidr, ip_version=constants.IP_VERSION_4,
ipv6_address_mode=constants.DHCPV6_STATEFUL)
self.assertEqual(webob.exc.HTTPClientError.code,
ctx_manager.exception.code)
@ -4348,7 +4366,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
mock.patch.object(orm.Session, 'add',
side_effect=db_ref_err_for_ipalloc,
autospec=True).start()
v6_subnet = {'ip_version': 6,
v6_subnet = {'ip_version': constants.IP_VERSION_6,
'cidr': 'fe80::/64',
'gateway_ip': 'fe80::1',
'tenant_id': v4_subnet['subnet']['tenant_id']}
@ -4366,10 +4384,11 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
ipam_driver.NeutronDbSubnet, '_verify_ip',
side_effect=ipam_exc.IpAddressAlreadyAllocated(
subnet_id=mock.ANY, ip=mock.ANY)).start()
v6_subnet = self._make_subnet(self.fmt, network, 'fe80::1',
'fe80::/64', ip_version=6,
ipv6_ra_mode=addr_mode,
ipv6_address_mode=addr_mode)
v6_subnet = self._make_subnet(
self.fmt, network, 'fe80::1', 'fe80::/64',
ip_version=constants.IP_VERSION_6,
ipv6_ra_mode=addr_mode,
ipv6_address_mode=addr_mode)
if (insert_db_reference_error or insert_address_allocated or
device_owner == constants.DEVICE_OWNER_ROUTER_SNAT or
device_owner in constants.ROUTER_INTERFACE_OWNERS):
@ -4469,7 +4488,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
with self.network() as network:
data = {'subnet': {'network_id': network['network']['id'],
'cidr': '10.0.2.0/24',
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'dns_nameservers': ['192.168.0.1'],
'host_routes': host_routes,
'tenant_id': network['network']['tenant_id']}}
@ -4512,7 +4531,8 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
# Create port on second network
network2 = self._make_network(self.fmt, 'net2', True)
self._make_subnet(self.fmt, network2, "1.1.1.1",
"1.1.1.0/24", ip_version=4)
"1.1.1.0/24",
ip_version=constants.IP_VERSION_4)
self._make_port(self.fmt, net_id=network2['network']['id'])
subnet = self._make_v6_subnet(
@ -4641,7 +4661,8 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
def test_update_subnet_inconsistent_ipv6_gatewayv4(self):
with self.network() as network:
with self.subnet(network=network,
ip_version=6, cidr='fe80::/48') as subnet:
ip_version=constants.IP_VERSION_6,
cidr='fe80::/48') as subnet:
data = {'subnet': {'gateway_ip': '10.1.1.1'}}
req = self.new_update_request('subnets', data,
subnet['subnet']['id'])
@ -4665,7 +4686,8 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
'nexthop': '10.0.2.20'}]
with self.network() as network:
with self.subnet(network=network,
ip_version=6, cidr='fe80::/48') as subnet:
ip_version=constants.IP_VERSION_6,
cidr='fe80::/48') as subnet:
data = {'subnet': {'host_routes': host_routes}}
req = self.new_update_request('subnets', data,
subnet['subnet']['id'])
@ -4678,7 +4700,8 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
'nexthop': 'fe80::1'}]
with self.network() as network:
with self.subnet(network=network,
ip_version=6, cidr='fe80::/48') as subnet:
ip_version=constants.IP_VERSION_6,
cidr='fe80::/48') as subnet:
data = {'subnet': {'host_routes': host_routes}}
req = self.new_update_request('subnets', data,
subnet['subnet']['id'])
@ -4700,7 +4723,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
res.status_int)
def test_update_subnet_ipv6_attributes_fails(self):
with self.subnet(ip_version=6, cidr='fe80::/64',
with self.subnet(ip_version=constants.IP_VERSION_6, cidr='fe80::/64',
ipv6_ra_mode=constants.IPV6_SLAAC,
ipv6_address_mode=constants.IPV6_SLAAC) as subnet:
data = {'subnet': {'ipv6_ra_mode': constants.DHCPV6_STATEFUL,
@ -4712,7 +4735,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
res.status_int)
def test_update_subnet_ipv6_ra_mode_fails(self):
with self.subnet(ip_version=6, cidr='fe80::/64',
with self.subnet(ip_version=constants.IP_VERSION_6, cidr='fe80::/64',
ipv6_ra_mode=constants.IPV6_SLAAC) as subnet:
data = {'subnet': {'ipv6_ra_mode': constants.DHCPV6_STATEFUL}}
req = self.new_update_request('subnets', data,
@ -4722,7 +4745,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
res.status_int)
def test_update_subnet_ipv6_address_mode_fails(self):
with self.subnet(ip_version=6, cidr='fe80::/64',
with self.subnet(ip_version=constants.IP_VERSION_6, cidr='fe80::/64',
ipv6_address_mode=constants.IPV6_SLAAC) as subnet:
data = {'subnet': {'ipv6_address_mode': constants.DHCPV6_STATEFUL}}
req = self.new_update_request('subnets', data,
@ -4732,7 +4755,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
res.status_int)
def test_update_subnet_ipv6_cannot_disable_dhcp(self):
with self.subnet(ip_version=6, cidr='fe80::/64',
with self.subnet(ip_version=constants.IP_VERSION_6, cidr='fe80::/64',
ipv6_ra_mode=constants.IPV6_SLAAC,
ipv6_address_mode=constants.IPV6_SLAAC) as subnet:
data = {'subnet': {'enable_dhcp': False}}
@ -5077,10 +5100,11 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
gateway_ip='10.0.1.1',
cidr='10.0.1.0/24') as v2:
subnets = (v1, v2)
query_params = 'ip_version=4&ip_version=6'
query_params = 'ip_version=%s&ip_version=%s' % (
constants.IP_VERSION_4, constants.IP_VERSION_6)
self._test_list_resources('subnet', subnets,
query_params=query_params)
query_params = 'ip_version=6'
query_params = 'ip_version=%s' % constants.IP_VERSION_6
self._test_list_resources('subnet', [],
query_params=query_params)
@ -5170,7 +5194,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
with self.network() as network:
data = {'subnet': {'network_id': network['network']['id'],
'cidr': 'invalid',
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'tenant_id': network['network']['tenant_id'],
'gateway_ip': '10.0.2.1'}}
@ -5182,7 +5206,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
with self.network() as network:
subnet = {'network_id': network['network']['id'],
'cidr': subnet_cidr,
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'enable_dhcp': True,
'tenant_id': network['network']['tenant_id']}
plugin = directory.get_plugin()
@ -5202,7 +5226,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
with self.network() as network:
data = {'subnet': {'network_id': network['network']['id'],
'cidr': '10.0.2.0/24',
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'tenant_id': network['network']['tenant_id'],
'gateway_ip': 'ipaddress'}}
@ -5214,7 +5238,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
with self.network() as network:
data = {'subnet': {'network_id': 'invalid-uuid',
'cidr': '10.0.2.0/24',
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'tenant_id': network['network']['tenant_id'],
'gateway_ip': '10.0.0.1'}}
@ -5249,7 +5273,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
dns_list = ['1.1.1.1', '2.2.2.2', '3.3.3.3']
data = {'subnet': {'network_id': network['network']['id'],
'cidr': '10.0.2.0/24',
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'tenant_id': network['network']['tenant_id'],
'gateway_ip': '10.0.0.1',
'dns_nameservers': dns_list}}
@ -5296,7 +5320,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
data = {'subnet': {'network_id': network['network']['id'],
'cidr': '10.0.2.0/24',
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'tenant_id': network['network']['tenant_id'],
'gateway_ip': '10.0.0.1',
'host_routes': host_routes}}
@ -5435,7 +5459,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
admin_state_up=True)
network = self.deserialize(self.fmt, res)
subnet = self._make_subnet(self.fmt, network, gateway_ip,
cidr, ip_version=4,
cidr, ip_version=constants.IP_VERSION_4,
dns_nameservers=dns_nameservers)
req = self.new_delete_request('subnets', subnet['subnet']['id'])
res = req.get_response(self.api)
@ -5451,7 +5475,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
admin_state_up=True)
network = self.deserialize(self.fmt, res)
subnet = self._make_subnet(self.fmt, network, gateway_ip,
cidr, ip_version=4,
cidr, ip_version=constants.IP_VERSION_4,
host_routes=host_routes)
req = self.new_delete_request('subnets', subnet['subnet']['id'])
res = req.get_response(self.api)
@ -5468,7 +5492,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
admin_state_up=True)
network = self.deserialize(self.fmt, res)
subnet = self._make_subnet(self.fmt, network, gateway_ip,
cidr, ip_version=4,
cidr, ip_version=constants.IP_VERSION_4,
dns_nameservers=dns_nameservers,
host_routes=host_routes)
req = self.new_delete_request('subnets', subnet['subnet']['id'])
@ -5508,7 +5532,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
with self.network() as network:
subnet = {'network_id': network['network']['id'],
'cidr': '10.0.2.0/24',
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'tenant_id': network['network']['tenant_id'],
'gateway_ip': '10.0.2.1',
'dns_nameservers': ['8.8.8.8'],
@ -5988,7 +6012,7 @@ class TestSubnetPoolsV2(NeutronDbPluginV2TestCase):
data = {'subnet': {'network_id': network['network']['id'],
'subnetpool_id': sp['subnetpool']['id'],
'prefixlen': 32,
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'tenant_id': network['network']['tenant_id']}}
req = self.new_create_request('subnets', data)
result = req.get_response(self.api)
@ -6005,7 +6029,7 @@ class TestSubnetPoolsV2(NeutronDbPluginV2TestCase):
data = {'subnet': {'network_id': network['network']['id'],
'subnetpool_id': sp['subnetpool']['id'],
'prefixlen': 24,
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'tenant_id': network['network']['tenant_id']}}
req = self.new_create_request('subnets', data)
res = self.deserialize(self.fmt, req.get_response(self.api))
@ -6028,7 +6052,7 @@ class TestSubnetPoolsV2(NeutronDbPluginV2TestCase):
# Request any subnet allocation using default prefix
data = {'subnet': {'network_id': network['network']['id'],
'subnetpool_id': sp['subnetpool']['id'],
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'tenant_id': network['network']['tenant_id']}}
req = self.new_create_request('subnets', data)
res = self.deserialize(self.fmt, req.get_response(self.api))
@ -6048,7 +6072,7 @@ class TestSubnetPoolsV2(NeutronDbPluginV2TestCase):
'subnetpool_id': sp['subnetpool']['id'],
'cidr': '10.10.1.0/24',
'prefixlen': 26,
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'tenant_id': network['network']['tenant_id']}}
req = self.new_create_request('subnets', data)
res = req.get_response(self.api)
@ -6065,7 +6089,7 @@ class TestSubnetPoolsV2(NeutronDbPluginV2TestCase):
'subnetpool_id': sp['subnetpool']['id'],
'cidr': '10.10.1.0/24',
'prefixlen': 24,
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'tenant_id': network['network']['tenant_id']}}
req = self.new_create_request('subnets', data)
res = req.get_response(self.api)
@ -6082,7 +6106,7 @@ class TestSubnetPoolsV2(NeutronDbPluginV2TestCase):
data = {'subnet': {'network_id': network['network']['id'],
'subnetpool_id': sp['subnetpool']['id'],
'cidr': '10.10.1.0/24',
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'tenant_id': network['network']['tenant_id']}}
req = self.new_create_request('subnets', data)
res = self.deserialize(self.fmt, req.get_response(self.api))
@ -6102,7 +6126,7 @@ class TestSubnetPoolsV2(NeutronDbPluginV2TestCase):
data = {'subnet': {'network_id': network['network']['id'],
'subnetpool_id': sp['subnetpool']['id'],
'cidr': '192.168.1.0/24',
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'tenant_id': network['network']['tenant_id']}}
req = self.new_create_request('subnets', data)
res = req.get_response(self.api)
@ -6119,7 +6143,7 @@ class TestSubnetPoolsV2(NeutronDbPluginV2TestCase):
data = {'subnet': {'network_id': network['network']['id'],
'subnetpool_id': sp['subnetpool']['id'],
'cidr': '10.10.10.0/24',
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'tenant_id': network['network']['tenant_id']}}
req = self.new_create_request('subnets', data)
# Allocate the subnet
@ -6141,7 +6165,7 @@ class TestSubnetPoolsV2(NeutronDbPluginV2TestCase):
data = {'subnet': {'network_id': network['network']['id'],
'subnetpool_id': sp['subnetpool']['id'],
'cidr': '10.10.0.0/20',
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'tenant_id': network['network']['tenant_id']}}
req = self.new_create_request('subnets', data)
res = req.get_response(self.api)
@ -6159,7 +6183,7 @@ class TestSubnetPoolsV2(NeutronDbPluginV2TestCase):
'subnetpool_id': sp['subnetpool']['id'],
'cidr': '10.10.1.0/24',
'gateway_ip': '10.10.1.254',
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'tenant_id': network['network']['tenant_id']}}
req = self.new_create_request('subnets', data)
res = self.deserialize(self.fmt, req.get_response(self.api))
@ -6179,7 +6203,7 @@ class TestSubnetPoolsV2(NeutronDbPluginV2TestCase):
'subnetpool_id': sp['subnetpool']['id'],
'cidr': '10.10.1.0/24',
'gateway_ip': '10.10.1.1',
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'allocation_pools': pools,
'tenant_id': network['network']['tenant_id']}}
req = self.new_create_request('subnets', data)
@ -6202,7 +6226,7 @@ class TestSubnetPoolsV2(NeutronDbPluginV2TestCase):
data = {'subnet': {'network_id': network['network']['id'],
'subnetpool_id': sp['subnetpool']['id'],
'prefixlen': '24',
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'allocation_pools': pools,
'tenant_id': network['network']['tenant_id']}}
req = self.new_create_request('subnets', data)
@ -6221,7 +6245,7 @@ class TestSubnetPoolsV2(NeutronDbPluginV2TestCase):
data = {'subnet': {'network_id': network['network']['id'],
'subnetpool_id': sp['subnetpool']['id'],
'cidr': '10.10.0.0/24',
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'tenant_id': network['network']['tenant_id']}}
req = self.new_create_request('subnets', data)
res = req.get_response(self.api)
@ -6237,7 +6261,7 @@ class TestSubnetPoolsV2(NeutronDbPluginV2TestCase):
data = {'subnet': {'network_id': network['network']['id'],
'subnetpool_id': sp['subnetpool']['id'],
'cidr': '10.10.0.0/24',
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'tenant_id': network['network']['tenant_id']}}
req = self.new_create_request('subnets', data)
req.get_response(self.api)
@ -6257,7 +6281,7 @@ class TestSubnetPoolsV2(NeutronDbPluginV2TestCase):
# Request a specific subnet allocation
data = {'subnet': {'network_id': network['network']['id'],
'subnetpool_id': sp['subnetpool']['id'],
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'prefixlen': 21,
'tenant_id': network['network']['tenant_id']}}
req = self.new_create_request('subnets', data)
@ -6278,7 +6302,7 @@ class TestSubnetPoolsV2(NeutronDbPluginV2TestCase):
# Request a specific subnet allocation
data = {'subnet': {'network_id': network['network']['id'],
'subnetpool_id': sp['subnetpool']['id'],
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'tenant_id': network['network']['tenant_id']}}
req = self.new_create_request('subnets', data)
res = req.get_response(self.api)
@ -6532,7 +6556,8 @@ class DbModelTenantTestCase(DbModelMixin, testlib_api.SqlTestCase):
def _make_subnet(self, ctx, network_id):
with db_api.context_manager.writer.using(ctx):
subnet = models_v2.Subnet(name="subsub", ip_version=4,
subnet = models_v2.Subnet(name="subsub",
ip_version=constants.IP_VERSION_4,
tenant_id='dbcheck',
cidr='turn_down_for_what',
network_id=network_id)
@ -6551,9 +6576,9 @@ class DbModelTenantTestCase(DbModelMixin, testlib_api.SqlTestCase):
def _make_subnetpool(self, ctx):
with db_api.context_manager.writer.using(ctx):
subnetpool = models_v2.SubnetPool(
ip_version=4, default_prefixlen=4, min_prefixlen=4,
max_prefixlen=4, shared=False, default_quota=4,
address_scope_id='f', tenant_id='dbcheck',
ip_version=constants.IP_VERSION_4, default_prefixlen=4,
min_prefixlen=4, max_prefixlen=4, shared=False,
default_quota=4, address_scope_id='f', tenant_id='dbcheck',
is_default=False
)
ctx.session.add(subnetpool)
@ -6571,7 +6596,8 @@ class DbModelProjectTestCase(DbModelMixin, testlib_api.SqlTestCase):
def _make_subnet(self, ctx, network_id):
with db_api.context_manager.writer.using(ctx):
subnet = models_v2.Subnet(name="subsub", ip_version=4,
subnet = models_v2.Subnet(name="subsub",
ip_version=constants.IP_VERSION_4,
project_id='dbcheck',
cidr='turn_down_for_what',
network_id=network_id)
@ -6590,9 +6616,9 @@ class DbModelProjectTestCase(DbModelMixin, testlib_api.SqlTestCase):
def _make_subnetpool(self, ctx):
with db_api.context_manager.writer.using(ctx):
subnetpool = models_v2.SubnetPool(
ip_version=4, default_prefixlen=4, min_prefixlen=4,
max_prefixlen=4, shared=False, default_quota=4,
address_scope_id='f', project_id='dbcheck',
ip_version=constants.IP_VERSION_4, default_prefixlen=4,
min_prefixlen=4, max_prefixlen=4, shared=False,
default_quota=4, address_scope_id='f', project_id='dbcheck',
is_default=False
)
ctx.session.add(subnetpool)
@ -6652,7 +6678,7 @@ class NeutronDbPluginV2AsMixinTestCase(NeutronDbPluginV2TestCase,
def test__validate_network_subnetpools(self):
network = models_v2.Network()
network.subnets = [models_v2.Subnet(subnetpool_id='test_id',
ip_version=4)]
ip_version=constants.IP_VERSION_4)]
new_subnetpool_id = None
self.assertRaises(n_exc.NetworkSubnetPoolAffinityError,
self.plugin.ipam._validate_network_subnetpools,

View File

@ -370,7 +370,7 @@ class TestDbBasePluginIpam(test_db_base.NeutronDbPluginV2TestCase):
context = mock.Mock()
pluggable_backend = ipam_pluggable_backend.IpamPluggableBackend()
with self.subnet(cidr=constants.PROVISIONAL_IPV6_PD_PREFIX,
ip_version=6) as subnet:
ip_version=constants.IP_VERSION_6) as subnet:
subnet = subnet['subnet']
fixed_ips = [{'subnet_id': subnet['id'],
'ip_address': '::1'}]
@ -406,7 +406,7 @@ class TestDbBasePluginIpam(test_db_base.NeutronDbPluginV2TestCase):
cfg.CONF.set_override('ipv6_pd_enabled', True)
cidr = constants.PROVISIONAL_IPV6_PD_PREFIX
allocation_pools = [netaddr.IPRange('::2', '::ffff:ffff:ffff:ffff')]
with self.subnet(cidr=None, ip_version=6,
with self.subnet(cidr=None, ip_version=constants.IP_VERSION_6,
subnetpool_id=constants.IPV6_PD_POOL_ID,
ipv6_ra_mode=constants.IPV6_SLAAC,
ipv6_address_mode=constants.IPV6_SLAAC):
@ -504,7 +504,7 @@ class TestDbBasePluginIpam(test_db_base.NeutronDbPluginV2TestCase):
admin_state_up=True)
network = self.deserialize(self.fmt, res)
subnet = self._make_subnet(self.fmt, network, gateway_ip,
cidr, ip_version=4)
cidr, ip_version=constants.IP_VERSION_4)
req = self.new_delete_request('subnets', subnet['subnet']['id'])
res = req.get_response(self.api)
self.assertEqual(webob.exc.HTTPNoContent.code, res.status_int)
@ -524,7 +524,7 @@ class TestDbBasePluginIpam(test_db_base.NeutronDbPluginV2TestCase):
admin_state_up=True)
network = self.deserialize(self.fmt, res)
subnet = self._make_subnet(self.fmt, network, gateway_ip,
cidr, ip_version=4)
cidr, ip_version=constants.IP_VERSION_4)
req = self.new_delete_request('subnets', subnet['subnet']['id'])
res = req.get_response(self.api)
self.assertEqual(webob.exc.HTTPServerError.code, res.status_int)
@ -711,7 +711,7 @@ class TestDbBasePluginIpam(test_db_base.NeutronDbPluginV2TestCase):
subnets = [{'id': ip_dict['subnet_id'],
'network_id': network_id,
'cidr': '192.1.1.0/24',
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'ipv6_address_mode': None,
'ipv6_ra_mode': None}]
get_subnets_mock = mock.Mock(return_value=subnets)

View File

@ -722,9 +722,9 @@ class L3DvrTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
{'network_id': ext_net_id}}})
with self.subnet(
network=net_int, cidr='20.0.0.0/24') as subnet_v4,\
self.subnet(
network=net_int, cidr='fe80::/64',
gateway_ip='fe80::1', ip_version=6) as subnet_v6:
self.subnet(network=net_int, cidr='fe80::/64',
gateway_ip='fe80::1', ip_version=const.IP_VERSION_6
) as subnet_v6:
self.mixin.add_router_interface(self.ctx, router['id'],
{'subnet_id': subnet_v4['subnet']['id']})
self.mixin.add_router_interface(self.ctx, router['id'],
@ -758,7 +758,8 @@ class L3DvrTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
self.mixin._core_plugin.update_port = update_port
with self.subnet(network=net, cidr='fe81::/64',
gateway_ip='fe81::1', ip_version=6) as subnet2_v6:
gateway_ip='fe81::1', ip_version=const.IP_VERSION_6
) as subnet2_v6:
self.mixin.add_router_interface(self.ctx, router['id'],
{'subnet_id': subnet2_v6['subnet']['id']})
if fail_revert:

View File

@ -997,7 +997,7 @@ class L3HAModeDbTestCase(L3HATestFramework):
def _create_subnet(self, plugin, ctx, network_id, cidr='10.0.0.0/8',
name='subnet', tenant_id='tenant1'):
subnet = {'subnet': {'name': name,
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'network_id': network_id,
'cidr': cidr,
'gateway_ip': constants.ATTR_NOT_SPECIFIED,

View File

@ -17,6 +17,7 @@ import socket
import mock
from neutron_lib.api.definitions import portbindings
from neutron_lib import constants
from oslo_config import cfg
from neutron.agent.linux import interface
@ -90,7 +91,7 @@ class TestDebugCommands(base.BaseTestCase):
'cidr': '10.0.0.0/24',
'allocation_pools': allocation_pools,
'enable_dhcp': True,
'ip_version': 4}}
'ip_version': constants.IP_VERSION_4}}
client_inst.list_ports.return_value = fake_ports
client_inst.create_port.return_value = fake_port
@ -324,7 +325,7 @@ class TestDebugCommands(base.BaseTestCase):
def test_ping_all_v6(self):
fake_subnet_v6 = {'subnet': {'name': 'fake_v6',
'ip_version': 6}}
'ip_version': constants.IP_VERSION_6}}
self.client.show_subnet.return_value = fake_subnet_v6
cmd = commands.PingAll(self.app, None)
cmd_parser = cmd.get_parser('ping_all')

View File

@ -382,7 +382,7 @@ class TestSubnetPoolsWithAddressScopes(AddressScopeTestCase):
'network_id': network['network']['id'],
'subnetpool_id': subnetpool_id,
'prefixlen': 24,
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'tenant_id': network['network']['tenant_id']}}
req = self.new_create_request('subnets', data)
subnet = self.deserialize(self.fmt,
@ -454,12 +454,12 @@ class TestSubnetPoolsWithAddressScopes(AddressScopeTestCase):
data = {'subnet': {
'network_id': network['network']['id'],
'subnetpool_id': v4_subnetpool_id,
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'tenant_id': network['network']['tenant_id']}}
req = self.new_create_request('subnets', data)
self.deserialize(self.fmt, req.get_response(self.api))
data['subnet']['subnetpool_id'] = v6_subnetpool_id
data['subnet']['ip_version'] = 6
data['subnet']['ip_version'] = constants.IP_VERSION_6
req = self.new_create_request('subnets', data)
self.deserialize(self.fmt, req.get_response(self.api))
result = self._show('networks', network['network']['id'])

View File

@ -58,7 +58,8 @@ class DefaultSubnetpoolsExtensionTestCase(
self).setUp(plugin=plugin, ext_mgr=ext_mgr)
def _create_subnet_using_default_subnetpool(
self, network_id, tenant_id, ip_version=4, **kwargs):
self, network_id, tenant_id, ip_version=constants.IP_VERSION_4,
**kwargs):
data = {'subnet': {
'network_id': network_id,
'ip_version': str(ip_version),
@ -147,7 +148,8 @@ class DefaultSubnetpoolsExtensionTestCase(
subnetpool_id = subnetpool['subnetpool']['id']
cfg.CONF.set_override('ipv6_pd_enabled', False)
subnet = self._create_subnet_using_default_subnetpool(
network['network']['id'], tenant_id, ip_version=6)
network['network']['id'], tenant_id,
ip_version=constants.IP_VERSION_6)
self.assertEqual(subnetpool_id, subnet['subnetpool_id'])
ip_net = netaddr.IPNetwork(subnet['cidr'])
self.assertIn(ip_net, netaddr.IPNetwork(subnetpool_prefix))
@ -157,7 +159,7 @@ class DefaultSubnetpoolsExtensionTestCase(
cfg.CONF.set_override('ipv6_pd_enabled', True)
with self.network() as network:
data = {'subnet': {'network_id': network['network']['id'],
'ip_version': '6',
'ip_version': constants.IP_VERSION_6,
'tenant_id': network['network']['tenant_id'],
'use_default_subnetpool': True}}
if ra_addr_mode:

View File

@ -384,17 +384,21 @@ class DnsExtensionTestCase(test_plugin.Ml2PluginV2TestCase):
network = self.deserialize(self.fmt, res)
sub_dicts = [
{'gateway': '10.0.0.1', 'cidr': '10.0.0.0/24',
'ip_version': 4, 'ra_addr_mode': None},
'ip_version': constants.IP_VERSION_4, 'ra_addr_mode': None},
{'gateway': '10.0.1.1', 'cidr': '10.0.1.0/24',
'ip_version': 4, 'ra_addr_mode': None},
'ip_version': constants.IP_VERSION_4, 'ra_addr_mode': None},
{'gateway': 'fe80::1', 'cidr': 'fe80::/64',
'ip_version': 6, 'ra_addr_mode': constants.IPV6_SLAAC},
'ip_version': constants.IP_VERSION_6,
'ra_addr_mode': constants.IPV6_SLAAC},
{'gateway': 'fe81::1', 'cidr': 'fe81::/64',
'ip_version': 6, 'ra_addr_mode': constants.IPV6_SLAAC},
'ip_version': constants.IP_VERSION_6,
'ra_addr_mode': constants.IPV6_SLAAC},
{'gateway': 'fe82::1', 'cidr': 'fe82::/64',
'ip_version': 6, 'ra_addr_mode': constants.DHCPV6_STATEFUL},
'ip_version': constants.IP_VERSION_6,
'ra_addr_mode': constants.DHCPV6_STATEFUL},
{'gateway': 'fe83::1', 'cidr': 'fe83::/64',
'ip_version': 6, 'ra_addr_mode': constants.DHCPV6_STATEFUL}]
'ip_version': constants.IP_VERSION_6,
'ra_addr_mode': constants.DHCPV6_STATEFUL}]
subnets = {}
for sub_dict in sub_dicts:
subnet = self._make_subnet(
@ -451,7 +455,7 @@ class DnsExtensionTestCase(test_plugin.Ml2PluginV2TestCase):
admin_state_up=True)
network = self.deserialize(self.fmt, res)
sub_dict = {'gateway': '10.0.0.1', 'cidr': '10.0.0.0/24',
'ip_version': 4, 'ra_addr_mode': None}
'ip_version': constants.IP_VERSION_4, 'ra_addr_mode': None}
self._make_subnet(self.fmt, network, gateway=sub_dict['gateway'],
cidr=sub_dict['cidr'],
ip_version=sub_dict['ip_version'],
@ -489,7 +493,7 @@ class DnsExtensionTestCase(test_plugin.Ml2PluginV2TestCase):
admin_state_up=True)
network = self.deserialize(self.fmt, res)
sub_dict = {'gateway': '10.0.0.1', 'cidr': '10.0.0.0/24',
'ip_version': 4, 'ra_addr_mode': None}
'ip_version': constants.IP_VERSION_4, 'ra_addr_mode': None}
self._make_subnet(self.fmt, network, gateway=sub_dict['gateway'],
cidr=sub_dict['cidr'],
ip_version=sub_dict['ip_version'],

View File

@ -16,6 +16,7 @@
import copy
from neutron_lib.api.definitions import extra_dhcp_opt as edo_ext
from neutron_lib import constants
import webob.exc
from neutron.db import db_base_plugin_v2
@ -65,7 +66,7 @@ class TestExtraDhcpOpt(ExtraDhcpOptDBTestCase):
for exp in expected:
if (name == exp['opt_name'] and
opt['ip_version'] == exp.get(
'ip_version', 4)):
'ip_version', constants.IP_VERSION_4)):
val = exp['opt_value']
break
self.assertEqual(val, opt['opt_value'])
@ -122,13 +123,13 @@ class TestExtraDhcpOpt(ExtraDhcpOptDBTestCase):
def test_create_port_with_extradhcpopts_ipv4_opt_version(self):
opt_list = [{'opt_name': 'bootfile-name',
'opt_value': 'pxelinux.0',
'ip_version': 4},
'ip_version': constants.IP_VERSION_4},
{'opt_name': 'server-ip-address',
'opt_value': '123.123.123.456',
'ip_version': 4},
'ip_version': constants.IP_VERSION_4},
{'opt_name': 'tftp-server',
'opt_value': '123.123.123.123',
'ip_version': 4}]
'ip_version': constants.IP_VERSION_4}]
params = {edo_ext.EXTRADHCPOPTS: opt_list,
'arg_list': (edo_ext.EXTRADHCPOPTS,)}
@ -140,10 +141,10 @@ class TestExtraDhcpOpt(ExtraDhcpOptDBTestCase):
def test_create_port_with_extradhcpopts_ipv6_opt_version(self):
opt_list = [{'opt_name': 'bootfile-name',
'opt_value': 'pxelinux.0',
'ip_version': 6},
'ip_version': constants.IP_VERSION_6},
{'opt_name': 'tftp-server',
'opt_value': '2001:192:168::1',
'ip_version': 6}]
'ip_version': constants.IP_VERSION_6}]
params = {edo_ext.EXTRADHCPOPTS: opt_list,
'arg_list': (edo_ext.EXTRADHCPOPTS,)}
@ -282,16 +283,16 @@ class TestExtraDhcpOpt(ExtraDhcpOptDBTestCase):
def test_update_port_with_blank_router_extradhcpopt(self):
opt_list = [{'opt_name': 'bootfile-name',
'opt_value': 'pxelinux.0',
'ip_version': 4},
'ip_version': constants.IP_VERSION_4},
{'opt_name': 'tftp-server',
'opt_value': '123.123.123.123',
'ip_version': 4},
'ip_version': constants.IP_VERSION_4},
{'opt_name': 'router',
'opt_value': '123.123.123.1',
'ip_version': 4}]
'ip_version': constants.IP_VERSION_4}]
upd_opts = [{'opt_name': 'router',
'opt_value': '',
'ip_version': 4}]
'ip_version': constants.IP_VERSION_4}]
expected_opts = copy.deepcopy(opt_list)
for i in expected_opts:
if i['opt_name'] == upd_opts[0]['opt_name']:
@ -304,13 +305,13 @@ class TestExtraDhcpOpt(ExtraDhcpOptDBTestCase):
def test_update_port_with_extradhcpopts_ipv6_change_value(self):
opt_list = [{'opt_name': 'bootfile-name',
'opt_value': 'pxelinux.0',
'ip_version': 6},
'ip_version': constants.IP_VERSION_6},
{'opt_name': 'tftp-server',
'opt_value': '2001:192:168::1',
'ip_version': 6}]
'ip_version': constants.IP_VERSION_6}]
upd_opts = [{'opt_name': 'tftp-server',
'opt_value': '2001:192:168::2',
'ip_version': 6}]
'ip_version': constants.IP_VERSION_6}]
expected_opts = copy.deepcopy(opt_list)
for i in expected_opts:
if i['opt_name'] == upd_opts[0]['opt_name']:
@ -322,13 +323,13 @@ class TestExtraDhcpOpt(ExtraDhcpOptDBTestCase):
def test_update_port_with_extradhcpopts_add_another_ver_opt(self):
opt_list = [{'opt_name': 'bootfile-name',
'opt_value': 'pxelinux.0',
'ip_version': 6},
'ip_version': constants.IP_VERSION_6},
{'opt_name': 'tftp-server',
'opt_value': '2001:192:168::1',
'ip_version': 6}]
'ip_version': constants.IP_VERSION_6}]
upd_opts = [{'opt_name': 'tftp-server',
'opt_value': '123.123.123.123',
'ip_version': 4}]
'ip_version': constants.IP_VERSION_4}]
expected_opts = copy.deepcopy(opt_list)
expected_opts.extend(upd_opts)
self._test_update_port_with_extradhcpopts(opt_list, upd_opts,

View File

@ -883,7 +883,8 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
def test_router_update_gateway_add_multiple_prefixes_ipv6(self):
with self.network() as n:
with self.subnet(network=n) as s1, \
self.subnet(network=n, ip_version=6, cidr='2001:db8::/32') \
self.subnet(network=n, ip_version=lib_constants.IP_VERSION_6,
cidr='2001:db8::/32') \
as s2, (self.router()) as r:
self._set_net_external(n['network']['id'])
res1 = self._add_external_gateway_to_router(
@ -924,7 +925,8 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
# ensure the router disappearing doesn't interfere with subnet
# creation
self._create_subnet(self.fmt, net_id=n['network']['id'],
ip_version=6, cidr='2001:db8::/32',
ip_version=lib_constants.IP_VERSION_6,
cidr='2001:db8::/32',
expected_res_status=(exc.HTTPCreated.code))
def test_router_update_gateway_upon_subnet_create_ipv6(self):
@ -938,7 +940,8 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
fip1 = (res1['router']['external_gateway_info']
['external_fixed_ips'][0])
sres = self._create_subnet(self.fmt, net_id=n['network']['id'],
ip_version=6, cidr='2001:db8::/32',
ip_version=lib_constants.IP_VERSION_6,
cidr='2001:db8::/32',
expected_res_status=(
exc.HTTPCreated.code))
s2 = self.deserialize(self.fmt, sres)
@ -962,7 +965,8 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
"""
with self.router() as r, self.network() as n:
with self.subnet(cidr='10.0.0.0/24', network=n) as s1, (
self.subnet(ip_version=6, cidr='2001:db8::/64',
self.subnet(ip_version=lib_constants.IP_VERSION_6,
cidr='2001:db8::/64',
network=n)) as s2:
self._set_net_external(n['network']['id'])
self._add_external_gateway_to_router(
@ -987,7 +991,8 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
s3 = self.deserialize(self.fmt,
self._create_subnet(self.fmt,
net_id=n['network']['id'],
ip_version=6, cidr='2001:db8:1::/64',
ip_version=lib_constants.IP_VERSION_6,
cidr='2001:db8:1::/64',
ipv6_ra_mode=lib_constants.IPV6_SLAAC,
ipv6_address_mode=lib_constants.IPV6_SLAAC))
res3 = self._show('routers', r['router']['id'])
@ -1111,7 +1116,8 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
fake_notifier.reset()
with self.router() as r, self.network() as n:
with self.subnet(network=n, cidr='fd00::1/64',
gateway_ip='fd00::1', ip_version=6,
gateway_ip='fd00::1',
ip_version=lib_constants.IP_VERSION_6,
ipv6_ra_mode=uc['ra_mode'],
ipv6_address_mode=uc['address_mode']) as s:
self._test_router_add_interface_subnet(r, s, uc['msg'])
@ -1148,9 +1154,11 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
to a router places them all on the same router interface.
"""
with self.router() as r, self.network() as n:
with (self.subnet(network=n, cidr='fd00::1/64', ip_version=6)
with (self.subnet(network=n, cidr='fd00::1/64',
ip_version=lib_constants.IP_VERSION_6)
) as s1, self.subnet(network=n, cidr='fd01::1/64',
ip_version=6) as s2:
ip_version=lib_constants.IP_VERSION_6
) as s2:
body = self._router_interface_action('add',
r['router']['id'],
s1['subnet']['id'],
@ -1180,9 +1188,11 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
to a router places them on different router interfaces.
"""
with self.router() as r, self.network() as n1, self.network() as n2:
with (self.subnet(network=n1, cidr='fd00::1/64', ip_version=6)
with (self.subnet(network=n1, cidr='fd00::1/64',
ip_version=lib_constants.IP_VERSION_6)
) as s1, self.subnet(network=n2, cidr='fd01::1/64',
ip_version=6) as s2:
ip_version=lib_constants.IP_VERSION_6
) as s2:
body = self._router_interface_action('add',
r['router']['id'],
s1['subnet']['id'],
@ -1219,7 +1229,8 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
for uc in use_cases:
with self.router() as r, self.network() as n:
with self.subnet(network=n, cidr='fd00::1/64',
gateway_ip='fd00::1', ip_version=6,
gateway_ip='fd00::1',
ip_version=lib_constants.IP_VERSION_6,
ipv6_ra_mode=uc['ra_mode'],
ipv6_address_mode=uc['address_mode']) as s:
exp_code = exc.HTTPBadRequest.code
@ -1232,7 +1243,8 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
def test_router_add_interface_ipv6_subnet_without_gateway_ip(self):
with self.router() as r:
with self.subnet(ip_version=6, cidr='fe80::/64',
with self.subnet(ip_version=lib_constants.IP_VERSION_6,
cidr='fe80::/64',
gateway_ip=None) as s:
error_code = exc.HTTPBadRequest.code
self._router_interface_action('add',
@ -1287,9 +1299,11 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
with self.network(shared=True) as n:
with self.subnet(network=n) as s1, (
self.subnet(network=n, cidr='fd00::/64',
ip_version=6)) as s2, (
ip_version=lib_constants.IP_VERSION_6)
) as s2, (
self.subnet(network=n, cidr='fd01::/64',
ip_version=6)) as s3:
ip_version=lib_constants.IP_VERSION_6)
) as s3:
fixed_ips = [{'subnet_id': s1['subnet']['id']},
{'subnet_id': s2['subnet']['id']},
{'subnet_id': s3['subnet']['id']}]
@ -1473,9 +1487,9 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
"""
with self.network() as n, self.router() as r:
with self.subnet(network=n, cidr='fd00::/64',
ip_version=6) as s1, (
ip_version=lib_constants.IP_VERSION_6) as s1, (
self.subnet(network=n, cidr='fd01::/64',
ip_version=6)) as s2:
ip_version=lib_constants.IP_VERSION_6)) as s2:
orig_update_port = self.plugin.update_port
with self.port(subnet=s1) as p, (
mock.patch.object(self.plugin,
@ -1504,9 +1518,9 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
"""
with self.network() as n, self.router() as r:
with self.subnet(network=n, cidr='fd00::/64',
ip_version=6) as s1, (
ip_version=lib_constants.IP_VERSION_6) as s1, (
self.subnet(network=n, cidr='fd01::/64',
ip_version=6)) as s2:
ip_version=lib_constants.IP_VERSION_6)) as s2:
fixed_ips = [{'subnet_id': s1['subnet']['id']},
{'subnet_id': s2['subnet']['id']}]
with self.port(subnet=s1, fixed_ips=fixed_ips) as p:
@ -1724,17 +1738,17 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
cidr='10.0.1.0/24', network=n)) as s2, (
self.subnet(
cidr='2001:db8::/64', network=n,
ip_version=6,
ip_version=lib_constants.IP_VERSION_6,
ipv6_ra_mode=lib_constants.IPV6_SLAAC,
ipv6_address_mode=lib_constants.IPV6_SLAAC)) as s3, (
self.subnet(
cidr='2001:db8:1::/64', network=n,
ip_version=6,
ip_version=lib_constants.IP_VERSION_6,
ipv6_ra_mode=lib_constants.DHCPV6_STATEFUL,
ipv6_address_mode=lib_constants.DHCPV6_STATEFUL)) as s4, (
self.subnet(
cidr='2001:db8:2::/64', network=n,
ip_version=6,
ip_version=lib_constants.IP_VERSION_6,
ipv6_ra_mode=lib_constants.DHCPV6_STATELESS,
ipv6_address_mode=lib_constants.DHCPV6_STATELESS)) as s5:
self._set_net_external(n['network']['id'])
@ -2053,9 +2067,11 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
removes the interface when it is the last subnet on the interface.
"""
with self.router() as r, self.network() as n:
with (self.subnet(network=n, cidr='fd00::1/64', ip_version=6)
with (self.subnet(network=n, cidr='fd00::1/64',
ip_version=lib_constants.IP_VERSION_6)
) as s1, self.subnet(network=n, cidr='fd01::1/64',
ip_version=6) as s2:
ip_version=lib_constants.IP_VERSION_6
) as s2:
body = self._router_interface_action('add', r['router']['id'],
s1['subnet']['id'],
None)
@ -3118,7 +3134,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
def _ipv6_subnet(self, mode):
return self.subnet(cidr='fd00::1/64', gateway_ip='fd00::1',
ip_version=6,
ip_version=lib_constants.IP_VERSION_6,
ipv6_ra_mode=mode,
ipv6_address_mode=mode)
@ -3222,7 +3238,8 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
self.assertEqual(router_req['router']['id'], result['id'])
def test_create_floatingip_ipv6_only_network_returns_400(self):
with self.subnet(cidr="2001:db8::/48", ip_version=6) as public_sub:
with self.subnet(cidr="2001:db8::/48",
ip_version=lib_constants.IP_VERSION_6) as public_sub:
self._set_net_external(public_sub['subnet']['network_id'])
res = self._create_floatingip(
self.fmt,
@ -3231,8 +3248,10 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
def test_create_floatingip_ipv6_and_ipv4_network_creates_ipv4(self):
with self.network() as n,\
self.subnet(cidr="2001:db8::/48", ip_version=6, network=n),\
self.subnet(cidr="192.168.1.0/24", ip_version=4, network=n):
self.subnet(cidr="2001:db8::/48",
ip_version=lib_constants.IP_VERSION_6, network=n),\
self.subnet(cidr="192.168.1.0/24",
ip_version=lib_constants.IP_VERSION_4, network=n):
self._set_net_external(n['network']['id'])
fip = self._make_floatingip(self.fmt, n['network']['id'])
fip_set = netaddr.IPSet(netaddr.IPNetwork("192.168.1.0/24"))
@ -3243,7 +3262,8 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
with self.subnet() as public_sub:
self._set_net_external(public_sub['subnet']['network_id'])
with self.subnet(cidr="2001:db8::/48",
ip_version=6) as private_sub:
ip_version=lib_constants.IP_VERSION_6
) as private_sub:
with self.port(subnet=private_sub) as private_port:
res = self._create_floatingip(
self.fmt,
@ -3254,7 +3274,8 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
def test_create_floatingip_with_assoc_to_ipv4_and_ipv6_port(self):
with self.network() as n,\
self.subnet(cidr='10.0.0.0/24', network=n) as s4,\
self.subnet(cidr='2001:db8::/64', ip_version=6, network=n),\
self.subnet(cidr='2001:db8::/64',
ip_version=lib_constants.IP_VERSION_6, network=n),\
self.port(subnet=s4) as p:
self.assertEqual(2, len(p['port']['fixed_ips']))
ipv4_address = next(i['ip_address'] for i in
@ -3437,7 +3458,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
'network_id': network['network']['id'],
'subnetpool_id': subnetpool_id,
'prefixlen': 24,
'ip_version': 4,
'ip_version': lib_constants.IP_VERSION_4,
'tenant_id': tenant_id}}
req = self.new_create_request('subnets', data)
subnet = self.deserialize(self.fmt, req.get_response(self.api))
@ -3534,7 +3555,8 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
exc.HTTPForbidden.code)
def test_associate_to_dhcp_port_fails(self):
with self.subnet(cidr="10.0.0.0/24", ip_version=4) as sub:
with self.subnet(cidr="10.0.0.0/24",
ip_version=lib_constants.IP_VERSION_4) as sub:
with self.port(subnet=sub,
device_owner=lib_constants.DEVICE_OWNER_DHCP) as p:
res = self._create_floatingip(

View File

@ -182,7 +182,7 @@ class TestL3GwModeMixin(testlib_api.SqlTestCase):
self.int_sub = subnet_obj.Subnet(self.context,
id=self.int_sub_id,
project_id=self.tenant_id,
ip_version=4,
ip_version=constants.IP_VERSION_4,
cidr=utils.AuthenticIPNetwork('3.3.3.0/24'),
gateway_ip=netaddr.IPAddress('3.3.3.1'),
network_id=self.int_net_id)

View File

@ -198,7 +198,7 @@ class TestNetworkIPAvailabilityAPI(
with self.network() as net:
with self.subnet(network=net):
# Get IPv4
params = 'ip_version=4'
params = 'ip_version=%s' % constants.IP_VERSION_4
request = self.new_list_request(API_RESOURCE, params=params)
response = self.deserialize(self.fmt,
request.get_response(self.ext_api))
@ -208,7 +208,7 @@ class TestNetworkIPAvailabilityAPI(
net, 0)
# Get IPv6 should return empty array
params = 'ip_version=6'
params = 'ip_version=%s' % constants.IP_VERSION_6
request = self.new_list_request(API_RESOURCE, params=params)
response = self.deserialize(self.fmt,
request.get_response(self.ext_api))
@ -218,10 +218,10 @@ class TestNetworkIPAvailabilityAPI(
with self.network() as net:
with self.subnet(
network=net, cidr='2607:f0d0:1002:51::/64',
ip_version=6,
ip_version=constants.IP_VERSION_6,
ipv6_address_mode=constants.DHCPV6_STATELESS):
# Get IPv6
params = 'ip_version=6'
params = 'ip_version=%s' % constants.IP_VERSION_6
request = self.new_list_request(API_RESOURCE, params=params)
response = self.deserialize(self.fmt,
request.get_response(self.ext_api))
@ -230,7 +230,7 @@ class TestNetworkIPAvailabilityAPI(
response[IP_AVAILS_KEY], net, 0, 18446744073709551614)
# Get IPv4 should return empty array
params = 'ip_version=4'
params = 'ip_version=%s' % constants.IP_VERSION_4
request = self.new_list_request(API_RESOURCE, params=params)
response = self.deserialize(self.fmt,
request.get_response(self.ext_api))
@ -240,7 +240,7 @@ class TestNetworkIPAvailabilityAPI(
with self.network() as net:
with self.subnet(
network=net, cidr='2607:f0d0:1002:51::/64',
ip_version=6,
ip_version=constants.IP_VERSION_6,
ipv6_address_mode=constants.DHCPV6_STATELESS) as subnet:
request = self.new_list_request(API_RESOURCE)
# Consume 3 ports
@ -350,10 +350,10 @@ class TestNetworkIPAvailabilityAPI(
self.network(name='net-v4-1') as net_v4_1, \
self.network(name='net-v4-2') as net_v4_2:
with self.subnet(network=net_v6_1, cidr='2607:f0d0:1002:51::/64',
ip_version=6) as s61, \
ip_version=constants.IP_VERSION_6) as s61, \
self.subnet(network=net_v6_2,
cidr='2607:f0d0:1003:52::/64',
ip_version=6) as s62, \
ip_version=constants.IP_VERSION_6) as s62, \
self.subnet(network=net_v4_1, cidr='10.0.0.0/24') as s41, \
self.subnet(network=net_v4_2, cidr='10.0.1.0/24') as s42:
with self.port(subnet=s61),\
@ -376,7 +376,8 @@ class TestNetworkIPAvailabilityAPI(
avails_list, net_v4_2, 2, 253)
# Query by IP versions. Ensure subnet versions match
for ip_ver in [4, 6]:
for ip_ver in [constants.IP_VERSION_4,
constants.IP_VERSION_6]:
params = 'ip_version=%i' % ip_ver
request = self.new_list_request(API_RESOURCE,
params=params)

View File

@ -569,9 +569,11 @@ class TestSegmentSubnetAssociation(SegmentTestCase):
segmentation_id=201)['segment']
with self.subnet(network=network, segment_id=None,
ip_version='4', cidr='10.0.0.0/24') as subnet1, \
ip_version=constants.IP_VERSION_4,
cidr='10.0.0.0/24') as subnet1, \
self.subnet(network=network, segment_id=None,
ip_version='4', cidr='10.0.1.0/24') as subnet2:
ip_version=constants.IP_VERSION_4,
cidr='10.0.1.0/24') as subnet2:
subnet1 = subnet1['subnet']
subnet2 = subnet2['subnet']
@ -1081,7 +1083,7 @@ class TestSegmentAwareIpam(SegmentAwareIpamTestCase):
"""Binding information is provided, subnets not on segments"""
with self.network() as network:
with self.subnet(network=network,
ip_version=6,
ip_version=constants.IP_VERSION_6,
cidr='2001:db8:0:0::/64') as subnet:
segment = self._test_create_segment(
network_id=network['network']['id'],
@ -2224,7 +2226,7 @@ class TestDhcpAgentSegmentScheduling(HostSegmentMappingTestCase):
def _test_create_subnet(self, network, segment, cidr=None,
enable_dhcp=True):
cidr = cidr or '10.0.0.0/24'
ip_version = 4
ip_version = constants.IP_VERSION_4
with self.subnet(network={'network': network},
segment_id=segment['id'],
ip_version=ip_version,

View File

@ -44,7 +44,8 @@ class TestNeutronDbIpamMixin(object):
created_network = plugin.create_network(ctx, network)
return (created_network, created_network['id'])
def _create_subnet(self, plugin, ctx, network_id, cidr, ip_version=4,
def _create_subnet(self, plugin, ctx, network_id, cidr,
ip_version=constants.IP_VERSION_4,
v6_address_mode=constants.ATTR_NOT_SPECIFIED,
allocation_pools=constants.ATTR_NOT_SPECIFIED):
subnet = {'subnet': {'name': 'sub',
@ -232,12 +233,13 @@ class TestNeutronDbIpamSubnet(testlib_api.SqlTestCase,
def _create_and_allocate_ipam_subnet(
self, cidr, allocation_pools=constants.ATTR_NOT_SPECIFIED,
ip_version=4, v6_auto_address=False, tenant_id=None):
ip_version=constants.IP_VERSION_4, v6_auto_address=False,
tenant_id=None):
v6_address_mode = constants.ATTR_NOT_SPECIFIED
if v6_auto_address:
# set ip version to 6 regardless of what's been passed to the
# method
ip_version = 6
ip_version = constants.IP_VERSION_6
v6_address_mode = n_const.IPV6_SLAAC
subnet = self._create_subnet(
self.plugin, self.ctx, self.net_id, cidr,
@ -342,7 +344,7 @@ class TestNeutronDbIpamSubnet(testlib_api.SqlTestCase,
def test_allocate_specific_address_in_use_fails(self):
ipam_subnet = self._create_and_allocate_ipam_subnet(
'fde3:abcd:4321:1::/64', ip_version=6)[0]
'fde3:abcd:4321:1::/64', ip_version=constants.IP_VERSION_6)[0]
addr_req = ipam_req.SpecificAddressRequest('fde3:abcd:4321:1::33')
ipam_subnet.allocate(addr_req)
self.assertRaises(ipam_exc.IpAddressAlreadyAllocated,
@ -352,7 +354,7 @@ class TestNeutronDbIpamSubnet(testlib_api.SqlTestCase,
def test_allocate_any_address_exhausted_pools_fails(self):
# Same as above, the ranges will be recalculated always
ipam_subnet = self._create_and_allocate_ipam_subnet(
'192.168.0.0/30', ip_version=4)[0]
'192.168.0.0/30', ip_version=constants.IP_VERSION_4)[0]
ipam_subnet.allocate(ipam_req.AnyAddressRequest)
# The second address generation request on a /30 for v4 net must fail
self.assertRaises(ipam_exc.IpAddressGenerationFailure,
@ -376,7 +378,7 @@ class TestNeutronDbIpamSubnet(testlib_api.SqlTestCase,
def test_allocate_unallocated_address_fails(self):
ipam_subnet = self._create_and_allocate_ipam_subnet(
'10.0.0.0/24', ip_version=4)[0]
'10.0.0.0/24', ip_version=constants.IP_VERSION_4)[0]
self.assertRaises(ipam_exc.IpAddressAllocationNotFound,
ipam_subnet.deallocate, '10.0.0.2')

View File

@ -320,7 +320,7 @@ class TestAddressRequestFactory(base.BaseTestCase):
class TestSubnetRequestFactory(IpamSubnetRequestTestCase):
def _build_subnet_dict(self, id=None, cidr='192.168.1.0/24',
prefixlen=8, ip_version=4):
prefixlen=8, ip_version=constants.IP_VERSION_4):
subnet = {'cidr': cidr,
'prefixlen': prefixlen,
'ip_version': ip_version,
@ -347,7 +347,8 @@ class TestSubnetRequestFactory(IpamSubnetRequestTestCase):
ipam_req.SpecificSubnetRequest)
def test_any_address_request_is_loaded_for_ipv4(self):
subnet, subnetpool = self._build_subnet_dict(cidr=None, ip_version=4)
subnet, subnetpool = self._build_subnet_dict(
cidr=None, ip_version=constants.IP_VERSION_4)
self.assertIsInstance(
ipam_req.SubnetRequestFactory.get_request(None,
subnet,
@ -355,7 +356,8 @@ class TestSubnetRequestFactory(IpamSubnetRequestTestCase):
ipam_req.AnySubnetRequest)
def test_any_address_request_is_loaded_for_ipv6(self):
subnet, subnetpool = self._build_subnet_dict(cidr=None, ip_version=6)
subnet, subnetpool = self._build_subnet_dict(
cidr=None, ip_version=constants.IP_VERSION_6)
self.assertIsInstance(
ipam_req.SubnetRequestFactory.get_request(None,
subnet,

View File

@ -1492,7 +1492,7 @@ class BaseDbObjectTestCase(_BaseObjectTestCase,
'project_id': uuidutils.generate_uuid(),
'name': 'test-subnet1',
'network_id': network_id,
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'cidr': netaddr.IPNetwork('10.0.0.0/24'),
'gateway_ip': '10.0.0.1',
'enable_dhcp': 1,

View File

@ -13,6 +13,7 @@
# License for the specific language governing permissions and limitations
# under the License.
from neutron_lib import constants
from oslo_utils import uuidutils
from neutron.objects import subnetpool
@ -25,7 +26,7 @@ class SubnetPoolTestMixin(object):
obj = subnetpool.SubnetPool(
self.context,
id=uuidutils.generate_uuid(),
ip_version=4,
ip_version=constants.IP_VERSION_4,
default_prefixlen=24,
min_prefixlen=0,
max_prefixlen=32,

View File

@ -359,7 +359,8 @@ class TunnelRpcCallbackTestMixin(object):
self._test_tunnel_sync(kwargs)
def test_tunnel_sync_called_with_host_passed_ipv6(self):
cfg.CONF.set_override('overlay_ip_version', 6, group='ml2')
cfg.CONF.set_override('overlay_ip_version', p_const.IP_VERSION_6,
group='ml2')
kwargs = {'tunnel_ip': TUNNEL_IPV6_ONE, 'tunnel_type': self.TYPE,
'host': HOST_ONE}
self._test_tunnel_sync(kwargs)
@ -402,13 +403,15 @@ class TunnelRpcCallbackTestMixin(object):
self._test_tunnel_sync_raises(kwargs)
def test_tunnel_sync_called_with_tunnel_overlay_mismatch(self):
cfg.CONF.set_override('overlay_ip_version', 6, group='ml2')
cfg.CONF.set_override('overlay_ip_version', p_const.IP_VERSION_6,
group='ml2')
kwargs = {'tunnel_ip': TUNNEL_IP_ONE, 'tunnel_type': self.TYPE,
'host': HOST_ONE}
self._test_tunnel_sync_raises(kwargs)
def test_tunnel_sync_called_with_tunnel_overlay_mismatch_ipv6(self):
cfg.CONF.set_override('overlay_ip_version', 4, group='ml2')
cfg.CONF.set_override('overlay_ip_version', p_const.IP_VERSION_4,
group='ml2')
kwargs = {'tunnel_ip': TUNNEL_IPV6_ONE, 'tunnel_type': self.TYPE,
'host': HOST_ONE}
self._test_tunnel_sync_raises(kwargs)

View File

@ -741,7 +741,7 @@ class TestL2PopulationRpcTestCase(test_plugin.Ml2PluginV2TestCase):
self.subnet(
self._network,
cidr='2001:db8::/64',
ip_version=6,
ip_version=constants.IP_VERSION_6,
gateway_ip='fe80::1',
ipv6_address_mode=constants.IPV6_SLAAC) as subnet2:
with self.port(

View File

@ -253,8 +253,10 @@ class QosLinuxbridgeAgentDriverTestCase(base.BaseTestCase):
iptables_manager.ipv4['mangle'].assert_has_calls(expected_calls)
iptables_manager.ipv6['mangle'].assert_has_calls(expected_calls)
iptables_manager.get_chain.assert_has_calls([
mock.call("mangle", dscp_chain_name, ip_version=4),
mock.call("mangle", dscp_chain_name, ip_version=6)
mock.call("mangle", dscp_chain_name,
ip_version=constants.IP_VERSION_4),
mock.call("mangle", dscp_chain_name,
ip_version=constants.IP_VERSION_6)
])
def test_delete_dscp_marking_chain_not_empty(self):
@ -274,8 +276,10 @@ class QosLinuxbridgeAgentDriverTestCase(base.BaseTestCase):
iptables_manager.ipv4['mangle'].assert_has_calls(expected_calls)
iptables_manager.ipv6['mangle'].assert_has_calls(expected_calls)
iptables_manager.get_chain.assert_has_calls([
mock.call("mangle", dscp_chain_name, ip_version=4),
mock.call("mangle", dscp_chain_name, ip_version=6)
mock.call("mangle", dscp_chain_name,
ip_version=constants.IP_VERSION_4),
mock.call("mangle", dscp_chain_name,
ip_version=constants.IP_VERSION_6)
])
iptables_manager.ipv4['mangle'].remove_chain.assert_not_called()
iptables_manager.ipv4['mangle'].remove_rule.assert_not_called()

View File

@ -296,7 +296,7 @@ class TestLinuxBridgeManager(base.BaseTestCase):
ipdict = dict(cidr='1.1.1.1/24',
broadcast='1.1.1.255',
scope='global',
ip_version=4,
ip_version=constants.IP_VERSION_4,
dynamic=False)
list_fn.return_value = ipdict
ret = self.lbm.get_interface_details("eth0", 4)
@ -477,7 +477,7 @@ class TestLinuxBridgeManager(base.BaseTestCase):
ipdict = dict(cidr='1.1.1.1/24',
broadcast='1.1.1.255',
scope='global',
ip_version=4,
ip_version=constants.IP_VERSION_4,
dynamic=False)
with mock.patch.object(ip_lib.IpAddrCommand, 'add') as add_fn,\
mock.patch.object(ip_lib.IpAddrCommand, 'delete') as del_fn,\

View File

@ -2579,7 +2579,7 @@ class TestOvsDvrNeutronAgent(object):
def _expected_install_dvr_process(self, lvid, port, ip_version,
gateway_ip, gateway_mac):
if ip_version == 4:
if ip_version == n_const.IP_VERSION_4:
ipvx_calls = [
mock.call.install_dvr_process_ipv4(
vlan_tag=lvid,
@ -2599,10 +2599,10 @@ class TestOvsDvrNeutronAgent(object):
),
]
def _test_port_bound_for_dvr_on_vlan_network(self, device_owner,
ip_version=4):
def _test_port_bound_for_dvr_on_vlan_network(
self, device_owner, ip_version=n_const.IP_VERSION_4):
self._setup_for_dvr_test()
if ip_version == 4:
if ip_version == n_const.IP_VERSION_4:
gateway_ip = '1.1.1.1'
cidr = '1.1.1.0/24'
else:
@ -2689,10 +2689,10 @@ class TestOvsDvrNeutronAgent(object):
self.assertFalse([], tun_br.mock_calls)
self.assertFalse([], phys_br.mock_calls)
def _test_port_bound_for_dvr_on_vxlan_network(self, device_owner,
ip_version=4):
def _test_port_bound_for_dvr_on_vxlan_network(
self, device_owner, ip_version=n_const.IP_VERSION_4):
self._setup_for_dvr_test()
if ip_version == 4:
if ip_version == n_const.IP_VERSION_4:
gateway_ip = '1.1.1.1'
cidr = '1.1.1.0/24'
else:
@ -2775,41 +2775,49 @@ class TestOvsDvrNeutronAgent(object):
self._test_port_bound_for_dvr_on_vlan_network(
device_owner=DEVICE_OWNER_COMPUTE)
self._test_port_bound_for_dvr_on_vlan_network(
device_owner=DEVICE_OWNER_COMPUTE, ip_version=6)
device_owner=DEVICE_OWNER_COMPUTE,
ip_version=n_const.IP_VERSION_6)
self._test_port_bound_for_dvr_on_vxlan_network(
device_owner=DEVICE_OWNER_COMPUTE)
self._test_port_bound_for_dvr_on_vxlan_network(
device_owner=DEVICE_OWNER_COMPUTE, ip_version=6)
device_owner=DEVICE_OWNER_COMPUTE,
ip_version=n_const.IP_VERSION_6)
def test_port_bound_for_dvr_with_lbaas_vip_ports(self):
self._test_port_bound_for_dvr_on_vlan_network(
device_owner=n_const.DEVICE_OWNER_LOADBALANCER)
self._test_port_bound_for_dvr_on_vlan_network(
device_owner=n_const.DEVICE_OWNER_LOADBALANCER, ip_version=6)
device_owner=n_const.DEVICE_OWNER_LOADBALANCER,
ip_version=n_const.IP_VERSION_6)
self._test_port_bound_for_dvr_on_vxlan_network(
device_owner=n_const.DEVICE_OWNER_LOADBALANCER)
self._test_port_bound_for_dvr_on_vxlan_network(
device_owner=n_const.DEVICE_OWNER_LOADBALANCER, ip_version=6)
device_owner=n_const.DEVICE_OWNER_LOADBALANCER,
ip_version=n_const.IP_VERSION_6)
def test_port_bound_for_dvr_with_lbaasv2_vip_ports(self):
self._test_port_bound_for_dvr_on_vlan_network(
device_owner=n_const.DEVICE_OWNER_LOADBALANCERV2)
self._test_port_bound_for_dvr_on_vlan_network(
device_owner=n_const.DEVICE_OWNER_LOADBALANCERV2, ip_version=6)
device_owner=n_const.DEVICE_OWNER_LOADBALANCERV2,
ip_version=n_const.IP_VERSION_6)
self._test_port_bound_for_dvr_on_vxlan_network(
device_owner=n_const.DEVICE_OWNER_LOADBALANCERV2)
self._test_port_bound_for_dvr_on_vxlan_network(
device_owner=n_const.DEVICE_OWNER_LOADBALANCERV2, ip_version=6)
device_owner=n_const.DEVICE_OWNER_LOADBALANCERV2,
ip_version=n_const.IP_VERSION_6)
def test_port_bound_for_dvr_with_dhcp_ports(self):
self._test_port_bound_for_dvr_on_vlan_network(
device_owner=n_const.DEVICE_OWNER_DHCP)
self._test_port_bound_for_dvr_on_vlan_network(
device_owner=n_const.DEVICE_OWNER_DHCP, ip_version=6)
device_owner=n_const.DEVICE_OWNER_DHCP,
ip_version=n_const.IP_VERSION_6)
self._test_port_bound_for_dvr_on_vxlan_network(
device_owner=n_const.DEVICE_OWNER_DHCP)
self._test_port_bound_for_dvr_on_vxlan_network(
device_owner=n_const.DEVICE_OWNER_DHCP, ip_version=6)
device_owner=n_const.DEVICE_OWNER_DHCP,
ip_version=n_const.IP_VERSION_6)
def test_port_bound_for_dvr_with_csnat_ports(self):
self._setup_for_dvr_test()
@ -2899,7 +2907,7 @@ class TestOvsDvrNeutronAgent(object):
'get_subnet_for_dvr',
return_value={'gateway_ip': '1.1.1.1',
'cidr': '1.1.1.0/24',
'ip_version': 4,
'ip_version': n_const.IP_VERSION_4,
'gateway_mac': 'aa:bb:cc:11:22:33'}),\
mock.patch.object(self.agent.dvr_agent.plugin_rpc,
'get_ports_on_host_by_subnet',
@ -2940,15 +2948,16 @@ class TestOvsDvrNeutronAgent(object):
def test_treat_devices_removed_for_dvr_interface(self):
self._test_treat_devices_removed_for_dvr_interface()
self._test_treat_devices_removed_for_dvr_interface(ip_version=6)
self._test_treat_devices_removed_for_dvr_interface(
ip_version=n_const.IP_VERSION_6)
self._test_treat_devices_removed_for_dvr_interface(network_type='vlan')
self._test_treat_devices_removed_for_dvr_interface(ip_version=6,
network_type='vlan')
self._test_treat_devices_removed_for_dvr_interface(
ip_version=n_const.IP_VERSION_6, network_type='vlan')
def _test_treat_devices_removed_for_dvr_interface(
self, ip_version=4, network_type='vxlan'):
self, ip_version=n_const.IP_VERSION_4, network_type='vxlan'):
self._setup_for_dvr_test()
if ip_version == 4:
if ip_version == n_const.IP_VERSION_4:
gateway_ip = '1.1.1.1'
cidr = '1.1.1.0/24'
else:
@ -3024,7 +3033,7 @@ class TestOvsDvrNeutronAgent(object):
failed_devices['removed'] = self.agent.treat_devices_removed(
[self._port.vif_id])
lvid = self.agent.vlan_manager.get(self._net_uuid).vlan
if ip_version == 4:
if ip_version == n_const.IP_VERSION_4:
expected = [
mock.call.delete_dvr_process_ipv4(
vlan_tag=lvid,
@ -3051,9 +3060,10 @@ class TestOvsDvrNeutronAgent(object):
self.assertEqual(expected, tun_br.mock_calls)
self.assertEqual([], phys_br.mock_calls)
def _test_treat_devices_removed_for_dvr(self, device_owner, ip_version=4):
def _test_treat_devices_removed_for_dvr(self, device_owner,
ip_version=n_const.IP_VERSION_4):
self._setup_for_dvr_test()
if ip_version == 4:
if ip_version == n_const.IP_VERSION_4:
gateway_ip = '1.1.1.1'
cidr = '1.1.1.0/24'
else:
@ -3152,25 +3162,28 @@ class TestOvsDvrNeutronAgent(object):
self._test_treat_devices_removed_for_dvr(
device_owner=DEVICE_OWNER_COMPUTE)
self._test_treat_devices_removed_for_dvr(
device_owner=DEVICE_OWNER_COMPUTE, ip_version=6)
device_owner=DEVICE_OWNER_COMPUTE, ip_version=n_const.IP_VERSION_6)
def test_treat_devices_removed_for_dvr_with_lbaas_vip_ports(self):
self._test_treat_devices_removed_for_dvr(
device_owner=n_const.DEVICE_OWNER_LOADBALANCER)
self._test_treat_devices_removed_for_dvr(
device_owner=n_const.DEVICE_OWNER_LOADBALANCER, ip_version=6)
device_owner=n_const.DEVICE_OWNER_LOADBALANCER,
ip_version=n_const.IP_VERSION_6)
def test_treat_devices_removed_for_dvr_with_lbaasv2_vip_ports(self):
self._test_treat_devices_removed_for_dvr(
device_owner=n_const.DEVICE_OWNER_LOADBALANCERV2)
self._test_treat_devices_removed_for_dvr(
device_owner=n_const.DEVICE_OWNER_LOADBALANCERV2, ip_version=6)
device_owner=n_const.DEVICE_OWNER_LOADBALANCERV2,
ip_version=n_const.IP_VERSION_6)
def test_treat_devices_removed_for_dvr_with_dhcp_ports(self):
self._test_treat_devices_removed_for_dvr(
device_owner=n_const.DEVICE_OWNER_DHCP)
self._test_treat_devices_removed_for_dvr(
device_owner=n_const.DEVICE_OWNER_DHCP, ip_version=6)
device_owner=n_const.DEVICE_OWNER_DHCP,
ip_version=n_const.IP_VERSION_6)
def test_treat_devices_removed_for_dvr_csnat_port(self):
self._setup_for_dvr_test()
@ -3183,7 +3196,7 @@ class TestOvsDvrNeutronAgent(object):
'get_subnet_for_dvr',
return_value={'gateway_ip': '1.1.1.1',
'cidr': '1.1.1.0/24',
'ip_version': 4,
'ip_version': n_const.IP_VERSION_4,
'gateway_mac': gateway_mac}),\
mock.patch.object(self.agent.dvr_agent.plugin_rpc,
'get_ports_on_host_by_subnet',

View File

@ -11,6 +11,7 @@
# under the License.
import mock
from neutron_lib import constants
from neutron_lib import context
from neutron_lib.plugins import directory
from oslo_config import cfg
@ -242,7 +243,7 @@ class DBExtensionDriverTestCase(test_plugin.Ml2PluginV2TestCase):
data = {'subnet':
{'network_id': network['network']['id'],
'cidr': '10.1.0.0/24',
'ip_version': '4',
'ip_version': constants.IP_VERSION_4,
'tenant_id': self._tenant_id,
'subnet_extension': 'abc'}}
req = self.new_create_request('subnets', data, self.fmt)

View File

@ -1096,7 +1096,7 @@ class TestMl2PortsV2(test_plugin.TestPortsV2, Ml2PluginV2TestCase):
with self.subnet(net,
gateway_ip=fake_gateway,
cidr=fake_prefix,
ip_version=6) as snet_v6,\
ip_version=constants.IP_VERSION_6) as snet_v6,\
mock.patch.object(
plugin.notifier,
'security_groups_member_updated') as m_upd:
@ -1325,7 +1325,7 @@ fixed_ips=ip_address_substr%%3D%s&fixed_ips=subnet_id%%3D%s
tenant_id=tenant_id,
net_id=net_id,
cidr='2607:f0d0:1002:51::/124',
ip_version=6,
ip_version=constants.IP_VERSION_6,
gateway_ip=constants.ATTR_NOT_SPECIFIED)
subnet2 = self.deserialize(self.fmt, res)
kwargs = {"fixed_ips":
@ -2569,7 +2569,7 @@ class TestFaultyMechansimDriver(Ml2PluginV2FaultyDriverTestCase):
net_id = network['network']['id']
data = {'subnet': {'network_id': net_id,
'cidr': '10.0.20.0/24',
'ip_version': '4',
'ip_version': constants.IP_VERSION_4,
'name': 'subnet1',
'tenant_id':
network['network']['tenant_id'],
@ -2598,7 +2598,7 @@ class TestFaultyMechansimDriver(Ml2PluginV2FaultyDriverTestCase):
data = {'subnet': {'network_id':
network['network']['id'],
'cidr': '10.0.20.0/24',
'ip_version': '4',
'ip_version': constants.IP_VERSION_4,
'name': 'subnet1',
'tenant_id':
network['network']['tenant_id'],
@ -2631,7 +2631,7 @@ class TestFaultyMechansimDriver(Ml2PluginV2FaultyDriverTestCase):
data = {'subnet': {'network_id':
network['network']['id'],
'cidr': '10.0.20.0/24',
'ip_version': '4',
'ip_version': constants.IP_VERSION_4,
'name': 'subnet1',
'tenant_id':
network['network']['tenant_id'],
@ -2797,7 +2797,7 @@ class TestML2PluggableIPAM(test_ipam.UseIpamMixin, TestMl2SubnetsV2):
with self.network() as network:
with self.subnet(network=network,
cidr='2001:100::0/64',
ip_version=6,
ip_version=constants.IP_VERSION_6,
ipv6_ra_mode=constants.IPV6_SLAAC) as subnet:
with self.port(subnet=subnet) as port:
with mock.patch(driver) as driver_mock:

View File

@ -331,7 +331,7 @@ class TestAutoScheduleSegments(test_plugin.Ml2PluginV2TestCase,
subnet = self.plugin.create_subnet(
self.ctx,
{'subnet': {'name': 'name',
'ip_version': 4,
'ip_version': constants.IP_VERSION_4,
'network_id': network_id,
'cidr': cidr,
'gateway_ip': constants.ATTR_NOT_SPECIFIED,

View File

@ -15,6 +15,7 @@ import mock
import testtools
from neutron_lib.api.definitions import constants as api_const
from neutron_lib import constants
from neutron_lib import context
from neutron_lib import exceptions as n_exc
from oslo_db import exception as db_exc
@ -219,7 +220,8 @@ class AutoAllocateTestCase(testlib_api.SqlTestCase):
with mock.patch.object(self.mixin, "_get_supported_subnetpools") as f,\
mock.patch.object(self.mixin, "_cleanup") as g:
f.return_value = (
[{'ip_version': 4, "id": uuidutils.generate_uuid()}])
[{'ip_version': constants.IP_VERSION_4,
'id': uuidutils.generate_uuid()}])
self.assertRaises(exceptions.AutoAllocationFailure,
self.mixin._provision_tenant_private_network,
self.ctx, 'foo_tenant')