Add Local IP API tests

Tests were verified on OVS environment.
API job definition is not changed, because OVN does not
support Local IP.

Depends-On: https://review.opendev.org/c/openstack/neutron/+/816435
Depends-On: https://review.opendev.org/c/openstack/neutron/+/818228
Change-Id: I4760db4dd9916ec895ef63573c49bde91727d142
This commit is contained in:
Nurmatov Mamatisa 2021-10-20 14:33:54 +03:00
parent 82cd7afa96
commit 3f2bbb5702
3 changed files with 331 additions and 0 deletions

View File

@ -118,6 +118,8 @@ class BaseNetworkTest(test.BaseTestCase):
cls.routers = []
cls.floating_ips = []
cls.port_forwardings = []
cls.local_ips = []
cls.local_ip_associations = []
cls.metering_labels = []
cls.service_profiles = []
cls.flavors = []
@ -167,6 +169,15 @@ class BaseNetworkTest(test.BaseTestCase):
for floating_ip in cls.floating_ips:
cls._try_delete_resource(cls.delete_floatingip, floating_ip)
# Clean up Local IP Associations
for association in cls.local_ip_associations:
cls._try_delete_resource(cls.delete_local_ip_association,
association)
# Clean up Local IPs
for local_ip in cls.local_ips:
cls._try_delete_resource(cls.delete_local_ip,
local_ip)
# Clean up conntrack helpers
for cth in cls.conntrack_helpers:
cls._try_delete_resource(cls.delete_conntrack_helper, cth)
@ -732,6 +743,98 @@ class BaseNetworkTest(test.BaseTestCase):
client = client or pf.get('client') or cls.client
client.delete_port_forwarding(pf['floatingip_id'], pf['id'])
def create_local_ip(cls, network_id=None,
client=None, **kwargs):
"""Creates a Local IP.
Create a Local IP and schedule it for later deletion.
If a client is passed, then it is used for deleting the IP too.
:param network_id: network ID where to create
By default this is 'CONF.network.public_network_id'.
:param client: network client to be used for creating and cleaning up
the Local IP.
:param **kwargs: additional creation parameters to be forwarded to
networking server.
"""
client = client or cls.client
network_id = (network_id or
cls.external_network_id)
local_ip = client.create_local_ip(network_id,
**kwargs)['local_ip']
# save client to be used later in cls.delete_local_ip
# for final cleanup
local_ip['client'] = client
cls.local_ips.append(local_ip)
return local_ip
@classmethod
def delete_local_ip(cls, local_ip, client=None):
"""Delete Local IP
:param client: Client to be used
If client is not given it will use the client used to create
the Local IP, or cls.client if unknown.
"""
client = client or local_ip.get('client') or cls.client
client.delete_local_ip(local_ip['id'])
@classmethod
def create_local_ip_association(cls, local_ip_id, fixed_port_id,
fixed_ip_address=None, client=None):
"""Creates a Local IP association.
Create a Local IP Association and schedule it for later deletion.
If a client is passed, then it is used for deleting the association
too.
:param local_ip_id: The ID of the Local IP.
:param fixed_port_id: The ID of the Neutron port
to be associated with the Local IP
:param fixed_ip_address: The fixed IPv4 address of the Neutron
port to be associated with the Local IP
:param client: network client to be used for creating and cleaning up
the Local IP Association.
"""
client = client or cls.client
association = client.create_local_ip_association(
local_ip_id, fixed_port_id,
fixed_ip_address)['port_association']
# save ID of Local IP for final cleanup
association['local_ip_id'] = local_ip_id
# save client to be used later in
# cls.delete_local_ip_association for final cleanup
association['client'] = client
cls.local_ip_associations.append(association)
return association
@classmethod
def delete_local_ip_association(cls, association, client=None):
"""Delete Local IP Association
:param client: Client to be used
If client is not given it will use the client used to create
the local IP association, or cls.client if unknown.
"""
client = client or association.get('client') or cls.client
client.delete_local_ip_association(association['local_ip_id'],
association['fixed_port_id'])
@classmethod
def create_router_interface(cls, router_id, subnet_id):
"""Wrapper utility that returns a router interface."""

View File

@ -0,0 +1,142 @@
# Copyright 2021 Huawei, Inc. 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.
#
from tempest.lib import decorators
from tempest.lib import exceptions
from neutron_tempest_plugin.api import base
from neutron_tempest_plugin import config
CONF = config.CONF
class LocalIPTestJSON(base.BaseNetworkTest):
credentials = ['primary', 'admin']
required_extensions = ['local_ip']
@classmethod
def resource_setup(cls):
super(LocalIPTestJSON, cls).resource_setup()
cls.ext_net_id = CONF.network.public_network_id
# Create network and subnet
cls.network = cls.create_network()
cls.subnet = cls.create_subnet(cls.network)
@decorators.idempotent_id('369257b0-521d-43f5-9482-50e18e87a472')
def test_local_ip_lifecycle(self):
port = self.create_port(self.network)
lip_description = 'Test Local IP description'
lip_name = 'test-local-ip'
created_local_ip = self.create_local_ip(
name=lip_name,
description=lip_description,
local_port_id=port['id'],
local_ip_address=port['fixed_ips'][0]['ip_address'])
self.assertEqual(self.network['id'], created_local_ip['network_id'])
self.assertEqual(lip_description, created_local_ip['description'])
self.assertEqual(lip_name, created_local_ip['name'])
self.assertEqual(port['id'], created_local_ip['local_port_id'])
self.assertEqual(port['fixed_ips'][0]['ip_address'],
created_local_ip['local_ip_address'])
# Show created local_ip
body = self.client.get_local_ip(created_local_ip['id'])
local_ip = body['local_ip']
self.assertEqual(lip_description, local_ip['description'])
self.assertEqual(lip_name, local_ip['name'])
# List local_ips
body = self.client.list_local_ips()
local_ip_ids = [lip['id'] for lip in body['local_ips']]
self.assertIn(created_local_ip['id'], local_ip_ids)
# Update local_ip
updated_local_ip = self.client.update_local_ip(
created_local_ip['id'],
name='updated_local_ip')
self.assertEqual('updated_local_ip',
updated_local_ip['local_ip']['name'])
self.delete_local_ip(created_local_ip)
self.assertRaises(exceptions.NotFound,
self.client.get_local_ip, created_local_ip['id'])
@decorators.idempotent_id('e32df8ac-4e29-4adf-8057-46ae8684eff2')
def test_create_local_ip_with_network(self):
local_ip = self.create_local_ip(self.network['id'])
self.assertEqual(self.network['id'], local_ip['network_id'])
class LocalIPAssociationTestJSON(base.BaseNetworkTest):
required_extensions = ['local_ip']
@classmethod
def resource_setup(cls):
super(LocalIPAssociationTestJSON, cls).resource_setup()
cls.ext_net_id = CONF.network.public_network_id
# Create network
cls.network = cls.create_network()
cls.subnet = cls.create_subnet(cls.network)
@decorators.idempotent_id('602d2874-49be-4c72-8799-b20c95853b6b')
def test_local_ip_association_lifecycle(self):
local_ip = self.create_local_ip(self.network['id'])
port = self.create_port(self.network)
local_ip_association = self.create_local_ip_association(
local_ip['id'],
fixed_port_id=port['id'])
self.assertEqual(local_ip['id'], local_ip_association['local_ip_id'])
self.assertEqual(port['id'], local_ip_association['fixed_port_id'])
# Test List Local IP Associations
body = self.client.list_local_ip_associations(local_ip['id'])
associations = body['port_associations']
self.assertEqual(local_ip['id'], associations[0]['local_ip_id'])
self.assertEqual(port['id'], associations[0]['fixed_port_id'])
# Show
body = self.client.get_local_ip_association(
local_ip['id'], port['id'])
association = body['port_association']
self.assertEqual(local_ip['id'], association['local_ip_id'])
self.assertEqual(port['id'], association['fixed_port_id'])
# Delete
self.client.delete_local_ip_association(local_ip['id'], port['id'])
self.assertRaises(exceptions.NotFound,
self.client.get_local_ip_association,
local_ip['id'], port['id'])
@decorators.idempotent_id('5d26edab-78d2-4cbd-9d0b-3c0b19f0f52d')
def test_local_ip_association_with_two_ips_on_port(self):
local_ip = self.create_local_ip(self.network['id'])
s = self.subnet
port = self.create_port(self.network)
# request another IP on the same subnet
port['fixed_ips'].append({'subnet_id': s['id']})
updated = self.client.update_port(port['id'],
fixed_ips=port['fixed_ips'])
port = updated['port']
local_ip_association = self.create_local_ip_association(
local_ip['id'],
fixed_port_id=port['id'],
fixed_ip_address=port['fixed_ips'][0]['ip_address'])
self.assertEqual(port['fixed_ips'][0]['ip_address'],
local_ip_association['fixed_ip'])

