mirror of
https://github.com/fhswf/aki_prj23_transparenzregister.git
synced 2025-06-22 00:23:55 +02:00
Add sql lite session (#71)
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.
This commit is contained in:
80
tests/utils/sql/connector_test.py
Normal file
80
tests/utils/sql/connector_test.py
Normal file
@ -0,0 +1,80 @@
|
||||
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))
|
4
tests/utils/sql/entities_test.py
Normal file
4
tests/utils/sql/entities_test.py
Normal file
@ -0,0 +1,4 @@
|
||||
def test_import() -> None:
|
||||
from aki_prj23_transparenzregister.utils.sql import entities
|
||||
|
||||
assert entities
|
Reference in New Issue
Block a user