64 lines
2.5 KiB
Python
64 lines
2.5 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 = 'colutotestmamul1'
|
|
fields = [
|
|
SqlTableField('measure_id', SqlTableField.Type.FIELD_TYPE_INT, 'unique identifier of the measurement', is_autoinc_index=True),
|
|
SqlTableField('measurement_time', SqlTableField.Type.FIELD_TYPE_TIME, 'the time (and date) at which this measurment has been made'),
|
|
SqlTableField('cpu_model', SqlTableField.Type.FIELD_TYPE_STRING, 'The exact model of the cpu running the benchmark eg "Intel(R) Core(TM) i5-8350U CPU @ 1.70GHz"'),
|
|
SqlTableField('duration', SqlTableField.Type.FIELD_TYPE_FLOAT, 'the duration of the benchmark (in seconds)'),
|
|
]
|
|
if sql_backend.table_exists(table_name):
|
|
sql_backend.delete_table(table_name)
|
|
sql_backend.create_table(table_name, fields)
|
|
measurements = [
|
|
{
|
|
'measurement_time': '1951+12-13 12:34:50',
|
|
'cpu_model': 'Intel(R) Core(TM) i5-8350U CPU @ 1.70GHz',
|
|
'duration': 0.42
|
|
}
|
|
]
|
|
for measurement in measurements:
|
|
sql_query = f'insert into {table_name}({", ".join([field.name for field in fields if not field.is_autoinc_index])}) values({", ".join([stringify(measurement[field.name]) for field in fields if not field.is_autoinc_index])});'
|
|
logging.debug('sql_query = %s', sql_query)
|
|
sql_backend.query(sql_query)
|
|
sql_backend.dump(Path('/tmp/toto.sql'))
|
|
sql_backend.delete_table(table_name)
|
|
|
|
|
|
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()
|