refactor: Align Python connector fallback and global region behavior with Go

- Reject 'global' region in direct PSC DNS resolution in DnsResolver to force CNAME/TXT fallback.
- Remove fallback to Private/Public IP in Connector when DNS fails and preferred IP is a hostname.
- Update test_connector.py to assert no fallback from PSC to Private IP when DNS fails.
- Add test_resolver.py test to verify global region skips direct resolution.

Author: Kokoro Connector Build

google/cloud/sql/connector/connector.py

@@ -18,7 +18,6 @@
 
 import asyncio
 from functools import partial
-import ipaddress
 import logging
 import os
 import socket
@@ -51,26 +50,8 @@
 logger = logging.getLogger(name=__name__)
 
 
-def _is_ip_address(ip: str) -> bool:
-    try:
-        ipaddress.ip_address(ip)
-        return True
-    except ValueError:
-        return False
 
 
-def _get_fallback_ips(
-    current_ips: list[str], ip_addresses: dict[str, list[str]]
-) -> list[str]:
-    if not current_ips:
-        return current_ips
-    if _is_ip_address(current_ips[0]):
-        return current_ips
-    fallback = ip_addresses.get("PRIVATE")
-    if not fallback:
-        fallback = ip_addresses.get("PRIMARY")
-    return fallback if fallback else current_ips
-
 ASYNC_DRIVERS = ["asyncpg"]
 SERVER_PROXY_PORT = 3307
 _DEFAULT_SCHEME = "https://"
@@ -430,25 +411,19 @@ async def connect_async(
                         "using it to connect"
                     )
                 else:
-                    fallback_ips = _get_fallback_ips(
-                        preferred_ips, conn_info.ip_addrs
-                    )
                     logger.debug(
                         f"['{instance_connection_string}']: Custom DNS name "
                         f"'{conn_info.conn_name.domain_name}' resolved but returned no "
-                        f"entries, using '{fallback_ips[0]}' from instance metadata"
+                        f"entries, using '{preferred_ips}' from instance metadata"
                     )
-                    targets.extend(fallback_ips)
+                    targets.extend(preferred_ips)
             except Exception as e:
-                fallback_ips = _get_fallback_ips(
-                    preferred_ips, conn_info.ip_addrs
-                )
                 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 '{fallback_ips[0]}' from instance metadata"
+                    f"address: {e}, using '{preferred_ips}' from instance metadata"
                 )
-                targets.extend(fallback_ips)
+                targets.extend(preferred_ips)
         else:
             targets.extend(preferred_ips)
 

google/cloud/sql/connector/resolver.py

@@ -69,19 +69,20 @@ async def resolve(self, dns: str) -> ConnectionName:  # type: ignore
             match = PSC_DNS_PATTERN.match(dns_normalized.lower())
             if match:
                 region = match.group(3)
-                if self._client is None:
-                    raise ValueError(
-                        "SQLAdmin client is not configured in the resolver."
+                if region != "global":
+                    if self._client is None:
+                        raise ValueError(
+                            "SQLAdmin client is not configured in the resolver."
+                        )
+
+                    dns_name_with_dot = dns_normalized + "."
+                    resp = await self._client.resolve_connect_settings(
+                        dns_name_with_dot, region
+                    )
+                    resolved_conn_name = resp["connectionName"]
+                    return _parse_connection_name_with_domain_name(
+                        resolved_conn_name, dns
                     )
-
-                dns_name_with_dot = dns_normalized + "."
-                resp = await self._client.resolve_connect_settings(
-                    dns_name_with_dot, region
-                )
-                resolved_conn_name = resp["connectionName"]
-                return _parse_connection_name_with_domain_name(
-                    resolved_conn_name, dns
-                )
 
             if not _is_valid_domain(current):
                 raise ValueError(

tests/unit/test_connector.py

@@ -660,10 +660,10 @@ async def test_Connector_connect_async_custom_dns_resolver_fallback(
 
 
 @pytest.mark.asyncio
-async def test_Connector_connect_async_custom_dns_resolver_fallback_psc_to_private_ip(
+async def test_Connector_connect_async_custom_dns_resolver_no_fallback_psc_to_private_ip(
     fake_credentials: Credentials, fake_client: CloudSQLClient
 ) -> None:
-    """Test that Connector.connect_async falls back to Private IP if CNAME/PSC DNS resolution fails."""
+    """Test that Connector.connect_async does not fall back to Private IP if CNAME/PSC DNS resolution fails."""
 
     with patch(
         "google.cloud.sql.connector.resolver.DnsResolver.resolve_a_record"
@@ -688,8 +688,10 @@ async def test_Connector_connect_async_custom_dns_resolver_fallback_psc_to_priva
                 connector._client = fake_client
 
                 original_ips = fake_client.instance.ip_addrs
-                # Configure instance to be PSC enabled, but also have a PRIVATE IP fallback!
+                original_dns_names = fake_client.instance.dns_names
+                # Configure instance to be PSC enabled, but also have a PRIVATE IP!
                 fake_client.instance.psc_enabled = True
+                fake_client.instance.dns_names = ["1ad3b5d73f10.3oxon2yfo9tob.us-east1.sql.goog"]
                 fake_client.instance.ip_addrs = {
                     "PSC": "1ad3b5d73f10.3oxon2yfo9tob.us-east1.sql.goog",
                     "PRIVATE": "10.0.0.1",
@@ -709,13 +711,14 @@ async def test_Connector_connect_async_custom_dns_resolver_fallback_psc_to_priva
                             db="my-db",
                         )
 
-                        # Verify mock_connect fell back to PRIVATE IP "10.0.0.1"!
+                        # Verify mock_connect used PSC DNS instead of falling back to PRIVATE IP "10.0.0.1"!
                         args, _ = mock_connect.call_args
-                        assert args[0] == "10.0.0.1"
+                        assert args[0] == "1ad3b5d73f10.3oxon2yfo9tob.us-east1.sql.goog"
                         assert connection is True
                 finally:
-                    # Restore original IPs
+                    # Restore original IPs and DNS names
                     fake_client.instance.ip_addrs = original_ips
+                    fake_client.instance.dns_names = original_dns_names
                     fake_client.instance.psc_enabled = False
 
 

tests/unit/test_resolver.py

@@ -277,4 +277,26 @@ async def test_DnsResolver_cname_loop_throws_error() -> None:
 
         with pytest.raises(DnsResolutionError) as exc_info:
             await resolver.resolve(dns_name)
-        assert "CNAME loop detected" in str(exc_info.value)
+        assert "CNAME loop detected" in str(exc_info.value)
+
+
+async def test_DnsResolver_global_region_skips_direct_resolution() -> None:
+    """Test DnsResolver skips direct resolution if the PSC DNS name has a global region."""
+    dns_name = "0123456789ab.fedcba9876543.global.sql-psc.goog"
+
+    mock_client = AsyncMock()
+    resolver = DnsResolver()
+    resolver.set_client(mock_client)
+
+    # Patch CNAME and TXT to fail, so it eventually raises DnsResolutionError
+    with patch.object(
+        resolver, "resolve_cname", AsyncMock(side_effect=DnsResolutionError("No CNAME"))
+    ), patch.object(
+        resolver, "resolve_txt", AsyncMock(side_effect=DnsResolutionError("No TXT"))
+    ):
+
+        with pytest.raises(DnsResolutionError):
+            await resolver.resolve(dns_name)
+
+    # Verify mock_client was NOT called because direct resolution was skipped
+    mock_client.resolve_connect_settings.assert_not_called()