Repaired the SQL copy and reduced the log volume a bit (#141)

- Added a cli interface to the SQL copy
- Repaired the SQL copy function
- Added the SQL copy function to the scripts
- Reduced the logging verbosity

tests/utils/sql/connector_test.py

@@ -4,19 +4,15 @@ 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,
 )
 
 
@@ -31,36 +27,6 @@ def test_get_engine_pg() -> None:
         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.

tests/utils/sql/copy_sql_test.py

@@ -0,0 +1,56 @@
+"""Test if the sql db can be copied."""
+import os
+from collections.abc import Generator
+
+import pandas as pd
+import pytest
+from sqlalchemy.engine import Engine
+from sqlalchemy.orm import Session
+
+from aki_prj23_transparenzregister.utils.sql.connector import Base, get_session, 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 = 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_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() -> None:
+    """Tests if the help argument exits the software gracefully."""
+    with pytest.raises(SystemExit):
+        copy_db_cli(["-h"])
+
+
+def test_copy_db_cli_help2() -> None:
+    """Tests if the help argument exits the software gracefully."""
+    with pytest.raises(SystemExit):
+        copy_db_cli(["eskse", "-h", "asdf"])