chore: use better fake for credentials in tests (#996)

Author: jackwotherspoon

google/cloud/sql/connector/refresh_utils.py

@@ -79,16 +79,11 @@ async def _get_metadata(
     :raises TypeError: If any of the arguments are not the specified type.
     """
 
-    if not isinstance(credentials, Credentials):
-        raise TypeError(
-            "credentials must be of type google.auth.credentials.Credentials,"
-            f" got {type(credentials)}"
-        )
-    elif not isinstance(project, str):
+    if not isinstance(project, str):
         raise TypeError(f"project must be of type str, got {type(project)}")
-    elif not isinstance(region, str):
+    if not isinstance(region, str):
         raise TypeError(f"region must be of type str, got {type(region)}")
-    elif not isinstance(instance, str):
+    if not isinstance(instance, str):
         raise TypeError(f"instance must be of type str, got {type(instance)}")
 
     if not credentials.valid:
@@ -172,16 +167,11 @@ async def _get_ephemeral(
 
     logger.debug(f"['{instance}']: Requesting ephemeral certificate")
 
-    if not isinstance(credentials, Credentials):
-        raise TypeError(
-            "credentials must be of type google.auth.credentials.Credentials,"
-            f" got {type(credentials)}"
-        )
-    elif not isinstance(project, str):
+    if not isinstance(project, str):
         raise TypeError(f"project must be of type str, got {type(project)}")
-    elif not isinstance(instance, str):
+    if not isinstance(instance, str):
         raise TypeError(f"instance must be of type str, got {type(instance)}")
-    elif not isinstance(pub_key, str):
+    if not isinstance(pub_key, str):
         raise TypeError(f"pub_key must be of type str, got {type(pub_key)}")
 
     if not credentials.valid:

tests/conftest.py

@@ -21,9 +21,8 @@
 
 from aioresponses import aioresponses
 from google.auth.credentials import Credentials
-from google.auth.credentials import with_scopes_if_required
-from google.oauth2 import service_account
 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
@@ -75,27 +74,8 @@ def connect_string() -> str:
 
 
 @pytest.fixture
-def fake_credentials() -> Credentials:
-    fake_service_account = {
-        "type": "service_account",
-        "project_id": "a-project-id",
-        "private_key_id": "a-private-key-id",
-        "private_key": "-----BEGIN PRIVATE KEY-----\nMIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQDY3E8o1NEFcjMM\nHW/5ZfFJw29/8NEqpViNjQIx95Xx5KDtJ+nWn9+OW0uqsSqKlKGhAdAo+Q6bjx2c\nuXVsXTu7XrZUY5Kltvj94DvUa1wjNXs606r/RxWTJ58bfdC+gLLxBfGnB6CwK0YQ\nxnfpjNbkUfVVzO0MQD7UP0Hl5ZcY0Puvxd/yHuONQn/rIAieTHH1pqgW+zrH/y3c\n59IGThC9PPtugI9ea8RSnVj3PWz1bX2UkCDpy9IRh9LzJLaYYX9RUd7++dULUlat\nAaXBh1U6emUDzhrIsgApjDVtimOPbmQWmX1S60mqQikRpVYZ8u+NDD+LNw+/Eovn\nxCj2Y3z1AgMBAAECggEAWDBzoqO1IvVXjBA2lqId10T6hXmN3j1ifyH+aAqK+FVl\nGjyWjDj0xWQcJ9ync7bQ6fSeTeNGzP0M6kzDU1+w6FgyZqwdmXWI2VmEizRjwk+/\n/uLQUcL7I55Dxn7KUoZs/rZPmQDxmGLoue60Gg6z3yLzVcKiDc7cnhzhdBgDc8vd\nQorNAlqGPRnm3EqKQ6VQp6fyQmCAxrr45kspRXNLddat3AMsuqImDkqGKBmF3Q1y\nxWGe81LphUiRqvqbyUlh6cdSZ8pLBpc9m0c3qWPKs9paqBIvgUPlvOZMqec6x4S6\nChbdkkTRLnbsRr0Yg/nDeEPlkhRBhasXpxpMUBgPywKBgQDs2axNkFjbU94uXvd5\nznUhDVxPFBuxyUHtsJNqW4p/ujLNimGet5E/YthCnQeC2P3Ym7c3fiz68amM6hiA\nOnW7HYPZ+jKFnefpAtjyOOs46AkftEg07T9XjwWNPt8+8l0DYawPoJgbM5iE0L2O\nx8TU1Vs4mXc+ql9F90GzI0x3VwKBgQDqZOOqWw3hTnNT07Ixqnmd3dugV9S7eW6o\nU9OoUgJB4rYTpG+yFqNqbRT8bkx37iKBMEReppqonOqGm4wtuRR6LSLlgcIU9Iwx\nyfH12UWqVmFSHsgZFqM/cK3wGev38h1WBIOx3/djKn7BdlKVh8kWyx6uC8bmV+E6\nOoK0vJD6kwKBgHAySOnROBZlqzkiKW8c+uU2VATtzJSydrWm0J4wUPJifNBa/hVW\ndcqmAzXC9xznt5AVa3wxHBOfyKaE+ig8CSsjNyNZ3vbmr0X04FoV1m91k2TeXNod\njMTobkPThaNm4eLJMN2SQJuaHGTGERWC0l3T18t+/zrDMDCPiSLX1NAvAoGBAN1T\nVLJYdjvIMxf1bm59VYcepbK7HLHFkRq6xMJMZbtG0ryraZjUzYvB4q4VjHk2UDiC\nlhx13tXWDZH7MJtABzjyg+AI7XWSEQs2cBXACos0M4Myc6lU+eL+iA+OuoUOhmrh\nqmT8YYGu76/IBWUSqWuvcpHPpwl7871i4Ga/I3qnAoGBANNkKAcMoeAbJQK7a/Rn\nwPEJB+dPgNDIaboAsh1nZhVhN5cvdvCWuEYgOGCPQLYQF0zmTLcM+sVxOYgfy8mV\nfbNgPgsP5xmu6dw2COBKdtozw0HrWSRjACd1N4yGu75+wPCcX/gQarcjRcXXZeEa\nNtBLSfcqPULqD+h7br9lEJio\n-----END PRIVATE KEY-----\n",
-        "client_email": "email@example.com",
-        "client_id": "12345",
-        "auth_uri": "https://accounts.google.com/o/oauth2/auth",
-        "token_uri": "https://oauth2.googleapis.com/token",
-        "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
-        "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/email%40example.com",
-    }
-
-    fake_credentials = service_account.Credentials.from_service_account_info(
-        fake_service_account
-    )
-    fake_credentials = with_scopes_if_required(fake_credentials, scopes=SCOPES)
-    # stub refresh of credentials
-    setattr(fake_credentials, "refresh", lambda *args: None)
-    return fake_credentials
+def fake_credentials() -> FakeCredentials:
+    return FakeCredentials()
 
 
 def mock_server(server_sock: socket.socket) -> None:

tests/unit/mocks.py

@@ -19,21 +19,61 @@
 import json
 import ssl
 from tempfile import TemporaryDirectory
-from typing import Any, Dict, Optional, Tuple
+from typing import Any, Callable, Dict, Optional, Tuple
 
 from cryptography import x509
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import hashes
 from cryptography.hazmat.primitives import serialization
 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
 
 
+class FakeCredentials:
+    def __init__(
+        self, token: Optional[str] = None, expiry: Optional[datetime.datetime] = None
+    ) -> None:
+        self.token = token
+        self.expiry = expiry
+
+    @property
+    def __class__(self) -> Credentials:
+        # set class type to google auth Credentials
+        return Credentials
+
+    def refresh(self, request: Callable) -> None:
+        """Refreshes the access token."""
+        self.token = "12345"
+        self.expiry = datetime.datetime.utcnow() + datetime.timedelta(minutes=60)
+
+    @property
+    def expired(self) -> bool:
+        """Checks if the credentials are expired.
+
+        Note that credentials can be invalid but not expired because
+        Credentials with expiry set to None are considered to never
+        expire.
+        """
+        if self.expiry is None:
+            return False
+        return False if self.expiry > datetime.datetime.utcnow() else True
+
+    @property
+    def valid(self) -> bool:
+        """Checks the validity of the credentials.
+
+        This is True if the credentials have a token and the token
+        is not expired.
+        """
+        return self.token is not None and not self.expired
+
+
 class MockInstance:
     _enable_iam_auth: bool
 
@@ -238,6 +278,10 @@ def generate_ephemeral(self, client_bytes: str) -> str:
             client_bytes.encode("UTF-8"), default_backend()
         )  # type: ignore
         ephemeral_cert = client_key_signed_cert(self.cert, self.key, client_key)
+        print(
+            "Test generate time: ",
+            datetime.datetime.utcnow() + datetime.timedelta(minutes=10),
+        )
         return json.dumps(
             {
                 "ephemeralCert": {

tests/unit/test_instance.py

@@ -264,7 +264,7 @@ async def test_perform_refresh(
 
 @pytest.mark.asyncio
 async def test_perform_refresh_expiration(
-    instance: Instance, fake_credentials: Credentials
+    instance: Instance,
 ) -> None:
     """
     Test that _perform_refresh returns ConnectionInfo with proper expiration.
@@ -274,15 +274,14 @@ async def test_perform_refresh_expiration(
     """
     # set credentials expiration to 1 minute from now
     expiration = datetime.datetime.utcnow() + datetime.timedelta(minutes=1)
+    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:
-        setattr(fake_credentials, "expiry", expiration)
-        mock_auth.return_value = fake_credentials
+        mock_auth.return_value = credentials
         instance_metadata = await instance._perform_refresh()
-
     # verify instance metadata object is returned
     assert isinstance(instance_metadata, ConnectionInfo)
     # verify instance metadata uses credentials expiration

tests/unit/test_refresh_utils.py

@@ -97,16 +97,6 @@ async def test_get_ephemeral_TypeError(credentials: Credentials) -> None:
     instance = "my-instance"
     pub_key = "key"
 
-    # incorrect credentials type
-    with pytest.raises(TypeError):
-        await _get_ephemeral(
-            client_session=client_session,
-            sqladmin_api_endpoint="https://sqladmin.googleapis.com",
-            credentials="bad-credentials",
-            project=project,
-            instance=instance,
-            pub_key=pub_key,
-        )
     # incorrect project type
     with pytest.raises(TypeError):
         await _get_ephemeral(
@@ -188,16 +178,6 @@ async def test_get_metadata_TypeError(credentials: Credentials) -> None:
     region = "my-region"
     instance = "my-instance"
 
-    # incorrect credentials type
-    with pytest.raises(TypeError):
-        await _get_metadata(
-            client_session=client_session,
-            sqladmin_api_endpoint="https://sqladmin.googleapis.com",
-            credentials="bad-credentials",
-            project=project,
-            region=region,
-            instance=instance,
-        )
     # incorrect project type
     with pytest.raises(TypeError):
         await _get_metadata(
@@ -287,22 +267,6 @@ async def test_is_valid_with_expired_metadata() -> None:
     assert not await _is_valid(task)
 
 
-# TODO: https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/issues/901
-# def test_downscope_credentials_service_account(fake_credentials: Credentials) -> None:
-#     """
-#     Test _downscope_credentials with google.oauth2.service_account.Credentials
-#     which mimics an authenticated service account.
-#     """
-#     # override actual refresh URI
-#     setattr(fake_credentials, "with_scopes", google.auth.credentials.Credentials(scopes=["https://www.googleapis.com/auth/sqlservice.login"]))
-#     credentials = _downscope_credentials(fake_credentials)
-#     # verify default credential scopes have not been altered
-#     assert fake_credentials.scopes == SCOPES
-#     # verify downscoped credentials have new scope
-#     assert credentials.scopes == ["https://www.googleapis.com/auth/sqlservice.login"]
-#     assert credentials != fake_credentials
-
-
 def test_downscope_credentials_user() -> None:
     """
     Test _downscope_credentials with google.oauth2.credentials.Credentials