Test floating IP attach/detach logs

Each time we have floating IP attached/detached from the VM we should
get those event logged.

Change-Id: I33235ec6af0ff9dc02722354d6d3876c89701f5f
This commit is contained in:
Alex Katz 2021-01-07 14:22:48 +02:00
parent b4fc2404b0
commit 3a73746836
4 changed files with 137 additions and 1 deletions

View File

@ -41,10 +41,15 @@ get_neutron_client = _client.get_neutron_client
find_subnet = _client.find_subnet
find_port = _client.find_port
list_ports = _client.list_ports
create_port = _client.create_port
delete_port = _client.delete_port
list_subnets = _client.list_subnets
list_subnet_cidrs = _client.list_subnet_cidrs
list_agents = _client.list_agents
get_floating_ip = _client.get_floating_ip
create_floating_ip = _client.create_floating_ip
delete_floating_ip = _client.delete_floating_ip
update_floating_ip = _client.update_floating_ip
get_router = _client.get_router
get_port = _client.get_port
get_subnet = _client.get_subnet
@ -53,6 +58,7 @@ find_l3_agent_hosting_router = _client.find_l3_agent_hosting_router
list_dhcp_agent_hosting_network = _client.list_dhcp_agent_hosting_network
NoSuchPort = _client.NoSuchPort
NoSuchFIP = _client.NoSuchPort
NoSuchRouter = _client.NoSuchRouter
NoSuchSubnet = _client.NoSuchSubnet

View File

@ -140,10 +140,39 @@ def list_subnet_cidrs(client=None, **params):
def get_floating_ip(floating_ip, client=None, **params):
floating_ip = neutron_client(client).show_floatingip(floating_ip, **params)
try:
floating_ip = neutron_client(client).show_floatingip(
floating_ip, **params)
except neutronclient.exceptions.NotFound as ex:
raise NoSuchFIP(id=floating_ip) from ex
return floating_ip['floatingip']
def create_floating_ip(floating_network_id=None, client=None, **params):
if floating_network_id is None:
from tobiko.openstack import stacks
floating_network_id = tobiko.setup_fixture(
stacks.FloatingNetworkStackFixture).external_id
if floating_network_id is not None:
params['floating_network_id'] = floating_network_id
floating_ip = neutron_client(client).create_floatingip(
body={'floatingip': params})
return floating_ip['floatingip']
def delete_floating_ip(floating_ip, client=None):
try:
neutron_client(client).delete_floatingip(floating_ip)
except neutronclient.exceptions.NotFound as ex:
raise NoSuchFIP(id=floating_ip) from ex
def update_floating_ip(floating_ip, client=None, **params):
fip = neutron_client(client).update_floatingip(
floating_ip, body={'floatingip': params})
return fip['floatingip']
def get_port(port, client=None, **params):
try:
return neutron_client(client).show_port(port, **params)['port']
@ -151,6 +180,18 @@ def get_port(port, client=None, **params):
raise NoSuchPort(id=port) from ex
def create_port(client=None, **params):
port = neutron_client(client).create_port(body={'port': params})
return port['port']
def delete_port(port, client=None):
try:
neutron_client(client).delete_port(port)
except neutronclient.exceptions.NotFound as ex:
raise NoSuchPort(id=port) from ex
def get_router(router, client=None, **params):
try:
return neutron_client(client).show_router(router, **params)['router']
@ -204,3 +245,7 @@ class NoSuchRouter(tobiko.ObjectNotFound):
class NoSuchSubnet(tobiko.ObjectNotFound):
message = "No such subnet found for {id!r}"
class NoSuchFIP(tobiko.ObjectNotFound):
message = "No such floating IP found for {id!r}"

View File

@ -54,6 +54,7 @@ L3haDifferentHostServerStackFixture = _l3ha.L3haDifferentHostServerStackFixture
L3haSameHostServerStackFixture = _l3ha.L3haSameHostServerStackFixture
NetworkStackFixture = _neutron.NetworkStackFixture
FloatingNetworkStackFixture = _neutron.FloatingNetworkStackFixture
NetworkWithNetMtuWriteStackFixture = (
_neutron.NetworkWithNetMtuWriteStackFixture)
SecurityGroupsFixture = _neutron.SecurityGroupsFixture

View File

