60 lines
2.6 KiB
Python
60 lines
2.6 KiB
Python
import pytest
|
|
from cuchulu.libraries.db.webmodel import WebModel
|
|
from cuchulu.libraries.db import corefields
|
|
from cuchulu.libraries.db.databases.sqlalchemy import SqlClass
|
|
|
|
|
|
def pytest_addoption(parser):
|
|
parser.addoption("--mysql_host", action="store", default="localhost", help="MySQL host: default localhost")
|
|
parser.addoption("--mysql_user", action="store", default="root", help="Mysql User for make the test: default root")
|
|
parser.addoption("--mysql_password", action="store", default="", help="Mysql password for make the test: default empty value")
|
|
parser.addoption("--mysql_db", action="store", default="test_paramecio_db", help="Mysql Database for execute the test: default test_paramecio_db")
|
|
parser.addoption("--mysql_type", action="store", default="pymysql", help="MySQL Python library used: options pymysql mysqldb")
|
|
|
|
final_conn=None
|
|
|
|
SqlClass.disable_pool=True
|
|
|
|
@pytest.fixture
|
|
def webmodel_conn(request):
|
|
options={'mysql_host': '', 'mysql_user': '', 'mysql_password': '', 'mysql_db': '', 'mysql_type': ''}
|
|
|
|
for k in options:
|
|
options[k]=request.config.getoption(k)
|
|
|
|
# Get connection
|
|
#'db': options['mysql_db'],
|
|
|
|
WebModel.connections={'default': {'name': 'default', 'host': options['mysql_host'], 'user': options['mysql_user'], 'password': options['mysql_password'], 'db': options['mysql_db'], 'charset': 'utf8mb4', 'set_connection': False, 'db_type': options['mysql_type']} }
|
|
|
|
#conn=WebModel.connection()
|
|
|
|
# Create database using raw mysql methods
|
|
|
|
if options['mysql_type']=='pymysql':
|
|
|
|
import pymysql.cursors
|
|
pymysql.install_as_MySQLdb
|
|
|
|
with pymysql.connect(host=options['mysql_host'], user=options['mysql_user'], passwd=options['mysql_password'], charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor) as conn:
|
|
conn.query( 'CREATE DATABASE {name} DEFAULT CHARACTER SET {charset}'.format( name=options['mysql_db'], charset='utf8mb4' ))
|
|
|
|
else:
|
|
import MySQLdb.cursors
|
|
|
|
with MySQLdb.connect(options['mysql_host'], user=options['mysql_user'], passwd=options['mysql_password'], charset='utf8mb4', cursorclass=MySQLdb.cursors.DictCursor) as conn:
|
|
conn.query( 'CREATE DATABASE {name} DEFAULT CHARACTER SET {charset}'.format( name=options['mysql_db'], charset='utf8mb4'))
|
|
|
|
#conn.close()
|
|
|
|
def drop_database():
|
|
|
|
final_conn.query('DROP DATABASE IF EXISTS %s' % options['mysql_db'])
|
|
final_conn.close()
|
|
print('Close db connection')
|
|
|
|
request.addfinalizer(drop_database)
|
|
|
|
final_conn=WebModel.connection()
|
|
|
|
return final_conn
|