update

google/cloud/storage/blob.py

@@ -829,11 +829,7 @@ def _get_download_url(
         """
         name_value_pairs = []
         if self.media_link is None:
-            hostname = (
-                client._connection.API_BASE_URL
-                if not hasattr(client._connection, "get_api_base_url_for_mtls")
-                else client._connection.get_api_base_url_for_mtls()
-            )
+            hostname = _get_host_name(client._connection)
             base_url = _DOWNLOAD_URL_TEMPLATE.format(hostname=hostname, path=self.path)
             if self.generation is not None:
                 name_value_pairs.append(("generation", "{:d}".format(self.generation)))
@@ -1686,11 +1682,7 @@ def _do_multipart_upload(
         info = self._get_upload_arguments(content_type)
         headers, object_metadata, content_type = info
 
-        hostname = (
-            client._connection.API_BASE_URL
-            if not hasattr(client._connection, "get_api_base_url_for_mtls")
-            else client._connection.get_api_base_url_for_mtls()
-        )
+        hostname = _get_host_name(client._connection)
         base_url = _MULTIPART_URL_TEMPLATE.format(
             hostname=hostname, bucket_path=self.bucket.path
         )
@@ -1872,11 +1864,7 @@ def _initiate_resumable_upload(
         if extra_headers is not None:
             headers.update(extra_headers)
 
-        hostname = (
-            client._connection.API_BASE_URL
-            if not hasattr(client._connection, "get_api_base_url_for_mtls")
-            else client._connection.get_api_base_url_for_mtls()
-        )
+        hostname = _get_host_name(client._connection)
         base_url = _RESUMABLE_URL_TEMPLATE.format(
             hostname=hostname, bucket_path=self.bucket.path
         )
@@ -3809,6 +3797,25 @@ def custom_time(self, value):
         self._patch_property("customTime", value)
 
 
+def _get_host_name(connection):
+    """Returns the host name from the given connection.
+
+    :type connection: :class:`~google.cloud.storage._http.Connection`
+    :param connection: The connection object.
+
+    :rtype: str
+    :returns: The host name.
+    """
+    # TODO: After google-cloud-core 1.6.0 is stable and we upgrade it
+    # to 1.6.0 in setup.py, we no longer need to check the attribute
+    # existence. We can simply return connection.get_api_base_url_for_mtls().
+    return (
+        connection.API_BASE_URL
+        if not hasattr(connection, "get_api_base_url_for_mtls")
+        else connection.get_api_base_url_for_mtls()
+    )
+
+
 def _get_encryption_headers(key, source=False):
     """Builds customer encryption key headers
 

tests/system/test_system.py

@@ -1867,6 +1867,12 @@ class TestStorageNotificationCRUD(unittest.TestCase):
     def setUpClass(cls):
         super(TestStorageNotificationCRUD, cls).setUpClass()
         if Config.TESTING_MTLS:
+            # mTLS is only available for python-pubsub >= 2.2.0. However, the
+            # system test uses python-pubsub < 2.0, so we skip those tests.
+            # Note that python-pubsub >= 2.0 no longer supports python 2.7, so
+            # we can only upgrade it after python 2.7 system test is removed.
+            # Since python-pubsub >= 2.0 has a new set of api, the test code
+            # also needs to be updated.
             raise unittest.SkipTest("Skip pubsub tests for mTLS testing")
 
     @property
@@ -2033,6 +2039,12 @@ def _kms_key_name(self, key_name=None):
     def setUpClass(cls):
         super(TestKMSIntegration, cls).setUpClass()
         if Config.TESTING_MTLS:
+            # mTLS is only available for python-kms >= 2.2.0. However, the
+            # system test uses python-kms < 2.0, so we skip those tests.
+            # Note that python-kms >= 2.0 no longer supports python 2.7, so
+            # we can only upgrade it after python 2.7 system test is removed.
+            # Since python-kms >= 2.0 has a new set of api, the test code
+            # also needs to be updated.
             raise unittest.SkipTest("Skip kms tests for mTLS testing")
 
         _empty_bucket(Config.CLIENT, cls.bucket)
@@ -2495,6 +2507,8 @@ def setUpClass(cls):
             type(Config.CLIENT._credentials)
             is not google.oauth2.service_account.Credentials
         ):
+            # mTLS only works for user credentials, it doesn't work for
+            # service account credentials.
             raise unittest.SkipTest("These tests require a service account credential")
 
     def setUp(self):