fix unit socket implementation, most tests should be fine now

Author: Nothing4You

.github/workflows/ci.yml

@@ -49,6 +49,8 @@ jobs:
         image: "${{ join(matrix.db, ':') }}"
         ports:
           - 3306:3306
+        volumes:
+          - "/tmp/run-${{ join(matrix.db, '-') }}/:/run/mysqld/"
         options: '--name=mysqld'
         env:
           MYSQL_ROOT_PASSWORD: rootpw
@@ -119,7 +121,7 @@ jobs:
       run: |
         # timeout ensures a more or less clean stop by sending a KeyboardInterrupt which will still provide useful logs
         timeout --preserve-status --signal=INT --verbose 5m \
-          pytest --color=yes --capture=no --verbosity 2 --cov-report term --cov-report xml --cov aiomysql ./tests
+          pytest --color=yes --capture=no --verbosity 2 --cov-report term --cov-report xml --cov aiomysql ./tests --mysql-unix-socket "${{ join(matrix.db, '') }}=/tmp/run-${{ join(matrix.db, '-') }}/mysql.sock" --mysql-address "${{ join(matrix.db, '') }}=127.0.0.1:3306"
       env:
         PYTHONUNBUFFERED: 1
         DB: '${{ matrix.db[0] }}'

aiomysql/connection.py

@@ -229,6 +229,7 @@ def __init__(self, host="localhost", user=None, password="",
         self._client_auth_plugin = auth_plugin
         self._server_auth_plugin = ""
         self._auth_plugin_used = ""
+        self._secure = False
         self.server_public_key = server_public_key
         self.salt = None
 
@@ -526,14 +527,15 @@ async def _connect(self):
         # raise OperationalError(CR.CR_SERVER_GONE_ERROR,
         # "MySQL server has gone away (%r)" % (e,))
         try:
-            if self._unix_socket and self._host in ('localhost', '127.0.0.1'):
+            if self._unix_socket:
                 self._reader, self._writer = await \
                     asyncio.wait_for(
                         _open_unix_connection(
                             self._unix_socket),
                         timeout=self.connect_timeout)
                 self.host_info = "Localhost via UNIX socket: " + \
                                  self._unix_socket
+                self._secure = True
             else:
                 self._reader, self._writer = await \
                     asyncio.wait_for(
@@ -743,7 +745,7 @@ async def _request_authentication(self):
         if self.user is None:
             raise ValueError("Did not specify a username")
 
-        if self._ssl_context:
+        if self._ssl_context and self.server_capabilities & CLIENT.SSL:
             # capablities, max packet, charset
             data = struct.pack('<IIB', self.client_flag, 16777216, 33)
             data += b'\x00' * (32 - len(data))
@@ -770,6 +772,8 @@ async def _request_authentication(self):
                 server_hostname=self._host
             )
 
+            self._secure = True
+
         charset_id = charset_by_name(self.charset).id
         if isinstance(self.user, str):
             _user = self.user.encode(self.encoding)
@@ -798,7 +802,7 @@ async def _request_authentication(self):
                 )
             # Else: empty password
         elif auth_plugin == 'sha256_password':
-            if self._ssl_context and self.server_capabilities & CLIENT.SSL:
+            if self._secure:
                 authresp = self._password.encode('latin1') + b'\0'
             elif self._password:
                 authresp = b'\1'  # request public key
@@ -960,7 +964,7 @@ async def caching_sha2_password_auth(self, pkt):
 
         logger.debug("caching sha2: Trying full auth...")
 
-        if self._ssl_context:
+        if self._secure:
             logger.debug("caching sha2: Sending plain "
                          "password via secure connection")
             self.write_packet(self._password.encode('latin1') + b'\0')
@@ -991,7 +995,7 @@ async def caching_sha2_password_auth(self, pkt):
         pkt.check_error()
 
     async def sha256_password_auth(self, pkt):
