Add Octavia volume-based Amphora test

Enable Amphora to run instances booted from
volume by flipping the config option
`enable-volume-based-amphora` to True and
checking volumes attached to the amphora
instances.

Depends-On: https://review.opendev.org/c/openstack/charm-octavia/+/931587
Related-Bug: https://launchpad.net/bugs/1901732
(cherry picked from commit d096096)
(cherry picked from commit f0f60a5)

unit_tests/utilities/test_zaza_utilities_os_versions.py

@@ -20,13 +20,13 @@
 class TestOpenStackUtils(ut_utils.BaseTestCase):
 
     def test_compare_openstack(self):
-        yoga = os_versions.CompareOpenStack('yoga')
         xena = os_versions.CompareOpenStack('xena')
         wallaby = os_versions.CompareOpenStack('wallaby')
-        self.assertGreater(yoga, xena)
-        self.assertLess(xena, yoga)
+        victoria = os_versions.CompareOpenStack('victoria')
+        self.assertGreater(xena, wallaby)
+        self.assertLess(wallaby, xena)
         self.assertGreaterEqual(xena, xena)
-        self.assertGreaterEqual(yoga, yoga)
-        self.assertGreaterEqual(yoga, wallaby)
+        self.assertGreaterEqual(wallaby, wallaby)
+        self.assertGreaterEqual(xena, victoria)
 
         self.assertEqual("CompareOpenStack<xena>", repr(xena))

zaza/openstack/charm_tests/octavia/tests.py

@@ -14,6 +14,7 @@
 
 """Encapsulate octavia testing."""
 
+import json
 import logging
 import subprocess
 import tenacity
@@ -30,6 +31,7 @@
     LoadBalancerUnexpectedState,
     LoadBalancerUnrecoverableError,
 )
+from zaza.openstack.utilities.os_versions import CompareOpenStack
 
 LBAAS_ADMIN_ROLE = 'load-balancer_admin'
 
@@ -140,12 +142,6 @@ def setUpClass(cls):
         cls.keystone_client = ObjectRetrierWraps(
             openstack_utils.get_keystone_session_client(cls.keystone_session))
 
-        if (openstack_utils.get_os_release() >=
-                openstack_utils.get_os_release('focal_wallaby')):
-            # add role to admin user for the duration of the test
-            grant_role_current_user(cls.keystone_client, cls.keystone_session,
-                                    LBAAS_ADMIN_ROLE)
-
         cls.neutron_client = ObjectRetrierWraps(
             openstack_utils.get_neutron_session_client(cls.keystone_session))
         cls.octavia_client = ObjectRetrierWraps(
@@ -163,6 +159,16 @@ def setUpClass(cls):
         # List of floating IPs created by this test
         cls.fips = []
 
+    def setUp(self):
+        """Configure the octavia test environment."""
+        super(LBAASv2Test, self).setUp()
+        if (openstack_utils.get_os_release() >=
+                openstack_utils.get_os_release('focal_wallaby')):
+            # add role to admin user for the duration of the test
+            grant_role_current_user(self.keystone_client,
+                                    self.keystone_session,
+                                    LBAAS_ADMIN_ROLE)
+
     def _remove_amphorae_instances(self):
         """Remove amphorae instances forcefully.
 
@@ -193,6 +199,7 @@ def resource_cleanup(self, only_local=False):
         :param only_local: When set to true do not call parent method
         :type only_local: bool
         """
+        logging.info("deleting loadbalancer(s): {}".format(self.loadbalancers))
         for lb in self.loadbalancers:
             try:
                 self.octavia_client.load_balancer_delete(
@@ -399,7 +406,7 @@ def _get_payload(ip):
              'http://{}/'.format(ip)],
             universal_newlines=True)
 
-    def test_create_loadbalancer(self):
+    def create_loadbalancer(self, ensure_volume_backed=False):
         """Create load balancer."""
         # Prepare payload instances
         # First we allow communication to port 80 by adding a security group
@@ -466,5 +473,47 @@ def test_create_loadbalancer(self):
                          .format(snippet, provider,
                                  lb_fp['floating_ip_address']))
 
+            if ensure_volume_backed:
+                amphora_list = self.octavia_client.amphora_list()
+                self.assertTrue(len(amphora_list) > 0)
+                attached_volumes = []
+                for amphora in amphora_list.get('amphorae', []):
+                    server_id = amphora['compute_id']
+                    logging.info("Checking amphora {} server {} for attached "
+                                 "volumes".format(amphora['id'], server_id))
+                    volumes = self.nova_client.volumes.get_server_volumes(
+                        server_id)
+                    logging.info('amphora {} server {} has volumes={}'.
+                                 format(amphora['id'],
+                                        server_id, volumes))
+                    attached_volumes.append(json.dumps(vars(volumes)))
+
+                self.assertTrue(len(attached_volumes) > 0)
+                logging.info("Amphora volumes creation successful: {}".format(
+                             attached_volumes))
+
         # If we get here, it means the tests passed
         self.run_resource_cleanup = True
+
+    def test_create_loadbalancer(self):
+        """Test creating a load balancer."""
+        self.create_loadbalancer()
+
+
+class OctaviaVolumeBackedAmphoraTest(LBAASv2Test):
+    """Octavia service tests."""
+
+    def test_volume_backed_amphora(self):
+        """Test volume-backed amphora load balancer."""
+        os_versions = openstack_utils.get_current_os_versions(['octavia'])
+        if CompareOpenStack(os_versions['octavia']) < 'ussuri':
+            self.skipTest('Run only for Openstack Ussuri and newer releases.')
+            return
+
+        """Test creating a load balancer that uses volume-based amphora."""
+        default_charm_config = {'enable-volume-based-amphora': False}
+        alternate_charm_config = {'enable-volume-based-amphora': True}
+        with self.config_change(default_charm_config,
+             alternate_charm_config, reset_to_charm_default=True):
+            logging.info("Testing create volume-backed amphora loadbalancer")
+            self.create_loadbalancer(ensure_volume_backed=True)