Philipp/aki_prj23_transparenzregister
1
0
Fork 0
mirror of https://github.com/fhswf/aki_prj23_transparenzregister.git synced 2025-06-21 23:43:55 +02:00
Code Issues Packages Projects Releases Wiki Activity

Replaced the bind with the connection method

Browse Source
This commit is contained in:
Philipp Horstenkamp
2024-01-15 20:24:38 +01:00
parent 3af9ff58d6
commit 27a95c1c23
7 changed files with 26 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
tests/utils/data_transfer_test.py
View File

@ -11,7 +11,6 @@ import pytest
import sqlalchemy as sa
from _pytest.monkeypatch import MonkeyPatch
from pytest_mock import MockerFixture
from sqlalchemy.engine import Engine
from sqlalchemy.orm import Session
from aki_prj23_transparenzregister.models.company import (
@ -714,10 +713,8 @@ def test_add_relationships_none(empty_relations: list, full_db: Session) -> None
def test_relationships(documents: list[dict[str, Any]], full_db: Session) -> None:
"""Testing to add lots of relations."""
data_transfer.add_relationships(documents, full_db)
bind = full_db.bind
assert isinstance(bind, Engine)
pd.testing.assert_frame_equal(
pd.read_sql_table("company", bind),
pd.read_sql_table("company", full_db.connection()),
pd.DataFrame(
{
"id": {0: 1, 1: 2, 2: 3},
@ -750,13 +747,13 @@ def test_relationships(documents: list[dict[str, Any]], full_db: Session) -> Non
}
),
)
assert len(pd.read_sql_table("company_relation", bind).index) == 0
assert len(pd.read_sql_table("company_relation", full_db.connection()).index) == 0
pd.testing.assert_frame_equal(
pd.read_sql_table("person_relation", bind),
pd.read_sql_table("person_relation", full_db.connection()),
pd.DataFrame({"id": {0: 1, 1: 2}, "person_id": {0: 6, 1: 7}}),
)
pd.testing.assert_frame_equal(
pd.read_sql_table("relation", bind),
pd.read_sql_table("relation", full_db.connection()),
pd.DataFrame(
{
"id": {0: 1, 1: 2},
@ -768,7 +765,7 @@ def test_relationships(documents: list[dict[str, Any]], full_db: Session) -> Non
),
)
pd.testing.assert_frame_equal(
pd.read_sql_table("person", bind),
pd.read_sql_table("person", full_db.connection()),
pd.DataFrame(
{
"id": {0: 1, 1: 2, 2: 3, 3: 4, 4: 5, 5: 6, 6: 7},
@ -976,7 +973,7 @@ def test_add_annual_report_empty(
) -> None:
"""Testing if the correct warning is thrown when the financial and auditor records are empty."""
df_prior = pd.read_sql_table(
entities.AnnualFinanceStatement.__tablename__, full_db.bind # type: ignore
str(entities.AnnualFinanceStatement.__tablename__), full_db.connection() # type: ignore
)
spy_warning = mocker.spy(data_transfer.logger, "debug")
@ -985,7 +982,9 @@ def test_add_annual_report_empty(
spy_warning.assert_called_once()
pd.testing.assert_frame_equal(
df_prior,
pd.read_sql_table(entities.AnnualFinanceStatement.__tablename__, full_db.bind), # type: ignore
pd.read_sql_table(
str(entities.AnnualFinanceStatement.__tablename__), full_db.connection() # type: ignore
),
)
@ -1031,7 +1030,7 @@ def test_add_annual_report(
)
full_db.commit()
df_prior = pd.read_sql_table(
entities.AnnualFinanceStatement.__tablename__, full_db.bind # type: ignore
str(entities.AnnualFinanceStatement.__tablename__), full_db.connection() # type: ignore
)
expected_results = pd.DataFrame(
finance_statements
@ -1158,7 +1157,7 @@ def test_company_relation_missing(empty_db: Session) -> None:
pd.testing.assert_frame_equal(
pd.read_sql_table(
entities.MissingCompany.__tablename__, empty_db.bind # type: ignore
entities.MissingCompany.__tablename__, empty_db.connection()
).set_index("name"),
pd.DataFrame(
[
@ -1204,7 +1203,7 @@ def test_company_relation_missing_reset(empty_db: Session) -> None:
empty_db.commit()
data_transfer.reset_relation_counter(empty_db)
queried_df = pd.read_sql_table(
entities.MissingCompany.__tablename__, empty_db.bind # type: ignore
entities.MissingCompany.__tablename__, empty_db.connection()
).set_index("name")
pd.testing.assert_frame_equal(
queried_df,