View File

@ -936,6 +936,92 @@ class NetworkClientJSON(service_client.RestClient):
self.expected_success(204, resp.status)
service_client.ResponseBody(resp, body)
def create_local_ip(self, network_id, **kwargs):
post_body = {'local_ip': {
'network_id': network_id}}
if kwargs:
post_body['local_ip'].update(kwargs)
body = jsonutils.dumps(post_body)
uri = '%s/local_ips' % self.uri_prefix
resp, body = self.post(uri, body)
self.expected_success(201, resp.status)
body = jsonutils.loads(body)
return service_client.ResponseBody(resp, body)
def list_local_ips(self, **kwargs):
uri = '%s/local_ips' % self.uri_prefix
if kwargs:
uri += '?' + urlparse.urlencode(kwargs, doseq=1)
resp, body = self.get(uri)
self.expected_success(200, resp.status)
body = jsonutils.loads(body)
return service_client.ResponseBody(resp, body)
def get_local_ip(self, local_ip_id):
uri = '%s/local_ips/%s' % (self.uri_prefix, local_ip_id)
get_resp, get_resp_body = self.get(uri)
self.expected_success(200, get_resp.status)
body = jsonutils.loads(get_resp_body)
return service_client.ResponseBody(get_resp, body)
def update_local_ip(self, local_ip_id, **kwargs):
uri = '%s/local_ips/%s' % (self.uri_prefix, local_ip_id)
get_resp, _ = self.get(uri)
self.expected_success(200, get_resp.status)
put_body = jsonutils.dumps({'local_ip': kwargs})
put_resp, resp_body = self.put(uri, put_body)
self.expected_success(200, put_resp.status)
body = jsonutils.loads(resp_body)
return service_client.ResponseBody(put_resp, body)
def delete_local_ip(self, local_ip_id):
uri = '%s/local_ips/%s' % (
self.uri_prefix, local_ip_id)
resp, body = self.delete(uri)
self.expected_success(204, resp.status)
return service_client.ResponseBody(resp, body)
def create_local_ip_association(self, local_ip_id, fixed_port_id,
fixed_ip=None):
post_body = {'port_association': {
'fixed_port_id': fixed_port_id}}
if fixed_ip:
post_body['port_association']['fixed_ip'] = (
fixed_ip)
body = jsonutils.dumps(post_body)
uri = '%s/local_ips/%s/port_associations' % (self.uri_prefix,
local_ip_id)
resp, body = self.post(uri, body)
self.expected_success(201, resp.status)
body = jsonutils.loads(body)
return service_client.ResponseBody(resp, body)
def get_local_ip_association(self, local_ip_id, fixed_port_id):
uri = '%s/local_ips/%s/port_associations/%s' % (self.uri_prefix,
local_ip_id,
fixed_port_id)
get_resp, get_resp_body = self.get(uri)
self.expected_success(200, get_resp.status)
body = jsonutils.loads(get_resp_body)
return service_client.ResponseBody(get_resp, body)
def list_local_ip_associations(self, local_ip_id):
uri = '%s/local_ips/%s/port_associations' % (self.uri_prefix,
local_ip_id)
resp, body = self.get(uri)
self.expected_success(200, resp.status)
body = jsonutils.loads(body)
return service_client.ResponseBody(resp, body)
def delete_local_ip_association(self, local_ip_id, fixed_port_id):
uri = '%s/local_ips/%s/port_associations/%s' % (self.uri_prefix,
local_ip_id,
fixed_port_id)
resp, body = self.delete(uri)
self.expected_success(204, resp.status)
service_client.ResponseBody(resp, body)
def create_conntrack_helper(self, router_id, helper, protocol, port):
post_body = {'conntrack_helper': {
'helper': helper,