-        if self._ssl_context:
+        if self._secure:
             logger.debug("sha256: Sending plain password")
             data = self._password.encode('latin1') + b'\0'
             self.write_packet(data)

tests/conftest.py

@@ -27,6 +27,48 @@ def pytest_generate_tests(metafunc):
         loop_type = ['asyncio', 'uvloop'] if uvloop else ['asyncio']
         metafunc.parametrize("loop_type", loop_type)
 
+    if "mysql_address" in metafunc.fixturenames:
+        mysql_addresses = []
+        ids = []
+
+        opt_mysql_unix_socket = \
+            list(metafunc.config.getoption("mysql_unix_socket"))
+        for i in range(len(opt_mysql_unix_socket)):
+            if "=" in opt_mysql_unix_socket[i]:
+                label, path = opt_mysql_unix_socket[i].rsplit("=", 1)
+                mysql_addresses.append(path)
+                ids.append(label)
+            else:
+                mysql_addresses.append(opt_mysql_unix_socket[i])
+                ids.append("socket{}".format(i))
+
+        opt_mysql_address = list(metafunc.config.getoption("mysql_address"))
+        for i in range(len(opt_mysql_address)):
+            if "=" in opt_mysql_address[i]:
+                label, addr = opt_mysql_unix_socket[i].rsplit("=", 1)
+                ids.append(label)
+            else:
+                addr = opt_mysql_address[i]
+                ids.append("socket{}".format(i))
+
+            if ":" in addr:
+                addr = addr.rsplit(":", 1)
+                mysql_addresses.append((addr[0], int(addr[1])))
+            else:
+                mysql_addresses.append((addr, 3306))
+            ids.append("address{}".format(i))
+
+        # default to connecting to localhost
+        if len(mysql_addresses) == 0:
+            mysql_addresses = [("127.0.0.1", 3306)]
+            ids = ["tcp-local"]
+
+        metafunc.parametrize("mysql_address",
+                             mysql_addresses,
+                             ids=ids,
+                             scope="session",
+                             )
+
 
 # This is here unless someone fixes the generate_tests bit
 @pytest.fixture(scope='session')
@@ -101,6 +143,21 @@ def pytest_configure(config):
     )
 
 
