[Yoga] Add Octavia volume-based Amphora test

unit_tests/utilities/test_zaza_utilities_os_versions.py

@@ -0,0 +1,32 @@
+# Copyright 2023 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.
+
+
+import unit_tests.utils as ut_utils
+from zaza.openstack.utilities import os_versions
+
+
+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)
+        self.assertGreaterEqual(xena, xena)
+        self.assertGreaterEqual(yoga, yoga)
+        self.assertGreaterEqual(yoga, wallaby)
+
+        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
@@ -33,6 +34,7 @@
     LoadBalancerUnexpectedState,
     LoadBalancerUnrecoverableError,
 )
+from zaza.openstack.utilities.os_versions import CompareOpenStack
 
 LBAAS_ADMIN_ROLE = 'load-balancer_admin'
 
@@ -191,12 +193,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(
@@ -214,6 +210,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.
 
@@ -244,6 +250,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(
@@ -450,7 +457,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
@@ -517,5 +524,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)

zaza/openstack/utilities/openstack.py

@@ -139,6 +139,10 @@
         'pkg': 'placement-common',
         'origin_setting': 'openstack-origin'
     },
+    'octavia': {
+        'pkg': 'octavia-common',
+        'origin_setting': 'openstack-origin'
+    },
 }
 
 # Older tests use the order the services appear in the list to imply
@@ -160,6 +164,7 @@
     {'name': 'ovn-central', 'type': CHARM_TYPES['ovn-central']},
     {'name': 'ceph-mon', 'type': CHARM_TYPES['ceph-mon']},
     {'name': 'placement', 'type': CHARM_TYPES['placement']},
+    {'name': 'octavia', 'type': CHARM_TYPES['octavia']},
 ]
 
 

zaza/openstack/utilities/os_versions.py

@@ -358,6 +358,10 @@ def __le__(self, other):
         """Do less than or equals."""
         return not self.__gt__(other)
 
+    def __repr__(self):
+        """Return the representation of CompareOpenStack."""
+        return "%s<%s>" % (self.__class__.__name__, self._list[self.index])
+
     def __str__(self):
         """Give back the item at the index.
 
@@ -383,3 +387,15 @@ class CompareHostReleases(BasicStringComparator):
     """
 
     _list = UBUNTU_RELEASES
+
+
+class CompareOpenStack(BasicStringComparator):
+    """Provide comparisons of OpenStack releases.
+
+    Use in the form of
+
+    if CompareOpenStack(release) > 'yoga':
+        # do something
+    """
+
+    _list = list(OPENSTACK_CODENAMES.values())