diff --git a/zaza/openstack/charm_tests/dragent/test.py b/zaza/openstack/charm_tests/dragent/test.py deleted file mode 100644 index 5b56fdc..0000000 --- a/zaza/openstack/charm_tests/dragent/test.py +++ /dev/null @@ -1,104 +0,0 @@ -#!/usr/bin/env python3 - -# Copyright 2018 Canonical Ltd. -# -# 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. - -"""Run BGP tests.""" - -import argparse -import logging -import sys -import tenacity - -from zaza import model -from zaza.openstack.utilities import ( - cli as cli_utils, - juju as juju_utils, - openstack as openstack_utils, -) - - -def test_bgp_routes(peer_application_name="osci-frr", keystone_session=None): - """Test BGP routes. - - :param peer_application_name: String name of BGP peer application - :type peer_application_name: string - :param keystone_session: Keystone session object for overcloud - :type keystone_session: keystoneauth1.session.Session object - :raises: AssertionError if expected BGP routes are not found - :returns: None - :rtype: None - """ - # If a session has not been provided, acquire one - if not keystone_session: - keystone_session = openstack_utils.get_overcloud_keystone_session() - - # Get authenticated clients - neutron_client = openstack_utils.get_neutron_session_client( - keystone_session) - - # Get the peer unit - peer_unit = model.get_units(peer_application_name)[0].entity_id - - # Get expected advertised routes - private_cidr = neutron_client.list_subnets( - name="private_subnet")["subnets"][0]["cidr"] - floating_ip_cidr = "{}/32".format( - neutron_client.list_floatingips() - ["floatingips"][0]["floating_ip_address"]) - - # This test may run immediately after configuration. It may take time for - # routes to propogate via BGP. Do a binary backoff. - @tenacity.retry(wait=tenacity.wait_exponential(multiplier=1, max=60), - reraise=True, stop=tenacity.stop_after_attempt(10)) - def _assert_cidr_in_peer_routing_table(peer_unit, cidr): - logging.debug("Checking for {} on BGP peer {}" - .format(cidr, peer_unit)) - # Run show ip route bgp on BGP peer - routes = juju_utils.remote_run( - peer_unit, remote_cmd='vtysh -c "show ip route bgp"') - logging.info(routes) - assert cidr in routes, ( - "CIDR, {}, not found in BGP peer's routing table: {}" - .format(cidr, routes)) - - _assert_cidr_in_peer_routing_table(peer_unit, private_cidr) - logging.info("Private subnet CIDR, {}, found in routing table" - .format(private_cidr)) - _assert_cidr_in_peer_routing_table(peer_unit, floating_ip_cidr) - logging.info("Floating IP CIDR, {}, found in routing table" - .format(floating_ip_cidr)) - - -def run_from_cli(): - """Run test for BGP routes from CLI. - - :returns: None - :rtype: None - """ - cli_utils.setup_logging() - parser = argparse.ArgumentParser() - parser.add_argument("--peer-application", "-a", - help="BGP Peer application name. Default: osci-frr", - default="osci-frr") - options = parser.parse_args() - - peer_application_name = cli_utils.parse_arg(options, - "peer_application") - - test_bgp_routes(peer_application_name) - - -if __name__ == "__main__": - sys.exit(run_from_cli()) diff --git a/zaza/openstack/charm_tests/dragent/tests.py b/zaza/openstack/charm_tests/dragent/tests.py index e95086d..67748c8 100644 --- a/zaza/openstack/charm_tests/dragent/tests.py +++ b/zaza/openstack/charm_tests/dragent/tests.py @@ -16,25 +16,122 @@ """Define class of BGP tests.""" +import logging +import tenacity import unittest +import zaza.openstack.charm_tests.neutron.tests as neutron_tests -from zaza.openstack.utilities import cli as cli_utils -from zaza.openstack.charm_tests.dragent import test +from zaza import model +from zaza.openstack.utilities import ( + cli as cli_utils, + juju as juju_utils, +) -class DRAgentTest(unittest.TestCase): +class DRAgentTest(neutron_tests.NeutronNetworkingBase): """Class to encapsulate BPG tests.""" BGP_PEER_APPLICATION = 'osci-frr' + def setUp(self): + """Run setup actions specific to the class.""" + super().setUp() + self._peer_unit = model.get_units( + self.BGP_PEER_APPLICATION)[0].entity_id + @classmethod def setUpClass(cls): - """Run setup for BGP tests.""" + """Run setup actions specific to the class.""" + super().setUpClass() cli_utils.setup_logging() + @staticmethod + @tenacity.retry(wait=tenacity.wait_exponential(multiplier=1, max=60), + reraise=True, stop=tenacity.stop_after_attempt(10)) + def _assert_cidr_in_peer_routing_table(peer_unit, cidr): + logging.debug("Checking for {} on BGP peer {}" + .format(cidr, peer_unit)) + # Run show ip route bgp on BGP peer + routes = juju_utils.remote_run( + peer_unit, remote_cmd='vtysh -c "show ip route bgp"') + logging.info(routes) + assert cidr in routes, ( + "CIDR, {}, not found in BGP peer's routing table: {}" + .format(cidr, routes)) + + @staticmethod + @tenacity.retry(wait=tenacity.wait_exponential(multiplier=1, max=60), + reraise=True, stop=tenacity.stop_after_attempt(10)) + def _assert_ip_reachable_via_peer(peer_unit, address): + logging.debug(f"Checking if {peer_unit} can reach {address} using " + f"routes adverised by NDR") + # Ping with -w specified will return an exit code if there is no + # response after the number of seconds specified. This is to ignore the + # first ping that may not arrive due to an ARP resolution. + juju_utils.remote_run(peer_unit, fatal=True, + remote_cmd=f'ping -w4 {address}') + def test_bgp_routes(self): - """Run bgp tests.""" - test.test_bgp_routes(peer_application_name=self.BGP_PEER_APPLICATION) + """Test BGP routes. + + A test that checks only the control plane of Neutron Dynamic Routing. + + :raises: AssertionError if expected BGP routes are not found + :returns: None + :rtype: None + """ + # Get expected advertised routes + private_cidr = self.project_subnet['cidr'] + floating_ip_cidr = "{}/32".format( + self.neutron_client.list_floatingips() + ["floatingips"][0]["floating_ip_address"]) + + # This test may run immediately after configuration. + # It may take time for routes to propagate via BGP. Do a + # binary backoff. + self._assert_cidr_in_peer_routing_table(self._peer_unit, private_cidr) + logging.info("Private subnet CIDR, {}, found in routing table" + .format(private_cidr)) + self._assert_cidr_in_peer_routing_table(self._peer_unit, + floating_ip_cidr) + logging.info("Floating IP CIDR, {}, found in routing table" + .format(floating_ip_cidr)) + + def test_instance_connectivity(self): + """Test connectivity to instances via dynamic routes. + + Make sure that with routes advertised via NDR it is actually possible + to reach instances from a unit that gets those routes programmed into + its routing table. + """ + # Get an instance but do not perform connectivity checks as the machine + # running those tests does not have the dynamic routes advertised to + # the peer unit by NDR. + fip_instance = self.launch_guest('fip-instance', instance_key='jammy', + perform_connectivity_check=False) + + @tenacity.retry(wait=tenacity.wait_exponential(multiplier=1, max=60), + reraise=True, stop=tenacity.stop_after_attempt(10)) + def get_fip(nova_client, instance_id): + """Try to get a FIP from an instance. + + Instance FIPs may not be immediately accessible from the Nova + object after the instance creation so a retry logic is necessary. + """ + # The reason for looking up an instance object again is that the + # Nova client does not refresh address information after + instance = nova_client.servers.find(id=instance_id) + fips = neutron_tests.floating_ips_from_instance(instance) + if not fips: + raise tenacity.TryAgain + return fips[0] + + fip = get_fip(self.nova_client, fip_instance.id) + + # First check that the FIP is present in the peer unit's routing table. + self._assert_cidr_in_peer_routing_table(self._peer_unit, f'{fip}/32') + # Once it is, check if it is actually reachable. + self._assert_ip_reachable_via_peer(self._peer_unit, fip) if __name__ == "__main__":