aki_prj23_transparenzregister/tests/conftest.py

"""Global configurations and definitions for pytest."""
import datetime
from collections.abc import Generator
from inspect import getmembers, isfunction
from typing import Any

import pytest
from sqlalchemy.orm import Session, sessionmaker

from aki_prj23_transparenzregister.config.config_template import SQLiteConnectionString
from aki_prj23_transparenzregister.models.company import CapitalTypeEnum
from aki_prj23_transparenzregister.utils import data_transfer
from aki_prj23_transparenzregister.utils.sql import entities
from aki_prj23_transparenzregister.utils.sql.connector import (
    get_engine,
    init_db,
)


@pytest.fixture(autouse=True)
def _clear_caches() -> Generator[None, None, None]:
    """A function that clears all caches after each test.

    All the modules containing the cached functions need to be listed in the modules tuple.
    """
    yield
    # https://stackoverflow.com/a/139198/11003343
    modules = (data_transfer,)
    functions = [
        function
        for module in modules
        for name, function in getmembers(module, isfunction)
        if function.__dict__.get("cache") is not None
    ]
    # https://cachetools.readthedocs.io/en/stable/?highlight=clear#memoizing-decorators
    for function in functions:
        function.cache.clear()  # type: ignore


@pytest.fixture()
def empty_db() -> Generator[Session, None, None]:
    """Generates a db Session to a sql_lite db."""
    engine = get_engine(SQLiteConnectionString(":memory:"))
    db = sessionmaker(autocommit=False, autoflush=False, bind=engine)()
    init_db(db)
    yield db
    db.close()
    engine.dispose()


@pytest.fixture()
def finance_statements() -> list[dict[str, Any]]:
    """Creates a list of finance statements."""
    return [
        {
            "id": 1,
            "company_id": 1,
            "date": datetime.date.fromisoformat("2023-01-01"),
            "ebit": 1000.0,
            "ebitda": 1000.0,
            "gross_profit": 1000.0,
            "equity": 1000.0,
            "current_liabilities": 1000.0,
            "assets": 1000.0,
            "net_income": 100.0,
            "long_term_debt": 1000.0,
            "short_term_debt": 1000.0,
            "revenue": 1000.0,
            "cash_flow": 1000.0,
            "current_assets": 1000.0,
            "liabilities": 0.0,
            "cash_and_cash_equivalents": 1.0,
            "dividends": 0.0,
        },
        {
            "id": 2,
            "company_id": 1,
            "date": datetime.date.fromisoformat("2022-01-01"),
            "revenue": 1100.0,
            "net_income": float("NaN"),
            "ebit": 1100.0,
            "ebitda": 1100.0,
            "gross_profit": 1100.0,
            "equity": 1100.0,
            "current_liabilities": 1100.0,
            "dividends": float("NaN"),
            "assets": 1100.0,
            "long_term_debt": 1100.0,
            "short_term_debt": 1100.0,
            "cash_flow": 1100.0,
            "current_assets": 1100.0,
            "operating_profit": 1.0,
        },
    ]


@pytest.fixture()
def full_db(empty_db: Session, finance_statements: list[dict[str, Any]]) -> Session:
    """Fills a db with some test data."""
    empty_db.add_all(
        [
            entities.DistrictCourt(name="Amtsgericht Bochum", city="Bochum"),
            entities.DistrictCourt(name="Amtsgericht Dortmund", city="Dortmund"),
            entities.Person(
                firstname="Max",
                lastname="Mustermann",
                date_of_birth=datetime.date(2023, 1, 1),
            ),
            entities.Person(
                firstname="Sabine",
                lastname="Mustermann",
                date_of_birth=datetime.date(2023, 1, 1),
            ),
            entities.Person(
                firstname="Some Firstname",
                lastname="Some Surname",
                date_of_birth=datetime.date(2023, 1, 1),
            ),
            entities.Person(
                firstname="Some Firstname",
                lastname="Some Surname",
                date_of_birth=datetime.date(2023, 1, 2),
            ),
            entities.Person(
                firstname="Other Firstname",
                lastname="Other Surname",
                date_of_birth=datetime.date(2023, 1, 2),
            ),
        ]
    )
    empty_db.commit()
    empty_db.add_all(
        [
            entities.Company(
                hr="HRB 123",
                court_id=2,
                name="Some Company GmbH",
                street="Sesamstr.",
                house_number="1",
                zip_code="58644",
                city="TV City",
                last_update=datetime.date.fromisoformat("2023-01-01"),
                latitude=51.3246,
                longitude=7.6968,
                pos_accuracy=4.0,
                founding_date=datetime.date(2010, 8, 7),
                capital_value=1000000,
                original_currency="DM",
                capital_type=CapitalTypeEnum.HAFTEINLAGE,
                business_purpose='Say "Hello World"',
            ),
            entities.Company(
                hr="HRB 123",
                court_id=1,
                name="Other Company GmbH",
                street="Sesamstr.",
                house_number="2",
                zip_code="58636",
                city="TV City",
                last_update=datetime.date.fromisoformat("2023-01-01"),
                latitude=51.38,
                longitude=7.7032,
                pos_accuracy=4.0,
                business_purpose="Some purpose",
            ),
            entities.Company(
                hr="HRB 12",
                court_id=2,
                name="Third Company GmbH",
                last_update=datetime.date.fromisoformat("2023-01-01"),
                sector="Electronic",
                capital_value=10000,
                original_currency="EUR",
                capital_type=CapitalTypeEnum.GRUNDKAPITAL,
            ),
        ]
    )
    empty_db.commit()

    empty_db.add_all(
        [
            entities.AnnualFinanceStatement(**finance_statement)
            for finance_statement in finance_statements
        ]
    )
    empty_db.commit()
    # print(pd.read_sql_table("company", empty_db.bind).to_string())
    return empty_db