city_retrofit/hub/unittests/test_db_factory.py

"""
Test db factory
SPDX - License - Identifier: LGPL - 3.0 - or -later
Copyright © 2022 Concordia CERC group
Project Coder Peter Yefi peteryefi@gmail.com
"""
import os
import unittest
from unittest import TestCase
from pathlib import Path
import sqlalchemy.exc

from hub.imports.geometry_factory import GeometryFactory
from hub.imports.db_factory import DBFactory as ImportDBFactory
from hub.imports.user_factory import UserFactory
from hub.exports.db_factory import DBFactory as ExportDBFactory
from hub.persistence.repository import Repository
from sqlalchemy import create_engine
from hub.persistence.models import City, Application, CityObject
from hub.persistence.models import User, UserRoles
from sqlalchemy.exc import ProgrammingError
import uuid

class Configure:

  _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 = str(Path("{}/.local/etc/hub/.env".format(os.path.expanduser('~'))).resolve())
    repository = Repository(db_name='hub_unittest', 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.execute('commit')
      connection.close()
    except ProgrammingError:
      print(f'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

    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)

    city_file = "tests_data/FZK_Haus_LoD_2.gml"
    self._city = GeometryFactory('citygml', city_file).city
    self._import_db_factory = ImportDBFactory(
      db_name=repository.configuration.db_name,
      app_env='TEST',
      dotenv_path=dotenv_path)
    self._export_db_factory = ExportDBFactory(
      db_name=repository.configuration.db_name,
      app_env='TEST',
      dotenv_path=dotenv_path)
    user_factory = UserFactory(
      db_name=repository.configuration.db_name,
      app_env='TEST',
      dotenv_path=dotenv_path)
    self._unique_id = str(uuid.uuid4())
    self._application = self.import_db_factory.persist_application("test", "test application", self.unique_id)
    self._user = user_factory.create_user("Admin", self.application.id, "Admin@123", UserRoles.Admin)
    self._pickle_path = 'tests_data/pickle_path.bz2'

  @property
  def import_db_factory(self):
    return self._import_db_factory

  @property
  def export_db_factory(self):
    return self._export_db_factory

  @property
  def unique_id(self):
    return self._unique_id

  @property
  def application(self):
    return self._application

  @property
  def user(self):
    return self._user

  @property
  def skip_test(self):
    return self._skip_test

  @property
  def skip_reason(self):
    return self._skip_reason

  @property
  def message(self):
    return self._skip_message

  @property
  def city(self):
    return self._city

  @property
  def pickle_path(self):
    return self._pickle_path

configure = Configure()

class TestDBFactory(TestCase):
  """
  TestDBFactory
  """

  @unittest.skipIf(configure.skip_test, configure.skip_reason)
  def test_save_application(self):
    self.assertEqual(configure.application.name, "test")
    self.assertEqual(configure.application.description, "test application")
    self.assertEqual(str(configure.application.application_uuid), configure.unique_id)

  @unittest.skipIf(configure.skip_test, configure.skip_reason)
  def test_save_city(self):
    configure.city.name = "Montreal"
    saved_city = configure.import_db_factory.persist_city(
      configure.city,
      configure.pickle_path,
      configure.application.id,
      configure.user.id)
    self.assertEqual(saved_city.name, 'Montreal')
    self.assertEqual(saved_city.pickle_path, self.pickle_path)
    self.assertEqual(saved_city.level_of_detail, self.city.level_of_detail.geometry)
    self._db_factory.delete_city(saved_city.id)

  @unittest.skipIf(configure.skip_test, configure.skip_reason)
  def test_get_city_by_name(self):
    city = self._db_factory.persist_city(self.city, self.pickle_path, self.application.id, self._user.id)
    retrieved_city = self._export_db_factory.get_city_by_name(city.name)
    self.assertEqual(retrieved_city[0].application_id, 1)
    self.assertEqual(retrieved_city[0].user_id, self._user.id)
    self._db_factory.delete_city(city.id)

  @unittest.skipIf(configure.skip_test, configure.skip_reason)
  def test_get_city_by_user(self):
    city = self._import_db_factory.persist_city(self.city, self.pickle_path, self.application.id, self._user.id)
    retrieved_city = self._export_db_factory.get_city_by_user(self._user.id)
    self.assertEqual(retrieved_city[0].pickle_path, self.pickle_path)
    self._db_factory.delete_city(city.id)

  @unittest.skipIf(configure.skip_test, configure.skip_reason)
  def test_get_city_by_id(self):
    city = self._db_factory.persist_city(self.city, self.pickle_path, self.application.id, self._user.id)
    retrieved_city = self._export_db_factory.get_city(city.id)
    self.assertEqual(retrieved_city.level_of_detail, self.city.level_of_detail.geometry)
    self._db_factory.delete_city(city.id)

  @unittest.skipIf(configure.skip_test, configure.skip_reason)
  def test_get_update_city(self):
    city = self._db_factory.persist_city(self.city, self.pickle_path, self.application.id, self._user.id)
    self.city.name = "Ottawa"
    self._db_factory.update_city(city.id, self.city)
    updated_city = self._export_db_factory.get_city(city.id)
    self.assertEqual(updated_city.name, self.city.name)
    self._db_factory.delete_city(city.id)