fix unit socket implementation, most tests should be fine now

configure custom socket path for mysql container, working around implicitly created volume folders being owned by root
we should probably just not use service containers for this to avoid having to do this patching

Author: Nothing4You

.github/workflows/ci.yml

@@ -49,6 +49,8 @@ jobs:
         image: '${{ matrix.db }}'
         ports:
           - 3306:3306
+        volumes:
+          - "/tmp/run-${{ join(matrix.db, '-') }}/:/socket-mount/"
         options: '--name=mysqld'
         env:
           MYSQL_ROOT_PASSWORD: rootpw
@@ -104,6 +106,19 @@ jobs:
         docker container stop mysqld
         docker container cp "${{ github.workspace }}/tests/ssl_resources/ssl" mysqld:/etc/mysql/ssl
         docker container cp "${{ github.workspace }}/tests/ssl_resources/tls.cnf" mysqld:/etc/mysql/conf.d/aiomysql-tls.cnf
+
+        # use custom socket path
+        # we need to ensure that the socket path is writable for the user running the DB process in the container
+        sudo chmod 0777 /tmp/run-${{ join(matrix.db, '-') }}
+
+        # mysql 5.7 container overrides the socket path in /etc/mysql/mysql.conf.d/mysqld.cnf
+        if [ "${{ join(matrix.db, '-') }}" = "mysql-5.7" ]
+        then
+          docker container cp "${{ github.workspace }}/tests/ssl_resources/socket.cnf" mysqld:/etc/mysql/mysql.conf.d/zz-aiomysql-socket.cnf
+        else
+          docker container cp "${{ github.workspace }}/tests/ssl_resources/socket.cnf" mysqld:/etc/mysql/conf.d/aiomysql-socket.cnf
+        fi
+
         docker container start mysqld
 
         # ensure server is started up
@@ -122,7 +137,7 @@ jobs:
 
         # 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 "unix-${{ join(matrix.db, '') }}=/tmp/run-${{ join(matrix.db, '-') }}/mysql.sock" --mysql-address "tcp-${{ join(matrix.db, '') }}=127.0.0.1:3306"
       env:
         PYTHONUNBUFFERED: 1
         MATRIX_DB: '${{ matrix.db }}'

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,51 @@ 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("unix{}".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_address[i].rsplit("=", 1)
+                ids.append(label)
+            else:
+                addr = opt_mysql_address[i]
+                ids.append("tcp{}".format(i))
+
+            if ":" in addr:
+                addr = addr.rsplit(":", 1)
+                mysql_addresses.append((addr[0], int(addr[1])))
+            else:
+                mysql_addresses.append((addr, 3306))
+
+        # default to connecting to localhost
+        if len(mysql_addresses) == 0:
+            mysql_addresses = [("127.0.0.1", 3306)]
+            ids = ["tcp-local"]
+
+        assert len(mysql_addresses) == len(set(mysql_addresses))
+        assert len(ids) == len(set(ids))
+        assert len(mysql_addresses) == len(ids)
+
+        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 +146,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 +265,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 +298,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

@@ -0,0 +1,2 @@
+[mysqld]
+socket          = /socket-mount/mysql.sock