diff --git a/zaza/openstack/charm_tests/ceph/tests.py b/zaza/openstack/charm_tests/ceph/tests.py index aa4b7a40e..5dc5f2314 100644 --- a/zaza/openstack/charm_tests/ceph/tests.py +++ b/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."""