From 570e975522139fd904ffdb1e05ef0ef6c46bedcb Mon Sep 17 00:00:00 2001 From: Marc Fischer Date: Mon, 15 Jun 2026 11:45:24 +0200 Subject: [PATCH] Add path_style support for GenericS3 backup and backup.get() - Add path_style parameter to HTTPBucket, S3Compatible and GenericS3 to support usePathStyleAddressing for S3-compatible storage backends such as generic S3 endpoints that require path-style bucket access - Add Backup.get() to retrieve the current backup configuration Co-Authored-By: Claude Sonnet 4.6 --- cterasdk/core/servers.py | 6 ++++++ cterasdk/core/types.py | 15 ++++++++------- tests/ut/core/admin/test_servers.py | 27 ++++++++++++++++++++++++++- 3 files changed, 40 insertions(+), 8 deletions(-) diff --git a/cterasdk/core/servers.py b/cterasdk/core/servers.py index aa3d928a..ac433241 100644 --- a/cterasdk/core/servers.py +++ b/cterasdk/core/servers.py @@ -116,6 +116,12 @@ def modify(self, name, server_name=None, app=None, preview=None, enable_public_i class Backup(BaseCommand): + def get(self): + """ + Retrieve the current backup configuration + """ + return self._core.servers.system_database.backupToBucket + def connected(self): """ Verify connectivity to the backup S3 bucket diff --git a/cterasdk/core/types.py b/cterasdk/core/types.py index 41f40ab1..a9b1663d 100644 --- a/cterasdk/core/types.py +++ b/cterasdk/core/types.py @@ -324,7 +324,7 @@ def to_server_object(self): class HTTPBucket(Bucket): - def __init__(self, bucket, driver, access_key, secret_key, endpoint, https, direct=False, verify_ssl=True): + def __init__(self, bucket, driver, access_key, secret_key, endpoint, https, direct=False, verify_ssl=True, path_style=False): super().__init__(bucket, driver) self.access_key = access_key self.secret_key = secret_key @@ -332,6 +332,7 @@ def __init__(self, bucket, driver, access_key, secret_key, endpoint, https, dire self.https = https self.direct = direct self.verify_ssl = verify_ssl + self.path_style = path_style @property def trust_all_certificates(self): @@ -348,7 +349,7 @@ def to_native_format_server_object(self): useHttps=self.https, trustAllCertificates=self.trust_all_certificates, masterHost=None, - usePathStyleAddressing=False + usePathStyleAddressing=self.path_style ) def to_database_backup_server_object(self): @@ -361,7 +362,7 @@ def to_database_backup_server_object(self): useHttps=self.https, trustAllCertificates=self.trust_all_certificates, masterHost=None, - usePathStyleAddressing=False + usePathStyleAddressing=self.path_style ) @@ -384,8 +385,8 @@ def to_server_object(self): class S3Compatible(HTTPBucket): def __init__(self, bucket, driver, access_key, secret_key, - endpoint, https, direct, verify_ssl): - super().__init__(bucket, driver, access_key, secret_key, endpoint, https, direct, verify_ssl) + endpoint, https, direct, verify_ssl, path_style=False): + super().__init__(bucket, driver, access_key, secret_key, endpoint, https, direct, verify_ssl, path_style) def to_server_object(self): param = super().to_server_object() @@ -444,8 +445,8 @@ def __init__(self, bucket, access_key, secret_key, class GenericS3(S3Compatible): def __init__(self, bucket, access_key, secret_key, - endpoint, https=False, direct=False, verify_ssl=True): - super().__init__(bucket, BucketType.GenericS3, access_key, secret_key, endpoint, https, direct, verify_ssl) + endpoint, https=False, direct=False, verify_ssl=True, path_style=False): + super().__init__(bucket, BucketType.GenericS3, access_key, secret_key, endpoint, https, direct, verify_ssl, path_style) class NetAppStorageGRID(S3Compatible): diff --git a/tests/ut/core/admin/test_servers.py b/tests/ut/core/admin/test_servers.py index 65f7df3e..22da0cb3 100644 --- a/tests/ut/core/admin/test_servers.py +++ b/tests/ut/core/admin/test_servers.py @@ -146,6 +146,23 @@ def test_enable_server_backup(self): }) self._assert_equal_objects(actual_param.backupToBucket, expected_param) + def test_enable_server_backup_path_style(self): + from cterasdk.core.types import GenericS3 + self._init_global_admin() + server_name = 'server' + with mock.patch("cterasdk.core.servers.Servers.system_database", new_callable=mock.PropertyMock) as query_mock: + query_mock.return_value = munch.Munch({'name': server_name, 'backupToBucket': None}) + bucket = GenericS3('portal-backup', 'access', 'secret', 'os.example.com', https=True, verify_ssl=True, path_style=True) + servers.Servers(self._global_admin).backup.enable(bucket, 5) + self._global_admin.api.put.assert_called_once_with(f'/servers/{server_name}', mock.ANY) + actual_param = self._global_admin.api.put.call_args[0][1] + expected_param = munch.Munch({ + 'enabled': True, + 'exportSchedulePeriod': 5, + 'details': TestCoreServers._create_database_backup_server_object(bucket) + }) + self._assert_equal_objects(actual_param.backupToBucket, expected_param) + @staticmethod def _create_database_backup_server_object(bucket): return munch.Munch({ @@ -157,7 +174,7 @@ def _create_database_backup_server_object(bucket): 'useHttps': bucket.https, 'trustAllCertificates': bucket.trust_all_certificates, 'masterHost': None, - 'usePathStyleAddressing': False + 'usePathStyleAddressing': bucket.path_style }) def test_disable_server_backup(self): @@ -170,6 +187,14 @@ def test_disable_server_backup(self): actual_param = self._global_admin.api.put.call_args[0][1] self._assert_equal_objects(actual_param.backupToBucket.enabled, False) + def test_get_server_backup(self): + self._init_global_admin() + backup_config = munch.Munch({'enabled': True, 'exportSchedulePeriod': 5, 'status': 'Connected'}) + with mock.patch("cterasdk.core.servers.Servers.system_database", new_callable=mock.PropertyMock) as query_mock: + query_mock.return_value = munch.Munch({'backupToBucket': backup_config}) + ret = servers.Servers(self._global_admin).backup.get() + self._assert_equal_objects(ret, backup_config) + def test_server_backup_status(self): self._init_global_admin() with mock.patch("cterasdk.core.servers.Servers.system_database", new_callable=mock.PropertyMock) as query_mock: