Class PyMySQLTestCase

pymysql/tests/base.py:10–112

class PyMySQLTestCase(unittest.TestCase):
    # You can specify your test environment creating a file named
    #  "databases.json" or editing the `databases` variable below.
    fname = os.path.join(os.path.dirname(__file__), "databases.json")
    if os.path.exists(fname):
        with open(fname) as f:
            databases = json.load(f)
    else:
        databases = [
            {
                "host": "localhost",
                "user": "root",
                "passwd": "",
                "database": "test1",
                "use_unicode": True,
                "local_infile": True,
            },
            {"host": "localhost", "user": "root", "passwd": "", "database": "test2"},
        ]

    def mysql_server_is(self, conn, version_tuple):
        """Return True if the given connection is on the version given or
        greater.

        This only checks the server version string provided when the
        connection is established, therefore any check for a version tuple
        greater than (5, 5, 5) will always fail on MariaDB, as it always
        starts with 5.5.5, e.g. 5.5.5-10.7.1-MariaDB-1:10.7.1+maria~focal.

        e.g.::

            if self.mysql_server_is(conn, (5, 6, 4)):
                # do something for MySQL 5.6.4 and above
        """
        server_version = conn.get_server_info()
        server_version_tuple = tuple(
            (int(dig) if dig is not None else 0)
            for dig in re.match(r"(\d+)\.(\d+)\.(\d+)", server_version).group(1, 2, 3)
        )
        return server_version_tuple >= version_tuple

    def get_mysql_vendor(self, conn):
        server_version = conn.get_server_info()

        if "MariaDB" in server_version:
            return "mariadb"

        return "mysql"

    _connections = None

    @property
    def connections(self):
        if self._connections is None:
            self._connections = []
            for params in self.databases:
                self._connections.append(pymysql.connect(**params))
            self.addCleanup(self._teardown_connections)