mirror of
https://github.com/fhswf/aki_prj23_transparenzregister.git
synced 2025-04-24 15:32:35 +02:00
Philipp Horstenkamp
2233b12468
I added an sql lite session generator. Changes The function `def get_session() -> Session` has changed to `def get_session(connect_to: ConfigProvider | str) -> Session` If a JsonFileConfig Is given the postgress conection is checked. If a string is given that starts with `sqlite:///` an sql db is created. The use should otherwise be the same.
81 lines
2.5 KiB
Python
81 lines
2.5 KiB
Python
import os.path
|
|
from collections.abc import Generator
|
|
from typing import Any
|
|
from unittest.mock import Mock, patch
|
|
|
|
import pytest
|
|
from sqlalchemy.engine import Engine
|
|
|
|
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 (
|
|
get_pg_engine,
|
|
get_session,
|
|
init_db,
|
|
)
|
|
|
|
|
|
def test_get_engine_pg() -> None:
|
|
conn_args = PostgreConnectionString("", "", "", "", 42)
|
|
with patch(
|
|
"aki_prj23_transparenzregister.utils.sql.connector.sa.create_engine"
|
|
) as mock_create_engine:
|
|
result = "someThing"
|
|
mock_create_engine.return_value = result
|
|
assert get_pg_engine(conn_args) == result
|
|
|
|
|
|
@pytest.fixture()
|
|
def delete_sqlite_table() -> Generator[str, None, None]:
|
|
"""Cleans a path before and deletes the table after a test.
|
|
|
|
Returns:
|
|
The path where the sqlite table is placed.
|
|
"""
|
|
sqlite_test_path = "test_db.db"
|
|
if os.path.exists(sqlite_test_path):
|
|
os.remove(sqlite_test_path)
|
|
|
|
yield sqlite_test_path
|
|
|
|
if os.path.exists(sqlite_test_path):
|
|
os.remove(sqlite_test_path)
|
|
|
|
|
|
def test_get_sqlite_init(delete_sqlite_table: str) -> None:
|
|
"""Tests if a sql table file can be initiated."""
|
|
assert not os.path.exists(delete_sqlite_table)
|
|
session = get_session(f"sqlite:///{delete_sqlite_table}")
|
|
init_db(session)
|
|
session.close()
|
|
engine = session.bind
|
|
assert isinstance(engine, Engine)
|
|
engine.dispose()
|
|
assert os.path.exists(delete_sqlite_table)
|
|
|
|
|
|
@pytest.mark.parametrize("connection", ["faulty-name", 0, 9.2, True])
|
|
def test_get_invalid_connection(connection: Any) -> None:
|
|
"""Tests if an error is thrown on a faulty connections."""
|
|
with pytest.raises(TypeError):
|
|
get_session(connection)
|
|
|
|
|
|
def test_init_pd_db() -> None:
|
|
"""Tests if a pg sql database can be connected and initiated to."""
|
|
with patch(
|
|
"aki_prj23_transparenzregister.utils.sql.connector.get_pg_engine"
|
|
) as mock_get_engine, patch(
|
|
"aki_prj23_transparenzregister.utils.sql.connector.declarative_base"
|
|
) as mock_declarative_base:
|
|
mock_get_engine.connect.return_value = {}
|
|
|
|
mock_value = Mock()
|
|
mock_value.metadata.create_all.return_value = None
|
|
mock_declarative_base.return_value = mock_value
|
|
|
|
mock_value = Mock(spec=JsonFileConfigProvider)
|
|
mock_value.get_postgre_connection_string.return_value = ""
|
|
|
|
init_db(get_session(mock_value))
|