dataset/seed/mariadb/tests/test_mariadb.py

81 lines
2.5 KiB
Python
Raw Normal View History

2022-07-05 22:09:16 +02:00
from yaml import load, SafeLoader
from os import environ
from pytest import raises
from pymysql import connect
from pymysql.err import OperationalError
def test_mariadb_wrong_password():
conf_file = f'{environ["MACHINE_TEST_DIR"]}/mariadb.yml'
with open(conf_file) as yaml:
data = load(yaml, Loader=SafeLoader)
with raises(OperationalError):
connect(data['address'], data['user'], 'a', data['dbname'])
def test_mariadb_connection():
conf_file = f'{environ["MACHINE_TEST_DIR"]}/mariadb.yml'
with open(conf_file) as yaml:
data = load(yaml, Loader=SafeLoader)
db = connect(data['address'], data['user'], data['password'], data['dbname'])
db.close()
def test_mariadb_migration():
conf_file = f'{environ["MACHINE_TEST_DIR"]}/mariadb.yml'
with open(conf_file) as yaml:
data = load(yaml, Loader=SafeLoader)
db = connect(data['address'], data['user'], data['password'], data['dbname'])
cursor = db.cursor()
if 'FIRST_RUN' in environ:
sql = """CREATE TABLE test (col CHAR(20) NOT NULL)"""
cursor.execute(sql)
sql = """INSERT INTO test (col) VALUES ("test")"""
cursor.execute(sql)
db.commit()
sql = """SELECT * FROM test"""
cursor.execute(sql)
results = cursor.fetchall()
assert len(results) == 1
results[0] == ('test',)
cursor.close()
db.close()
def test_mariadb_insert():
conf_file = f'{environ["MACHINE_TEST_DIR"]}/mariadb.yml'
with open(conf_file) as yaml:
data = load(yaml, Loader=SafeLoader)
db = connect(data['address'], data['user'], data['password'], data['dbname'])
cursor = db.cursor()
sql = """INSERT INTO test (col) VALUES ("test2")"""
cursor.execute(sql)
db.commit()
#
sql = """SELECT * FROM test WHERE col = 'test2'"""
cursor.execute(sql)
results = cursor.fetchall()
assert len(results) == 1
results[0] == ('test2',)
cursor.close()
db.close()
def test_mariadb_delete():
conf_file = f'{environ["MACHINE_TEST_DIR"]}/mariadb.yml'
with open(conf_file) as yaml:
data = load(yaml, Loader=SafeLoader)
db = connect(data['address'], data['user'], data['password'], data['dbname'])
cursor = db.cursor()
sql = """DELETE FROM test WHERE col = 'test2'"""
cursor.execute(sql)
db.commit()
#
sql = """SELECT * FROM test WHERE col = 'test2'"""
cursor.execute(sql)
results = cursor.fetchall()
assert len(results) == 0
cursor.close()
db.close()