mirror of
https://github.com/fhswf/aki_prj23_transparenzregister.git
synced 2025-04-25 08:52:35 +02:00
Philipp Horstenkamp
d2d4a436f8
- [x] add a cli to the webserver to take env variables into account - [x] add a cli to the data processing that takes enviromental variable as a valid source into account - [x] rework the cli for the reset sql command - [x] rework the cli for the copying of sql data from one db to another
72 lines
2.2 KiB
Python
72 lines
2.2 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)
|
|
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()
|