mirror of
https://github.com/fhswf/aki_prj23_transparenzregister.git
synced 2025-04-25 07:52:35 +02:00
68 lines
2.1 KiB
Python
68 lines
2.1 KiB
Python
"""Test if the sql db can be copied."""
|
|
import os
|
|
import sys
|
|
from collections.abc import Generator
|
|
|
|
import pandas as pd
|
|
import pytest
|
|
from _pytest.monkeypatch import MonkeyPatch
|
|
from sqlalchemy.engine import Engine
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
from aki_prj23_transparenzregister.config.config_template import SQLiteConnectionString
|
|
from aki_prj23_transparenzregister.utils.sql.connector import (
|
|
Base,
|
|
get_engine,
|
|
init_db,
|
|
)
|
|
from aki_prj23_transparenzregister.utils.sql.copy_sql import (
|
|
copy_db_cli,
|
|
transfer_db_function,
|
|
)
|
|
|
|
|
|
@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 = sessionmaker(
|
|
autocommit=False,
|
|
autoflush=False,
|
|
bind=get_engine(SQLiteConnectionString("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_function(source=full_db, destination=destination_db)
|
|
assert Base.metadata.sorted_tables
|
|
for table in Base.metadata.sorted_tables + ["company"]:
|
|
pd.testing.assert_frame_equal(
|
|
pd.read_sql_table(str(table), destination_db.connection()),
|
|
pd.read_sql_table(str(table), full_db.connection()),
|
|
)
|
|
|
|
|
|
def test_copy_db_cli_help1(monkeypatch: MonkeyPatch) -> None:
|
|
"""Tests if the help argument exits the software gracefully."""
|
|
with monkeypatch.context() as m, pytest.raises(SystemExit): # noqa: PT012
|
|
m.setattr(sys, "argv", [sys.argv[0], "-h"])
|
|
copy_db_cli()
|
|
|
|
|
|
def test_copy_db_cli_help2(monkeypatch: MonkeyPatch) -> None:
|
|
"""Tests if the help argument exits the software gracefully."""
|
|
with monkeypatch.context() as m, pytest.raises(SystemExit): # noqa: PT012
|
|
m.setattr(sys, "argv", [sys.argv[0], "eskse", "-h", "asdf"])
|
|
copy_db_cli()
|