Add Neutron dhcp agent faults test cases

This new test case checks if dnsmasq process spawned for network
works fine when neutron-dhcp-agent is stopped and than started again.
Dnsmasq process should be running fine for all this time and should
not change its PID.

Change-Id: Iad7317761830d6dc07a84bce07a0437ca7e9d602
This commit is contained in:
Slawek Kaplonski 2020-02-07 02:27:44 +01:00
parent f54c14b4b4
commit 6319ca9fbd
6 changed files with 123 additions and 0 deletions

View File

@ -38,6 +38,8 @@ get_port = _client.get_port
get_subnet = _client.get_subnet
list_l3_agent_hosting_routers = _client.list_l3_agent_hosting_routers
find_l3_agent_hosting_router = _client.find_l3_agent_hosting_router
list_dhcp_agent_hosting_network = _client.list_dhcp_agent_hosting_network
NoSuchNetwork = _client.NoSuchNetwork
NoSuchPort = _client.NoSuchPort
NoSuchRouter = _client.NoSuchRouter

View File

@ -185,6 +185,14 @@ def find_l3_agent_hosting_router(router, client=None, unique=False,
return default
def list_dhcp_agent_hosting_network(network, client=None, **params):
agents = neutron_client(client).list_dhcp_agent_hosting_networks(
network, **params)
if isinstance(agents, collections.Mapping):
agents = agents['agents']
return tobiko.select(agents)
class NoSuchNetwork(tobiko.ObjectNotFound):
message = "No such network found for {id!r}"

View File

@ -153,6 +153,10 @@ class OpenStackTopology(tobiko.SharedFixture):
config = tobiko.required_setup_fixture(OpenStackTopologyConfig)
agent_to_service_name_mappings = {
'neutron-dhcp-agent': 'devstack@q-dhcp',
}
def __init__(self):
super(OpenStackTopology, self).__init__()
self._reachable_ips = set()
@ -171,6 +175,12 @@ class OpenStackTopology(tobiko.SharedFixture):
self._nodes_by_ips.clear()
self._nodes_by_group.clear()
def get_agent_service_name(self, agent_name):
try:
return self.agent_to_service_name_mappings[agent_name]
except KeyError:
return None
def discover_nodes(self):
self.discover_configured_nodes()
self.discover_controller_nodes()

View File

View File

@ -0,0 +1,99 @@
# Copyright (c) 2020 Red Hat
# 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 __future__ import absolute_import
import testtools
import tobiko
from tobiko.openstack import neutron
from tobiko.openstack import stacks
from tobiko.openstack import topology
from tobiko.shell import sh
class DHCPAgentTest(testtools.TestCase):
#: Resources stack with Nova server to send messages to
stack = tobiko.required_setup_fixture(stacks.CirrosPeerServerStackFixture)
def setUp(self):
super(DHCPAgentTest, self).setUp()
os_topology = topology.get_openstack_topology()
self.dhcp_agent_service_name = os_topology.get_agent_service_name(
"neutron-dhcp-agent")
if not self.dhcp_agent_service_name:
self.skip("Neutron DHCP agent's service name not defined for "
"the topology %s" % os_topology)
def test_stop_dhcp_agent(self):
network_dhcp_agents = neutron.list_dhcp_agent_hosting_network(
self.stack.network)
network_dnsmasq_pids = self._get_dnsmasq_pids_for_network(
self.stack.network, network_dhcp_agents)
self._stop_dhcp_agent_on_hosts(network_dhcp_agents)
# Now check if dnsmasq processes are still run and have got same pids
# like before dhcp agent's stop
self.assertEqual(
network_dnsmasq_pids,
self._get_dnsmasq_pids_for_network(self.stack.network,
network_dhcp_agents))
self._start_dhcp_agent_on_hosts(network_dhcp_agents)
# And finally check if dnsmasq processes are still run and have got
# same pids like at the beginning of the test
self.assertEqual(
network_dnsmasq_pids,
self._get_dnsmasq_pids_for_network(self.stack.network,
network_dhcp_agents))
def _get_dnsmasq_pids_for_network(self, network_id, agents):
dnsmasq_pids_per_agent = {}
for agent in agents:
agent_host = topology.get_openstack_node(hostname=agent['host'])
dnsmasq_processes_on_host = sh.list_processes(
command="dnsmasq", ssh_client=agent_host.ssh_client)
dnsmasq_pid = self._get_dnsmasq_pid_for_network(
agent_host.ssh_client, network_id,
dnsmasq_processes_on_host)
if not dnsmasq_pid:
self.fail("Dnsmasq process for network: %(network_id)s "
"not found on host %(host)s" % {
'network_id': network_id,
'host': agent['host']})
dnsmasq_pids_per_agent[agent['host']] = dnsmasq_pid
return dnsmasq_pids_per_agent
def _get_dnsmasq_pid_for_network(self, ssh_client, network_id, processes):
for process in processes:
cmdline_result = sh.execute(
"cat /proc/%s/cmdline" % process.pid, ssh_client=ssh_client)
if network_id in cmdline_result.stdout:
return process.pid
return None
def _stop_dhcp_agent_on_hosts(self, agents):
for agent in agents:
agent_host = topology.get_openstack_node(hostname=agent['host'])
sh.execute(
"sudo systemctl stop %s" % self.dhcp_agent_service_name,
ssh_client=agent_host.ssh_client)
def _start_dhcp_agent_on_hosts(self, agents):
for agent in agents:
agent_host = topology.get_openstack_node(hostname=agent['host'])
sh.execute(
"sudo systemctl start %s" % self.dhcp_agent_service_name,
ssh_client=agent_host.ssh_client)

View File

@ -25,6 +25,10 @@ LOG = log.getLogger(__name__)
class TripleoTopology(topology.OpenStackTopology):
agent_to_service_name_mappings = {
'neutron-dhcp-agent': 'tripleo_neutron_dhcp',
}
def discover_nodes(self):
self.discover_undercloud_nodes()
self.discover_overcloud_nodes()