+def pytest_addoption(parser):
+    parser.addoption(
+        "--mysql-address",
+        action="append",
+        default=[],
+        help="list of addresses to connect to",
+    )
+    parser.addoption(
+        "--mysql-unix-socket",
+        action="append",
+        default=[],
+        help="list of unix sockets to connect to",
+    )
+
+
 @pytest.fixture
 def mysql_params(mysql_server):
     params = {**mysql_server['conn_params'],
@@ -205,24 +262,31 @@ def ensure_mysql_version(request, mysql_image, mysql_tag):
 
 
 @pytest.fixture(scope='session')
-def mysql_server(mysql_image, mysql_tag):
-    ssl_directory = os.path.join(os.path.dirname(__file__),
-                                 'ssl_resources', 'ssl')
-    ca_file = os.path.join(ssl_directory, 'ca.pem')
+def mysql_server(mysql_image, mysql_tag, mysql_address):
+    unix_socket = type(mysql_address) is str
+
+    if not unix_socket:
+        ssl_directory = os.path.join(os.path.dirname(__file__),
+                                     'ssl_resources', 'ssl')
+        ca_file = os.path.join(ssl_directory, 'ca.pem')
 
-    ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
-    ctx.check_hostname = False
-    ctx.load_verify_locations(cafile=ca_file)
-    # ctx.verify_mode = ssl.CERT_NONE
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
+        ctx.check_hostname = False
+        ctx.load_verify_locations(cafile=ca_file)
+        # ctx.verify_mode = ssl.CERT_NONE
 
     server_params = {
-        'host': '127.0.0.1',
-        'port': 3306,
         'user': 'root',
         'password': os.environ.get("MYSQL_ROOT_PASSWORD"),
-        'ssl': ctx,
     }
 
+    if unix_socket:
+        server_params["unix_socket"] = mysql_address
+    else:
+        server_params["host"] = mysql_address[0]
+        server_params["port"] = mysql_address[1]
+        server_params["ssl"] = ctx
+
     try:
         connection = pymysql.connect(
             db='mysql',
@@ -231,21 +295,22 @@ def mysql_server(mysql_image, mysql_tag):
             **server_params)
 
         with connection.cursor() as cursor:
-            cursor.execute("SHOW VARIABLES LIKE '%ssl%';")
+            if not unix_socket:
+                cursor.execute("SHOW VARIABLES LIKE '%ssl%';")
 
-            result = cursor.fetchall()
-            result = {item['Variable_name']:
-                      item['Value'] for item in result}
+                result = cursor.fetchall()
+                result = {item['Variable_name']:
+                          item['Value'] for item in result}
 
-            assert result['have_ssl'] == "YES", \
-                "SSL Not Enabled on MySQL"
+                assert result['have_ssl'] == "YES", \
+                    "SSL Not Enabled on MySQL"
 
-            cursor.execute("SHOW STATUS LIKE 'Ssl_version%'")
+                cursor.execute("SHOW STATUS LIKE 'Ssl_version%'")
 
-            result = cursor.fetchone()
-            # As we connected with TLS, it should start with that :D
-            assert result['Value'].startswith('TLS'), \
-                "Not connected to the database with TLS"
+                result = cursor.fetchone()
+                # As we connected with TLS, it should start with that :D
+                assert result['Value'].startswith('TLS'), \
+                    "Not connected to the database with TLS"
 
             # Drop possibly existing old databases
             cursor.execute('DROP DATABASE IF EXISTS test_pymysql;')

tests/fixtures/my.cnf.tmpl renamed to tests/fixtures/my.cnf.tcp.tmpl

@@ -0,0 +1,16 @@
+#
+# The MySQL database server configuration file.
+#
+[client]
+user = {user}
+socket = {unix_socket}
+password = {password}
+database = {db}
+default-character-set = utf8
+
+[client_with_unix_socket]
+user = {user}
+socket = {unix_socket}
+password = {password}
+database = {db}
+default-character-set = utf8

tests/fixtures/my.cnf.unix.tmpl

@@ -15,12 +15,19 @@
 @pytest.fixture()
 def make_engine(mysql_params, connection):
     async def _make_engine(**kwargs):
+        if "unix_socket" in mysql_params:
+            conn_args = {"unix_socket": mysql_params["unix_socket"]}
+        else:
+            conn_args = {
+                "host": mysql_params['host'],
+                "port": mysql_params['port'],
+            }
+
         return (await sa.create_engine(db=mysql_params['db'],
                                        user=mysql_params['user'],
                                        password=mysql_params['password'],
-                                       host=mysql_params['host'],
-                                       port=mysql_params['port'],
                                        minsize=10,
+                                       **conn_args,
                                        **kwargs))
 
     return _make_engine

tests/sa/test_sa_compiled_cache.py

@@ -22,12 +22,19 @@
 @pytest.fixture()
 def make_engine(mysql_params, connection):
     async def _make_engine(**kwargs):
+        if "unix_socket" in mysql_params:
+            conn_args = {"unix_socket": mysql_params["unix_socket"]}
+        else:
+            conn_args = {
+                "host": mysql_params['host'],
+                "port": mysql_params['port'],
+            }
+
         return (await sa.create_engine(db=mysql_params['db'],
                                        user=mysql_params['user'],
                                        password=mysql_params['password'],
-                                       host=mysql_params['host'],
-                                       port=mysql_params['port'],
                                        minsize=10,
+                                       **conn_args,
                                        **kwargs))
 
     return _make_engine

tests/sa/test_sa_default.py

@@ -15,12 +15,19 @@
 @pytest.fixture()
 def make_engine(connection, mysql_params):
     async def _make_engine(**kwargs):
+        if "unix_socket" in mysql_params:
+            conn_args = {"unix_socket": mysql_params["unix_socket"]}
+        else:
+            conn_args = {
+                "host": mysql_params['host'],
+                "port": mysql_params['port'],
+            }
+
         return (await sa.create_engine(db=mysql_params['db'],
                                        user=mysql_params['user'],
                                        password=mysql_params['password'],
-                                       host=mysql_params['host'],
-                                       port=mysql_params['port'],
                                        minsize=10,
+                                       **conn_args,
                                        **kwargs))
     return _make_engine
 

tests/sa/test_sa_engine.py

@@ -10,7 +10,13 @@
 @pytest.fixture()
 def fill_my_cnf(mysql_params):
     tests_root = os.path.abspath(os.path.dirname(__file__))
-    path1 = os.path.join(tests_root, 'fixtures/my.cnf.tmpl')
+
+    if "unix_socket" in mysql_params:
+        tmpl_path = "fixtures/my.cnf.unix.tmpl"
+    else:
+        tmpl_path = "fixtures/my.cnf.tcp.tmpl"
+
+    path1 = os.path.join(tests_root, tmpl_path)
     path2 = os.path.join(tests_root, 'fixtures/my.cnf')
     with open(path1) as f1:
         tmpl = f1.read()
@@ -31,8 +37,11 @@ async def test_config_file(fill_my_cnf, connection_creator, mysql_params):
     path = os.path.join(tests_root, 'fixtures/my.cnf')
     conn = await connection_creator(read_default_file=path)
 
-    assert conn.host == mysql_params['host']
-    assert conn.port == mysql_params['port']
+    if "unix_socket" in mysql_params:
+        assert conn.unix_socket == mysql_params["unix_socket"]
+    else:
+        assert conn.host == mysql_params['host']
+        assert conn.port == mysql_params['port']
     assert conn.user, mysql_params['user']
 
     # make sure connection is working
@@ -167,12 +176,15 @@ async def test_connection_gone_away(connection_creator):
 
 
 @pytest.mark.run_loop
-async def test_connection_info_methods(connection_creator):
+async def test_connection_info_methods(connection_creator, mysql_params):
     conn = await connection_creator()
     # trhead id is int
     assert isinstance(conn.thread_id(), int)
     assert conn.character_set_name() in ('latin1', 'utf8mb4')
-    assert str(conn.port) in conn.get_host_info()
+    if "unix_socket" in mysql_params:
+        assert mysql_params["unix_socket"] in conn.get_host_info()
+    else:
+        assert str(conn.port) in conn.get_host_info()
     assert isinstance(conn.get_server_info(), str)
     # protocol id is int
     assert isinstance(conn.get_proto_info(), int)
@@ -200,8 +212,11 @@ async def test_connection_ping(connection_creator):
 @pytest.mark.run_loop
 async def test_connection_properties(connection_creator, mysql_params):
     conn = await connection_creator()
-    assert conn.host == mysql_params['host']
-    assert conn.port == mysql_params['port']
+    if "unix_socket" in mysql_params:
+        assert conn.unix_socket == mysql_params["unix_socket"]
+    else:
+        assert conn.host == mysql_params['host']
+        assert conn.port == mysql_params['port']
     assert conn.user == mysql_params['user']
     assert conn.db == mysql_params['db']
     assert conn.echo is False

tests/test_connection.py

@@ -184,7 +184,7 @@ async def test_issue_17(connection, connection_creator, mysql_params):
 async def test_issue_34(connection_creator):
     try:
         await connection_creator(host="localhost", port=1237,
-                                 user="root")
+                                 user="root", unix_socket=None)
         pytest.fail()
     except aiomysql.OperationalError as e:
         assert 2003 == e.args[0]