forked from stove/dataset
80 lines
2.5 KiB
Python
80 lines
2.5 KiB
Python
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()
|