772 lines
36 KiB
Python
772 lines
36 KiB
Python
# Copyright 2013 Openstack Foundation
|
|
# All Rights Reserved.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
|
|
import contextlib
|
|
|
|
import mock
|
|
from mock import patch
|
|
from neutron.common import exceptions
|
|
from oslo_config import cfg
|
|
import webob
|
|
|
|
from quark.db import ip_types
|
|
from quark.db import models
|
|
from quark import exceptions as quark_exceptions
|
|
from quark.plugin_modules import ip_addresses
|
|
from quark.tests import test_quark_plugin
|
|
|
|
|
|
def _port_associate_stub(context, ports, address, **kwargs):
|
|
for port in ports:
|
|
assoc = models.PortIpAssociation()
|
|
assoc.port_id = port.id
|
|
assoc.ip_address_id = address.id
|
|
assoc.port = port
|
|
# NOTE(thomasem): This causes address['associations'] to gain this
|
|
# PortIpAssocation instance.
|
|
assoc.ip_address = address
|
|
assoc.enabled = address.address_type == "fixed"
|
|
return address
|
|
|
|
|
|
def _port_disassociate_stub(context, ports, address):
|
|
port_ids = [port.id for port in ports]
|
|
for idx, assoc in enumerate(address['associations']):
|
|
if assoc.port_id in port_ids:
|
|
address.associations.pop(idx)
|
|
return address
|
|
|
|
|
|
def _ip_deallocate_stub(context, address):
|
|
address['deallocated'] = 1
|
|
address['address_type'] = None
|
|
|
|
|
|
class TestIpAddresses(test_quark_plugin.TestQuarkPlugin):
|
|
@contextlib.contextmanager
|
|
def _stubs(self, port, addr):
|
|
port_model = None
|
|
addr_model = None
|
|
if port:
|
|
port_model = models.Port()
|
|
port_model.update(port)
|
|
if addr:
|
|
addr_model = models.IPAddress()
|
|
addr_model.update(addr)
|
|
|
|
def _alloc_ip(context, new_addr, net_id, port_m, *args, **kwargs):
|
|
new_addr.extend([addr_model])
|
|
|
|
with contextlib.nested(
|
|
mock.patch("quark.db.api.network_find"),
|
|
mock.patch("quark.db.api.port_find"),
|
|
mock.patch(
|
|
"quark.plugin_modules.ip_addresses.ipam_driver"),
|
|
mock.patch(
|
|
"quark.plugin_modules.ip_addresses.db_api"
|
|
".port_associate_ip"),
|
|
mock.patch(
|
|
"quark.plugin_modules.ip_addresses"
|
|
".validate_and_fetch_segment")
|
|
) as (net_f, port_find, mock_ipam, mock_port_associate_ip, validate):
|
|
port_find.return_value = port_model
|
|
mock_ipam.allocate_ip_address.side_effect = _alloc_ip
|
|
mock_port_associate_ip.side_effect = _port_associate_stub
|
|
yield
|
|
|
|
def test_create_ip_address_by_network_and_device(self):
|
|
old_cfg = cfg.CONF.QUARK.ipaddr_allow_fixed_ip
|
|
cfg.CONF.set_override('ipaddr_allow_fixed_ip', True, "QUARK")
|
|
port = dict(id=1, network_id=2, ip_addresses=[])
|
|
ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",
|
|
subnet_id=1, network_id=2, version=4, used_by_tenant_id=1)
|
|
with self._stubs(port=port, addr=ip):
|
|
ip_address = {"network_id": ip["network_id"],
|
|
"version": 4, 'device_ids': [2]}
|
|
response = self.plugin.create_ip_address(
|
|
self.context, dict(ip_address=ip_address))
|
|
self.assertIsNotNone(response["id"])
|
|
self.assertEqual(response["network_id"], ip_address["network_id"])
|
|
self.assertEqual(response["port_ids"], [port["id"]])
|
|
self.assertEqual(response["subnet_id"], ip["subnet_id"])
|
|
self.assertEqual(response['type'], None)
|
|
self.assertEqual(response["version"], 4)
|
|
self.assertEqual(response["address"], "192.168.1.100")
|
|
self.assertEqual(response["tenant_id"], 1)
|
|
cfg.CONF.set_override('ipaddr_allow_fixed_ip', old_cfg, "QUARK")
|
|
|
|
def test_create_ip_address_with_port(self):
|
|
old_cfg = cfg.CONF.QUARK.ipaddr_allow_fixed_ip
|
|
cfg.CONF.set_override('ipaddr_allow_fixed_ip', True, "QUARK")
|
|
port = dict(id=1, network_id=2, ip_addresses=[])
|
|
ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",
|
|
subnet_id=1, network_id=2, version=4)
|
|
with self._stubs(port=port, addr=ip):
|
|
ip_address = dict(port_ids=[port["id"]])
|
|
ip_address['version'] = 4
|
|
ip_address['network_id'] = 2
|
|
response = self.plugin.create_ip_address(
|
|
self.context, dict(ip_address=ip_address))
|
|
|
|
self.assertIsNotNone(response['id'])
|
|
self.assertEqual(response['network_id'], ip["network_id"])
|
|
self.assertEqual(response['port_ids'], [port["id"]])
|
|
self.assertEqual(response['subnet_id'], ip['id'])
|
|
cfg.CONF.set_override('ipaddr_allow_fixed_ip', old_cfg, "QUARK")
|
|
|
|
def test_fail_create_ip_address_with_port_when_disallowed(self):
|
|
old_cfg = cfg.CONF.QUARK.ipaddr_allow_fixed_ip
|
|
cfg.CONF.set_override('ipaddr_allow_fixed_ip', False, "QUARK")
|
|
port = dict(id=1, network_id=2, ip_addresses=[])
|
|
ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",
|
|
subnet_id=1, network_id=2, version=4)
|
|
with self._stubs(port=port, addr=ip):
|
|
ip_address = dict(port_ids=[port["id"]])
|
|
ip_address['version'] = 4
|
|
ip_address['network_id'] = 2
|
|
with self.assertRaises(exceptions.BadRequest):
|
|
self.plugin.create_ip_address(self.context,
|
|
dict(ip_address=ip_address))
|
|
cfg.CONF.set_override('ipaddr_allow_fixed_ip', old_cfg, "QUARK")
|
|
|
|
def test_create_ip_address_by_device_no_network_fails(self):
|
|
with self._stubs(port={}, addr=None):
|
|
ip_address = dict(device_ids=[4], version=4)
|
|
with self.assertRaises(exceptions.BadRequest):
|
|
self.plugin.create_ip_address(self.context,
|
|
dict(ip_address=ip_address))
|
|
|
|
def test_create_ip_address_invalid_missing_port_and_device_list(self):
|
|
with self._stubs(port=None, addr=None):
|
|
with self.assertRaises(exceptions.BadRequest):
|
|
ip_address = {'ip_address': {'network_id': 'fake',
|
|
'version': 4}}
|
|
self.plugin.create_ip_address(self.context, ip_address)
|
|
|
|
def test_create_ip_address_invalid_port(self):
|
|
old_cfg = cfg.CONF.QUARK.ipaddr_allow_fixed_ip
|
|
cfg.CONF.set_override('ipaddr_allow_fixed_ip', True, "QUARK")
|
|
with self._stubs(port=None, addr=None):
|
|
with self.assertRaises(exceptions.PortNotFound):
|
|
ip_address = {
|
|
'ip_address': {
|
|
'port_ids': ['fake'],
|
|
'version': 4,
|
|
'network_id': 'fake'
|
|
}
|
|
}
|
|
self.plugin.create_ip_address(self.context, ip_address)
|
|
cfg.CONF.set_override('ipaddr_allow_fixed_ip', old_cfg, "QUARK")
|
|
|
|
|
|
@mock.patch("quark.plugin_modules.ip_addresses.v")
|
|
@mock.patch("quark.plugin_modules.ip_addresses.ipam_driver")
|
|
@mock.patch("quark.plugin_modules.ip_addresses.db_api")
|
|
class TestQuarkSharedIPAddressCreate(test_quark_plugin.TestQuarkPlugin):
|
|
def setUp(self):
|
|
super(TestQuarkSharedIPAddressCreate, self).setUp()
|
|
patcher = patch('quark.plugin_modules.ip_addresses.'
|
|
'validate_and_fetch_segment')
|
|
segment_check = patcher.start()
|
|
segment_check.return_value = None
|
|
self.addCleanup(patcher.stop)
|
|
|
|
def _alloc_stub(self, ip_model):
|
|
def _alloc_ip(context, addr, *args, **kwargs):
|
|
addr.append(ip_model)
|
|
return _alloc_ip
|
|
|
|
def test_create_ip_address_calls_port_associate_ip(self, mock_dbapi,
|
|
mock_ipam, *args):
|
|
old_cfg = cfg.CONF.QUARK.ipaddr_allow_fixed_ip
|
|
cfg.CONF.set_override('ipaddr_allow_fixed_ip', True, "QUARK")
|
|
port = dict(id=1, network_id=2, ip_addresses=[])
|
|
ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",
|
|
subnet_id=1, network_id=2, version=4, tenant_id=1)
|
|
port_model = models.Port()
|
|
port_model.update(port)
|
|
ip_model = models.IPAddress()
|
|
ip_model.update(ip)
|
|
|
|
mock_dbapi.port_find.return_value = port_model
|
|
mock_ipam.allocate_ip_address.side_effect = (
|
|
self._alloc_stub(ip_model))
|
|
ip_address = {"network_id": ip["network_id"],
|
|
"version": 4, 'device_ids': [2],
|
|
"port_ids": [1]}
|
|
|
|
self.plugin.create_ip_address(self.context,
|
|
dict(ip_address=ip_address))
|
|
mock_dbapi.port_associate_ip.assert_called_once_with(
|
|
self.context, [port_model], ip_model)
|
|
cfg.CONF.set_override('ipaddr_allow_fixed_ip', old_cfg, "QUARK")
|
|
|
|
def test_create_ip_address_address_type_shared(self, mock_dbapi, mock_ipam,
|
|
*args):
|
|
cfg.CONF.set_override('ipam_reuse_after', 100, "QUARK")
|
|
ports = [dict(id=1, network_id=2, ip_addresses=[]),
|
|
dict(id=2, network_id=2, ip_addresses=[])]
|
|
ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",
|
|
subnet_id=1, network_id=2, version=4, tenant_id=1)
|
|
port_models = [models.Port(**p) for p in ports]
|
|
ip_model = models.IPAddress()
|
|
ip_model.update(ip)
|
|
mock_dbapi.port_find.side_effect = port_models
|
|
mock_ipam.allocate_ip_address.side_effect = (
|
|
self._alloc_stub(ip_model))
|
|
|
|
ip_address = {"network_id": ip["network_id"],
|
|
"version": 4, 'device_ids': [2],
|
|
"port_ids": [pm.id for pm in port_models]}
|
|
self.plugin.create_ip_address(self.context,
|
|
dict(ip_address=ip_address))
|
|
# NOTE(thomasem): Having to assert that [ip_model] was passed instead
|
|
# of an empty list due to the expected behavior of this method being
|
|
# that it mutates the passed in list. So, after it's run, the list
|
|
# has already been mutated and it's a reference to that list that
|
|
# we're checking. This method ought to be changed to return the new
|
|
# IP and let the caller mutate the list, not the other way around.
|
|
mock_ipam.allocate_ip_address.assert_called_once_with(
|
|
self.context, [ip_model], ip['network_id'], None, 100,
|
|
version=ip_address['version'], ip_addresses=[],
|
|
address_type="shared", segment_id=None)
|
|
|
|
def test_create_ip_address_address_type_fixed(self, mock_dbapi, mock_ipam,
|
|
*args):
|
|
cfg.CONF.set_override('ipam_reuse_after', 100, "QUARK")
|
|
old_cfg = cfg.CONF.QUARK.ipaddr_allow_fixed_ip
|
|
cfg.CONF.set_override('ipaddr_allow_fixed_ip', True, "QUARK")
|
|
ports = [dict(id=1, network_id=2, ip_addresses=[])]
|
|
ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",
|
|
subnet_id=1, network_id=2, version=4, tenant_id=1)
|
|
port_models = [models.Port(**p) for p in ports]
|
|
ip_model = models.IPAddress()
|
|
ip_model.update(ip)
|
|
mock_dbapi.port_find.side_effect = port_models
|
|
mock_ipam.allocate_ip_address.side_effect = (
|
|
self._alloc_stub(ip_model))
|
|
|
|
ip_address = {"network_id": ip["network_id"],
|
|
"version": 4, 'device_ids': [2],
|
|
"port_ids": [pm.id for pm in port_models]}
|
|
self.plugin.create_ip_address(self.context,
|
|
dict(ip_address=ip_address))
|
|
# NOTE(thomasem): Having to assert that [ip_model] was passed instead
|
|
# of an empty list due to the expected behavior of this method being
|
|
# that it mutates the passed in list. So, after it's run, the list
|
|
# has already been mutated and it's a reference to that list that
|
|
# we're checking. This method ought to be changed to return the new
|
|
# IP and let the caller mutate the list, not the other way around.
|
|
mock_ipam.allocate_ip_address.assert_called_once_with(
|
|
self.context, [ip_model], ip['network_id'], None, 100,
|
|
version=ip_address['version'], ip_addresses=[],
|
|
segment_id=None, address_type="fixed")
|
|
cfg.CONF.set_override('ipaddr_allow_fixed_ip', old_cfg, "QUARK")
|
|
|
|
|
|
class TestQuarkSharedIPAddressPortsValid(test_quark_plugin.TestQuarkPlugin):
|
|
def test_validate_ports_on_network_raise_segment(self):
|
|
mock_ports = [models.Port(id="1", network_id="2"),
|
|
models.Port(id="2", network_id="2")]
|
|
mock_subnets = [models.Subnet(id="1", segment_id="2"),
|
|
models.Subnet(id="2", segment_id="3")]
|
|
for i, subnet in enumerate(mock_subnets):
|
|
mock_address = models.IPAddress(id="2", network_id="2")
|
|
mock_address.subnet = subnet
|
|
mock_ports[i].ip_addresses.append(mock_address)
|
|
|
|
with self.assertRaises(exceptions.BadRequest):
|
|
ip_addresses.validate_and_fetch_segment(
|
|
mock_ports, "2")
|
|
|
|
def test_validate_ports_on_network_raise_segment_multiple_ips(self):
|
|
mock_ports = [models.Port(id="1", network_id="2"),
|
|
models.Port(id="2", network_id="2")]
|
|
mock_subnets = [models.Subnet(id="1", segment_id="2"),
|
|
models.Subnet(id="2", segment_id="3")]
|
|
for i, subnet in enumerate(mock_subnets):
|
|
mock_address = models.IPAddress(id="2", network_id="2")
|
|
mock_address.subnet = subnet
|
|
for x in xrange(i + 1):
|
|
mock_ports[x].ip_addresses.append(mock_address)
|
|
|
|
with self.assertRaises(exceptions.BadRequest):
|
|
ip_addresses.validate_and_fetch_segment(
|
|
mock_ports, "2")
|
|
|
|
def test_validate_ports_on_network_raise_network(self):
|
|
mock_ports = [models.Port(id="1", network_id="2"),
|
|
models.Port(id="2", network_id="3")]
|
|
mock_addresses = [models.IPAddress(id="1", network_id="2"),
|
|
models.IPAddress(id="2", network_id="3")]
|
|
|
|
for i, ip_address in enumerate(mock_addresses):
|
|
ip_address.subnet = models.Subnet(id="1", segment_id="2")
|
|
mock_ports[i].ip_addresses.append(ip_address)
|
|
|
|
with self.assertRaises(exceptions.BadRequest):
|
|
ip_addresses.validate_and_fetch_segment(
|
|
mock_ports, "2")
|
|
|
|
def test_validate_ports_on_network_valid(self):
|
|
mock_ports = [models.Port(id="1", network_id="2"),
|
|
models.Port(id="2", network_id="2")]
|
|
for p in mock_ports:
|
|
p.ip_addresses.append(models.IPAddress(id="1", network_id="2"))
|
|
p.ip_addresses[-1].subnet = models.Subnet(id="1", segment_id="1")
|
|
|
|
r = ip_addresses.validate_and_fetch_segment(
|
|
mock_ports, "2")
|
|
self.assertEqual(r, "1")
|
|
|
|
|
|
class TestQuarkSharedIPAddress(test_quark_plugin.TestQuarkPlugin):
|
|
def test_shared_ip_request(self):
|
|
ip_address_mock = {"ip_address": {"port_ids": [1, 2, 3]}}
|
|
r = ip_addresses._shared_ip_request(ip_address_mock)
|
|
self.assertTrue(r)
|
|
|
|
def test_shared_ip_request_false(self):
|
|
ip_address_mock = {"ip_address": {"port_ids": [1]}}
|
|
r = ip_addresses._shared_ip_request(ip_address_mock)
|
|
self.assertFalse(r)
|
|
|
|
def test_can_be_shared(self):
|
|
mock_address = models.IPAddress(id="1", address=3232235876,
|
|
address_readable="192.168.1.100",
|
|
subnet_id="1", network_id="2",
|
|
version=4, used_by_tenant_id="1")
|
|
mock_assocs = []
|
|
for x in xrange(3):
|
|
assoc = models.PortIpAssociation()
|
|
assoc.ip_address_id = mock_address.id
|
|
assoc.ip_address = mock_address
|
|
assoc.enabled = [False, False, False][x]
|
|
mock_assocs.append(assoc)
|
|
r = ip_addresses._can_be_shared(mock_address)
|
|
self.assertTrue(r)
|
|
|
|
def test_can_be_shared_false(self):
|
|
mock_address = models.IPAddress(id="1", address=3232235876,
|
|
address_readable="192.168.1.100",
|
|
subnet_id="1", network_id="2",
|
|
version=4, used_by_tenant_id="1")
|
|
mock_assocs = []
|
|
for x in xrange(3):
|
|
assoc = models.PortIpAssociation()
|
|
assoc.ip_address_id = mock_address.id
|
|
assoc.ip_address = mock_address
|
|
assoc.enabled = [False, True, False][x]
|
|
mock_assocs.append(assoc)
|
|
r = ip_addresses._can_be_shared(mock_address)
|
|
self.assertFalse(r)
|
|
|
|
@mock.patch("quark.plugin_modules.ip_addresses._shared_ip_request")
|
|
@mock.patch("quark.plugin_modules.ip_addresses._can_be_shared")
|
|
def test_raise_if_shared_and_enabled(self, can_be_shared_mock,
|
|
shared_ip_request_mock):
|
|
can_be_shared_mock.return_value = False
|
|
shared_ip_request_mock.return_value = True
|
|
obj = mock.MagicMock()
|
|
with self.assertRaises(exceptions.BadRequest):
|
|
ip_addresses._raise_if_shared_and_enabled(obj, obj)
|
|
|
|
@mock.patch("quark.plugin_modules.ip_addresses._shared_ip_request")
|
|
@mock.patch("quark.plugin_modules.ip_addresses._can_be_shared")
|
|
def test_raise_if_shared_and_enabled_noraise(self, can_be_shared_mock,
|
|
shared_ip_request_mock):
|
|
can_be_shared_mock.return_value = True
|
|
shared_ip_request_mock.return_value = True
|
|
obj = mock.MagicMock()
|
|
r = ip_addresses._raise_if_shared_and_enabled(obj, obj)
|
|
self.assertEqual(r, None)
|
|
|
|
@mock.patch("quark.plugin_modules.ip_addresses._shared_ip_request")
|
|
@mock.patch("quark.plugin_modules.ip_addresses._can_be_shared")
|
|
def test_raise_if_shared_and_enabled_fixed_request(self,
|
|
can_be_shared_mock,
|
|
shared_ip_request_mock):
|
|
can_be_shared_mock.return_value = True
|
|
shared_ip_request_mock.return_value = False
|
|
obj = mock.MagicMock()
|
|
r = ip_addresses._raise_if_shared_and_enabled(obj, obj)
|
|
self.assertEqual(r, None)
|
|
|
|
@mock.patch("quark.plugin_modules.ip_addresses._shared_ip_request")
|
|
@mock.patch("quark.plugin_modules.ip_addresses._can_be_shared")
|
|
def test_raise_if_shared_and_enabled_fixed_request_and_not_shareable(
|
|
self, can_be_shared_mock, shared_ip_request_mock):
|
|
can_be_shared_mock.return_value = False
|
|
shared_ip_request_mock.return_value = False
|
|
obj = mock.MagicMock()
|
|
r = ip_addresses._raise_if_shared_and_enabled(obj, obj)
|
|
self.assertEqual(r, None)
|
|
|
|
|
|
class TestQuarkUpdateIPAddress(test_quark_plugin.TestQuarkPlugin):
|
|
|
|
@contextlib.contextmanager
|
|
def _stubs(self, ports, addr, addr_ports=False):
|
|
port_models = []
|
|
addr_model = None
|
|
|
|
for port in ports:
|
|
port_model = models.Port()
|
|
port_model.update(port)
|
|
port_models.append(port_model)
|
|
if addr:
|
|
addr_model = models.IPAddress()
|
|
addr_model.update(addr)
|
|
if addr_ports:
|
|
addr_model.ports = port_models
|
|
|
|
db_mod = "quark.db.api"
|
|
with contextlib.nested(
|
|
mock.patch("%s.port_find" % db_mod),
|
|
mock.patch("%s.ip_address_find" % db_mod),
|
|
mock.patch("%s.port_associate_ip" % db_mod),
|
|
mock.patch("%s.port_disassociate_ip" % db_mod),
|
|
mock.patch("quark.plugin_modules.ip_addresses"
|
|
".validate_and_fetch_segment"),
|
|
mock.patch("quark.plugin_modules.ip_addresses.ipam_driver")
|
|
) as (port_find, ip_find, port_associate_ip,
|
|
port_disassociate_ip, val, mock_ipam):
|
|
port_find.return_value = port_models
|
|
ip_find.return_value = addr_model
|
|
port_associate_ip.side_effect = _port_associate_stub
|
|
port_disassociate_ip.side_effect = _port_disassociate_stub
|
|
mock_ipam.deallocate_ip_address.side_effect = (
|
|
_ip_deallocate_stub)
|
|
yield
|
|
|
|
def test_update_ip_address_does_not_exist(self):
|
|
with self._stubs(ports=[], addr=None):
|
|
with self.assertRaises(exceptions.NotFound):
|
|
self.plugin.update_ip_address(self.context,
|
|
'no_ip_address_id',
|
|
{'ip_address': {'port_ids': []}})
|
|
|
|
def test_update_ip_address_port_not_found(self):
|
|
ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",
|
|
subnet_id=1, network_id=2, version=4)
|
|
with self._stubs(ports=[], addr=ip):
|
|
with self.assertRaises(exceptions.NotFound):
|
|
ip_address = {'ip_address': {'port_ids': ['fake']}}
|
|
self.plugin.update_ip_address(self.context,
|
|
ip["id"],
|
|
ip_address)
|
|
|
|
def test_update_ip_address_specify_ports(self):
|
|
port = dict(id=1, network_id=2, ip_addresses=[])
|
|
ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",
|
|
subnet_id=1, network_id=2, version=4)
|
|
with self._stubs(ports=[port], addr=ip):
|
|
ip_address = {'ip_address': {'port_ids': [port['id']],
|
|
'network_id': 2}}
|
|
response = self.plugin.update_ip_address(self.context,
|
|
ip['id'],
|
|
ip_address)
|
|
self.assertEqual(response['port_ids'], [port['id']])
|
|
|
|
def test_bad_request_fixed_update_multiple_ports(self):
|
|
port1 = dict(id=1, network_id=2, ip_addresses=[])
|
|
port2 = dict(id=2, network_id=2, ip_addresses=[])
|
|
ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",
|
|
subnet_id=1, network_id=2, version=4,
|
|
address_type=ip_types.FIXED)
|
|
with self._stubs(ports=[port1, port2], addr=ip):
|
|
ip_address = {'ip_address': {'port_ids': [port1['id'],
|
|
port2['id']],
|
|
'network_id': 2}}
|
|
with self.assertRaises(exceptions.BadRequest):
|
|
self.plugin.update_ip_address(self.context, ip['id'],
|
|
ip_address)
|
|
|
|
def _create_patch(self, path):
|
|
patcher = patch(path)
|
|
mocked = patcher.start()
|
|
self.addCleanup(patcher.stop)
|
|
return mocked
|
|
|
|
def test_update_ip_address_update_deallocated_at(self):
|
|
port = dict(id=1, network_id=2, ip_addresses=[])
|
|
ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",
|
|
subnet_id=1, network_id=2, version=4, deallocated=1,
|
|
deallocated_at='2020-01-01 00:00:00')
|
|
|
|
path = 'quark.plugin_modules.ip_addresses'
|
|
lookup = self._create_patch('%s._get_deallocated_override' % path)
|
|
|
|
with self._stubs(ports=[port], addr=ip):
|
|
ip_address = {'ip_address': {'reset_allocation_time': True}}
|
|
self.plugin.update_ip_address(self.admin_context, ip['id'],
|
|
ip_address)
|
|
self.assertTrue(lookup.called)
|
|
|
|
def test_update_ip_address_update_deallocated_at_not_deallocated(self):
|
|
port = dict(id=1, network_id=2, ip_addresses=[])
|
|
ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",
|
|
subnet_id=1, network_id=2, version=4, deallocated=0,
|
|
deallocated_at='2020-01-01 00:00:00')
|
|
|
|
path = 'quark.plugin_modules.ip_addresses'
|
|
lookup = self._create_patch('%s._get_deallocated_override' % path)
|
|
|
|
with self._stubs(ports=[port], addr=ip):
|
|
ip_address = {'ip_address': {'reset_allocation_time': True}}
|
|
self.plugin.update_ip_address(self.admin_context, ip['id'],
|
|
ip_address)
|
|
self.assertFalse(lookup.called)
|
|
|
|
def test_update_ip_address_update_deallocated_at_not_admin(self):
|
|
port = dict(id=1, network_id=2, ip_addresses=[])
|
|
ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",
|
|
subnet_id=1, network_id=2, version=4, deallocated=1,
|
|
deallocated_at='2020-01-01 00:00:00')
|
|
|
|
with self._stubs(ports=[port], addr=ip):
|
|
ip_address = {'ip_address': {'reset_allocation_time': True}}
|
|
with self.assertRaises(webob.exc.HTTPForbidden):
|
|
self.plugin.update_ip_address(self.context, ip['id'],
|
|
ip_address)
|
|
|
|
def test_update_ip_address_no_ports(self):
|
|
port = dict(id=1, network_id=2, ip_addresses=[])
|
|
ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",
|
|
subnet_id=1, network_id=2, version=4)
|
|
with self._stubs(ports=[port], addr=ip):
|
|
ip_address = {'ip_address': {}}
|
|
response = self.plugin.update_ip_address(self.context,
|
|
ip['id'],
|
|
ip_address)
|
|
self.assertEqual(response['port_ids'], [])
|
|
|
|
def test_update_ip_address_empty_ports_does_not_delete_but_errors(self):
|
|
port = dict(id=1, network_id=2, ip_addresses=[])
|
|
ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",
|
|
subnet_id=1, network_id=2, version=4)
|
|
with self._stubs(ports=[port], addr=ip, addr_ports=True):
|
|
ip_address = {'ip_address': {'port_ids': []}}
|
|
with self.assertRaises(exceptions.BadRequest):
|
|
self.plugin.update_ip_address(self.context, ip['id'],
|
|
ip_address)
|
|
|
|
|
|
class TestQuarkGetIpAddress(test_quark_plugin.TestQuarkPlugin):
|
|
@contextlib.contextmanager
|
|
def _stubs(self, ips, ports):
|
|
with mock.patch("quark.db.api.ip_address_find") as ip_find:
|
|
port_models = []
|
|
for port in ports:
|
|
p = models.Port()
|
|
p.update(port)
|
|
port_models.append(p)
|
|
if ips:
|
|
version = ips.pop("version")
|
|
ip_mod = models.IPAddress()
|
|
ip_mod.update(ips)
|
|
ip_mod.version = version
|
|
ip_mod.ports = port_models
|
|
# Set up Port to IP associations
|
|
assoc = models.PortIpAssociation()
|
|
assoc.port = p
|
|
assoc.port_id = p.id
|
|
assoc.ip_address = ip_mod
|
|
assoc.ip_address_id = ip_mod.id
|
|
ip_mod.associations.append(assoc)
|
|
ip_find.return_value = ip_mod
|
|
else:
|
|
ip_find.return_value = ips
|
|
yield
|
|
|
|
def test_get_ip_address(self):
|
|
port = dict(id=100)
|
|
ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",
|
|
subnet_id=1, network_id=2, version=4)
|
|
with self._stubs(ips=ip, ports=[port]):
|
|
res = self.plugin.get_ip_address(self.context, 1)
|
|
self.assertEqual(ip["id"], res["id"])
|
|
self.assertEqual(ip["subnet_id"], res["subnet_id"])
|
|
self.assertEqual(ip["address_readable"], res["address"])
|
|
self.assertEqual(res["port_ids"][0], port["id"])
|
|
|
|
def test_get_ip_address_no_ip_fails(self):
|
|
port = dict(id=100)
|
|
with self._stubs(ips=None, ports=[port]):
|
|
with self.assertRaises(quark_exceptions.IpAddressNotFound):
|
|
self.plugin.get_ip_address(self.context, 1)
|
|
|
|
|
|
class TestQuarkGetIpAddresses(test_quark_plugin.TestQuarkPlugin):
|
|
@contextlib.contextmanager
|
|
def _stubs(self, ips, ports):
|
|
with mock.patch("quark.db.api.ip_address_find") as ip_find:
|
|
ip_models = []
|
|
port_models = []
|
|
for port in ports:
|
|
p = models.Port()
|
|
p.update(port)
|
|
port_models.append(p)
|
|
for ip in ips:
|
|
version = ip.pop("version")
|
|
ip_mod = models.IPAddress()
|
|
ip_mod.update(ip)
|
|
ip_mod.version = version
|
|
ip_mod.ports = port_models
|
|
# Set up Port to IP associations
|
|
assoc = models.PortIpAssociation()
|
|
assoc.port = p
|
|
assoc.port_id = p.id
|
|
assoc.ip_address = ip_mod
|
|
assoc.ip_address_id = ip_mod.id
|
|
ip_mod.associations.append(assoc)
|
|
ip_models.append(ip_mod)
|
|
ip_find.return_value = ip_models
|
|
yield
|
|
|
|
def test_get_ip_addresses(self):
|
|
port = dict(id=100, device_id="foobar")
|
|
ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",
|
|
subnet_id=1, network_id=2, version=4)
|
|
with self._stubs(ips=[ip], ports=[port]):
|
|
res = self.plugin.get_ip_addresses(self.context)
|
|
addr_res = res[0]
|
|
self.assertEqual(ip["id"], addr_res["id"])
|
|
self.assertEqual(ip["subnet_id"], addr_res["subnet_id"])
|
|
self.assertEqual(ip["address_readable"], addr_res["address"])
|
|
self.assertEqual(addr_res["port_ids"][0], port["id"])
|
|
|
|
def test_get_ip_addresses_multiple(self):
|
|
port = dict(id=100, device_id="foobar")
|
|
ips = [dict(id=1, address=3232235876, address_readable="192.168.1.100",
|
|
subnet_id=1, network_id=2, version=4),
|
|
dict(id=2, address=3232235878, address_readable="192.168.1.101",
|
|
subnet_id=1, network_id=2, version=4)]
|
|
with self._stubs(ips=ips, ports=[port]):
|
|
res = self.plugin.get_ip_addresses(self.context)
|
|
self.assertEqual(len(res), 2)
|
|
for i, addr in enumerate(sorted(res, key=lambda x: x['id'])):
|
|
self.assertEqual(ips[i]["id"], addr["id"])
|
|
self.assertEqual(ips[i]["subnet_id"], addr["subnet_id"])
|
|
self.assertEqual(ips[i]["address_readable"], addr["address"])
|
|
|
|
|
|
@mock.patch("quark.plugin_modules.ip_addresses"
|
|
".validate_and_fetch_segment")
|
|
@mock.patch("quark.plugin_modules.ip_addresses.ipam_driver")
|
|
@mock.patch("quark.plugin_modules.ip_addresses.db_api")
|
|
class TestQuarkGetIpAddressPort(test_quark_plugin.TestQuarkPlugin):
|
|
def _alloc_stub(self, ip_model):
|
|
def _alloc_ip(context, addr, *args, **kwargs):
|
|
addr.append(ip_model)
|
|
return _alloc_ip
|
|
|
|
def test_get_ip_address_ports(self, mock_dbapi, mock_ipam, *args):
|
|
port = dict(mac_address="AA:BB:CC:DD:EE:FF", network_id=1,
|
|
tenant_id=self.context.tenant_id, device_id=2,
|
|
bridge="xenbr0", device_owner='network:dhcp', id=100)
|
|
ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",
|
|
subnet_id=1, network_id=2, version=4)
|
|
port_model = models.Port()
|
|
port_model.update(port)
|
|
ip_model = models.IPAddress()
|
|
ip_model.update(ip)
|
|
|
|
mock_dbapi.port_find.return_value = [port_model]
|
|
mock_ipam.allocate_ip_address.side_effect = (
|
|
self._alloc_stub(ip_model))
|
|
|
|
res = self.plugin.get_ports_for_ip_address(self.context, 1)[0]
|
|
mock_dbapi.port_find.assert_called_with(self.context, None, None, None,
|
|
join_security_groups=True,
|
|
fields=None,
|
|
ip_address_id=[1])
|
|
|
|
self.assertEqual(port["id"], res["id"])
|
|
self.assertEqual(port["device_id"], res["device_id"])
|
|
self.assertFalse('mac_address' in res)
|
|
self.assertFalse('network_id' in res)
|
|
self.assertFalse('bridge' in res)
|
|
self.assertFalse('tenant_id' in res)
|
|
|
|
def test_get_ip_address_port(self, mock_dbapi, mock_ipam, *args):
|
|
port = dict(mac_address="AA:BB:CC:DD:EE:FF", network_id=1,
|
|
tenant_id=self.context.tenant_id, device_id=2,
|
|
bridge="xenbr0", device_owner='network:dhcp', id=100)
|
|
port2 = dict(mac_address="AA:BB:CC:DD:EE:FF", network_id=1,
|
|
tenant_id=self.context.tenant_id, device_id=2,
|
|
bridge="xenbr0", device_owner='network:dhcp', id=100)
|
|
ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",
|
|
subnet_id=1, network_id=2, version=4)
|
|
port_model = models.Port()
|
|
port_model.update(port)
|
|
port_model2 = models.Port()
|
|
port_model2.update(port2)
|
|
ip_model = models.IPAddress()
|
|
ip_model.update(ip)
|
|
|
|
mock_dbapi.port_find.return_value = port_model
|
|
mock_ipam.allocate_ip_address.side_effect = (
|
|
self._alloc_stub(ip_model))
|
|
mock_dbapi.ip_address_find.return_value = ip_model
|
|
|
|
res = self.plugin.get_port_for_ip_address(self.context, 1, 100)
|
|
|
|
mock_dbapi.ip_address_find.assert_called_with(self.context,
|
|
scope=mock_dbapi.ONE,
|
|
id=1)
|
|
mock_dbapi.port_find.assert_called_with(self.context, fields=None,
|
|
id=100, ip_address_id=[1],
|
|
scope=mock_dbapi.ONE)
|
|
self.assertEqual(port["id"], res["id"])
|
|
self.assertEqual(port["device_id"], res["device_id"])
|
|
self.assertFalse('mac_address' in res)
|
|
self.assertFalse('network_id' in res)
|
|
self.assertFalse('bridge' in res)
|
|
self.assertFalse('tenant_id' in res)
|
|
|
|
def test_deleting_inactive_shared_ip(self, mock_dbapi, mock_ipam, *args):
|
|
port = dict(id=100, network_id=2,
|
|
backend_key="derp", device_id="y")
|
|
port2 = dict(id=101, network_id=2,
|
|
backend_key="derp", device_id="x")
|
|
ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",
|
|
subnet_id=1, network_id=2, version=4, address_type="shared")
|
|
port_model = models.Port()
|
|
port_model2 = models.Port()
|
|
port_model.update(port)
|
|
port_model2.update(port2)
|
|
ip_model = models.IPAddress()
|
|
ip_model.update(ip)
|
|
ip_model.ports = [port_model, port_model2]
|
|
|
|
mock_dbapi.port_find.return_value = port_model
|
|
mock_dbapi.ip_address_find.return_value = ip_model
|
|
mock_ipam.allocate_ip_address.side_effect = (
|
|
self._alloc_stub(ip_model))
|
|
self.plugin.delete_ip_address(self.context, 1)
|
|
self.assertFalse(mock_dbapi.ip_address_delete.called)
|
|
self.assertTrue(mock_ipam.deallocate_ip_address.called)
|
|
|
|
def test_get_ip_address_no_ip_fails(self, mock_dbapi, mock_ipam, *args):
|
|
mock_dbapi.ip_address_find.return_value = []
|
|
with self.assertRaises(quark_exceptions.IpAddressNotFound):
|
|
self.plugin.get_port_for_ip_address(self.context, 123, 100)
|
|
|
|
def test_get_ip_address_no_port_fails(self, mock_dbapi, mock_ipam, *args):
|
|
mock_dbapi.port_find.return_value = []
|
|
with self.assertRaises(exceptions.PortNotFound):
|
|
self.plugin.get_port_for_ip_address(self.context, 123, 100)
|