chore: update unit tests

Author: jackwotherspoon

google/cloud/sql/connector/pg8000.py

@@ -17,8 +17,6 @@
 import ssl
 from typing import Any, TYPE_CHECKING
 
-SERVER_PROXY_PORT = 3307
-
 if TYPE_CHECKING:
     import pg8000
 

google/cloud/sql/connector/pymysql.py

@@ -17,8 +17,6 @@
 import ssl
 from typing import Any, TYPE_CHECKING
 
-SERVER_PROXY_PORT = 3307
-
 if TYPE_CHECKING:
     import pymysql
 

google/cloud/sql/connector/pytds.py

@@ -20,8 +20,6 @@
 
 from google.cloud.sql.connector.exceptions import PlatformNotSupportedError
 
-SERVER_PROXY_PORT = 3307
-
 if TYPE_CHECKING:
     import pytds
 

tests/unit/test_pg8000.py

@@ -14,7 +14,7 @@
 limitations under the License.
 """
 
-from functools import partial
+import socket
 from typing import Any
 
 from mock import patch
@@ -31,15 +31,14 @@ async def test_pg8000(kwargs: Any) -> None:
     ip_addr = "127.0.0.1"
     # build ssl.SSLContext
     context = await create_ssl_context()
-    # force all wrap_socket calls to have do_handshake_on_connect=False
-    setattr(
-        context,
-        "wrap_socket",
-        partial(context.wrap_socket, do_handshake_on_connect=False),
+    sock = context.wrap_socket(
+        socket.create_connection((ip_addr, 3307)),
+        server_hostname=ip_addr,
+        do_handshake_on_connect=False,
     )
     with patch("pg8000.dbapi.connect") as mock_connect:
         mock_connect.return_value = True
-        connection = connect(ip_addr, context, **kwargs)
+        connection = connect(ip_addr, sock, **kwargs)
         assert connection is True
         # verify that driver connection call would be made
         assert mock_connect.assert_called_once

tests/unit/test_pymysql.py

@@ -14,7 +14,7 @@
 limitations under the License.
 """
 
-from functools import partial
+import socket
 import ssl
 from typing import Any
 
@@ -40,15 +40,14 @@ async def test_pymysql(kwargs: Any) -> None:
     ip_addr = "127.0.0.1"
     # build ssl.SSLContext
     context = await create_ssl_context()
-    # force all wrap_socket calls to have do_handshake_on_connect=False
-    setattr(
-        context,
-        "wrap_socket",
-        partial(context.wrap_socket, do_handshake_on_connect=False),
+    sock = context.wrap_socket(
+        socket.create_connection((ip_addr, 3307)),
+        server_hostname=ip_addr,
+        do_handshake_on_connect=False,
     )
     kwargs["timeout"] = 30
     with patch("pymysql.Connection") as mock_connect:
         mock_connect.return_value = MockConnection
-        pymysql_connect(ip_addr, context, **kwargs)
+        pymysql_connect(ip_addr, sock, **kwargs)
         # verify that driver connection call would be made
         assert mock_connect.assert_called_once

tests/unit/test_pytds.py

@@ -14,8 +14,8 @@
 limitations under the License.
 """
 
-from functools import partial
 import platform
+import socket
 from typing import Any
 
 from mock import patch
@@ -43,16 +43,15 @@ async def test_pytds(kwargs: Any) -> None:
     ip_addr = "127.0.0.1"
     # build ssl.SSLContext
     context = await create_ssl_context()
-    # force all wrap_socket calls to have do_handshake_on_connect=False
-    setattr(
-        context,
-        "wrap_socket",
-        partial(context.wrap_socket, do_handshake_on_connect=False),
+    sock = context.wrap_socket(
+        socket.create_connection((ip_addr, 3307)),
+        server_hostname=ip_addr,
+        do_handshake_on_connect=False,
     )
 
     with patch("pytds.connect") as mock_connect:
         mock_connect.return_value = True
-        connection = connect(ip_addr, context, **kwargs)
+        connection = connect(ip_addr, sock, **kwargs)
         # verify that driver connection call would be made
         assert connection is True
         assert mock_connect.assert_called_once
@@ -68,17 +67,16 @@ async def test_pytds_platform_error(kwargs: Any) -> None:
     assert platform.system() == "Linux"
     # build ssl.SSLContext
     context = await create_ssl_context()
-    # force all wrap_socket calls to have do_handshake_on_connect=False
-    setattr(
-        context,
-        "wrap_socket",
-        partial(context.wrap_socket, do_handshake_on_connect=False),
+    sock = context.wrap_socket(
+        socket.create_connection((ip_addr, 3307)),
+        server_hostname=ip_addr,
+        do_handshake_on_connect=False,
     )
     # add active_directory_auth to kwargs
     kwargs["active_directory_auth"] = True
     # verify that error is thrown with Linux and active_directory_auth
     with pytest.raises(PlatformNotSupportedError):
-        connect(ip_addr, context, **kwargs)
+        connect(ip_addr, sock, **kwargs)
 
 
 @pytest.mark.usefixtures("server")
@@ -94,11 +92,10 @@ async def test_pytds_windows_active_directory_auth(kwargs: Any) -> None:
     assert platform.system() == "Windows"
     # build ssl.SSLContext
     context = await create_ssl_context()
-    # force all wrap_socket calls to have do_handshake_on_connect=False
-    setattr(
-        context,
-        "wrap_socket",
-        partial(context.wrap_socket, do_handshake_on_connect=False),
+    sock = context.wrap_socket(
+        socket.create_connection((ip_addr, 3307)),
+        server_hostname=ip_addr,
+        do_handshake_on_connect=False,
     )
     # add active_directory_auth and server_name to kwargs
     kwargs["active_directory_auth"] = True
@@ -107,7 +104,7 @@ async def test_pytds_windows_active_directory_auth(kwargs: Any) -> None:
         mock_connect.return_value = True
         with patch("pytds.login.SspiAuth") as mock_login:
             mock_login.return_value = True
-            connection = connect(ip_addr, context, **kwargs)
+            connection = connect(ip_addr, sock, **kwargs)
         # verify that driver connection call would be made
         assert mock_login.assert_called_once
         assert connection is True