#!/usr/bin/env python -O # -*- coding: utf-8 -*- import os import unittest import mariadb from test.base_test import create_connection, is_skysql, is_maxscale from test.conf_test import conf from mariadb.constants import STATUS import platform from packaging.version import parse as parse_version from packaging import version import traceback, sys class TestConnection(unittest.TestCase): def setUp(self): self.connection = create_connection() def tearDown(self): del self.connection def test_conpy36(self): if platform.system() == "Windows": self.skipTest("unix_socket not supported on Windows") default_conf = conf() try: mariadb.connect(user=default_conf["user"], unix_socket="/does_not_exist/x.sock", port=default_conf["port"], host=default_conf["host"]) except (mariadb.OperationalError,): # make asan happy tb = sys.exc_info()[2] traceback.clear_frames(tb) pass def test_connection_default_file(self): if os.path.exists("client.cnf"): os.remove("client.cnf") default_conf = conf() f = open("client.cnf", "w+") f.write("[client]\n") f.write("host =%s\n" % default_conf["host"]) f.write("port =%i\n" % default_conf["port"]) f.write("user =%s\n" % default_conf["user"]) if "password" in default_conf: f.write("password =%s\n" % default_conf["password"]) f.write("database =%s\n" % default_conf["database"]) f.close() new_conn = mariadb.connect(user=default_conf["user"], ssl=True, default_file="./client.cnf") self.assertEqual(new_conn.database, default_conf["database"]) del new_conn os.remove("client.cnf") def test_autocommit(self): conn = self.connection conn.autocommit = False self.assertEqual(conn.autocommit, False) # revert conn.autocommit = True self.assertEqual(conn.autocommit, True) conn.close() def test_local_infile(self): default_conf = conf() new_conn = mariadb.connect(**default_conf, local_infile=False) cursor = new_conn.cursor() cursor.execute("CREATE TEMPORARY TABLE t1 (a int)") try: cursor.execute("LOAD DATA LOCAL INFILE 'x.x' INTO TABLE t1") except (mariadb.OperationalError,): # make asan happy if mariadb._have_asan: tb = sys.exc_info()[2] traceback.clear_frames(tb) pass del cursor del new_conn def test_tls_version(self): if is_maxscale(): self.skipTest("MAXSCALE test has no SSL on port by default") default_conf = conf() conn = mariadb.connect(**default_conf, tls_version="TLSv1.2") cursor = conn.cursor() cursor.execute("SHOW STATUS LIKE 'ssl_version'") row = cursor.fetchone() self.assertEqual(row[1], "TLSv1.2") cursor.close() conn.close() def test_init_command(self): default_conf = conf() new_conn = mariadb.connect(**default_conf, init_command="SET @a:=1") cursor = new_conn.cursor() cursor.execute("SELECT @a") row = cursor.fetchone() self.assertEqual(row[0], 1) del cursor del new_conn def test_compress(self): default_conf = conf() new_conn = mariadb.connect(**default_conf, compress=True) cursor = new_conn.cursor() cursor.execute("SHOW SESSION STATUS LIKE 'compression'") row = cursor.fetchone() if is_maxscale(): self.assertEqual(row[1], "OFF") else: self.assertEqual(row[1], "ON") del cursor del new_conn def test_schema(self): if self.connection.server_version < 100202: self.skipTest("session tracking not supported") if is_maxscale(): self.skipTest("MAXSCALE doesn't tell schema change for now") default_conf = conf() conn = create_connection() self.assertEqual(conn.database, default_conf["database"]) cursor = conn.cursor() cursor.execute("DROP SCHEMA IF EXISTS test1") cursor.execute("CREATE SCHEMA test1") cursor.execute("USE test1") self.assertEqual(conn.database, "test1") conn.database = default_conf["database"] self.assertEqual(conn.database, default_conf["database"]) cursor.close() conn.close() def test_ping(self): if is_maxscale(): self.skipTest("MAXSCALE wrong thread id") with create_connection() as conn: with conn.cursor() as cursor: oldid = conn.connection_id try: cursor.execute("KILL {id}" . format(id=oldid)) except mariadb.Error: pass conn.auto_reconnect = True conn.ping() self.assertNotEqual(oldid, conn.connection_id) self.assertNotEqual(0, conn.connection_id) def test_ed25519(self): if is_skysql(): self.skipTest("Test fail on SkySQL") default_conf = conf() if is_maxscale(): self.skipTest("MAXSCALE doesn't support ed25519 for now") if self.connection.server_version < 100122: self.skipTest("ed25519 not supported") conn = create_connection() curs = conn.cursor(buffered=True) if self.connection.server_name == "localhost": curs.execute("select * from information_schema.plugins where " "plugin_name ='unix_socket' and " "plugin_status ='ACTIVE'") if curs.rowcount > 0: del curs self.skipTest("unix_socket is active") cursor = conn.cursor() try: cursor.execute("INSTALL SONAME 'auth_ed25519'") except (mariadb.DatabaseError, mariadb.OperationalError): self.skipTest("Server couldn't load auth_ed25519") cursor.execute("DROP USER IF EXISTS eduser") if self.connection.server_version < 100400: cursor.execute("CREATE USER eduser@'%' IDENTIFIED VIA ed25519 " "USING " "'6aW9C7ENlasUfymtfMvMZZtnkCVlcb1ssxOLJ0kj/AA'") else: cursor.execute("CREATE USER eduser@'%' IDENTIFIED VIA ed25519 " "USING PASSWORD('MySup8%rPassw@ord')") cursor.execute("GRANT ALL on " + default_conf["database"] + ".* to eduser@'%'") conn2 = create_connection({"user": "eduser", "password": "MySup8%rPassw@ord"}) cursor.execute("DROP USER IF EXISTS eduser") try: create_connection({"user": "eduser", "password": "MySup8%rPassw@ord", "plugin_dir": "wrong_plugin_dir"}) self.fail("wrong plugin directory, must not have found " "authentication plugin") except (mariadb.OperationalError): pass cursor.execute("DROP USER IF EXISTS eduser") del cursor, conn2, conn def test_conpy46(self): with create_connection() as con: with con.cursor() as cursor: cursor.execute("SELECT 'foo'") row = cursor.fetchone() self.assertEqual(row[0], "foo") try: cursor.execute("SELECT 'bar'") except mariadb.ProgrammingError: pass try: cursor = con.cursor() except mariadb.ProgrammingError: pass def test_conpy101(self): default_conf = conf() c1 = mariadb.connect(**default_conf) self.assertEqual(c1.autocommit, False) c1.close() c1 = mariadb.connect(**default_conf, autocommit=True) self.assertEqual(c1.autocommit, True) c1.close() def test_db_attribute(self): with create_connection() as con: with con.cursor() as cursor: db = con.database try: cursor.execute("create schema test123") except mariadb.Error: pass con.database = "test123" cursor.execute("select database()", buffered=True) row = cursor.fetchone() self.assertEqual(row[0], "test123") con.database = db cursor.execute("select database()", buffered=True) row = cursor.fetchone() self.assertEqual(row[0], db) self.assertEqual(row[0], con.database) cursor.execute("drop schema test123") def test_server_status(self): con = create_connection() self.assertTrue(not con.server_status & STATUS.AUTOCOMMIT) con.autocommit = True self.assertTrue(con.server_status & STATUS.AUTOCOMMIT) con.autocommit = False self.assertTrue(not con.server_status & STATUS.AUTOCOMMIT) def test_conpy175(self): default_conf = conf() conn = mariadb.connect(**default_conf) str = "Bob's" cursor= conn.cursor() cursor.execute("SET session sql_mode='NO_BACKSLASH_ESCAPES'") newstr = conn.escape_string(str) self.assertEqual(newstr, "Bob''s") cursor.execute("SET session sql_mode=''") newstr = conn.escape_string(str) self.assertEqual(newstr, "Bob\\'s") conn.close() def test_closed(self): default_conf = conf() conn = mariadb.connect(**default_conf) conn.close() try: conn.cursor() except (mariadb.ProgrammingError): pass def test_multi_host(self): default_conf = conf() default_conf["host"] = "non_existant," + default_conf["host"] try: mariadb.connect(**default_conf) except mariadb.ProgrammingError: self.assertLess(parse_version(mariadb.mariadbapi_version), parse_version('3.3.0')) pass def test_tls_verification(self): if is_maxscale(): self.skipTest("MAXSCALE test has no SSL on port by default") if version.Version(mariadb.mariadbapi_version) <\ version.Version('3.4.2'): self.skipTest("Requires C/C 3.4.2 or newer") default_conf= conf() default_conf["ssl"] = False conn= mariadb.connect(**default_conf) self.assertEqual(conn._tls_verify_status, None) conn.close() default_conf= conf() default_conf["ssl"] = True conn= mariadb.connect(**default_conf) self.assertNotEqual(conn._tls_verify_status, None) conn.close() def test_tls_fp(self): if is_maxscale(): self.skipTest("MAXSCALE test has no SSL on port by default") if version.Version(mariadb.mariadbapi_version) <\ version.Version('3.4.2'): self.skipTest("Requires C/C 3.4.2 or newer") default_conf= conf() default_conf["ssl"] = True conn= mariadb.connect(**default_conf) self.assertEqual(conn._tls, True) x509_info= conn.tls_peer_cert_info if not x509_info: conn.close() self.skipTest("Peer certificate information not supported") fp= x509_info["fingerprint"] self.assertEqual(len(fp), 64) conn.close() default_conf= conf() default_conf["tls_fp"] = fp conn= mariadb.connect(**default_conf) self.assertEqual(conn._tls, True) x509_info= conn.tls_peer_cert_info self.assertEqual(fp, x509_info["fingerprint"]) conn.close() def test_conpy278(self): if is_maxscale(): self.skipTest("MAXSCALE bug MXS-4961") with create_connection({"reconnect" : True}) as conn: old_id= conn.connection_id try: conn.kill(conn.connection_id) except mariadb.OperationalError: conn.ping() self.assertNotEqual(old_id, conn.connection_id) with create_connection({"reconnect" : True}) as conn: old_id= conn.connection_id try: conn.kill(conn.connection_id) except mariadb.OperationalError: conn.ping() self.assertNotEqual(old_id, conn.connection_id) with create_connection({"reconnect" : True}) as conn: old_id= conn.connection_id try: conn.kill(conn.connection_id) except mariadb.OperationalError: pass with conn.cursor() as cursor: try: cursor.execute("set @a:=1") except mariadb.InterfaceError: pass cursor.execute("set @a:=1") self.assertNotEqual(old_id, conn.connection_id) old_id= conn.connection_id conn.reconnect() self.assertNotEqual(old_id, conn.connection_id) if __name__ == '__main__': unittest.main()