add test for virtual hosted bucket

zaza/openstack/charm_tests/ceph/tests.py

@@ -685,6 +685,14 @@ def multisite(self):
         except KeyError:
             return False
 
+    @property
+    def virtual_hosted_bucket_enabled(self):
+        """Determine whether virtual hosted bucket is enabled."""
+        try:
+            return zaza_model.get_application_config(self.primary_rgw_app)["virtual-hosted-bucket-enabled"]["value"]
+        except KeyError:
+            return False
+
     def get_rgwadmin_cmd_skeleton(self, unit_name):
         """
         Get radosgw-admin cmd skeleton with rgw.hostname populated key.
@@ -1196,6 +1204,43 @@ def test_004_migration_and_multisite_failover(self):
         self.purge_bucket(self.secondary_rgw_app, 'zaza-container')
         self.purge_bucket(self.secondary_rgw_app, 'failover-container')
 
+    def test_005_virtual_hosted_bucket(self):
+        """Test virtual hosted bucket."""
+        if not self.virtual_hosted_bucket_enabled:
+            logging.info('Skipping virtual hosted bucket test')
+            return
+        logging.info('Testing virtual hosted bucket')
+        container_name = 'zaza-bucket'
+        obj_data = 'Test content from Zaza'
+        obj_name = 'testfile'
+
+         # 1. Fetch Primary Endpoint Details
+        primary_endpoint = self.get_rgw_endpoint(self.primary_rgw_unit)
+        self.assertNotEqual(primary_endpoint, None)
+
+        # 2. Create RGW Client and perform IO
+        access_key, secret_key = self.get_client_keys()
+        primary_client = boto3.resource("s3",
+                                        verify=False,
+                                        endpoint_url=primary_endpoint,
+                                        aws_access_key_id=access_key,
+                                        aws_secret_access_key=secret_key)
+        primary_client.Bucket(container_name).create()
+        primary_object_one = primary_client.Object(
+            container_name,
+            obj_name
+        )
+        primary_object_one.put(Body=obj_data)
+        primary_client.Bucket(container_name).Acl().put(ACL='public-read')
+        primary_client.Object(container_name,obj_name).Acl().put(ACL='public-read')
+
+        # 3. Test if we can get content via virtual hosted bucket name
+        public_hostname = zaza_model.get_application_config(self.primary_rgw_app)["os-public-hostname"]["value"]
+        url = f"{primary_endpoint}/{obj_name}"
+        headers = {'host': f"{container_name}.{public_hostname}"}
+        f = requests.get(url, headers=headers, verify=False)
+        self.assertEqual(f.text, obj_data)
+
 
 class CephProxyTest(unittest.TestCase):
     """Test ceph via proxy."""