@ -14,17 +14,21 @@
# under the License.
from __future__ import absolute_import
from oslo_log import log
import testtools
import tobiko
from tobiko import config
from tobiko.shell import files
from tobiko.shell import ping
from tobiko.shell import sh
from tobiko.openstack import neutron
from tobiko.openstack import stacks
from tobiko.openstack import topology
CONF = config.CONF
LOG = log.getLogger(__name__)
class FloatingIPTest(testtools.TestCase):
@ -237,3 +241,83 @@ class FloatingIpWithMtuWritableTest(FloatingIPTest):
class FloatingIpWithL3HATest(FloatingIPTest):
#: Resources stack with floating IP and Nova server
stack = tobiko.required_setup_fixture(stacks.L3haServerStackFixture)
@topology.skip_unless_osp_version('16.1', higher=True)
class TestFloatingIPLogging(testtools.TestCase):
stack = tobiko.required_setup_fixture(stacks.NetworkStackFixture)
def setUp(self):
super(TestFloatingIPLogging, self).setUp()
net = self.stack.network_id
self.port = neutron.create_port(**{'network_id': net})
self.addCleanup(self.cleanup_port)
self.fip = neutron.create_floating_ip()
self.addCleanup(self.cleanup_floatingip)
log_filename = '/var/log/containers/neutron/server.log'
self.log_digger = files.MultihostLogFileDigger(filename=log_filename,
sudo=True)
for node in topology.list_openstack_nodes(group='controller'):
self.log_digger.add_host(hostname=node.hostname,
ssh_client=node.ssh_client)
def cleanup_port(self):
try:
neutron.delete_port(self.port['id'])
except neutron.NoSuchPort:
pass
def cleanup_floatingip(self):
try:
neutron.delete_floating_ip(self.fip['id'])
except neutron.NoSuchFIP:
pass
def test_fip_attach_log(self):
'''Test log that FIP is attached to VM'''
pattern = f'{self.fip["id"]}.*associated'
self.log_digger.find_lines(pattern=pattern)
neutron.update_floating_ip(
self.fip['id'], **{'port_id': self.port['id']})
new_logs = self.log_digger.find_new_lines(pattern=pattern)
self.assertEqual(len(new_logs), 1)
self.assertIn(self.port['id'], new_logs[0][1])
self.assertIn(self.fip['floating_ip_address'], new_logs[0][1])
def test_fip_detach_log(self):
'''Test log that FIP is dettached from VM'''
neutron.update_floating_ip(
self.fip['id'], **{'port_id': self.port['id']})
pattern = f'{self.fip["id"]}.*disassociated'
self.log_digger.find_lines(pattern=pattern)
neutron.update_floating_ip(self.fip['id'], **{'port_id': None})
new_logs = self.log_digger.find_new_lines(pattern=pattern)
self.assertEqual(len(new_logs), 1)
self.assertIn(self.port['id'], new_logs[0][1])
self.assertIn(self.fip['floating_ip_address'], new_logs[0][1])
@tobiko.skip('Skipped because of bz1542122')
def test_fip_delete_detach_log(self):
'''Test log that FIP is dettached from VM if FIP is deleted'''
neutron.update_floating_ip(
self.fip['id'], **{'port_id': self.port['id']})
pattern = f'{self.fip["id"]}.*disassociated'
self.log_digger.find_lines(pattern=pattern)
neutron.delete_floating_ip(self.fip['id'])
new_logs = self.log_digger.find_new_lines(pattern=pattern)
self.assertEqual(len(new_logs), 1)
self.assertIn(self.port['id'], new_logs[0][1])
self.assertIn(self.fip['floating_ip_address'], new_logs[0][1])
def test_port_delete_fip_detach_log(self):
'''Test log that FIP is dettached from port if port is deleted'''
neutron.update_floating_ip(
self.fip['id'], **{'port_id': self.port['id']})
pattern = f'{self.fip["id"]}.*disassociated'
self.log_digger.find_lines(pattern=pattern)
neutron.delete_port(self.port['id'])
new_logs = self.log_digger.find_new_lines(pattern=pattern)
self.assertEqual(len(new_logs), 1)
self.assertIn(self.port['id'], new_logs[0][1])
self.assertIn(self.fip['floating_ip_address'], new_logs[0][1])