"""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) sbind = full_db.bind dbind = destination_db.bind assert isinstance(sbind, Engine) assert isinstance(dbind, Engine) assert Base.metadata.sorted_tables for table in Base.metadata.sorted_tables + ["company"]: pd.testing.assert_frame_equal( pd.read_sql_table(str(table), dbind), pd.read_sql_table(str(table), sbind), ) 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()