129 lines
3.6 KiB
Python
129 lines
3.6 KiB
Python
"""
|
|
Test db retrieve
|
|
SPDX - License - Identifier: LGPL - 3.0 - or -later
|
|
Copyright © 2022 Concordia CERC group
|
|
Project Coder Ruben Sanchez ruben.sanchez@mail.concordia.ca
|
|
"""
|
|
import datetime
|
|
import distutils.spawn
|
|
import os
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import TestCase
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy_utils import database_exists, create_database, drop_database
|
|
|
|
from cerc_persistence.db_control import DBControl
|
|
from cerc_persistence.repository import Repository
|
|
from cerc_persistence.models import City, Application, CityObject, SimulationResults, User
|
|
|
|
|
|
class Control:
|
|
_skip_test = False
|
|
_skip_reason = 'PostgreSQL not properly installed in host machine'
|
|
|
|
def __init__(self):
|
|
"""
|
|
Test
|
|
setup
|
|
:return: None
|
|
"""
|
|
self._skip_test = False
|
|
# Create test database
|
|
dotenv_path = Path("{}/.local/etc/hub/.env".format(os.path.expanduser('~'))).resolve()
|
|
if not dotenv_path.exists():
|
|
self._skip_test = True
|
|
self._skip_reason = f'.env file missing at {dotenv_path}'
|
|
return
|
|
dotenv_path = str(dotenv_path)
|
|
repository = Repository(db_name='persistence_test_db', app_env='TEST', dotenv_path=dotenv_path)
|
|
engine = create_engine(repository.configuration.connection_string)
|
|
if database_exists(engine.url):
|
|
drop_database(engine.url)
|
|
create_database(engine.url)
|
|
Application.__table__.create(bind=engine, checkfirst=True)
|
|
User.__table__.create(bind=engine, checkfirst=True)
|
|
City.__table__.create(bind=engine, checkfirst=True)
|
|
CityObject.__table__.create(bind=engine, checkfirst=True)
|
|
SimulationResults.__table__.create(bind=engine, checkfirst=True)
|
|
|
|
self._database = DBControl(
|
|
db_name=repository.configuration.db_name,
|
|
app_env='TEST',
|
|
dotenv_path=dotenv_path)
|
|
|
|
self._application_uuid = '60b7fc1b-f389-4254-9ffd-22a4cf32c7a3'
|
|
self._application_id = 1
|
|
self._user_id = 1
|
|
self._pickle_path = 'tests_data/pickle_path.bz2'
|
|
|
|
@property
|
|
def database(self):
|
|
return self._database
|
|
|
|
@property
|
|
def application_uuid(self):
|
|
return self._application_uuid
|
|
|
|
@property
|
|
def application_id(self):
|
|
return self._application_id
|
|
|
|
@property
|
|
def user_id(self):
|
|
return self._user_id
|
|
|
|
@property
|
|
def skip_test(self):
|
|
return self._skip_test
|
|
|
|
@property
|
|
def insel(self):
|
|
return distutils.spawn.find_executable('insel')
|
|
|
|
@property
|
|
def sra(self):
|
|
return distutils.spawn.find_executable('sra')
|
|
|
|
@property
|
|
def skip_insel_test(self):
|
|
return self.insel is None
|
|
|
|
@property
|
|
def skip_reason(self):
|
|
return self._skip_reason
|
|
|
|
@property
|
|
def message(self):
|
|
return self._skip_reason
|
|
|
|
@property
|
|
def pickle_path(self):
|
|
return self._pickle_path
|
|
|
|
|
|
control = Control()
|
|
|
|
|
|
class TestDBFactory(TestCase):
|
|
"""
|
|
TestDBFactory
|
|
"""
|
|
|
|
@unittest.skipIf(control.skip_test, control.skip_reason)
|
|
def test_retrieve_results(self):
|
|
datetime.datetime.now()
|
|
request_values = {
|
|
"scenarios": [
|
|
{"current status": ["01002777", "01002773", "01036804"]},
|
|
{"skin retrofit": ["01002777", "01002773", "01036804"]},
|
|
{"system retrofit and pv": ["01002777", "01002773", "01036804"]},
|
|
{"skin and system retrofit with pv": ["01002777", "01002773", "01036804"]}
|
|
]
|
|
}
|
|
results = control.database.results(control.user_id, control.application_id, request_values)
|
|
scenarios = ['current status', 'skin retrofit', 'system retrofit and pv', 'skin and system retrofit with pv']
|
|
for key, value in results.items():
|
|
self.assertTrue(key in scenarios, 'Wrong key value')
|
|
self.assertEqual(len(value), 0, 'wrong number of results') |