cocluto/test/test_simpadb.py

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()