1 # Copyright (c) 2012 OpenStack Foundation.
4 # Licensed under the Apache License, Version 2.0 (the "License"); you may
5 # not use this file except in compliance with the License. You may obtain
6 # a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13 # License for the specific language governing permissions and limitations
17 from oslo_config import cfg
19 from neutron.api.v2 import attributes
20 from neutron.common import constants
21 from neutron.common import exceptions as n_exc
22 from neutron.common import ipv6_utils
23 from neutron.db import db_base_plugin_common
24 from neutron.db import db_base_plugin_v2
25 from neutron.db import ipam_non_pluggable_backend as non_ipam
26 from neutron.db import models_v2
27 from neutron.tests import base
30 class TestIpamNonPluggableBackend(base.BaseTestCase):
31 """Unit Tests for non pluggable IPAM Logic."""
33 def test_generate_ip(self):
34 with mock.patch.object(non_ipam.IpamNonPluggableBackend,
35 '_try_generate_ip') as generate:
36 with mock.patch.object(non_ipam.IpamNonPluggableBackend,
37 '_rebuild_availability_ranges') as rebuild:
39 non_ipam.IpamNonPluggableBackend._generate_ip('c', 's')
41 generate.assert_called_once_with('c', 's')
42 self.assertEqual(0, rebuild.call_count)
44 def test_generate_ip_exhausted_pool(self):
45 with mock.patch.object(non_ipam.IpamNonPluggableBackend,
46 '_try_generate_ip') as generate:
47 with mock.patch.object(non_ipam.IpamNonPluggableBackend,
48 '_rebuild_availability_ranges') as rebuild:
50 exception = n_exc.IpAddressGenerationFailure(net_id='n')
51 # fail first call but not second
52 generate.side_effect = [exception, None]
53 non_ipam.IpamNonPluggableBackend._generate_ip('c', 's')
55 self.assertEqual(2, generate.call_count)
56 rebuild.assert_called_once_with('c', 's')
58 def _validate_rebuild_availability_ranges(self, pools, allocations,
61 ip_qry.with_lockmode.return_value = ip_qry
62 ip_qry.filter_by.return_value = allocations
64 pool_qry = mock.Mock()
65 pool_qry.options.return_value = pool_qry
66 pool_qry.with_lockmode.return_value = pool_qry
67 pool_qry.filter_by.return_value = pools
69 def return_queries_side_effect(*args, **kwargs):
70 if args[0] == models_v2.IPAllocation:
72 if args[0] == models_v2.IPAllocationPool:
76 context.session.query.side_effect = return_queries_side_effect
77 subnets = [mock.MagicMock()]
79 non_ipam.IpamNonPluggableBackend._rebuild_availability_ranges(
82 actual = [[args[0].allocation_pool_id,
83 args[0].first_ip, args[0].last_ip]
84 for _name, args, _kwargs in context.session.add.mock_calls]
85 self.assertEqual(expected, actual)
87 def test_rebuild_availability_ranges(self):
89 'first_ip': '192.168.1.3',
90 'last_ip': '192.168.1.10'},
92 'first_ip': '192.168.1.100',
93 'last_ip': '192.168.1.120'}]
95 allocations = [{'ip_address': '192.168.1.3'},
96 {'ip_address': '192.168.1.78'},
97 {'ip_address': '192.168.1.7'},
98 {'ip_address': '192.168.1.110'},
99 {'ip_address': '192.168.1.11'},
100 {'ip_address': '192.168.1.4'},
101 {'ip_address': '192.168.1.111'}]
103 expected = [['a', '192.168.1.5', '192.168.1.6'],
104 ['a', '192.168.1.8', '192.168.1.10'],
105 ['b', '192.168.1.100', '192.168.1.109'],
106 ['b', '192.168.1.112', '192.168.1.120']]
108 self._validate_rebuild_availability_ranges(pools, allocations,
111 def test_rebuild_ipv6_availability_ranges(self):
113 'first_ip': '2001::1',
114 'last_ip': '2001::50'},
116 'first_ip': '2001::100',
117 'last_ip': '2001::ffff:ffff:ffff:fffe'}]
119 allocations = [{'ip_address': '2001::10'},
120 {'ip_address': '2001::45'},
121 {'ip_address': '2001::60'},
122 {'ip_address': '2001::111'},
123 {'ip_address': '2001::200'},
124 {'ip_address': '2001::ffff:ffff:ffff:ff10'},
125 {'ip_address': '2001::ffff:ffff:ffff:f2f0'}]
127 expected = [['a', '2001::1', '2001::f'],
128 ['a', '2001::11', '2001::44'],
129 ['a', '2001::46', '2001::50'],
130 ['b', '2001::100', '2001::110'],
131 ['b', '2001::112', '2001::1ff'],
132 ['b', '2001::201', '2001::ffff:ffff:ffff:f2ef'],
133 ['b', '2001::ffff:ffff:ffff:f2f1',
134 '2001::ffff:ffff:ffff:ff0f'],
135 ['b', '2001::ffff:ffff:ffff:ff11',
136 '2001::ffff:ffff:ffff:fffe']]
138 self._validate_rebuild_availability_ranges(pools, allocations,
141 def _test__allocate_ips_for_port(self, subnets, port, expected):
142 # this test is incompatible with pluggable ipam, because subnets
143 # were not actually created, so no ipam_subnet exists
144 cfg.CONF.set_override("ipam_driver", None)
145 plugin = db_base_plugin_v2.NeutronDbPluginV2()
146 with mock.patch.object(db_base_plugin_common.DbBasePluginCommon,
147 '_get_subnets') as get_subnets:
148 with mock.patch.object(non_ipam.IpamNonPluggableBackend,
149 '_check_unique_ip') as check_unique:
150 context = mock.Mock()
151 get_subnets.return_value = subnets
152 check_unique.return_value = True
153 actual = plugin.ipam._allocate_ips_for_port(context, port)
154 self.assertEqual(expected, actual)
156 def test__allocate_ips_for_port_2_slaac_subnets(self):
159 'cidr': u'2001:100::/64',
161 'gateway_ip': u'2001:100::1',
162 'id': u'd1a28edd-bd83-480a-bd40-93d036c89f13',
163 'network_id': 'fbb9b578-95eb-4b79-a116-78e5c4927176',
165 'ipv6_address_mode': None,
166 'ipv6_ra_mode': u'slaac'},
168 'cidr': u'2001:200::/64',
170 'gateway_ip': u'2001:200::1',
171 'id': u'dc813d3d-ed66-4184-8570-7325c8195e28',
172 'network_id': 'fbb9b578-95eb-4b79-a116-78e5c4927176',
174 'ipv6_address_mode': None,
175 'ipv6_ra_mode': u'slaac'}]
177 'network_id': 'fbb9b578-95eb-4b79-a116-78e5c4927176',
178 'fixed_ips': attributes.ATTR_NOT_SPECIFIED,
179 'mac_address': '12:34:56:78:44:ab',
180 'device_owner': 'compute'}}
182 for subnet in subnets:
183 addr = str(ipv6_utils.get_ipv6_addr_by_EUI64(
184 subnet['cidr'], port['port']['mac_address']))
185 expected.append({'ip_address': addr, 'subnet_id': subnet['id']})
187 self._test__allocate_ips_for_port(subnets, port, expected)
189 def test__allocate_ips_for_port_2_slaac_pd_subnets(self):
192 'cidr': constants.PROVISIONAL_IPV6_PD_PREFIX,
195 'id': 'd1a28edd-bd83-480a-bd40-93d036c89f13',
196 'network_id': 'fbb9b578-95eb-4b79-a116-78e5c4927176',
198 'ipv6_address_mode': None,
199 'ipv6_ra_mode': 'slaac'},
201 'cidr': constants.PROVISIONAL_IPV6_PD_PREFIX,
204 'id': 'dc813d3d-ed66-4184-8570-7325c8195e28',
205 'network_id': 'fbb9b578-95eb-4b79-a116-78e5c4927176',
207 'ipv6_address_mode': None,
208 'ipv6_ra_mode': 'slaac'}]
210 'network_id': 'fbb9b578-95eb-4b79-a116-78e5c4927176',
211 'fixed_ips': attributes.ATTR_NOT_SPECIFIED,
212 'mac_address': '12:34:56:78:44:ab',
213 'device_owner': 'compute'}}
215 for subnet in subnets:
216 addr = str(ipv6_utils.get_ipv6_addr_by_EUI64(
217 subnet['cidr'], port['port']['mac_address']))
218 expected.append({'ip_address': addr, 'subnet_id': subnet['id']})
220 self._test__allocate_ips_for_port(subnets, port, expected)