fix: post merge conflict changes and linting

Author: thameezb

google/cloud/sql/connector/connector.py

@@ -503,59 +503,52 @@ async def connect_async(
         monitored_cache = await self._get_cache(
             instance_connection_string, enable_iam_auth, ip_type, driver
         )
-        conn_info = await monitored_cache.connect_info()
-        ip_address = conn_info.get_preferred_ip(ip_type)
 
         try:
             conn_info = await monitored_cache.connect_info()
             # validate driver matches intended database engine
             DriverMapping.validate_engine(driver, conn_info.database_version)
             ip_address = conn_info.get_preferred_ip(ip_type)
-        except Exception:
-            # with an error from Cloud SQL Admin API call or IP type, invalidate
-            # the cache and re-raise the error
-            await self._remove_cached(str(conn_name), enable_iam_auth)
-            raise
 
-        # If the connector is configured with a custom DNS name, attempt to use
-        # that DNS name to connect to the instance. Fall back to the metadata IP
-        # address if the DNS name does not resolve to an IP address.
-        if conn_info.conn_name.domain_name and isinstance(self._resolver, DnsResolver):
-            try:
-                ips = await self._resolver.resolve_a_record(conn_info.conn_name.domain_name)
-                if ips:
-                    ip_address = ips[0]
+            # If the connector is configured with a custom DNS name, attempt to use
+            # that DNS name to connect to the instance. Fall back to the metadata IP
+            # address if the DNS name does not resolve to an IP address.
+            if conn_info.conn_name.domain_name and isinstance(self._resolver, DnsResolver):
+                try:
+                    ips = await self._resolver.resolve_a_record(conn_info.conn_name.domain_name)
+                    if ips:
+                        ip_address = ips[0]
+                        logger.debug(
+                            f"['{instance_connection_string}']: Custom DNS name "
+                            f"'{conn_info.conn_name.domain_name}' resolved to '{ip_address}', "
+                            "using it to connect"
+                        )
+                    else:
+                        logger.debug(
+                            f"['{instance_connection_string}']: Custom DNS name "
+                            f"'{conn_info.conn_name.domain_name}' resolved but returned no "
+                            f"entries, using '{ip_address}' from instance metadata"
+                        )
+                except Exception as e:
                     logger.debug(
                         f"['{instance_connection_string}']: Custom DNS name "
-                        f"'{conn_info.conn_name.domain_name}' resolved to '{ip_address}', "
-                        "using it to connect"
+                        f"'{conn_info.conn_name.domain_name}' did not resolve to an IP "
+                        f"address: {e}, using '{ip_address}' from instance metadata"
                     )
-                else:
-                    logger.debug(
-                        f"['{instance_connection_string}']: Custom DNS name "
-                        f"'{conn_info.conn_name.domain_name}' resolved but returned no "
-                        f"entries, using '{ip_address}' from instance metadata"
-                    )
-            except Exception as e:
-                logger.debug(
-                    f"['{instance_connection_string}']: Custom DNS name "
-                    f"'{conn_info.conn_name.domain_name}' did not resolve to an IP "
-                    f"address: {e}, using '{ip_address}' from instance metadata"
-                )
 
-        logger.debug(f"['{conn_info.conn_name}']: Connecting to {ip_address}:3307")
-        # format `user` param for automatic IAM database authn
-        if enable_iam_auth:
-            formatted_user = format_database_user(
-                conn_info.database_version, kwargs["user"]
-            )
-            if formatted_user != kwargs["user"]:
-                logger.debug(
-                    f"['{instance_connection_string}']: "
-                    "Truncated IAM database username from "
-                    f"{kwargs['user']} to {formatted_user}"
+            logger.debug(f"['{conn_info.conn_name}']: Connecting to {ip_address}:3307")
+            # format `user` param for automatic IAM database authn
+            if enable_iam_auth:
+                formatted_user = format_database_user(
+                    conn_info.database_version, kwargs["user"]
                 )
-                kwargs["user"] = formatted_user
+                if formatted_user != kwargs["user"]:
+                    logger.debug(
+                        f"['{instance_connection_string}']: "
+                        "Truncated IAM database username from "
+                        f"{kwargs['user']} to {formatted_user}"
+                    )
+                    kwargs["user"] = formatted_user
 
             ctx = await conn_info.create_ssl_context(enable_iam_auth)
             # async drivers are unblocking and can be awaited directly
@@ -576,11 +569,9 @@ async def connect_async(
                 return await self._loop.run_in_executor(None, connect_partial)
 
         except Exception:
-            # with any exception, we attempt a force refresh, then throw the error
-            monitored_cache = await self._get_cache(
-                instance_connection_string, enable_iam_auth, ip_type, driver
-            )
-            await monitored_cache.force_refresh()
+            # with an error from Cloud SQL Admin API call or connection, invalidate
+            # the cache and re-raise the error
+            await self._remove_cached(str(conn_name), enable_iam_auth)
             raise
 
     async def start_unix_socket_proxy_async(

google/cloud/sql/connector/local_unix_socket.py

@@ -15,7 +15,8 @@
 """
 
 import ssl
-from typing import Any, TYPE_CHECKING
+from typing import Any
+
 
 def connect(
     host: str, sock: ssl.SSLSocket, **kwargs: Any

google/cloud/sql/connector/proxy.py

@@ -224,7 +224,7 @@ def _handle_server_connection_lost(
         
         :return: None
         """
-        logger.debug(f"Closing proxy server due to lost connection")
+        logger.debug("Closing proxy server due to lost connection")
         self._loop.create_task(self.close())
 
     async def _create_db_instance_connection(self, conn: ProxyClientConnection) -> None:

tests/system/test_psycopg_connection.py

@@ -16,19 +16,16 @@
 
 import asyncio
 from datetime import datetime
+import logging
 import os
 
 # [START cloud_sql_connector_postgres_psycopg]
-from typing import Union
-
 from psycopg import Connection
 import pytest
-import logging
 import sqlalchemy
 
 from google.cloud.sql.connector import Connector
 from google.cloud.sql.connector import DefaultResolver
-from google.cloud.sql.connector import DnsResolver
 
 SERVER_PROXY_PORT = 3307
 

tests/unit/test_connector.py

@@ -14,8 +14,8 @@
 import asyncio
 import logging
 import os
-from threading import Thread
 import socket
+from threading import Thread
 from typing import Union
 
 from aiohttp import ClientResponseError
@@ -34,7 +34,6 @@
 from google.cloud.sql.connector.exceptions import IncompatibleDriverError
 from google.cloud.sql.connector.instance import RefreshAheadCache
 from google.cloud.sql.connector.resolver import DnsResolver
-from google.cloud.sql.connector.proxy import start_local_proxy
 
 logger = logging.getLogger(name=__name__)
 
@@ -679,40 +678,6 @@ async def test_Connector_connect_async_custom_dns_resolver_fallback(
                     # Restore original IPs
                     fake_client.instance.ip_addrs = original_ips
 
-
-async def test_Connector_start_unix_socket_proxy_async(
-    fake_credentials: Credentials,
-    fake_client: CloudSQLClient,
-    proxy_server_async: None,
-) -> 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
-        # Open proxy connection
-        # start the proxy server
-        await connector.start_unix_socket_proxy_async(
-            "test-project:test-region:test-instance",
-            "/tmp/csql-python/proxytest/.s.PGSQL.5432",
-            driver="asyncpg",
-            user="my-user",
-            password="my-pass",
-            db="my-db",
-        )
-        # Wait for server to start
-        await asyncio.sleep(0.5)
-
-        reader, writer = await asyncio.open_unix_connection(
-            "/tmp/csql-python/proxytest/.s.PGSQL.5432"
-        )
-        writer.write("hello\n".encode())
-        await writer.drain()
-        await asyncio.sleep(0.5)
-        msg = await reader.readline()
-        assert msg.decode("utf-8") == "world\n"
-
-
 class TestProtocol(asyncio.Protocol):
     """
     A protocol to proxy data between two transports.

tests/unit/test_local_unix_socket.py

@@ -18,8 +18,6 @@
 import ssl
 from typing import Any
 
-from mock import patch
-from mock import PropertyMock
 import pytest
 
 from google.cloud.sql.connector.local_unix_socket import connect

tests/unit/test_proxy.py

@@ -22,7 +22,8 @@
 
 import pytest
 
-from google.cloud.sql.connector.proxy import Proxy, ServerConnectionFactory
+from google.cloud.sql.connector.proxy import Proxy
+from google.cloud.sql.connector.proxy import ServerConnectionFactory
 
 
 @pytest.fixture
@@ -156,7 +157,7 @@ async def test_proxy_server_connect_fails(proxy_server):
     # wait for server connection to be attempted
     await connector.connect_called.wait()
 
-    assert os.path.exists(socket_path) == True
+    assert os.path.exists(socket_path)
 
     # The client connection should be closed by the proxy
     # Reading should return EOF
@@ -165,7 +166,7 @@ async def test_proxy_server_connect_fails(proxy_server):
 
     await asyncio.sleep(1)  # give proxy a chance to shut down
 
-    assert os.path.exists(socket_path) == False
+    assert os.path.exists(socket_path)
 
 
 @pytest.mark.asyncio
@@ -336,7 +337,7 @@ async def test_tcp_proxy_server_connection_refused(tcp_proxy_server_with_no_tcp_
     await writer.drain()
 
     await asyncio.sleep(1.5)
-    assert os.path.exists(socket_path) == False
+    assert os.path.exists(socket_path)
 
 
 
@@ -355,7 +356,7 @@ async def test_tcp_proxy_server_unexpected_closed(tcp_proxy_server_with_closing_
     assert data == b""
 
     await asyncio.sleep(0.5)  # give event loop a chance to run
-    assert os.path.exists(socket_path) == False
+    assert os.path.exists(socket_path)