diff --git a/barman/cloud_providers/google_cloud_storage.py b/barman/cloud_providers/google_cloud_storage.py index f24c4369b..e8da7922e 100644 --- a/barman/cloud_providers/google_cloud_storage.py +++ b/barman/cloud_providers/google_cloud_storage.py @@ -131,7 +131,12 @@ def _reinit_session(self): Creates a client using "GOOGLE_APPLICATION_CREDENTIALS" env. An error will be raised if the variable is missing. """ - self.client = storage.Client() + client_options = None + universe_domain = os.getenv("GOOGLE_CLOUD_UNIVERSE_DOMAIN") + if universe_domain: + client_options = {"universe_domain": universe_domain} + + self.client = storage.Client(client_options=client_options) self.container_client = self.client.bucket(self.bucket_name) def test_connectivity(self): diff --git a/tests/test_cloud.py b/tests/test_cloud.py index 1cb37e717..eb2775e76 100644 --- a/tests/test_cloud.py +++ b/tests/test_cloud.py @@ -2658,6 +2658,53 @@ def test_connectivity_failure(self, gcs_client_mock): container_client_mock.exists.side_effect = GoogleAPIError("error") assert cloud_interface.test_connectivity() is False + @mock.patch.dict( + os.environ, {"GOOGLE_CLOUD_UNIVERSE_DOMAIN": "custom.universe.domain"} + ) + @mock.patch("barman.cloud_providers.google_cloud_storage.storage.Client") + def test_universe_domain_from_environment(self, gcs_client_mock): + """ + Test that the universe domain is properly loaded from GOOGLE_CLOUD_UNIVERSE_DOMAIN + environment variable and passed to the storage client + """ + # GIVEN a GoogleCloudInterface instance is created + cloud_interface = GoogleCloudInterface( + "https://console.cloud.google.com/storage/browser/barman-test/test" + ) + + # THEN the storage.Client should be called with the universe_domain in client_options + gcs_client_mock.assert_called_once_with( + client_options={"universe_domain": "custom.universe.domain"} + ) + + # AND the cloud interface should be properly initialized + assert cloud_interface.client == gcs_client_mock.return_value + assert ( + cloud_interface.container_client + == gcs_client_mock.return_value.bucket.return_value + ) + + @mock.patch("barman.cloud_providers.google_cloud_storage.storage.Client") + def test_no_universe_domain_environment(self, gcs_client_mock): + """ + Test that when GOOGLE_CLOUD_UNIVERSE_DOMAIN is not set, the storage client + is created without client_options + """ + # GIVEN a GoogleCloudInterface instance is created without universe domain env var + cloud_interface = GoogleCloudInterface( + "https://console.cloud.google.com/storage/browser/barman-test/test" + ) + + # THEN the storage.Client should be called with client_options=None + gcs_client_mock.assert_called_once_with(client_options=None) + + # AND the cloud interface should be properly initialized + assert cloud_interface.client == gcs_client_mock.return_value + assert ( + cloud_interface.container_client + == gcs_client_mock.return_value.bucket.return_value + ) + @mock.patch("barman.cloud_providers.google_cloud_storage.storage.Client") def test_setup_bucket(self, gcs_client_mock): """