Repository: GoogleCloudPlatform/cloud-sql-python-connector
Pull request #981 (2024-01-25 14:29:48)

chore(main): release 1.7.0
:robot: I have created a release *beep* *boop*
---


## [1.7.0](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/compare/v1.6.0...v1.7.0) (2024-02-13)


### Features

* add support for a custom user agent ([#986](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/issues/986)) ([82da410](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/commit/82da41050992de2eddfeae5a81a434192e88d134))
* add support for Python 3.12 ([#905](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/issues/905)) ([7501d17](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/commit/7501d17ee05d41c1e009e4ca44ad7f8edaf18001))


### Dependencies

* Update dependency aiohttp to v3.9.3 ([#993](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/issues/993)) ([168b1f4](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/commit/168b1f43b2d0e4b376d8858aa32bc54e26d3e069))
* Update dependency cryptography to v42.0.2 ([#994](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/issues/994)) ([6b762df](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/commit/6b762df593ffbdc81a7ae98386230af9dc570147))
* Update dependency google-auth to v2.27.0 ([#980](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/issues/980)) ([def011a](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/commit/def011a0f7eee9d6288b441de5df416f429d227f))

---
This PR was generated with [Release Please](https://github.com/googleapis/release-please). See [documentation](https://github.com/googleapis/release-please#release-please).


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 4f49f69..11a00a6 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,20 @@
 # Changelog
 
+## [1.7.0](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/compare/v1.6.0...v1.7.0) (2024-02-13)
+
+
+### Features
+
+* add support for a custom user agent ([#986](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/issues/986)) ([82da410](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/commit/82da41050992de2eddfeae5a81a434192e88d134))
+* add support for Python 3.12 ([#905](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/issues/905)) ([7501d17](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/commit/7501d17ee05d41c1e009e4ca44ad7f8edaf18001))
+
+
+### Dependencies
+
+* Update dependency aiohttp to v3.9.3 ([#993](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/issues/993)) ([168b1f4](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/commit/168b1f43b2d0e4b376d8858aa32bc54e26d3e069))
+* Update dependency cryptography to v42.0.2 ([#994](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/issues/994)) ([6b762df](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/commit/6b762df593ffbdc81a7ae98386230af9dc570147))
+* Update dependency google-auth to v2.27.0 ([#980](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/issues/980)) ([def011a](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/commit/def011a0f7eee9d6288b441de5df416f429d227f))
+
 ## [1.6.0](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/compare/v1.5.0...v1.6.0) (2024-01-17)
 
 
diff --git a/google/cloud/sql/connector/client.py b/google/cloud/sql/connector/client.py
new file mode 100644
index 0000000..f28faa6
--- /dev/null
+++ b/google/cloud/sql/connector/client.py
@@ -0,0 +1,221 @@
+# Copyright 2024 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+import datetime
+import logging
+from typing import Any, Dict, Optional, Tuple, TYPE_CHECKING
+
+import aiohttp
+from cryptography.hazmat.backends import default_backend
+from cryptography.x509 import load_pem_x509_certificate
+import google.auth.transport.requests
+
+from google.cloud.sql.connector.refresh_utils import _downscope_credentials
+from google.cloud.sql.connector.version import __version__ as version
+
+if TYPE_CHECKING:
+    from google.auth.credentials import Credentials
+
+USER_AGENT: str = f"cloud-sql-python-connector/{version}"
+API_VERSION: str = "v1beta4"
+
+logger = logging.getLogger(name=__name__)
+
+
+def _format_user_agent(driver: Optional[str], custom: Optional[str]) -> str:
+    agent = f"{USER_AGENT}+{driver}" if driver else USER_AGENT
+    if custom:
+        agent = f"{agent} {custom}"
+    return agent
+
+
+class CloudSQLClient:
+    def __init__(
+        self,
+        sqladmin_api_endpoint: str,
+        quota_project: Optional[str],
+        credentials: Credentials,
+        client: Optional[aiohttp.ClientSession] = None,
+        driver: Optional[str] = None,
+        user_agent: Optional[str] = None,
+    ) -> None:
+        """
+        Establish the client to be used for Cloud SQL Admin API requests.
+
+        Args:
+            sqladmin_api_endpoint (str): Base URL to use when calling
+                the Cloud SQL Admin API endpoints.
+            quota_project (str): The Project ID for an existing Google Cloud
+                project. The project specified is used for quota and
+                billing purposes.
+            credentials (google.auth.credentials.Credentials):
+                A credentials object created from the google-auth Python library.
+                Must have the Cloud SQL Admin scopes. For more info check out
+                https://google-auth.readthedocs.io/en/latest/.
+            client (aiohttp.ClientSession): Async client used to make requests to
+                Cloud SQL Admin APIs.
+                Optional, defaults to None and creates new client.
+            driver (str): Database driver to be used by the client.
+        """
+        user_agent = _format_user_agent(driver, user_agent)
+        headers = {
+            "x-goog-api-client": user_agent,
+            "User-Agent": user_agent,
+            "Content-Type": "application/json",
+        }
+        if quota_project:
+            headers["x-goog-user-project"] = quota_project
+
+        self._client = client if client else aiohttp.ClientSession(headers=headers)
+        self._credentials = credentials
+        self._sqladmin_api_endpoint = sqladmin_api_endpoint
+        self._user_agent = user_agent
+
+    async def _get_metadata(
+        self,
+        project: str,
+        region: str,
+        instance: str,
+    ) -> Dict[str, Any]:
+        """Requests metadata from the Cloud SQL Instance
+        and returns a dictionary containing the IP addresses and certificate
+        authority of the Cloud SQL Instance.
+
+        :type project: str
+        :param project:
+            A string representing the name of the project.
+
+        :type region: str
+        :param region : A string representing the name of the region.
+
+        :type instance: str
+        :param instance: A string representing the name of the instance.
+
+        :rtype: Dict[str: Union[Dict, str]]
+        :returns: Returns a dictionary containing a dictionary of all IP
+            addresses and their type and a string representing the
+            certificate authority.
+        """
+        if not self._credentials.valid:
+            request = google.auth.transport.requests.Request()
+            self._credentials.refresh(request)
+
+        headers = {
+            "Authorization": f"Bearer {self._credentials.token}",
+        }
+
+        url = f"{self._sqladmin_api_endpoint}/sql/{API_VERSION}/projects/{project}/instances/{instance}/connectSettings"
+
+        logger.debug(f"['{instance}']: Requesting metadata")
+
+        resp = await self._client.get(url, headers=headers, raise_for_status=True)
+        ret_dict = await resp.json()
+
+        if ret_dict["region"] != region:
+            raise ValueError(
+                f'[{project}:{region}:{instance}]: Provided region was mismatched - got region {region}, expected {ret_dict["region"]}.'
+            )
+
+        ip_addresses = (
+            {ip["type"]: ip["ipAddress"] for ip in ret_dict["ipAddresses"]}
+            if "ipAddresses" in ret_dict
+            else {}
+        )
+        if "dnsName" in ret_dict:
+            ip_addresses["PSC"] = ret_dict["dnsName"]
+
+        return {
+            "ip_addresses": ip_addresses,
+            "server_ca_cert": ret_dict["serverCaCert"]["cert"],
+            "database_version": ret_dict["databaseVersion"],
+        }
+
+    async def _get_ephemeral(
+        self,
+        project: str,
+        instance: str,
+        pub_key: str,
+        enable_iam_auth: bool = False,
+    ) -> Tuple[str, datetime.datetime]:
+        """Asynchronously requests an ephemeral certificate from the Cloud SQL Instance.
+
+        :type project: str
+        :param project : A string representing the name of the project.
+
+        :type instance: str
+        :param instance: A string representing the name of the instance.
+
+        :type pub_key:
+        :param str: A string representing PEM-encoded RSA public key.
+
+        :type enable_iam_auth: bool
+        :param enable_iam_auth
+            Enables automatic IAM database authentication for Postgres or MySQL
+            instances.
+
+        :rtype: str
+        :returns: An ephemeral certificate from the Cloud SQL instance that allows
+            authorized connections to the instance.
+        """
+
+        logger.debug(f"['{instance}']: Requesting ephemeral certificate")
+
+        if not self._credentials.valid:
+            request = google.auth.transport.requests.Request()
+            self._credentials.refresh(request)
+
+        headers = {
+            "Authorization": f"Bearer {self._credentials.token}",
+        }
+
+        url = f"{self._sqladmin_api_endpoint}/sql/{API_VERSION}/projects/{project}/instances/{instance}:generateEphemeralCert"
+
+        data = {"public_key": pub_key}
+
+        if enable_iam_auth:
+            # down-scope credentials with only IAM login scope (refreshes them too)
+            login_creds = _downscope_credentials(self._credentials)
+            data["access_token"] = login_creds.token
+
+        resp = await self._client.post(
+            url, headers=headers, json=data, raise_for_status=True
+        )
+
+        ret_dict = await resp.json()
+
+        ephemeral_cert: str = ret_dict["ephemeralCert"]["cert"]
+
+        # decode cert to read expiration
+        x509 = load_pem_x509_certificate(
+            ephemeral_cert.encode("UTF-8"), default_backend()
+        )
+        expiration = x509.not_valid_after_utc
+        # for IAM authentication OAuth2 token is embedded in cert so it
+        # must still be valid for successful connection
+        if enable_iam_auth:
+            token_expiration: datetime.datetime = login_creds.expiry
+            # google.auth library strips timezone info for backwards compatibality
+            # reasons with Python 2. Add it back to allow timezone aware datetimes.
+            # Ref: https://github.com/googleapis/google-auth-library-python/blob/49a5ff7411a2ae4d32a7d11700f9f961c55406a9/google/auth/_helpers.py#L93-L99
+            token_expiration = token_expiration.replace(tzinfo=datetime.timezone.utc)
+
+            if expiration > token_expiration:
+                expiration = token_expiration
+        return ephemeral_cert, expiration
+
+    async def close(self) -> None:
+        """Close CloudSQLClient gracefully."""
+        await self._client.close()
diff --git a/google/cloud/sql/connector/connector.py b/google/cloud/sql/connector/connector.py
index 11765df..96da705 100755
--- a/google/cloud/sql/connector/connector.py
+++ b/google/cloud/sql/connector/connector.py
@@ -13,6 +13,7 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.
 """
+
 from __future__ import annotations
 
 import asyncio
@@ -28,6 +29,7 @@ from google.auth.credentials import Credentials
 from google.auth.credentials import with_scopes_if_required
 
 import google.cloud.sql.connector.asyncpg as asyncpg
+from google.cloud.sql.connector.client import CloudSQLClient
 from google.cloud.sql.connector.exceptions import ConnectorLoopError
 from google.cloud.sql.connector.exceptions import DnsNameResolutionError
 from google.cloud.sql.connector.instance import Instance
@@ -109,11 +111,12 @@ class Connector:
                 loop=self._loop,
             )
         self._instances: Dict[str, Instance] = {}
+        self._client: Optional[CloudSQLClient] = None
 
         # initialize credentials
         scopes = ["https://www.googleapis.com/auth/sqlservice.admin"]
         if credentials:
-            # verfiy custom credentials are proper type
+            # verify custom credentials are proper type
             # and atleast base class of google.auth.credentials
             if not isinstance(credentials, Credentials):
                 raise TypeError(
@@ -207,27 +210,31 @@ class Connector:
         # Use the Instance to establish an SSL Connection.
         #
         # Return a DBAPI connection
+        if self._client is None:
+            # lazy init client as it has to be initialized in async context
+            self._client = CloudSQLClient(
+                self._sqladmin_api_endpoint,
+                self._quota_project,
+                self._credentials,
+                user_agent=self._user_agent,
+                driver=driver,
+            )
         enable_iam_auth = kwargs.pop("enable_iam_auth", self._enable_iam_auth)
         if instance_connection_string in self._instances:
             instance = self._instances[instance_connection_string]
             if enable_iam_auth != instance._enable_iam_auth:
                 raise ValueError(
-                    f"connect() called with `enable_iam_auth={enable_iam_auth}`, "
-                    f"but previously used enable_iam_auth={instance._enable_iam_auth}`. "
+                    f"connect() called with 'enable_iam_auth={enable_iam_auth}', "
+                    f"but previously used 'enable_iam_auth={instance._enable_iam_auth}'. "
                     "If you require both for your use case, please use a new "
                     "connector.Connector object."
                 )
         else:
             instance = Instance(
                 instance_connection_string,
-                driver,
+                self._client,
                 self._keys,
-                self._loop,
-                self._credentials,
                 enable_iam_auth,
-                self._quota_project,
-                self._sqladmin_api_endpoint,
-                user_agent=self._user_agent,
             )
             self._instances[instance_connection_string] = instance
 
@@ -329,8 +336,8 @@ class Connector:
             close_future = asyncio.run_coroutine_threadsafe(
                 self.close_async(), loop=self._loop
             )
-            # Will attempt to safely shut down tasks for 5s
-            close_future.result(timeout=5)
+            # Will attempt to safely shut down tasks for 3s
+            close_future.result(timeout=3)
         # if background thread exists for Connector, clean it up
         if self._thread:
             if self._loop.is_running():
@@ -345,12 +352,8 @@ class Connector:
         await asyncio.gather(
             *[instance.close() for instance in self._instances.values()]
         )
-
-    def __del__(self) -> None:
-        """Close Connector as part of garbage collection"""
-        # only want to call destructor when used for sync connections
-        if self._thread:
-            self.close()
+        if self._client:
+            await self._client.close()
 
 
 async def create_async_connector(
diff --git a/google/cloud/sql/connector/instance.py b/google/cloud/sql/connector/instance.py
index 6b64741..dda5eb9 100644
--- a/google/cloud/sql/connector/instance.py
+++ b/google/cloud/sql/connector/instance.py
@@ -13,6 +13,7 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.
 """
+
 from __future__ import annotations
 
 import asyncio
@@ -21,21 +22,18 @@ import logging
 import re
 import ssl
 from tempfile import TemporaryDirectory
-from typing import Any, Dict, Optional, Tuple, TYPE_CHECKING
+from typing import Any, Dict, Tuple, TYPE_CHECKING
 
 import aiohttp
-from google.auth.credentials import Credentials
 
+from google.cloud.sql.connector.client import CloudSQLClient
 from google.cloud.sql.connector.exceptions import AutoIAMAuthNotSupported
 from google.cloud.sql.connector.exceptions import CloudSQLIPTypeError
 from google.cloud.sql.connector.exceptions import TLSVersionError
 from google.cloud.sql.connector.rate_limiter import AsyncRateLimiter
-from google.cloud.sql.connector.refresh_utils import _get_ephemeral
-from google.cloud.sql.connector.refresh_utils import _get_metadata
 from google.cloud.sql.connector.refresh_utils import _is_valid
 from google.cloud.sql.connector.refresh_utils import _seconds_until_refresh
 from google.cloud.sql.connector.utils import write_to_file
-from google.cloud.sql.connector.version import __version__ as version
 
 if TYPE_CHECKING:
     import datetime
@@ -132,13 +130,6 @@ class ConnectionInfo:
         )
 
 
-def _format_user_agent(version: str, driver: str, custom: Optional[str]) -> str:
-    agent = f"{APPLICATION_NAME}/{version}+{driver}"
-    if custom:
-        agent = f"{agent} {custom}"
-    return agent
-
-
 class Instance:
     """A class to manage the details of the connection to a Cloud SQL
     instance, including refreshing the credentials.
@@ -148,67 +139,15 @@ class Instance:
         string.
     :type instance_connection_string: str
 
-    :param user_agent_string:
-        The user agent string to append to SQLAdmin API requests
-    :type user_agent_string: str
-
-    :type credentials: google.auth.credentials.Credentials
-    :param credentials
-        Credentials object used to authenticate connections to Cloud SQL server.
-
     :param enable_iam_auth
         Enables automatic IAM database authentication for Postgres or MySQL
         instances.
     :type enable_iam_auth: bool
-
-    :param loop:
-        A new event loop for the refresh function to run in.
-    :type loop: asyncio.AbstractEventLoop
-
-    :type quota_project: str
-    :param quota_project
-        The Project ID for an existing Google Cloud project. The project specified
-        is used for quota and billing purposes. If not specified, defaults to
-        project sourced from environment.
-
-    :type sqladmin_api_endpoint: str
-    :param sqladmin_api_endpoint:
-        Base URL to use when calling the Cloud SQL Admin API endpoint.
-        Defaults to "https://sqladmin.googleapis.com", this argument should
-        only be used in development.
     """
 
-    # asyncio.AbstractEventLoop is used because the default loop,
-    # SelectorEventLoop, is usable on both Unix and Windows but has limited
-    # functionality on Windows. It is recommended to use ProactorEventLoop
-    # while developing on Windows.
-    # Link to Github issue:
-    # https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/issues/22
-    _loop: asyncio.AbstractEventLoop
-
     _enable_iam_auth: bool
-
-    __client_session: Optional[aiohttp.ClientSession] = None
-
-    @property
-    def _client_session(self) -> aiohttp.ClientSession:
-        if self.__client_session is None:
-            headers = {
-                "x-goog-api-client": self._user_agent_string,
-                "User-Agent": self._user_agent_string,
-                "Content-Type": "application/json",
-            }
-            if self._quota_project:
-                headers["x-goog-user-project"] = self._quota_project
-            self.__client_session = aiohttp.ClientSession(headers=headers)
-        return self.__client_session
-
-    _credentials: Credentials
     _keys: asyncio.Future
-
     _instance_connection_string: str
-    _user_agent_string: str
-    _sqladmin_api_endpoint: str
     _instance: str
     _project: str
     _region: str
@@ -221,14 +160,9 @@ class Instance:
     def __init__(
         self,
         instance_connection_string: str,
-        driver_name: str,
+        client: CloudSQLClient,
         keys: asyncio.Future,
-        loop: asyncio.AbstractEventLoop,
-        credentials: Credentials,
         enable_iam_auth: bool = False,
-        quota_project: Optional[str] = None,
-        sqladmin_api_endpoint: str = "https://sqladmin.googleapis.com",
-        user_agent: Optional[str] = None,
     ) -> None:
         # validate and parse instance connection name
         self._project, self._region, self._instance = _parse_instance_connection_name(
@@ -237,19 +171,11 @@ class Instance:
         self._instance_connection_string = instance_connection_string
 
         self._enable_iam_auth = enable_iam_auth
-
-        self._user_agent_string = _format_user_agent(
-            version,
-            driver_name,
-            user_agent,
-        )
-        self._quota_project = quota_project
-        self._sqladmin_api_endpoint = sqladmin_api_endpoint
-        self._loop = loop
         self._keys = keys
-        self._credentials = credentials
+        self._client = client
         self._refresh_rate_limiter = AsyncRateLimiter(
-            max_capacity=2, rate=1 / 30, loop=self._loop
+            max_capacity=2,
+            rate=1 / 30,
         )
         self._refresh_in_progress = asyncio.locks.Event()
         self._current = self._schedule_refresh(0)
@@ -287,22 +213,16 @@ class Instance:
 
             logger.debug(f"['{self._instance_connection_string}']: Creating context")
 
-            metadata_task = self._loop.create_task(
-                _get_metadata(
-                    self._client_session,
-                    self._sqladmin_api_endpoint,
-                    self._credentials,
+            metadata_task = asyncio.create_task(
+                self._client._get_metadata(
                     self._project,
                     self._region,
                     self._instance,
                 )
             )
 
-            ephemeral_task = self._loop.create_task(
-                _get_ephemeral(
-                    self._client_session,
-                    self._sqladmin_api_endpoint,
-                    self._credentials,
+            ephemeral_task = asyncio.create_task(
+                self._client._get_ephemeral(
                     self._project,
                     self._instance,
                     pub_key,
@@ -374,7 +294,7 @@ class Instance:
                 logger.debug(f"['{self._instance_connection_string}']: Entering sleep")
                 if delay > 0:
                     await asyncio.sleep(delay)
-                refresh_task = self._loop.create_task(self._perform_refresh())
+                refresh_task = asyncio.create_task(self._perform_refresh())
                 refresh_data = await refresh_task
             except asyncio.CancelledError:
                 logger.debug(
@@ -405,7 +325,7 @@ class Instance:
             return refresh_data
 
         # schedule refresh task and return it
-        scheduled_task = self._loop.create_task(_refresh_task(self, delay))
+        scheduled_task = asyncio.create_task(_refresh_task(self, delay))
         return scheduled_task
 
     async def connect_info(
@@ -451,4 +371,6 @@ class Instance:
         logger.debug(
             f"['{self._instance_connection_string}']: Waiting for _client_session to close"
         )
-        await self._client_session.close()
+        # gracefully wait for tasks to cancel
+        tasks = asyncio.gather(self._current, self._next, return_exceptions=True)
+        await asyncio.wait_for(tasks, timeout=2.0)
diff --git a/google/cloud/sql/connector/rate_limiter.py b/google/cloud/sql/connector/rate_limiter.py
index 6698471..38e7b94 100644
--- a/google/cloud/sql/connector/rate_limiter.py
+++ b/google/cloud/sql/connector/rate_limiter.py
@@ -13,6 +13,7 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.
 """
+
 import asyncio
 from typing import Optional
 
diff --git a/google/cloud/sql/connector/refresh_utils.py b/google/cloud/sql/connector/refresh_utils.py
index 18ce222..1d678c5 100644
--- a/google/cloud/sql/connector/refresh_utils.py
+++ b/google/cloud/sql/connector/refresh_utils.py
@@ -13,197 +13,26 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.
 """
+
 from __future__ import annotations
 
 import asyncio
 import copy
 import datetime
 import logging
-from typing import Any, Dict, List, Tuple, TYPE_CHECKING
+from typing import List
 
-from cryptography.hazmat.backends import default_backend
-from cryptography.x509 import load_pem_x509_certificate
 from google.auth.credentials import Credentials
 from google.auth.credentials import Scoped
 import google.auth.transport.requests
 
-if TYPE_CHECKING:
-    import aiohttp
-
 logger = logging.getLogger(name=__name__)
 
-_sql_api_version: str = "v1beta4"
-
 # _refresh_buffer is the amount of time before a refresh's result expires
 # that a new refresh operation begins.
 _refresh_buffer: int = 4 * 60  # 4 minutes
 
 
-async def _get_metadata(
-    client_session: aiohttp.ClientSession,
-    sqladmin_api_endpoint: str,
-    credentials: Credentials,
-    project: str,
-    region: str,
-    instance: str,
-) -> Dict[str, Any]:
-    """Requests metadata from the Cloud SQL Instance
-    and returns a dictionary containing the IP addresses and certificate
-    authority of the Cloud SQL Instance.
-
-    :type sqladmin_api_endpoint: str
-    :param sqladmin_api_endpoint:
-        Base URL to use when calling the Cloud SQL Admin API endpoint.
-
-    :type credentials: google.auth.credentials.Credentials
-    :param credentials:
-        A credentials object created from the google-auth Python library.
-        Must have the SQL Admin API scopes. For more info check out
-        https://google-auth.readthedocs.io/en/latest/.
-
-    :type project: str
-    :param project:
-        A string representing the name of the project.
-
-    :type region: str
-    :param region : A string representing the name of the region.
-
-    :type instance: str
-    :param instance: A string representing the name of the instance.
-
-    :rtype: Dict[str: Union[Dict, str]]
-    :returns: Returns a dictionary containing a dictionary of all IP
-          addresses and their type and a string representing the
-          certificate authority.
-
-    :raises TypeError: If any of the arguments are not the specified type.
-    """
-
-    if not credentials.valid:
-        request = google.auth.transport.requests.Request()
-        credentials.refresh(request)
-
-    headers = {
-        "Authorization": f"Bearer {credentials.token}",
-    }
-
-    url = f"{sqladmin_api_endpoint}/sql/{_sql_api_version}/projects/{project}/instances/{instance}/connectSettings"
-
-    logger.debug(f"['{instance}']: Requesting metadata")
-
-    resp = await client_session.get(url, headers=headers, raise_for_status=True)
-    ret_dict = await resp.json()
-
-    if ret_dict["region"] != region:
-        raise ValueError(
-            f'[{project}:{region}:{instance}]: Provided region was mismatched - got region {region}, expected {ret_dict["region"]}.'
-        )
-
-    ip_addresses = (
-        {ip["type"]: ip["ipAddress"] for ip in ret_dict["ipAddresses"]}
-        if "ipAddresses" in ret_dict
-        else {}
-    )
-    if "dnsName" in ret_dict:
-        ip_addresses["PSC"] = ret_dict["dnsName"]
-
-    metadata = {
-        "ip_addresses": ip_addresses,
-        "server_ca_cert": ret_dict["serverCaCert"]["cert"],
-        "database_version": ret_dict["databaseVersion"],
-    }
-
-    return metadata
-
-
-async def _get_ephemeral(
-    client_session: aiohttp.ClientSession,
-    sqladmin_api_endpoint: str,
-    credentials: Credentials,
-    project: str,
-    instance: str,
-    pub_key: str,
-    enable_iam_auth: bool = False,
-) -> Tuple[str, datetime.datetime]:
-    """Asynchronously requests an ephemeral certificate from the Cloud SQL Instance.
-
-    :type sqladmin_api_endpoint: str
-    :param sqladmin_api_endpoint:
-        Base URL to use when calling the Cloud SQL Admin API endpoint.
-
-    :type credentials: google.auth.credentials.Credentials
-    :param credentials: A credentials object
-        created from the google-auth library. Must be
-        using the SQL Admin API scopes. For more info, check out
-        https://google-auth.readthedocs.io/en/latest/.
-
-    :type project: str
-    :param project : A string representing the name of the project.
-
-    :type instance: str
-    :param instance: A string representing the name of the instance.
-
-    :type pub_key:
-    :param str: A string representing PEM-encoded RSA public key.
-
-    :type enable_iam_auth: bool
-    :param enable_iam_auth
-        Enables automatic IAM database authentication for Postgres or MySQL
-        instances.
-
-    :rtype: str
-    :returns: An ephemeral certificate from the Cloud SQL instance that allows
-          authorized connections to the instance.
-
-    :raises TypeError: If one of the arguments passed in is None.
-    """
-
-    logger.debug(f"['{instance}']: Requesting ephemeral certificate")
-
-    if not isinstance(pub_key, str):
-        raise TypeError(f"pub_key must be of type str, got {type(pub_key)}")
-
-    if not credentials.valid:
-        request = google.auth.transport.requests.Request()
-        credentials.refresh(request)
-
-    headers = {
-        "Authorization": f"Bearer {credentials.token}",
-    }
-
-    url = f"{sqladmin_api_endpoint}/sql/{_sql_api_version}/projects/{project}/instances/{instance}:generateEphemeralCert"
-
-    data = {"public_key": pub_key}
-
-    if enable_iam_auth:
-        # down-scope credentials with only IAM login scope (refreshes them too)
-        login_creds = _downscope_credentials(credentials)
-        data["access_token"] = login_creds.token
-
-    resp = await client_session.post(
-        url, headers=headers, json=data, raise_for_status=True
-    )
-
-    ret_dict = await resp.json()
-
-    ephemeral_cert: str = ret_dict["ephemeralCert"]["cert"]
-
-    # decode cert to read expiration
-    x509 = load_pem_x509_certificate(ephemeral_cert.encode("UTF-8"), default_backend())
-    expiration = x509.not_valid_after_utc
-    # for IAM authentication OAuth2 token is embedded in cert so it
-    # must still be valid for successful connection
-    if enable_iam_auth:
-        token_expiration: datetime.datetime = login_creds.expiry
-        # google.auth library strips timezone info for backwards compatibality
-        # reasons with Python 2. Add it back to allow timezone aware datetimes.
-        # Ref: https://github.com/googleapis/google-auth-library-python/blob/49a5ff7411a2ae4d32a7d11700f9f961c55406a9/google/auth/_helpers.py#L93-L99
-        token_expiration = token_expiration.replace(tzinfo=datetime.timezone.utc)
-        if expiration > token_expiration:
-            expiration = token_expiration
-    return ephemeral_cert, expiration
-
-
 def _seconds_until_refresh(
     expiration: datetime.datetime,
 ) -> int:
diff --git a/google/cloud/sql/connector/version.py b/google/cloud/sql/connector/version.py
index 6cd7b95..e4f6eef 100644
--- a/google/cloud/sql/connector/version.py
+++ b/google/cloud/sql/connector/version.py
@@ -12,4 +12,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-__version__ = "1.6.0"
+__version__ = "1.7.0"


Test patch:
diff --git a/requirements-test.txt b/requirements-test.txt
index 201d95f..24324d7 100644
--- a/requirements-test.txt
+++ b/requirements-test.txt
@@ -10,3 +10,4 @@ pg8000==1.30.4
 asyncpg==0.29.0
 python-tds==1.15.0
 aioresponses==0.7.6
+pytest-aiohttp==1.0.5
diff --git a/tests/conftest.py b/tests/conftest.py
index f2d671f..8f3d780 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -17,15 +17,14 @@ import asyncio
 import os
 import socket
 from threading import Thread
-from typing import Any, AsyncGenerator, Generator, Tuple
+from typing import Any, AsyncGenerator, Generator
 
-from aioresponses import aioresponses
-from google.auth.credentials import Credentials
+from aiohttp import web
 import pytest  # noqa F401 Needed to run the tests
 from unit.mocks import FakeCredentials  # type: ignore
 from unit.mocks import FakeCSQLInstance  # type: ignore
 
-from google.cloud.sql.connector import Connector
+from google.cloud.sql.connector.client import CloudSQLClient
 from google.cloud.sql.connector.instance import Instance
 from google.cloud.sql.connector.utils import generate_keys
 
@@ -106,98 +105,44 @@ def kwargs() -> Any:
     return kwargs
 
 
-@pytest.fixture(scope="module")
-def mock_instance() -> FakeCSQLInstance:
-    mock_instance = FakeCSQLInstance("my-project", "my-region", "my-instance")
-    return mock_instance
+@pytest.fixture(scope="session")
+def fake_instance() -> FakeCSQLInstance:
+    return FakeCSQLInstance()
 
 
 @pytest.fixture
-async def instance(
-    mock_instance: FakeCSQLInstance,
-    fake_credentials: Credentials,
-) -> AsyncGenerator[Instance, None]:
-    """
-    Instance with mocked API calls.
-    """
-    loop = asyncio.get_running_loop()
-    # generate client key pair
-    keys = asyncio.create_task(generate_keys())
-    _, client_key = await keys
-
-    # mock Cloud SQL Admin API calls
-    with aioresponses() as mocked:
-        mocked.get(
-            f"https://sqladmin.googleapis.com/sql/v1beta4/projects/{mock_instance.project}/instances/{mock_instance.name}/connectSettings",
-            status=200,
-            body=mock_instance.connect_settings(),
-            repeat=True,
-        )
-        mocked.post(
-            f"https://sqladmin.googleapis.com/sql/v1beta4/projects/{mock_instance.project}/instances/{mock_instance.name}:generateEphemeralCert",
-            status=200,
-            body=mock_instance.generate_ephemeral(client_key),
-            repeat=True,
-        )
-
-        instance = Instance(
-            f"{mock_instance.project}:{mock_instance.region}:{mock_instance.name}",
-            "pg8000",
-            keys,
-            loop,
-            fake_credentials,
-        )
-
-        yield instance
-        await instance.close()
+async def fake_client(
+    fake_credentials: FakeCredentials,
+    fake_instance: FakeCSQLInstance,
+    aiohttp_client: Any,
+) -> CloudSQLClient:
+    app = web.Application()
+    # add SQL Server instance for additional tests
+    # TODO: remove when SQL Server supports IAM authN
+    sqlserver_instance = FakeCSQLInstance(
+        name="sqlserver-instance", db_version="SQLSERVER_2019_STANDARD"
+    )
+    metadata_uri = f"/sql/v1beta4/projects/{fake_instance.project}/instances/{fake_instance.name}/connectSettings"
+    app.router.add_get(metadata_uri, fake_instance.connect_settings)
+    sqlserver_metadata_uri = f"/sql/v1beta4/projects/{sqlserver_instance.project}/instances/{sqlserver_instance.name}/connectSettings"
+    app.router.add_get(sqlserver_metadata_uri, sqlserver_instance.connect_settings)
+    client_cert_uri = f"/sql/v1beta4/projects/{fake_instance.project}/instances/{fake_instance.name}:generateEphemeralCert"
+    app.router.add_post(client_cert_uri, fake_instance.generate_ephemeral)
+    sqlserver_client_cert_uri = f"/sql/v1beta4/projects/{sqlserver_instance.project}/instances/{sqlserver_instance.name}:generateEphemeralCert"
+    app.router.add_post(
+        sqlserver_client_cert_uri, sqlserver_instance.generate_ephemeral
+    )
+    client_session = await aiohttp_client(app)
+    return CloudSQLClient("", "", fake_credentials, client=client_session)
 
 
 @pytest.fixture
-async def connector(fake_credentials: Credentials) -> AsyncGenerator[Connector, None]:
-    instance_connection_name = "my-project:my-region:my-instance"
-    project, region, instance_name = instance_connection_name.split(":")
-    # initialize connector
-    connector = Connector(credentials=fake_credentials)
-    # mock Cloud SQL Admin API calls
-    mock_instance = FakeCSQLInstance(project, region, instance_name)
-
-    async def wait_for_keys(future: asyncio.Future) -> Tuple[bytes, str]:
-        """
-        Helper method to await keys of Connector in tests prior to
-        initializing an Instance object.
-        """
-        return await future
-
-    # converting asyncio.Future into concurrent.Future
-    # await keys in background thread so that .result() is set
-    # required because keys are needed for mocks, but are not awaited
-    # in the code until Instance() is initialized
-    _, client_key = asyncio.run_coroutine_threadsafe(
-        wait_for_keys(connector._keys), connector._loop
-    ).result()
-    with aioresponses() as mocked:
-        mocked.get(
-            f"https://sqladmin.googleapis.com/sql/v1beta4/projects/{project}/instances/{instance_name}/connectSettings",
-            status=200,
-            body=mock_instance.connect_settings(),
-            repeat=True,
-        )
-        mocked.post(
-            f"https://sqladmin.googleapis.com/sql/v1beta4/projects/{project}/instances/{instance_name}:generateEphemeralCert",
-            status=200,
-            body=mock_instance.generate_ephemeral(client_key),
-            repeat=True,
-        )
-        # initialize Instance using mocked API calls
-        instance = Instance(
-            instance_connection_name,
-            "pg8000",
-            connector._keys,
-            connector._loop,
-            fake_credentials,
-        )
-
-        connector._instances[instance_connection_name] = instance
-
-        yield connector
-        connector.close()
+async def instance(fake_client: CloudSQLClient) -> AsyncGenerator[Instance, None]:
+    keys = asyncio.create_task(generate_keys())
+    instance = Instance(
+        "test-project:test-region:test-instance",
+        client=fake_client,
+        keys=keys,
+    )
+    yield instance
+    await instance.close()
diff --git a/tests/unit/mocks.py b/tests/unit/mocks.py
index 57d4642..babaf0e 100644
--- a/tests/unit/mocks.py
+++ b/tests/unit/mocks.py
@@ -21,6 +21,7 @@ import ssl
 from tempfile import TemporaryDirectory
 from typing import Any, Callable, Dict, Optional, Tuple
 
+from aiohttp import web
 from cryptography import x509
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import hashes
@@ -29,7 +30,6 @@ from cryptography.hazmat.primitives.asymmetric import rsa
 from cryptography.x509.oid import NameOID
 from google.auth.credentials import Credentials
 
-from google.cloud.sql.connector import IPTypes
 from google.cloud.sql.connector.instance import ConnectionInfo
 from google.cloud.sql.connector.utils import generate_keys
 from google.cloud.sql.connector.utils import write_to_file
@@ -78,25 +78,6 @@ class FakeCredentials:
         return self.token is not None and not self.expired
 
 
-class MockInstance:
-    _enable_iam_auth: bool
-
-    def __init__(
-        self,
-        enable_iam_auth: bool = False,
-    ) -> None:
-        self._enable_iam_auth = enable_iam_auth
-
-    # mock connect_info
-    async def connect_info(
-        self,
-        driver: str,
-        ip_type: IPTypes,
-        **kwargs: Any,
-    ) -> Any:
-        return True
-
-
 class BadRefresh(Exception):
     pass
 
@@ -128,7 +109,11 @@ async def instance_metadata_error(*args: Any, **kwargs: Any) -> None:
 
 
 def generate_cert(
-    project: str, name: str
+    project: str,
+    name: str,
+    cert_before: datetime.datetime = datetime.datetime.now(datetime.timezone.utc),
+    cert_after: datetime.datetime = datetime.datetime.now(datetime.timezone.utc)
+    + datetime.timedelta(hours=1),
 ) -> Tuple[x509.CertificateBuilder, rsa.RSAPrivateKey]:
     """
     Generate a private key and cert object to be used in testing.
@@ -153,12 +138,8 @@ def generate_cert(
         .issuer_name(issuer)
         .public_key(key.public_key())
         .serial_number(x509.random_serial_number())
-        .not_valid_before(datetime.datetime.now(datetime.timezone.utc))
-        .not_valid_after(
-            # cert valid for 10 mins
-            datetime.datetime.now(datetime.timezone.utc)
-            + datetime.timedelta(minutes=60)
-        )
+        .not_valid_before(cert_before)
+        .not_valid_after(cert_after)
     )
     return cert, key
 
@@ -197,7 +178,7 @@ def client_key_signed_cert(
         .issuer_name(issuer)
         .public_key(client_key)
         .serial_number(x509.random_serial_number())
-        .not_valid_before(datetime.datetime.now(datetime.timezone.utc))
+        .not_valid_before(cert._not_valid_before)
         .not_valid_after(cert._not_valid_after)  # type: ignore
     )
     return (
@@ -230,72 +211,68 @@ async def create_ssl_context() -> ssl.SSLContext:
 
 
 class FakeCSQLInstance:
+    """Fake Cloud SQL instance to use for testing"""
+
     def __init__(
         self,
-        project: str = "my-project",
-        region: str = "my-region",
-        name: str = "my-instance",
-        db_version: str = "POSTGRES_14",
+        project: str = "test-project",
+        region: str = "test-region",
+        name: str = "test-instance",
+        db_version: str = "POSTGRES_15",
+        ip_addrs: Dict = {
+            "PRIMARY": "127.0.0.1",
+            "PRIVATE": "10.0.0.1",
+        },
+        cert_before: datetime = datetime.datetime.now(datetime.timezone.utc),
+        cert_expiry: datetime = datetime.datetime.now(datetime.timezone.utc)
+        + datetime.timedelta(hours=1),
     ) -> None:
         self.project = project
         self.region = region
         self.name = name
         self.db_version = db_version
-        self.ip_addrs = {"PRIMARY": "0.0.0.0", "PRIVATE": "1.1.1.1"}
-        self.backend_type = "SECOND_GEN"
-
-        # generate server private key and cert
-        cert, key = generate_cert(project, name)
-        self.key = key
-        self.cert = cert
-
-    def connect_settings(self, ip_addrs: Optional[Dict] = None) -> str:
-        """
-        Mock data for the following API:
-        https://sqladmin.googleapis.com/sql/v1beta4/projects/{project}/instances/{instance}/connectSettings
-        """
-        server_ca_cert = self_signed_cert(self.cert, self.key)
-        ip_addrs = ip_addrs if ip_addrs else self.ip_addrs
-        ip_addresses = [
-            {"type": key, "ipAddress": value} for key, value in ip_addrs.items()
-        ]
-        return json.dumps(
-            {
-                "kind": "sql#connectSettings",
-                "serverCaCert": {
-                    "cert": server_ca_cert,
-                    "instance": self.name,
-                    "expirationTime": str(
-                        datetime.datetime.now(datetime.timezone.utc)
-                        + datetime.timedelta(minutes=10)
-                    ),
-                },
-                "dnsName": "abcde.12345.us-central1.sql.goog",
-                "ipAddresses": ip_addresses,
-                "region": self.region,
-                "databaseVersion": self.db_version,
-                "backendType": self.backend_type,
-            }
+        self.ip_addrs = ip_addrs
+        self.cert_before = cert_before
+        self.cert_expiry = cert_expiry
+        # create self signed CA cert
+        self.server_ca, self.server_key = generate_cert(
+            self.project, self.name, cert_before, cert_expiry
         )
-
-    def generate_ephemeral(self, client_bytes: str) -> str:
-        """
-        Mock data for the following API:
-        https://sqladmin.googleapis.com/sql/v1beta4/projects/{project}/instances/{instance}:generateEphemeralCert
-        """
+        self.server_cert = self.server_ca.sign(self.server_key, hashes.SHA256())
+        self.server_cert_pem = self.server_cert.public_bytes(
+            encoding=serialization.Encoding.PEM
+        ).decode("UTF-8")
+
+    async def connect_settings(self, request: Any) -> web.Response:
+        ip_addrs = [{"type": k, "ipAddress": v} for k, v in self.ip_addrs.items()]
+        response = {
+            "kind": "sql#connectSettings",
+            "serverCaCert": {
+                "cert": self.server_cert_pem,
+                "instance": self.name,
+                "expirationTime": str(self.server_cert.not_valid_after_utc),
+            },
+            "dnsName": "abcde.12345.us-central1.sql.goog",
+            "ipAddresses": ip_addrs,
+            "region": self.region,
+            "databaseVersion": self.db_version,
+        }
+        return web.Response(content_type="application/json", body=json.dumps(response))
+
+    async def generate_ephemeral(self, request: Any) -> web.Response:
+        body = await request.json()
+        pub_key = body["public_key"]
         client_key: rsa.RSAPublicKey = serialization.load_pem_public_key(
-            client_bytes.encode("UTF-8"), default_backend()
+            pub_key.encode("UTF-8"), default_backend()
         )  # type: ignore
-        ephemeral_cert = client_key_signed_cert(self.cert, self.key, client_key)
-        return json.dumps(
-            {
-                "ephemeralCert": {
-                    "kind": "sql#sslCert",
-                    "cert": ephemeral_cert,
-                    "expirationTime": str(
-                        datetime.datetime.now(datetime.timezone.utc)
-                        + datetime.timedelta(minutes=10)
-                    ),
-                }
-            }
+        ephemeral_cert = client_key_signed_cert(
+            self.server_ca, self.server_key, client_key
         )
+        response = {
+            "ephemeralCert": {
+                "kind": "sql#sslCert",
+                "cert": ephemeral_cert,
+                "expirationTime": str(self.server_cert.not_valid_after_utc),
+            }
+        }
+        return web.Response(content_type="application/json", body=json.dumps(response))
diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py
new file mode 100644
index 0000000..da1c97e
--- /dev/null
+++ b/tests/unit/test_client.py
@@ -0,0 +1,120 @@
+# Copyright 2024 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import datetime
+from typing import Optional
+
+from mocks import FakeCredentials
+import pytest
+
+from google.cloud.sql.connector.client import CloudSQLClient
+from google.cloud.sql.connector.utils import generate_keys
+from google.cloud.sql.connector.version import __version__ as version
+
+
+@pytest.mark.asyncio
+async def test_get_metadata(fake_client: CloudSQLClient) -> None:
+    """
+    Test _get_metadata returns successfully.
+    """
+    resp = await fake_client._get_metadata(
+        "test-project",
+        "test-region",
+        "test-instance",
+    )
+    assert resp["database_version"] == "POSTGRES_15"
+    assert resp["ip_addresses"] == {
+        "PRIMARY": "127.0.0.1",
+        "PRIVATE": "10.0.0.1",
+        "PSC": "abcde.12345.us-central1.sql.goog",
+    }
+    assert isinstance(resp["server_ca_cert"], str)
+
+
+@pytest.mark.asyncio
+async def test_get_ephemeral(fake_client: CloudSQLClient) -> None:
+    """
+    Test _get_ephemeral returns successfully.
+    """
+    keys = await generate_keys()
+    client_cert, expiration = await fake_client._get_ephemeral(
+        "test-project", "test-instance", keys[1]
+    )
+    assert isinstance(client_cert, str)
+    assert expiration > datetime.datetime.now(datetime.timezone.utc)
+
+
+@pytest.mark.asyncio
+async def test_CloudSQLClient_init_(fake_credentials: FakeCredentials) -> None:
+    """
+    Test to check whether the __init__ method of CloudSQLClient
+    can correctly initialize a client.
+    """
+    driver = "pg8000"
+    client = CloudSQLClient(
+        "www.test-endpoint.com", "my-quota-project", fake_credentials, driver=driver
+    )
+    # verify base endpoint is set
+    assert client._sqladmin_api_endpoint == "www.test-endpoint.com"
+    # verify proper headers are set
+    assert (
+        client._client.headers["User-Agent"]
+        == f"cloud-sql-python-connector/{version}+{driver}"
+    )
+    assert client._client.headers["x-goog-user-project"] == "my-quota-project"
+    # close client
+    await client.close()
+
+
+@pytest.mark.asyncio
+async def test_CloudSQLClient_init_custom_user_agent(
+    fake_credentials: FakeCredentials,
+) -> None:
+    """
+    Test to check that custom user agents are included in HTTP requests.
+    """
+    client = CloudSQLClient(
+        "www.test-endpoint.com",
+        "my-quota-project",
+        fake_credentials,
+        user_agent="custom-agent/v1.0.0 other-agent/v2.0.0",
+    )
+    assert (
+        client._client.headers["User-Agent"]
+        == f"cloud-sql-python-connector/{version} custom-agent/v1.0.0 other-agent/v2.0.0"
+    )
+    await client.close()
+
+
+@pytest.mark.parametrize(
+    "driver",
+    [None, "pg8000", "asyncpg", "pymysql", "pytds"],
+)
+@pytest.mark.asyncio
+async def test_CloudSQLClient_user_agent(
+    driver: Optional[str], fake_credentials: FakeCredentials
+) -> None:
+    """
+    Test to check whether the __init__ method of CloudSQLClient
+    properly sets user agent when passed a database driver.
+    """
+    client = CloudSQLClient(
+        "www.test-endpoint.com", "my-quota-project", fake_credentials, driver=driver
+    )
+    if driver is None:
+        assert client._user_agent == f"cloud-sql-python-connector/{version}"
+    else:
+        assert client._user_agent == f"cloud-sql-python-connector/{version}+{driver}"
+    # close client
+    await client.close()
diff --git a/tests/unit/test_connector.py b/tests/unit/test_connector.py
index f402fe0..a9c57de 100644
--- a/tests/unit/test_connector.py
+++ b/tests/unit/test_connector.py
@@ -17,47 +17,48 @@ import asyncio
 
 from google.auth.credentials import Credentials
 from mock import patch
-from mocks import MockInstance
 import pytest  # noqa F401 Needed to run the tests
 
 from google.cloud.sql.connector import Connector
 from google.cloud.sql.connector import create_async_connector
 from google.cloud.sql.connector import IPTypes
+from google.cloud.sql.connector.client import CloudSQLClient
 from google.cloud.sql.connector.exceptions import ConnectorLoopError
+from google.cloud.sql.connector.instance import Instance
 
 
-def test_connect_enable_iam_auth_error(fake_credentials: Credentials) -> None:
+def test_connect_enable_iam_auth_error(
+    fake_credentials: Credentials, instance: Instance
+) -> None:
     """Test that calling connect() with different enable_iam_auth
     argument values throws error."""
-    connect_string = "my-project:my-region:my-instance"
-    # create mock instance with enable_iam_auth=False
-    instance = MockInstance(enable_iam_auth=False)
-    mock_instances = {}
-    mock_instances[connect_string] = instance
-    # init Connector
+    connect_string = "test-project:test-region:test-instance"
     connector = Connector(credentials=fake_credentials)
-    with patch.dict(connector._instances, mock_instances):
-        # try to connect using enable_iam_auth=True, should raise error
-        pytest.raises(
-            ValueError,
-            connector.connect,
-            connect_string,
-            "pg8000",
-            enable_iam_auth=True,
-        )
-    # remove mock_instance to avoid destructor warnings
+    # set instance
+    connector._instances[connect_string] = instance
+    # try to connect using enable_iam_auth=True, should raise error
+    with pytest.raises(ValueError) as exc_info:
+        connector.connect(connect_string, "pg8000", enable_iam_auth=True)
+    assert (
+        exc_info.value.args[0] == "connect() called with 'enable_iam_auth=True', "
+        "but previously used 'enable_iam_auth=False'. "
+        "If you require both for your use case, please use a new "
+        "connector.Connector object."
+    )
+    # remove instance to avoid destructor warnings
     connector._instances = {}
 
 
-def test_connect_with_unsupported_driver(connector: Connector) -> None:
-    # try to connect using unsupported driver, should raise KeyError
-    with pytest.raises(KeyError) as exc_info:
-        connector.connect(
-            "my-project:my-region:my-instance",
-            "bad_driver",
-        )
-    # assert custom error message for unsupported driver is present
-    assert exc_info.value.args[0] == "Driver 'bad_driver' is not supported."
+def test_connect_with_unsupported_driver(fake_credentials: Credentials) -> None:
+    with Connector(credentials=fake_credentials) as connector:
+        # try to connect using unsupported driver, should raise KeyError
+        with pytest.raises(KeyError) as exc_info:
+            connector.connect(
+                "my-project:my-region:my-instance",
+                "bad_driver",
+            )
+        # assert custom error message for unsupported driver is present
+        assert exc_info.value.args[0] == "Driver 'bad_driver' is not supported."
 
 
 @pytest.mark.asyncio
@@ -129,17 +130,27 @@ async def test_Connector_Init_async_context_manager(
         assert connector._loop == loop
 
 
-def test_Connector_connect(connector: Connector) -> None:
-    """Test that Connector.connect can properly return a DB API connection."""
-    connect_string = "my-project:my-region:my-instance"
-    # patch db connection creation
-    with patch("google.cloud.sql.connector.pg8000.connect") as mock_connect:
-        mock_connect.return_value = True
-        connection = connector.connect(
-            connect_string, "pg8000", user="my-user", password="my-pass", db="my-db"
-        )
-        # verify connector made connection call
-        assert connection is True
+@pytest.mark.asyncio
+async def test_Connector_connect_async(
+    fake_credentials: Credentials, fake_client: CloudSQLClient
+) -> None:
+    """Test that Connector.connect_async can properly return a DB API connection."""
+    async with Connector(
+        credentials=fake_credentials, loop=asyncio.get_running_loop()
+    ) as connector:
+        connector._client = fake_client
+        # patch db connection creation
+        with patch("google.cloud.sql.connector.asyncpg.connect") as mock_connect:
+            mock_connect.return_value = True
+            connection = await connector.connect_async(
+                "test-project:test-region:test-instance",
+                "asyncpg",
+                user="my-user",
+                password="my-pass",
+                db="my-db",
+            )
+            # verify connector made connection call
+            assert connection is True
 
 
 @pytest.mark.asyncio
diff --git a/tests/unit/test_instance.py b/tests/unit/test_instance.py
index 8731554..c083d5b 100644
--- a/tests/unit/test_instance.py
+++ b/tests/unit/test_instance.py
@@ -25,6 +25,7 @@ from mock import patch
 import mocks
 import pytest  # noqa F401 Needed to run the tests
 
+from google.cloud.sql.connector.client import CloudSQLClient
 from google.cloud.sql.connector.exceptions import AutoIAMAuthNotSupported
 from google.cloud.sql.connector.exceptions import CloudSQLIPTypeError
 from google.cloud.sql.connector.instance import _parse_instance_connection_name
@@ -33,7 +34,6 @@ from google.cloud.sql.connector.instance import Instance
 from google.cloud.sql.connector.instance import IPTypes
 from google.cloud.sql.connector.rate_limiter import AsyncRateLimiter
 from google.cloud.sql.connector.utils import generate_keys
-from google.cloud.sql.connector.version import __version__ as version
 
 
 @pytest.fixture
@@ -71,39 +71,18 @@ def test_parse_instance_connection_name_bad_conn_name() -> None:
 
 @pytest.mark.asyncio
 async def test_Instance_init(
-    fake_credentials: Credentials,
+    instance: Instance,
 ) -> None:
     """
     Test to check whether the __init__ method of Instance
     can tell if the connection string that's passed in is formatted correctly.
     """
-    event_loop = asyncio.get_running_loop()
-    connect_string = "test-project:test-region:test-instance"
-    keys = asyncio.wrap_future(
-        asyncio.run_coroutine_threadsafe(generate_keys(), event_loop), loop=event_loop
-    )
-    instance = Instance(
-        connect_string,
-        "pymysql",
-        keys,
-        event_loop,
-        credentials=fake_credentials,
-        user_agent="custom/v1.0.0",
-    )
-    project_result = instance._project
-    region_result = instance._region
-    instance_result = instance._instance
-    assert (
-        project_result == "test-project"
-        and region_result == "test-region"
-        and instance_result == "test-instance"
-    )
     assert (
-        instance._user_agent_string
-        == f"cloud-sql-python-connector/{version}+pymysql custom/v1.0.0"
+        instance._project == "test-project"
+        and instance._region == "test-region"
+        and instance._instance == "test-instance"
     )
-    # cleanup instance
-    await instance.close()
+    assert instance._enable_iam_auth is False
 
 
 @pytest.mark.asyncio
@@ -193,7 +172,6 @@ async def test_schedule_refresh_replaces_invalid_result(
     # check that current is now valid MockMetadata
     assert instance._current.result() == refresh_metadata
     assert isinstance(instance._current.result(), mocks.MockMetadata)
-    await instance.close()
 
 
 @pytest.mark.asyncio
@@ -234,19 +212,17 @@ async def test_Instance_close(instance: Instance) -> None:
     await instance._current
     assert instance._current.cancelled() is False
     assert instance._next.cancelled() is False
-    assert instance._client_session.closed is False
     # run close() to cancel tasks and close ClientSession
     await instance.close()
     # verify tasks are cancelled and ClientSession is closed
     assert (instance._current.done() or instance._current.cancelled()) is True
     assert instance._next.cancelled() is True
-    assert instance._client_session.closed is True
 
 
 @pytest.mark.asyncio
 async def test_perform_refresh(
     instance: Instance,
-    mock_instance: mocks.FakeCSQLInstance,
+    fake_instance: mocks.FakeCSQLInstance,
 ) -> None:
     """
     Test that _perform_refresh returns valid ConnectionInfo object.
@@ -256,12 +232,7 @@ async def test_perform_refresh(
     # verify instance metadata object is returned
     assert isinstance(instance_metadata, ConnectionInfo)
     # verify instance metadata expiration
-    assert (
-        mock_instance.cert._not_valid_after.replace(
-            tzinfo=datetime.timezone.utc, microsecond=0  # type: ignore
-        )
-        == instance_metadata.expiration
-    )
+    assert fake_instance.server_cert.not_valid_after_utc == instance_metadata.expiration
 
 
 @pytest.mark.asyncio
@@ -281,11 +252,10 @@ async def test_perform_refresh_expiration(
     credentials = mocks.FakeCredentials(token="my-token", expiry=expiration)
     setattr(instance, "_enable_iam_auth", True)
     # set downscoped credential to mock
-    with patch(
-        "google.cloud.sql.connector.refresh_utils._downscope_credentials"
-    ) as mock_auth:
+    with patch("google.cloud.sql.connector.client._downscope_credentials") as mock_auth:
         mock_auth.return_value = credentials
         instance_metadata = await instance._perform_refresh()
+    mock_auth.assert_called_once()
     # verify instance metadata object is returned
     assert isinstance(instance_metadata, ConnectionInfo)
     # verify instance metadata uses credentials expiration
@@ -303,33 +273,7 @@ async def test_connect_info(
 
     # verify metadata and ip address
     assert isinstance(instance_metadata, ConnectionInfo)
-    assert ip_addr == "0.0.0.0"
-    # cleanup instance
-    await instance.close()
-
-
-@pytest.mark.asyncio
-async def test_get_preferred_ip(instance: Instance) -> None:
-    """
-    Test that get_preferred_ip returns proper IP address
-    for both Public and Private IP addresses.
-    """
-    instance_metadata: ConnectionInfo = await instance._current
-
-    # test public IP as preferred IP for connection
-    ip_addr = instance_metadata.get_preferred_ip(IPTypes.PUBLIC)
-    # verify public ip address is preferred
-    assert ip_addr == "0.0.0.0"
-
-    # test private IP as preferred IP for connection
-    ip_addr = instance_metadata.get_preferred_ip(IPTypes.PRIVATE)
-    # verify private ip address is preferred
-    assert ip_addr == "1.1.1.1"
-
-    # test PSC as preferred IP type for connection
-    ip_addr = instance_metadata.get_preferred_ip(IPTypes.PSC)
-    # verify PSC ip address is preferred
-    assert ip_addr == "abcde.12345.us-central1.sql.goog"
+    assert ip_addr == "127.0.0.1"
 
 
 @pytest.mark.asyncio
@@ -361,10 +305,12 @@ async def test_ClientResponseError(
     """
     Test that detailed error message is applied to ClientResponseError.
     """
-    event_loop = asyncio.get_running_loop()
     # mock Cloud SQL Admin API calls with exceptions
-    keys = asyncio.wrap_future(
-        asyncio.run_coroutine_threadsafe(generate_keys(), event_loop), loop=event_loop
+    keys = asyncio.create_task(generate_keys())
+    client = CloudSQLClient(
+        sqladmin_api_endpoint="https://sqladmin.googleapis.com",
+        quota_project=None,
+        credentials=fake_credentials,
     )
     get_url = "https://sqladmin.googleapis.com/sql/v1beta4/projects/my-project/instances/my-instance/connectSettings"
     post_url = "https://sqladmin.googleapis.com/sql/v1beta4/projects/my-project/instances/my-instance:generateEphemeralCert"
@@ -385,15 +331,12 @@ async def test_ClientResponseError(
             ),
             repeat=True,
         )
-        # verify that error is raised with detailed error message
+        instance = Instance(
+            "my-project:my-region:my-instance",
+            client,
+            keys,
+        )
         try:
-            instance = Instance(
-                "my-project:my-region:my-instance",
-                "pymysql",
-                keys,
-                event_loop,
-                credentials=fake_credentials,
-            )
             await instance._current
         except ClientResponseError as e:
             assert e.status == 403
@@ -405,20 +348,22 @@ async def test_ClientResponseError(
             )
         finally:
             await instance.close()
+            await client.close()
 
 
 @pytest.mark.asyncio
-@pytest.mark.parametrize(
-    "mock_instance",
-    [
-        mocks.FakeCSQLInstance(db_version="SQLSERVER_2019_STANDARD"),
-    ],
-)
-async def test_AutoIAMAuthNotSupportedError(instance: Instance) -> None:
+async def test_AutoIAMAuthNotSupportedError(fake_client: CloudSQLClient) -> None:
     """
     Test that AutoIAMAuthNotSupported exception is raised
-    for SQL Server and MySQL instances.
+    for SQL Server instances.
     """
-    instance._enable_iam_auth = True
+    # generate client key pair
+    keys = asyncio.create_task(generate_keys())
+    instance = Instance(
+        "test-project:test-region:sqlserver-instance",
+        client=fake_client,
+        keys=keys,
+        enable_iam_auth=True,
+    )
     with pytest.raises(AutoIAMAuthNotSupported):
         await instance._current
diff --git a/tests/unit/test_refresh_utils.py b/tests/unit/test_refresh_utils.py
index d798fa2..9b76b03 100644
--- a/tests/unit/test_refresh_utils.py
+++ b/tests/unit/test_refresh_utils.py
@@ -15,27 +15,21 @@ limitations under the License.
 """
 import asyncio
 import datetime
-from typing import Any, no_type_check
+from typing import no_type_check
 
-import aiohttp
-from aioresponses import aioresponses
 from conftest import SCOPES  # type: ignore
 import google.auth
 from google.auth.credentials import Credentials
 import google.oauth2.credentials
 from mock import Mock
 from mock import patch
-from mocks import FakeCSQLInstance  # type: ignore
 from mocks import instance_metadata_expired
 from mocks import instance_metadata_success
 import pytest  # noqa F401 Needed to run the tests
 
 from google.cloud.sql.connector.refresh_utils import _downscope_credentials
-from google.cloud.sql.connector.refresh_utils import _get_ephemeral
-from google.cloud.sql.connector.refresh_utils import _get_metadata
 from google.cloud.sql.connector.refresh_utils import _is_valid
 from google.cloud.sql.connector.refresh_utils import _seconds_until_refresh
-from google.cloud.sql.connector.utils import generate_keys
 
 
 @pytest.fixture
@@ -46,136 +40,6 @@ def credentials() -> Credentials:
     return credentials
 
 
-@pytest.mark.asyncio
-async def test_get_ephemeral(
-    mock_instance: FakeCSQLInstance, credentials: Credentials
-) -> None:
-    """
-    Test to check whether _get_ephemeral runs without problems given valid
-    parameters.
-    """
-    project = "my-project"
-    instance = "my-instance"
-    _, pub_key = await generate_keys()
-    # mock Cloud SQL Admin API call
-    with aioresponses() as mocked:
-        mocked.post(
-            "https://sqladmin.googleapis.com/sql/v1beta4/projects/my-project/instances/my-instance:generateEphemeralCert",
-            status=200,
-            body=mock_instance.generate_ephemeral(pub_key),
-            repeat=True,
-        )
-        async with aiohttp.ClientSession() as client_session:
-            result: Any = await _get_ephemeral(
-                client_session,
-                "https://sqladmin.googleapis.com",
-                credentials,
-                project,
-                instance,
-                pub_key,
-            )
-    cert, _ = result
-    cert = cert.strip()  # remove any trailing whitespace
-    cert = cert.split("\n")
-
-    assert (
-        cert[0] == "-----BEGIN CERTIFICATE-----"
-        and cert[len(cert) - 1] == "-----END CERTIFICATE-----"
-    )
-
-
-@pytest.mark.asyncio
-@no_type_check
-async def test_get_ephemeral_TypeError(credentials: Credentials) -> None:
-    """
-    Test to check whether _get_ephemeral throws proper TypeError
-    when given incorrect input arg types.
-    """
-    client_session = Mock(aiohttp.ClientSession)
-    # incorrect pub_key type
-    with pytest.raises(TypeError):
-        await _get_ephemeral(
-            client_session=client_session,
-            sqladmin_api_endpoint="https://sqladmin.googleapis.com",
-            credentials=credentials,
-            project="my-project",
-            instance="my-instance",
-            pub_key=12345,
-        )
-
-
-@pytest.mark.asyncio
-async def test_get_metadata(
-    mock_instance: FakeCSQLInstance, credentials: Credentials
-) -> None:
-    """
-    Test to check whether _get_metadata runs without problems given valid
-    parameters.
-    """
-    project = "my-project"
-    region = "my-region"
-    instance = "my-instance"
-    # mock Cloud SQL Admin API call
-    with aioresponses() as mocked:
-        mocked.get(
-            f"https://sqladmin.googleapis.com/sql/v1beta4/projects/{project}/instances/{instance}/connectSettings",
-            status=200,
-            body=mock_instance.connect_settings(),
-            repeat=True,
-        )
-
-        async with aiohttp.ClientSession() as client_session:
-            result = await _get_metadata(
-                client_session,
-                "https://sqladmin.googleapis.com",
-                credentials,
-                project,
-                region,
-                instance,
-            )
-
-    assert (
-        result["ip_addresses"] is not None
-        and result["database_version"] == "POSTGRES_14"
-        and isinstance(result["server_ca_cert"], str)
-    )
-
-
-@pytest.mark.asyncio
-@no_type_check
-async def test_get_metadata_region_mismatch(
-    mock_instance: FakeCSQLInstance, credentials: Credentials
-) -> None:
-    """
-    Test to check whether _get_metadata throws proper ValueError
-    when given mismatched region.
-    """
-    client_session = Mock(aiohttp.ClientSession)
-    project = "my-project"
-    region = "bad-region"
-    instance = "my-instance"
-
-    # mock Cloud SQL Admin API call
-    with aioresponses() as mocked:
-        mocked.get(
-            f"https://sqladmin.googleapis.com/sql/v1beta4/projects/{project}/instances/{instance}/connectSettings",
-            status=200,
-            body=mock_instance.connect_settings(),
-            repeat=True,
-        )
-
-        async with aiohttp.ClientSession() as client_session:
-            with pytest.raises(ValueError):
-                await _get_metadata(
-                    client_session=client_session,
-                    sqladmin_api_endpoint="https://sqladmin.googleapis.com",
-                    credentials=credentials,
-                    project=project,
-                    region=region,
-                    instance=instance,
-                )
-
-
 @pytest.mark.asyncio
 @no_type_check
 async def test_is_valid_with_valid_metadata() -> None: