mirror of
https://github.com/fhswf/aki_prj23_transparenzregister.git
synced 2025-04-22 22:02:54 +02:00
Philipp Horstenkamp
ba44b082b8
Transfers data betwenn two sql instances. Limited in data volume. Should be good enough for now. --------- Co-authored-by: Tim <tim.ronneburg@outlook.de>
117 lines
3.6 KiB
Python
117 lines
3.6 KiB
Python
"""Tests the sql connector."""
|
|
import os.path
|
|
from collections.abc import Generator
|
|
from typing import Any
|
|
from unittest.mock import Mock, patch
|
|
|
|
import pandas as pd
|
|
import pytest
|
|
from sqlalchemy.engine import Engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from aki_prj23_transparenzregister.config.config_providers import JsonFileConfigProvider
|
|
from aki_prj23_transparenzregister.config.config_template import PostgreConnectionString
|
|
from aki_prj23_transparenzregister.utils.sql.connector import (
|
|
Base,
|
|
get_pg_engine,
|
|
get_session,
|
|
init_db,
|
|
transfer_db,
|
|
)
|
|
|
|
|
|
def test_get_engine_pg() -> None:
|
|
"""Tests the creation of a postgre engine."""
|
|
conn_args = PostgreConnectionString("", "", "", "", 42)
|
|
with patch(
|
|
"aki_prj23_transparenzregister.utils.sql.connector.sa.create_engine"
|
|
) as mock_create_engine:
|
|
result = "someThing"
|
|
mock_create_engine.return_value = result
|
|
assert get_pg_engine(conn_args) == result
|
|
|
|
|
|
@pytest.fixture()
|
|
def destination_db() -> Generator[Session, None, None]:
|
|
"""Generates a db Session to a sqlite db to copy data to."""
|
|
if os.path.exists("secondary.db"):
|
|
os.remove("secondary.db")
|
|
db = get_session("sqlite:///secondary.db")
|
|
init_db(db)
|
|
yield db
|
|
db.close()
|
|
bind = db.bind
|
|
assert isinstance(bind, Engine)
|
|
bind.dispose()
|
|
os.remove("secondary.db")
|
|
|
|
|
|
def test_transfer_db(full_db: Session, destination_db: Session) -> None:
|
|
"""Tests if the data transfer between two sql tables works."""
|
|
transfer_db(source=full_db, destination=destination_db)
|
|
sbind = full_db.bind
|
|
dbind = destination_db.bind
|
|
assert isinstance(sbind, Engine)
|
|
assert isinstance(dbind, Engine)
|
|
|
|
for table in Base.metadata.sorted_tables:
|
|
pd.testing.assert_frame_equal(
|
|
pd.read_sql_table(str(table), dbind),
|
|
pd.read_sql_table(str(table), sbind),
|
|
)
|
|
|
|
|
|
@pytest.fixture()
|
|
def delete_sqlite_table() -> Generator[str, None, None]:
|
|
"""Cleans a path before and deletes the table after a test.
|
|
|
|
Returns:
|
|
The path where the sqlite table is placed.
|
|
"""
|
|
sqlite_test_path = "test_db.db"
|
|
if os.path.exists(sqlite_test_path):
|
|
os.remove(sqlite_test_path)
|
|
|
|
yield sqlite_test_path
|
|
|
|
if os.path.exists(sqlite_test_path):
|
|
os.remove(sqlite_test_path)
|
|
|
|
|
|
def test_get_sqlite_init(delete_sqlite_table: str) -> None:
|
|
"""Tests if a sql table file can be initiated."""
|
|
assert not os.path.exists(delete_sqlite_table)
|
|
session = get_session(f"sqlite:///{delete_sqlite_table}")
|
|
init_db(session)
|
|
session.close()
|
|
engine = session.bind
|
|
assert isinstance(engine, Engine)
|
|
engine.dispose()
|
|
assert os.path.exists(delete_sqlite_table)
|
|
|
|
|
|
@pytest.mark.parametrize("connection", ["faulty-name", 0, 9.2, True])
|
|
def test_get_invalid_connection(connection: Any) -> None:
|
|
"""Tests if an error is thrown on a faulty connections."""
|
|
with pytest.raises(TypeError):
|
|
get_session(connection)
|
|
|
|
|
|
def test_init_pd_db() -> None:
|
|
"""Tests if a pg sql database can be connected and initiated to."""
|
|
with patch(
|
|
"aki_prj23_transparenzregister.utils.sql.connector.get_pg_engine"
|
|
) as mock_get_engine, patch(
|
|
"aki_prj23_transparenzregister.utils.sql.connector.declarative_base"
|
|
) as mock_declarative_base:
|
|
mock_get_engine.connect.return_value = {}
|
|
|
|
mock_value = Mock()
|
|
mock_value.metadata.create_all.return_value = None
|
|
mock_declarative_base.return_value = mock_value
|
|
|
|
mock_value = Mock(spec=JsonFileConfigProvider)
|
|
mock_value.get_postgre_connection_string.return_value = ""
|
|
|
|
init_db(get_session(mock_value))
|