update: 2026-03-28 20:59
This commit is contained in:
81
test/test_mysql_connect.py
Normal file
81
test/test_mysql_connect.py
Normal file
@@ -0,0 +1,81 @@
|
||||
#!/usr/bin/env python3
|
||||
import sys
|
||||
|
||||
PROFILES = [
|
||||
{
|
||||
'label': 'development',
|
||||
'host': '10.1.1.100',
|
||||
'port': 3306,
|
||||
'name': 'infogenie-test',
|
||||
'user': 'infogenie-test',
|
||||
'password': 'infogenie-test',
|
||||
},
|
||||
{
|
||||
'label': 'production',
|
||||
'host': '192.168.1.100',
|
||||
'port': 3306,
|
||||
'name': 'infogenie',
|
||||
'user': 'infogenie',
|
||||
'password': 'infogenie',
|
||||
},
|
||||
]
|
||||
|
||||
try:
|
||||
import pymysql
|
||||
DRIVER_NAME = 'pymysql'
|
||||
except ImportError:
|
||||
try:
|
||||
import mysql.connector as mysql_connector
|
||||
DRIVER_NAME = 'mysql.connector'
|
||||
except ImportError:
|
||||
print('missing_driver: install pymysql or mysql-connector-python')
|
||||
sys.exit(2)
|
||||
|
||||
|
||||
def connect_and_check(profile):
|
||||
label = profile['label']
|
||||
host = profile['host']
|
||||
port = profile['port']
|
||||
name = profile['name']
|
||||
user = profile['user']
|
||||
password = profile['password']
|
||||
|
||||
print(f'[{label}] target={host}:{port} db={name} user={user}')
|
||||
|
||||
try:
|
||||
if DRIVER_NAME == 'pymysql':
|
||||
conn = pymysql.connect(
|
||||
host=host,
|
||||
port=port,
|
||||
user=user,
|
||||
password=password,
|
||||
database=name,
|
||||
connect_timeout=5,
|
||||
charset='utf8mb4',
|
||||
)
|
||||
else:
|
||||
conn = mysql_connector.connect(
|
||||
host=host,
|
||||
port=port,
|
||||
user=user,
|
||||
password=password,
|
||||
database=name,
|
||||
connection_timeout=5,
|
||||
)
|
||||
|
||||
with conn.cursor() as cursor:
|
||||
cursor.execute('SELECT DATABASE(), CURRENT_USER(), VERSION()')
|
||||
row = cursor.fetchone()
|
||||
conn.close()
|
||||
print(f'[{label}] mysql_ok driver={DRIVER_NAME} result={row!r}')
|
||||
return True
|
||||
except Exception as exc:
|
||||
print(f'[{label}] mysql_fail driver={DRIVER_NAME} err={exc}')
|
||||
return False
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
ok = True
|
||||
for profile in PROFILES:
|
||||
ok = connect_and_check(profile) and ok
|
||||
sys.exit(0 if ok else 1)
|
||||
Reference in New Issue
Block a user