s_ranjbar/summer_course_2024
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Correct unittests

Browse Source
This commit is contained in:
Guille Gutierrez 2024-04-16 07:01:51 +02:00
parent b52e66c263
commit d9a941a97a
3 changed files with 153 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
requirements.txt
View File

@ -24,3 +24,4 @@ triangle
psycopg2-binary
Pillow
pathlib
sqlalchemy_utils

81
tests/test_db_factory.py
View File

@ -4,18 +4,18 @@ SPDX - License - Identifier: LGPL - 3.0 - or -later
Copyright © 2022 Concordia CERC group
Project Coder Peter Yefi peteryefi@gmail.com
"""
import distutils.spawn
import distutils
import glob
import logging
import os
import subprocess
import unittest
from pathlib import Path
from unittest import TestCase
import sqlalchemy
import sqlalchemy.exc
from sqlalchemy import create_engine
from sqlalchemy.exc import ProgrammingError
from sqlalchemy_utils import database_exists, create_database, drop_database
import hub.helpers.constants as cte
from hub.exports.energy_building_exports_factory import EnergyBuildingsExportsFactory
@ -28,8 +28,8 @@ from hub.imports.results_factory import ResultFactory
from hub.imports.usage_factory import UsageFactory
from hub.imports.weather_factory import WeatherFactory
from hub.persistence.db_control import DBControl
from hub.persistence.models import City, Application, CityObject, SimulationResults
from hub.persistence.models import User, UserRoles
from hub.persistence.models import City, Application, CityObject, SimulationResults, UserRoles
from hub.persistence.models import User
from hub.persistence.repository import Repository
@ -51,27 +51,29 @@ class Control:
self._skip_reason = f'.env file missing at {dotenv_path}'
return
dotenv_path = str(dotenv_path)
repository = Repository(db_name='test_db', app_env='TEST', dotenv_path=dotenv_path)
repository = Repository(db_name='persistence_test_db', app_env='TEST', dotenv_path=dotenv_path)
engine = create_engine(repository.configuration.connection_string)
try:
# delete test database if it exists
connection = engine.connect()
connection.close()
except ProgrammingError:
logging.info('Database does not exist. Nothing to delete')
except sqlalchemy.exc.OperationalError as operational_error:
self._skip_test = True
self._skip_reason = f'{operational_error}'
return
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)
Application.__table__.create(bind=repository.engine, checkfirst=True)
User.__table__.create(bind=repository.engine, checkfirst=True)
City.__table__.create(bind=repository.engine, checkfirst=True)
CityObject.__table__.create(bind=repository.engine, checkfirst=True)
SimulationResults.__table__.create(bind=repository.engine, checkfirst=True)
self._database = DBControl(
db_name=repository.configuration.db_name,
app_env='TEST',
dotenv_path=dotenv_path)
city_file = Path('tests_data/test.geojson').resolve()
output_path = Path('tests_outputs/').resolve()
example_path = (Path(__file__).parent / 'tests_data').resolve()
city_file = (example_path / 'test.geojson').resolve()
output_path = (Path(__file__).parent / 'tests_outputs').resolve()
self._application_uuid = '60b7fc1b-f389-4254-9ffd-22a4cf32c7a3'
self._application_id = 1
self._user_id = 1
self._pickle_path = 'tests_data/pickle_path.bz2'
self._city = GeometryFactory('geojson',
city_file,
height_field='citygml_me',
@ -105,15 +107,23 @@ class Control:
self._application_uuid = 'b9e0ce80-1218-410c-8a64-9d9b7026aad8'
self._application_id = 1
self._user_id = 1
try:
self._application_id = self._database.persist_application(
'test',
'test',
self.application_uuid
)
self._user_id = self._database.create_user('test', self._application_id, 'test', UserRoles.Admin)
except sqlalchemy.exc.SQLAlchemyError:
self._application_id = self._database.application_info(self.application_uuid).id
self._pickle_path = Path('tests_data/pickle_path.bz2').resolve()
try:
self._user_id = self._database.create_user('test', self._application_id, 'test', UserRoles.Admin)
except sqlalchemy.exc.SQLAlchemyError:
self._user_id = self._database.user_info(name='test', password='test', application_id=self._application_id).id
self._pickle_path = (example_path / 'pickle_path.bz2').resolve()
self._city
@property
def database(self):
@ -155,14 +165,14 @@ class Control:
def message(self):
return self._skip_reason
@property
def city(self):
return self._city
@property
def pickle_path(self):
return self._pickle_path
@property
def city(self):
return self._city
control = Control()
@ -210,6 +220,9 @@ TestDBFactory
control.application_id,
control.user_id)
city_objects_id = []
request_values = {
'scenarios': [{'current status': ['1']}]
}
for building in control.city.buildings:
_building = control.database.building_info(building.name, city_id)
if cte.MONTH not in building.cooling_demand:
@ -286,6 +299,11 @@ TestDBFactory
self.assertEqual(17, len(city_objects_id), 'wrong number of results')
self.assertIsNotNone(city_objects_id[0], 'city_object_id is None')
results = control.database.results(control.user_id, control.application_id, request_values)
self.assertEqual(
28,
len(results['current status'][0]['insel meb'].keys()),
'wrong number of results after retrieve'
)
for _id in city_objects_id:
control.database.delete_results_by_name('insel meb', city_object_id=_id)
@ -297,3 +315,6 @@ TestDBFactory
control.database.delete_application(control.application_uuid)
control.database.delete_user(control.user_id)
os.unlink(control.pickle_path)
output_files = glob.glob('./tests_outputs/*')
for output_file in output_files:
os.unlink(output_file)

113
tests/test_db_retrieve.py
View File

@ -5,17 +5,31 @@ Copyright © 2022 Concordia CERC group
Project Coder Peter Yefi peteryefi@gmail.com
"""
import distutils.spawn
import logging
import glob
import os
import subprocess
import unittest
from datetime import datetime
from pathlib import Path
from unittest import TestCase
import sqlalchemy
import sqlalchemy.exc
from hub.exports.energy_building_exports_factory import EnergyBuildingsExportsFactory
from hub.exports.exports_factory import ExportsFactory
from hub.helpers.data.montreal_function_to_hub_function import MontrealFunctionToHubFunction
from hub.imports.construction_factory import ConstructionFactory
from hub.imports.energy_systems_factory import EnergySystemsFactory
from hub.imports.geometry_factory import GeometryFactory
from hub.imports.results_factory import ResultFactory
from hub.imports.usage_factory import UsageFactory
from hub.imports.weather_factory import WeatherFactory
from sqlalchemy import create_engine
from sqlalchemy.exc import ProgrammingError
from sqlalchemy_utils import database_exists, create_database, drop_database
from hub.persistence.db_control import DBControl
from hub.persistence.models import City, Application, CityObject, SimulationResults, UserRoles
from hub.persistence.models import User
from hub.persistence.repository import Repository
@ -37,28 +51,79 @@ class Control:
self._skip_reason = f'.env file missing at {dotenv_path}'
return
dotenv_path = str(dotenv_path)
repository = Repository(db_name='montreal_retrofit_test', app_env='TEST', dotenv_path=dotenv_path)
repository = Repository(db_name='persistence_test_db', app_env='TEST', dotenv_path=dotenv_path)
engine = create_engine(repository.configuration.connection_string)
try:
# delete test database if it exists
connection = engine.connect()
connection.close()
except ProgrammingError:
logging.info('Database does not exist. Nothing to delete')
except sqlalchemy.exc.OperationalError as operational_error:
self._skip_test = True
self._skip_reason = f'{operational_error}'
return
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)
example_path = (Path(__file__).parent / 'tests_data').resolve()
city_file = (example_path / 'test.geojson').resolve()
output_path = (Path(__file__).parent / 'tests_outputs').resolve()
self._application_uuid = '60b7fc1b-f389-4254-9ffd-22a4cf32c7a3'
self._application_id = 1
self._user_id = 1
self._pickle_path = 'tests_data/pickle_path.bz2'
self._city = GeometryFactory('geojson',
city_file,
height_field='citygml_me',
year_of_construction_field='ANNEE_CONS',
aliases_field=['ID_UEV', 'CIVIQUE_DE', 'NOM_RUE'],
function_field='CODE_UTILI',
function_to_hub=MontrealFunctionToHubFunction().dictionary).city
ConstructionFactory('nrcan', self._city).enrich()
UsageFactory('nrcan', self._city).enrich()
WeatherFactory('epw', self._city).enrich()
ExportsFactory('sra', self._city, output_path).export()
sra_file = str((output_path / f'{self._city.name}_sra.xml').resolve())
subprocess.run([self.sra, sra_file], stdout=subprocess.DEVNULL)
ResultFactory('sra', self._city, output_path).enrich()
for building in self._city.buildings:
building.energy_systems_archetype_name = 'system 1 gas pv'
EnergySystemsFactory('montreal_custom', self._city).enrich()
EnergyBuildingsExportsFactory('insel_monthly_energy_balance', self._city, output_path).export()
_insel_files = glob.glob(f'{output_path}/*.insel')
for insel_file in _insel_files:
subprocess.run([self.insel, str(insel_file)], stdout=subprocess.DEVNULL)
ResultFactory('insel_monthly_energy_balance', self._city, output_path).enrich()
self._database = DBControl(
db_name=repository.configuration.db_name,
app_env='TEST',
dotenv_path=dotenv_path)
self._application_uuid = 'b9e0ce80-1218-410c-8a64-9d9b7026aad8'
self._application_id = 1
self._user_id = 1
try:
self._application_id = self._database.persist_application(
'test',
'test',
self.application_uuid
)
except sqlalchemy.exc.SQLAlchemyError:
self._application_id = self._database.application_info(self.application_uuid).id
try:
self._user_id = self._database.create_user('test', self._application_id, 'test', UserRoles.Admin)
except sqlalchemy.exc.SQLAlchemyError:
self._user_id = self._database.user_info(name='test', password='test', application_id=self._application_id).id
self._pickle_path = (example_path / 'pickle_path.bz2').resolve()
self._city
@property
def database(self):
@ -104,6 +169,10 @@ class Control:
def pickle_path(self):
return self._pickle_path
@property
def city(self):
return self._city
control = Control()
@ -114,12 +183,18 @@ TestDBFactory
"""
@unittest.skipIf(control.skip_test, control.skip_reason)
def test_buildings_info(self):
def test_retrieve_results(self):
datetime.now()
request_values = {
"buildings": [
"01002777", "01002773", "01036804"
"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.buildings_info(control.user_id, control.application_id, request_values)
self.assertEqual(12, len(results), 'wrong number of results')
print(results)
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')