70 lines
2.4 KiB
Python
70 lines
2.4 KiB
Python
|
from typing import Any
|
||
|
|
import unittest
|
||
|
|
import logging
|
||
|
|
from pathlib import Path
|
||
|
|
from cocluto.SimpaDbUtil import ISqlDatabaseBackend, SqliteDb, SqlTableField, SshAccessedMysqlDb
|
||
|
|
|
||
|
|
|
||
|
|
def stringify(value: Any):
|
||
|
|
return f"'{str(value)}'"
|
||
|
|
|
||
|
|
|
||
|
|
def test_sql_backend(sql_backend: ISqlDatabaseBackend):
|
||
|
|
table_name = 'dummy'
|
||
|
|
fields = [
|
||
|
|
SqlTableField('family_name', SqlTableField.Type.FIELD_TYPE_STRING, 'family name'),
|
||
|
|
SqlTableField('birth_area_code', SqlTableField.Type.FIELD_TYPE_INT, 'the number encoding the birth department'),
|
||
|
|
SqlTableField('height', SqlTableField.Type.FIELD_TYPE_FLOAT, 'height (in meters)'),
|
||
|
|
SqlTableField('birth_date', SqlTableField.Type.FIELD_TYPE_TIME, 'birth date')
|
||
|
|
]
|
||
|
|
if sql_backend.table_exists(table_name):
|
||
|
|
sql_backend.delete_table(table_name)
|
||
|
|
sql_backend.create_table(table_name, fields)
|
||
|
|
persons = [
|
||
|
|
{
|
||
|
|
'family_name': 'Dupont',
|
||
|
|
'birth_area_code': 35,
|
||
|
|
'height': 1.75,
|
||
|
|
'birth_date': '1950-11-12 01:23:45'
|
||
|
|
},
|
||
|
|
{
|
||
|
|
'family_name': 'Dupond',
|
||
|
|
'birth_area_code': 44,
|
||
|
|
'height': 1.74,
|
||
|
|
'birth_date': '1951+12-13 12:34:50'
|
||
|
|
}
|
||
|
|
]
|
||
|
|
for person in persons:
|
||
|
|
sql_query = f'insert into {table_name}({", ".join([field.name for field in fields])}) values({", ".join([stringify(person[field.name]) for field in fields])});'
|
||
|
|
logging.debug('sql_query = %s', sql_query)
|
||
|
|
sql_backend.query(sql_query)
|
||
|
|
sql_backend.dump(Path('/tmp/toto.sql'))
|
||
|
|
|
||
|
|
|
||
|
|
class SimpadbTestCase(unittest.TestCase):
|
||
|
|
|
||
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
||
|
|
|
||
|
|
def setUp(self) -> None:
|
||
|
|
return super().setUp()
|
||
|
|
|
||
|
|
def test_sqlite_backend(self):
|
||
|
|
logging.info('test_sqlite_backend')
|
||
|
|
backend = SqliteDb(Path('/tmp/toto.sqlite'))
|
||
|
|
test_sql_backend(backend)
|
||
|
|
# self.assertIsInstance(job_state, JobsState)
|
||
|
|
|
||
|
|
def test_ssh_accessed_mysql_backend(self):
|
||
|
|
logging.info('test_ssh_accessed_mysql_backend')
|
||
|
|
db_server_fqdn = 'iprbenchsdb.ipr.univ-rennes1.fr'
|
||
|
|
db_user = 'test_iprbenchw'
|
||
|
|
db_name = 'test_iprbenchs'
|
||
|
|
ssh_user = 'test_iprbenchw'
|
||
|
|
|
||
|
|
backend = SshAccessedMysqlDb(db_server_fqdn, db_user, db_name, ssh_user)
|
||
|
|
test_sql_backend(backend)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
unittest.main()
|