fix unit socket implementation, most tests should be fine now

Author: Nothing4You

.github/workflows/ci.yml

@@ -28,14 +28,14 @@ jobs:
           # - '3.10'
           # - '3.11.0-alpha.3'
         db:
-          - 'mysql:5.7'
-          - 'mysql:8.0'
-          - 'mariadb:10.2'
-          - 'mariadb:10.3'
-          - 'mariadb:10.4'
-          - 'mariadb:10.5'
-          - 'mariadb:10.6'
-          - 'mariadb:10.7'
+          - [mysql, '5.7']
+          - [mysql, '8.0']
+          - [mariadb, '10.2']
+          - [mariadb, '10.3']
+          - [mariadb, '10.4']
+          - [mariadb, '10.5']
+          - [mariadb, '10.6']
+          - [mariadb, '10.7']
 
       fail-fast: false
     runs-on: ${{ matrix.os }}
@@ -46,9 +46,11 @@ jobs:
 
     services:
       mysql:
-        image: '${{ matrix.db }}'
+        image: "${{ join(matrix.db, ':') }}"
         ports:
           - 3306:3306
+        volumes:
+          - "/tmp/run-${{ join(matrix.db, '-') }}/:/run/mysqld/"
         options: '--name=mysqld'
         env:
           MYSQL_ROOT_PASSWORD: rootpw
@@ -117,29 +119,18 @@ jobs:
 
     - name: Run tests
       run: |
-        export DB="${MATRIX_DB%%:*}"
-        export DBTAG="${MATRIX_DB##*:}"
-
         # 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
-        MATRIX_DB: '${{ matrix.db }}'
+        DB: '${{ matrix.db[0] }}'
+        DBTAG: '${{ matrix.db[1] }}'
       timeout-minutes: 6
 
-    - name: Build coverage flag
-      run: |
-        COVERAGE_FLAG="${MATRIX_OS}_${MATRIX_PY}_${MATRIX_DB//:/-}"
-        echo "COVERAGE_FLAG=$COVERAGE_FLAG" | tee -a "$GITHUB_ENV"
-      env:
-        MATRIX_OS: '${{ matrix.os }}'
-        MATRIX_PY: '${{ matrix.py }}'
-        MATRIX_DB: '${{ matrix.db }}'
-
     - name: Upload coverage
       uses: codecov/[email protected]
       with:
         file: ./coverage.xml
-        flags: "${{ env.COVERAGE_FLAG }}"
+        flags: "${{ matrix.os }}_${{ matrix.py }}_${{ join(matrix.db, '-') }}"
         fail_ci_if_error: true

aiomysql/connection.py

@@ -226,6 +226,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
 
@@ -523,14 +524,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(
@@ -740,7 +742,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))
@@ -767,6 +769,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)
@@ -795,7 +799,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
@@ -957,7 +961,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')
@@ -988,7 +992,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'],
@@ -209,24 +266,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',
@@ -235,21 +299,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