CERC/cerc_persistence
19
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 7 Wiki Activity

remove test_db_retrieve

Browse Source
This commit is contained in:
Ruben Sanchez 2023-12-06 11:47:27 -05:00
parent 5bb032500c
commit f49e2f3afa
4 changed files with 58 additions and 209 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cerc_persistence/configuration.py
View File

@ -16,7 +16,7 @@ Models = declarative_base()
class Configuration:
"""
Configuration class to hold common cerc_persistence configuration
Configuration class to hold common persistence configuration
"""
def __init__(self, db_name: str, dotenv_path: str, app_env='TEST'):

2
cerc_persistence/db_setup.py
View File

@ -48,7 +48,7 @@ class DBSetup:
@staticmethod
def _create_admin_app(application_repo, application_uuid):
name = 'AdminTool'
description = 'Admin tool to control city cerc_persistence and to test the API v1.4'
description = 'Admin tool to control city persistence and to test the API v1.4'
logging.info('Creating default admin tool application...')
application = application_repo.insert(name, description, application_uuid)

126
tests/test_db_factory.py
View File

@ -17,6 +17,7 @@ from unittest import TestCase
import sqlalchemy.exc
from sqlalchemy import create_engine
from sqlalchemy.exc import ProgrammingError
from sqlalchemy.ext.declarative import declarative_base
import hub.helpers.constants as cte
from hub.exports.energy_building_exports_factory import EnergyBuildingsExportsFactory
@ -45,17 +46,19 @@ class Control:
:return: None
"""
self._skip_test = False
# Create test database
# 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)
# Check if database exists.
repository = Repository(db_name='test_db', app_env='TEST', dotenv_path=dotenv_path)
engine = create_engine(repository.configuration.connection_string)
try:
# delete test database if it exists
engine = create_engine(repository.configuration.connection_string)
connection = engine.connect()
connection.close()
except ProgrammingError:
@ -65,12 +68,14 @@ class Control:
self._skip_reason = f'{operational_error}'
return
# Create tables if they don't exist.
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)
# Generate files for the persistence tests.
city_file = (Path(__file__).parent / 'tests_data/test.geojson').resolve()
output_path = (Path(__file__).parent / 'tests_outputs').resolve()
self._city = GeometryFactory('geojson',
@ -98,15 +103,13 @@ class Control:
subprocess.run([self.insel, str(insel_file)], stdout=subprocess.DEVNULL)
ResultFactory('insel_monthly_energy_balance', self._city, output_path).enrich()
# Create test application and user.
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
self._application_id = self._database.persist_application(
'test',
'test',
@ -186,6 +189,7 @@ TestDBFactory
@unittest.skipIf(control.skip_test, control.skip_reason)
def test_get_update_city(self):
# Create and update city name
city_id = control.database.persist_city(control.city,
control.pickle_path,
control.city.name,
@ -193,101 +197,83 @@ TestDBFactory
control.user_id)
control.city.name = "Ottawa"
control.database.update_city(city_id, control.city)
# Check the city was updated
cities = control.database.cities_by_user_and_application(
control.user_id,
control.application_id)
control.application_id) # TODO: woudln't this be better wih a search by city id?
for updated_city in cities:
if updated_city.id == city_id:
self.assertEqual(updated_city.name, control.city.name)
break
# Delete extra city created
control.database.delete_city(city_id)
@unittest.skipIf(control.skip_test, control.skip_reason)
@unittest.skipIf(control.skip_insel_test, 'insel is not installed')
def test_save_results(self):
# Create city
city_id = control.database.persist_city(control.city,
control.pickle_path,
'current status',
control.application_id,
control.user_id)
# Create city objects
city_objects_id = []
for building in control.city.buildings:
_building = control.database.building_info(building.name, city_id)
building_info = control.database.building_info(building.name, city_id)
if cte.MONTH not in building.cooling_demand:
print(f'building {building.name} not calculated')
continue
monthly_cooling_peak_load = building.cooling_peak_load[cte.MONTH]
yearly_cooling_peak_load = building.cooling_peak_load[cte.YEAR]
monthly_heating_peak_load = building.heating_peak_load[cte.MONTH]
yearly_heating_peak_load = building.heating_peak_load[cte.YEAR]
monthly_lighting_peak_load = building.lighting_peak_load[cte.MONTH]
yearly_lighting_peak_load = building.lighting_peak_load[cte.YEAR]
monthly_appliances_peak_load = building.appliances_peak_load[cte.MONTH]
yearly_appliances_peak_load = building.appliances_peak_load[cte.YEAR]
monthly_cooling_demand = building.cooling_demand[cte.MONTH]
yearly_cooling_demand = building.cooling_demand[cte.YEAR]
monthly_heating_demand = building.heating_demand[cte.MONTH]
yearly_heating_demand = building.heating_demand[cte.YEAR]
monthly_lighting_electrical_demand = building.lighting_electrical_demand[cte.MONTH]
yearly_lighting_electrical_demand = building.lighting_electrical_demand[cte.YEAR]
monthly_appliances_electrical_demand = building.appliances_electrical_demand[cte.MONTH]
yearly_appliances_electrical_demand = building.appliances_electrical_demand[cte.YEAR]
monthly_domestic_hot_water_heat_demand = building.domestic_hot_water_heat_demand[cte.MONTH]
yearly_domestic_hot_water_heat_demand = building.domestic_hot_water_heat_demand[cte.YEAR]
monthly_heating_consumption = building.heating_consumption[cte.MONTH]
yearly_heating_consumption = building.heating_consumption[cte.YEAR]
monthly_cooling_consumption = building.cooling_consumption[cte.MONTH]
yearly_cooling_consumption = building.cooling_consumption[cte.YEAR]
monthly_domestic_hot_water_consumption = building.domestic_hot_water_consumption[cte.MONTH]
yearly_domestic_hot_water_consumption = building._domestic_hot_water_consumption[cte.YEAR]
monthly_distribution_systems_electrical_consumption = building.distribution_systems_electrical_consumption[
cte.MONTH]
yearly_distribution_systems_electrical_consumption = building.distribution_systems_electrical_consumption[
cte.YEAR]
monthly_on_site_electrical_production = [x * cte.WATTS_HOUR_TO_JULES
for x in building.onsite_electrical_production[cte.MONTH]]
yearly_on_site_electrical_production = [x * cte.WATTS_HOUR_TO_JULES
for x in building.onsite_electrical_production[cte.YEAR]]
demands = ['cooling_demand', 'heating_demand', 'lighting_electrical_demand',
'appliances_electrical_demand', 'domestic_hot_water_heat_demand',
'heating_consumption', 'cooling_consumption',
'domestic_hot_water_consumption', 'distribution_systems_electrical_consumption']
results = {cte.INSEL_MEB: {
'monthly_cooling_peak_load': monthly_cooling_peak_load,
'yearly_cooling_peak_load': yearly_cooling_peak_load,
'monthly_heating_peak_load': monthly_heating_peak_load,
'yearly_heating_peak_load': yearly_heating_peak_load,
'monthly_lighting_peak_load': monthly_lighting_peak_load,
'yearly_lighting_peak_load': yearly_lighting_peak_load,
'monthly_appliances_peak_load': monthly_appliances_peak_load,
'yearly_appliances_peak_load': yearly_appliances_peak_load,
'monthly_cooling_demand': monthly_cooling_demand,
'yearly_cooling_demand': yearly_cooling_demand,
'monthly_heating_demand': monthly_heating_demand,
'yearly_heating_demand': yearly_heating_demand,
'monthly_lighting_electrical_demand': monthly_lighting_electrical_demand,
'yearly_lighting_electrical_demand': yearly_lighting_electrical_demand,
'monthly_appliances_electrical_demand': monthly_appliances_electrical_demand,
'yearly_appliances_electrical_demand': yearly_appliances_electrical_demand,
'monthly_domestic_hot_water_heat_demand': monthly_domestic_hot_water_heat_demand,
'yearly_domestic_hot_water_heat_demand': yearly_domestic_hot_water_heat_demand,
'monthly_heating_consumption': monthly_heating_consumption,
'yearly_heating_consumption': yearly_heating_consumption,
'monthly_cooling_consumption': monthly_cooling_consumption,
'yearly_cooling_consumption': yearly_cooling_consumption,
'monthly_domestic_hot_water_consumption': monthly_domestic_hot_water_consumption,
'yearly_domestic_hot_water_consumption': yearly_domestic_hot_water_consumption,
'monthly_distribution_systems_electrical_consumption': monthly_distribution_systems_electrical_consumption,
'yearly_distribution_systems_electrical_consumption': yearly_distribution_systems_electrical_consumption,
'monthly_on_site_electrical_production': monthly_on_site_electrical_production,
'yearly_on_site_electrical_production': yearly_on_site_electrical_production
f'{period}_{demand}': getattr(building, demand)[cte.MONTH if 'monthly' in period else cte.YEAR]
for demand in demands for period in ['monthly', 'yearly']
}}
db_building_id = _building.id
results[cte.INSEL_MEB].update({
'monthly_on_site_electrical_production': [x * cte.WATTS_HOUR_TO_JULES for x in
building.onsite_electrical_production[cte.MONTH]],
'yearly_on_site_electrical_production': [x * cte.WATTS_HOUR_TO_JULES for x in
building.onsite_electrical_production[cte.YEAR]]
})
db_building_id = building_info.id
city_objects_id.append(db_building_id)
control.database.add_simulation_results(
cte.INSEL_MEB,
results, city_object_id=db_building_id)
# Verify 17 city objects have been created
self.assertEqual(17, len(city_objects_id), 'wrong number of results')
self.assertIsNotNone(city_objects_id[0], 'city_object_id is None')
# Verify results have been saved
# TODO: How do I get city object ids?
request_values = {
"scenarios": [
{
"insel meb": ["01002777", "01002773", "01036804"]
},
{
"skin retrofit": ["01002777", "01002773", "01036804"]
},
]
}
results = control.database.results(control.user_id, control.application_id, request_values)
print(results)
for _id in city_objects_id:
control.database.delete_results_by_name('insel meb', city_object_id=_id)
control.database.delete_city(city_id)
@classmethod

137
tests/test_db_retrieve.py
View File

@ -1,137 +0,0 @@
"""
Test db factory
SPDX - License - Identifier: LGPL - 3.0 - or -later
Copyright © 2022 Concordia CERC group
Project Coder Peter Yefi peteryefi@gmail.com
"""
import distutils.spawn
import logging
import os
import unittest
from pathlib import Path
from unittest import TestCase
import sqlalchemy.exc
from sqlalchemy import create_engine
from sqlalchemy.exc import ProgrammingError
from cerc_persistence.db_control import DBControl
from cerc_persistence.repository import Repository
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='montreal_retrofit_test', 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
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 = (Path(__file__).parent / 'tests_data/pickle_path.bz2').resolve()
@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):
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)
print(results)