Merge "Enabled platform interfaces to add ip address(es)"

This commit is contained in:
Zuul 2020-02-25 20:51:54 +00:00 committed by Gerrit Code Review
commit 5b865b33d4
5 changed files with 80 additions and 23 deletions

View File

@ -244,9 +244,6 @@ class AddressController(rest.RestController):
def _check_interface_type(self, interface_id):
interface = pecan.request.dbapi.iinterface_get(interface_id)
if (interface['ifclass'] == constants.INTERFACE_CLASS_PLATFORM and
interface['networktypelist'] is None):
raise exception.InterfaceNetworkNotSet()
for nt in interface['networktypelist']:
if nt not in ALLOWED_NETWORK_TYPES:
raise exception.UnsupportedInterfaceNetworkType(

View File

@ -1949,6 +1949,5 @@ def _is_interface_address_allowed(interface):
elif interface['ifclass'] == constants.INTERFACE_CLASS_DATA:
return True
elif interface['ifclass'] == constants.INTERFACE_CLASS_PLATFORM:
if any(nt in address.ALLOWED_NETWORK_TYPES for nt in interface['networktypelist'] or []):
return True
return True
return False

View File

@ -646,6 +646,13 @@ def get_interface_address_method(context, iface, network_id=None):
# natively supported in vswitch or need to be shared with the kernel
# because of a platform VLAN should be left as manual config
return MANUAL_METHOD
elif (iface.ifclass == constants.INTERFACE_CLASS_PLATFORM and
networktype is None and
(iface.ipv4_mode == constants.IPV4_STATIC or
iface.ipv6_mode == constants.IPV6_STATIC)):
# Allow platform-class interface with ipv4 mode set to static to
# have static ip address
return STATIC_METHOD
elif not iface.ifclass or iface.ifclass == constants.INTERFACE_CLASS_NONE \
or not networktype:
# Interfaces that are configured purely as a dependency from other

View File

@ -77,20 +77,24 @@ class AddressTestCase(base.FunctionalTest, dbbase.BaseHostTestCase):
self.assertNotIn(field, api_object)
def get_post_object(self, name='test_address', ip_address='127.0.0.1',
prefix=8, address_pool_id=None, interface_id=None):
prefix=8, address_pool_id=None, interface_uuid=None):
addr = netaddr.IPAddress(ip_address)
addr_db = dbutils.get_test_address(
address=str(addr),
prefix=prefix,
name=name,
address_pool_id=address_pool_id,
interface_id=interface_id,
)
if self.oam_subnet.version == 6:
addr_db["enable_dad"] = True
# pool_uuid in api corresponds to address_pool_id in db
addr_db['pool_uuid'] = addr_db.pop('address_pool_id')
addr_db['interface_uuid'] = addr_db.pop('interface_id')
addr_db.pop('family')
addr_db['interface_uuid'] = interface_uuid
del addr_db['family']
del addr_db['interface_id']
return addr_db
@ -99,15 +103,16 @@ class TestPostMixin(AddressTestCase):
def setUp(self):
super(TestPostMixin, self).setUp()
self.worker = self._create_test_host(constants.WORKER,
administrative=constants.ADMIN_LOCKED)
def _test_create_address_success(self, name, ip_address, prefix,
address_pool_id, interface_id):
address_pool_id, interface_uuid):
# Test creation of object
addr_db = self.get_post_object(name=name, ip_address=ip_address,
prefix=prefix,
address_pool_id=address_pool_id,
interface_id=interface_id)
interface_uuid=interface_uuid)
response = self.post_json(self.API_PREFIX,
addr_db,
headers=self.API_HEADERS)
@ -121,14 +126,14 @@ class TestPostMixin(AddressTestCase):
addr_db[self.COMMON_FIELD])
def _test_create_address_fail(self, name, ip_address, prefix,
address_pool_id, interface_id,
status_code, error_message):
address_pool_id, status_code,
error_message, interface_uuid=None):
# Test creation of object
addr_db = self.get_post_object(name=name, ip_address=ip_address,
prefix=prefix,
address_pool_id=address_pool_id,
interface_id=interface_id)
interface_uuid=interface_uuid)
response = self.post_json(self.API_PREFIX,
addr_db,
headers=self.API_HEADERS,
@ -143,8 +148,7 @@ class TestPostMixin(AddressTestCase):
self._test_create_address_success(
"fake-address",
str(self.oam_subnet[25]), self.oam_subnet.prefixlen,
address_pool_id=self.address_pools[2].uuid,
interface_id=None,
address_pool_id=self.address_pools[2].uuid, interface_uuid=None
)
def test_create_address_wrong_address_pool(self):
@ -152,7 +156,6 @@ class TestPostMixin(AddressTestCase):
"fake-address",
str(self.oam_subnet[25]), self.oam_subnet.prefixlen,
address_pool_id=self.address_pools[1].uuid,
interface_id=None,
status_code=http_client.CONFLICT,
error_message="does not match pool network",
)
@ -162,7 +165,6 @@ class TestPostMixin(AddressTestCase):
"fake-address",
str(self.oam_subnet[25]), self.oam_subnet.prefixlen - 1,
address_pool_id=self.address_pools[2].uuid,
interface_id=None,
status_code=http_client.CONFLICT,
error_message="does not match pool network",
)
@ -174,7 +176,6 @@ class TestPostMixin(AddressTestCase):
"fake-address",
str(self.oam_subnet[25]), 0,
address_pool_id=self.address_pools[2].uuid,
interface_id=None,
status_code=http_client.INTERNAL_SERVER_ERROR,
error_message=error_message,
)
@ -189,7 +190,6 @@ class TestPostMixin(AddressTestCase):
"fake-address",
zero_address, self.oam_subnet.prefixlen,
address_pool_id=self.address_pools[2].uuid,
interface_id=None,
status_code=http_client.INTERNAL_SERVER_ERROR,
error_message=error_message,
)
@ -199,7 +199,6 @@ class TestPostMixin(AddressTestCase):
"fake_address",
str(self.oam_subnet[25]), self.oam_subnet.prefixlen,
address_pool_id=self.address_pools[2].uuid,
interface_id=None,
status_code=http_client.BAD_REQUEST,
error_message="Please configure valid hostname.",
)
@ -209,11 +208,36 @@ class TestPostMixin(AddressTestCase):
"fake-address",
str(self.multicast_subnet[1]), self.oam_subnet.prefixlen,
address_pool_id=self.address_pools[2].uuid,
interface_id=None,
status_code=http_client.INTERNAL_SERVER_ERROR,
error_message="Address must be a unicast address",
)
def test_create_address_platform_interface(self):
if self.oam_subnet.version == 4:
ipv4_mode, ipv6_mode = (constants.IPV4_STATIC, constants.IPV6_DISABLED)
else:
ipv4_mode, ipv6_mode = (constants.IPV4_DISABLED, constants.IPV6_STATIC)
# Create platform interface, patch to make static
interface = dbutils.create_test_interface(
ifname="platformip",
ifclass=constants.INTERFACE_CLASS_PLATFORM,
forihostid=self.worker.id,
ihost_uuid=self.worker.uuid)
response = self.patch_dict_json(
'%s/%s' % (self.IFACE_PREFIX, interface['uuid']),
ipv4_mode=ipv4_mode, ipv6_mode=ipv6_mode)
self.assertEqual('application/json', response.content_type)
self.assertEqual(response.status_code, http_client.OK)
self.assertEqual(response.json['ifclass'], 'platform')
self.assertEqual(response.json['ipv4_mode'], ipv4_mode)
self.assertEqual(response.json['ipv6_mode'], ipv6_mode)
# Verify an address associated with the interface can be created
self._test_create_address_success('platformtest',
str(self.oam_subnet[25]), self.oam_subnet.prefixlen,
None, interface.uuid)
class TestDelete(AddressTestCase):
""" Tests deletion.

View File

@ -719,6 +719,36 @@ class InterfaceTestCase(InterfaceTestCaseMixin, dbbase.BaseHostTestCase):
self.context, self.iface, network.id)
self.assertEqual(method, 'static')
def test_get_interface_address_method_for_platform_ipv4(self):
self.iface['ifclass'] = constants.INTERFACE_CLASS_PLATFORM
self.iface['ipv4_mode'] = constants.IPV4_STATIC
self.iface['networktype'] = constants.NETWORK_TYPE_NONE
method = interface.get_interface_address_method(
self.context, self.iface)
self.assertEqual(method, 'static')
def test_get_interface_address_method_for_platform_ipv6(self):
self.iface['ifclass'] = constants.INTERFACE_CLASS_PLATFORM
self.iface['ipv6_mode'] = constants.IPV6_STATIC
self.iface['networktype'] = constants.NETWORK_TYPE_NONE
method = interface.get_interface_address_method(
self.context, self.iface)
self.assertEqual(method, 'static')
def test_get_interface_address_method_for_platform_invalid(self):
self.iface['ifclass'] = constants.INTERFACE_CLASS_PLATFORM
self.iface['ipv4_mode'] = constants.IPV4_STATIC
self.iface['networktype'] = constants.NETWORK_TYPE_OAM
self.iface['networks'] = self._get_network_ids_by_type(
constants.NETWORK_TYPE_OAM)
self.host['personality'] = constants.WORKER
self._update_context()
network = self.dbapi.network_get_by_type(
constants.NETWORK_TYPE_OAM)
method = interface.get_interface_address_method(
self.context, self.iface, network.id)
self.assertEqual(method, 'dhcp')
def test_get_interface_traffic_classifier_for_mgmt(self):
self.iface['ifclass'] = constants.INTERFACE_CLASS_PLATFORM
self.iface['networktypelist'] = [constants.NETWORK_TYPE_MGMT]