fix(psycopg): add platform guard for platforms without AF_UNIX support

google/cloud/sql/connector/psycopg.py

@@ -142,6 +142,11 @@ def connect(
             'Unable to import module "psycopg." Please install and try again.'
         )
 
+    if not hasattr(socket, "AF_UNIX"):
+        raise NotImplementedError(
+            "Unix domain sockets (AF_UNIX) are not supported on this platform"
+        )
+
     tmpdir = tempfile.mkdtemp()
     socket_path = os.path.join(tmpdir, ".s.PGSQL.5432")
     logger.debug("psycopg: created Unix socket at %s", socket_path)

tests/system/test_psycopg_connection.py

@@ -18,6 +18,7 @@
 import asyncio
 from datetime import datetime
 import os
+import socket
 
 import pytest
 import sqlalchemy
@@ -26,6 +27,11 @@
 from google.cloud.sql.connector import DefaultResolver
 from google.cloud.sql.connector import DnsResolver
 
+pytestmark = pytest.mark.skipif(
+    not hasattr(socket, "AF_UNIX"),
+    reason="Unix domain sockets (AF_UNIX) not available on this platform",
+)
+
 
 def create_sqlalchemy_engine(
     instance_connection_name: str,

tests/system/test_psycopg_iam_auth.py

@@ -16,17 +16,24 @@
 
 from datetime import datetime
 import os
+import socket
 
 import pytest
 import sqlalchemy
 
 from google.cloud.sql.connector import Connector
 
-# Skip all tests in this file if POSTGRES_IAM_USER is not set
-pytestmark = pytest.mark.skipif(
-    not os.environ.get("POSTGRES_IAM_USER"),
-    reason="POSTGRES_IAM_USER env var not set for IAM Authn tests",
-)
+# Skip all tests in this file if POSTGRES_IAM_USER is not set or AF_UNIX is not available
+pytestmark = [
+    pytest.mark.skipif(
+        not os.environ.get("POSTGRES_IAM_USER"),
+        reason="POSTGRES_IAM_USER env var not set for IAM Authn tests",
+    ),
+    pytest.mark.skipif(
+        not hasattr(socket, "AF_UNIX"),
+        reason="Unix domain sockets (AF_UNIX) not available on this platform",
+    ),
+]
 
 
 def create_sqlalchemy_engine(

tests/unit/test_psycopg.py

@@ -388,6 +388,18 @@ def test_connect_import_error() -> None:
         connect("127.0.0.1", mock_remote_sock)
 
 
+def test_connect_unsupported_platform() -> None:
+    """Test that connect raises NotImplementedError when AF_UNIX is not available."""
+    mock_remote_sock = MagicMock(spec=ssl.SSLSocket)
+    with (
+        patch("google.cloud.sql.connector.psycopg.hasattr", return_value=False),
+        pytest.raises(
+            NotImplementedError, match="Unix domain sockets \\(AF_UNIX\\)"
+        ),
+    ):
+        connect("127.0.0.1", mock_remote_sock)
+
+
 def test_connect_cleanup_errors() -> None:
     """Test that connect ignores OSErrors when removing temp files/dirs during cleanup."""
     mock_remote_sock = MagicMock(spec=ssl.SSLSocket)