mirror of
https://github.com/fhswf/aki_prj23_transparenzregister.git
synced 2025-06-22 00:23:55 +02:00
Add a cli interface to choose a configuration (#163)
- [x] add a cli to the webserver to take env variables into account - [x] add a cli to the data processing that takes enviromental variable as a valid source into account - [x] rework the cli for the reset sql command - [x] rework the cli for the copying of sql data from one db to another
This commit is contained in:
@ -1,12 +1,19 @@
|
||||
"""Tests the config provers."""
|
||||
import json
|
||||
from pathlib import Path
|
||||
from unittest.mock import mock_open, patch
|
||||
|
||||
import pytest
|
||||
from _pytest.monkeypatch import MonkeyPatch
|
||||
|
||||
from aki_prj23_transparenzregister.config.config_providers import (
|
||||
EnvironmentConfigProvider,
|
||||
JsonFileConfigProvider,
|
||||
get_config_provider,
|
||||
)
|
||||
from aki_prj23_transparenzregister.config.config_template import (
|
||||
PostgreConnectionString,
|
||||
SQLiteConnectionString,
|
||||
)
|
||||
|
||||
|
||||
@ -19,7 +26,7 @@ def test_json_provider_init_fail() -> None:
|
||||
def test_json_provider_init_no_json() -> None:
|
||||
"""Tests if a non json file throws the correct error."""
|
||||
with patch("os.path.isfile") as mock_isfile, patch(
|
||||
"builtins.open", mock_open(read_data="fhdaofhdoas")
|
||||
"builtins.open", mock_open(read_data="fh1da23of4!hdo4s")
|
||||
):
|
||||
mock_isfile.return_value = True
|
||||
with pytest.raises(TypeError):
|
||||
@ -30,20 +37,47 @@ def test_json_provider_init() -> None:
|
||||
"""Tests the JsonFileConfigProvider creation."""
|
||||
data = {"hello": "world"}
|
||||
input_data = json.dumps(data)
|
||||
with patch("os.path.isfile") as mock_isfile, patch(
|
||||
"builtins.open", mock_open(read_data=input_data)
|
||||
):
|
||||
mock_isfile.return_value = True
|
||||
provider = JsonFileConfigProvider("someWhere")
|
||||
assert provider.__data__ == data
|
||||
|
||||
|
||||
@pytest.mark.parametrize("as_path", [True, False])
|
||||
def test_get_config_provider_json(as_path: bool) -> None:
|
||||
"""Tests the JsonFileConfigProvider creation by the factory method."""
|
||||
data = {"hello": "world"}
|
||||
input_data = json.dumps(data)
|
||||
with patch("os.path.isfile") as mock_isfile:
|
||||
mock_isfile.return_value = True
|
||||
with patch("builtins.open", mock_open(read_data=input_data)):
|
||||
provider = JsonFileConfigProvider("someWhere")
|
||||
assert provider.__data__ == data
|
||||
config_provider = get_config_provider(
|
||||
Path("some_where.json") if as_path else "some_where.json"
|
||||
)
|
||||
assert isinstance(config_provider, JsonFileConfigProvider)
|
||||
assert config_provider.__data__ == data
|
||||
|
||||
|
||||
def test_json_provider_get_postgres() -> None:
|
||||
@pytest.mark.parametrize("args", ["ENV", "env", "", None, "PYTHON_", "ELSE_"])
|
||||
def test_get_config_provider_env(args: str | None) -> None:
|
||||
env_provider = get_config_provider(args)
|
||||
assert isinstance(env_provider, EnvironmentConfigProvider)
|
||||
|
||||
|
||||
def test_get_config_provider_fail() -> None:
|
||||
with pytest.raises(ValueError, match="No configuration found"):
|
||||
get_config_provider("something-else")
|
||||
|
||||
|
||||
def test_json_provider_get_sql() -> None:
|
||||
"""Tests if the config provider can return the postgre config string."""
|
||||
data = {
|
||||
"postgres": {
|
||||
"username": "user",
|
||||
"password": "pass",
|
||||
"host": "locahost",
|
||||
"host": "localhost",
|
||||
"database": "postgres",
|
||||
"port": 420,
|
||||
}
|
||||
@ -52,7 +86,8 @@ def test_json_provider_get_postgres() -> None:
|
||||
with patch("os.path.isfile") as mock_isfile:
|
||||
mock_isfile.return_value = True
|
||||
with patch("builtins.open", mock_open(read_data=input_data)):
|
||||
config = JsonFileConfigProvider("someWhere").get_postgre_connection_string()
|
||||
config = JsonFileConfigProvider("someWhere").get_sql_connection_string()
|
||||
assert isinstance(config, PostgreConnectionString)
|
||||
assert config.username == data["postgres"]["username"]
|
||||
assert config.password == data["postgres"]["password"]
|
||||
assert config.host == data["postgres"]["host"]
|
||||
@ -67,14 +102,39 @@ def test_json_provider_get_postgres() -> None:
|
||||
assert "Mongo" not in str(JsonFileConfigProvider("someWhere"))
|
||||
|
||||
|
||||
@pytest.mark.parametrize("additional_data", [True, False])
|
||||
def test_json_provider_get_sqlit3(additional_data: bool) -> None:
|
||||
"""Tests if the config provider can return the sqlite config string."""
|
||||
data = {
|
||||
"sqlite": "some.db",
|
||||
}
|
||||
if additional_data:
|
||||
data |= {
|
||||
"postgres": { # type: ignore
|
||||
"username": "user",
|
||||
"password": "pass",
|
||||
"host": "localhost",
|
||||
"database": "postgres",
|
||||
"port": 420,
|
||||
}
|
||||
}
|
||||
input_data = json.dumps(data)
|
||||
with patch("os.path.isfile") as mock_isfile:
|
||||
mock_isfile.return_value = True
|
||||
with patch("builtins.open", mock_open(read_data=input_data)):
|
||||
assert JsonFileConfigProvider(
|
||||
"someWhere"
|
||||
).get_sql_connection_string() == SQLiteConnectionString("some.db")
|
||||
|
||||
|
||||
def test_json_provider_get_mongo() -> None:
|
||||
"""Tests the JsonConfigProvider for the mongo db."""
|
||||
data = {
|
||||
"mongo": {
|
||||
"username": "user",
|
||||
"password": "pass",
|
||||
"host": "locahost",
|
||||
"database": "postgres",
|
||||
"host": "localhost",
|
||||
"database": "mongo",
|
||||
"port": 420,
|
||||
}
|
||||
}
|
||||
@ -98,35 +158,59 @@ def test_json_provider_get_mongo() -> None:
|
||||
assert "Postgre" not in str(JsonFileConfigProvider("someWhere"))
|
||||
|
||||
|
||||
def test_env_provider_constructor() -> None:
|
||||
with patch("aki_prj23_transparenzregister.config.config_providers.os") as mock_os:
|
||||
keys = {"PYTHON_TEST": "test", "NOT_PYTHON_TEST": ""}
|
||||
mock_os.environ = keys
|
||||
provider = EnvironmentConfigProvider()
|
||||
assert provider.__data__ == {"TEST": "test"}
|
||||
|
||||
|
||||
def test_env_provider_postgres() -> None:
|
||||
def test_env_provider_constructor(monkeypatch: MonkeyPatch) -> None:
|
||||
env_configs = {"PYTHON_TEST": "test", "NOT_PYTHON_TEST": ""}
|
||||
for key, value in env_configs.items():
|
||||
monkeypatch.setenv(key, value)
|
||||
provider = EnvironmentConfigProvider()
|
||||
env_data = {
|
||||
"POSTGRES_USERNAME": "postgres",
|
||||
"POSTGRES_PASSWORD": "postgres",
|
||||
"POSTGRES_HOST": "localhost",
|
||||
"POSTGRES_DATABASE": "postgres",
|
||||
"POSTGRES_PORT": "5432",
|
||||
assert "TEST" in provider.__data__
|
||||
assert provider.__data__["TEST"] == "test"
|
||||
assert "NOT_PYTHON_TEST" not in provider.__data__
|
||||
|
||||
|
||||
def test_env_provider_postgres(monkeypatch: MonkeyPatch) -> None:
|
||||
env_configs = {
|
||||
"PYTHON_POSTGRES_USERNAME": "postgres-user",
|
||||
"PYTHON_POSTGRES_PASSWORD": "postgres-pw",
|
||||
"PYTHON_POSTGRES_HOST": "localhost",
|
||||
"PYTHON_POSTGRES_DATABASE": "postgres",
|
||||
"PYTHON_POSTGRES_PORT": "5432",
|
||||
}
|
||||
provider.__data__ = env_data
|
||||
conn_string = provider.get_postgre_connection_string()
|
||||
for env_config in env_configs.items():
|
||||
monkeypatch.setenv(*env_config)
|
||||
|
||||
assert conn_string.database == env_data["POSTGRES_DATABASE"]
|
||||
assert conn_string.host == env_data["POSTGRES_HOST"]
|
||||
assert conn_string.password == env_data["POSTGRES_PASSWORD"]
|
||||
assert conn_string.port == env_data["POSTGRES_PORT"]
|
||||
assert conn_string.username == env_data["POSTGRES_USERNAME"]
|
||||
|
||||
|
||||
def test_env_provider_mongodb() -> None:
|
||||
provider = EnvironmentConfigProvider()
|
||||
|
||||
conn_string = provider.get_sql_connection_string()
|
||||
assert isinstance(conn_string, PostgreConnectionString)
|
||||
|
||||
assert conn_string.database == env_configs["PYTHON_POSTGRES_DATABASE"]
|
||||
assert conn_string.host == env_configs["PYTHON_POSTGRES_HOST"]
|
||||
assert conn_string.password == env_configs["PYTHON_POSTGRES_PASSWORD"]
|
||||
assert conn_string.port == env_configs["PYTHON_POSTGRES_PORT"]
|
||||
assert conn_string.username == env_configs["PYTHON_POSTGRES_USERNAME"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("additional_args", [True, False])
|
||||
def test_env_provider_sqlite(additional_args: bool, monkeypatch: MonkeyPatch) -> None:
|
||||
env_configs = {"PYTHON_SQLITE_PATH": "some.db"}
|
||||
if additional_args:
|
||||
env_configs |= {
|
||||
"PYTHON_POSTGRES_USERNAME": "postgres-user",
|
||||
"PYTHON_POSTGRES_PASSWORD": "postgres-pw",
|
||||
"PYTHON_POSTGRES_HOST": "localhost",
|
||||
"PYTHON_POSTGRES_DATABASE": "postgres",
|
||||
"PYTHON_POSTGRES_PORT": "5432",
|
||||
}
|
||||
for env_config in env_configs.items():
|
||||
monkeypatch.setenv(*env_config)
|
||||
|
||||
provider = EnvironmentConfigProvider()
|
||||
assert provider.get_sql_connection_string() == SQLiteConnectionString("some.db")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("prefix", ["", "NON_PYTHON", "ELSE"])
|
||||
def test_env_provider_mongodb(prefix: str, monkeypatch: MonkeyPatch) -> None:
|
||||
env_data = {
|
||||
"MONGO_USERNAME": "username",
|
||||
"MONGO_HOST": "localhost",
|
||||
@ -134,6 +218,13 @@ def test_env_provider_mongodb() -> None:
|
||||
"MONGO_PORT": 27017,
|
||||
"MONGO_DATABASE": "transparenzregister",
|
||||
}
|
||||
for key, value in env_data.items():
|
||||
monkeypatch.setenv(f"{prefix if prefix else 'PYTHON_'}{key}", str(value))
|
||||
if prefix:
|
||||
provider = EnvironmentConfigProvider(prefix)
|
||||
else:
|
||||
provider = EnvironmentConfigProvider()
|
||||
|
||||
provider.__data__ = env_data
|
||||
conn_string = provider.get_mongo_connection_string()
|
||||
|
||||
|
34
tests/config/config_template_test.py
Normal file
34
tests/config/config_template_test.py
Normal file
@ -0,0 +1,34 @@
|
||||
"""Test for config templates."""
|
||||
from aki_prj23_transparenzregister.config.config_template import (
|
||||
MongoConnection,
|
||||
SQLiteConnectionString,
|
||||
)
|
||||
|
||||
|
||||
def test_sqlite_connection_string() -> None:
|
||||
"""Tests if the sqlite protocol is correctly added to a path."""
|
||||
assert str(SQLiteConnectionString("some-path-to.db")) == "sqlite:///some-path-to.db"
|
||||
assert (
|
||||
str(SQLiteConnectionString("other-path-to.db")) == "sqlite:///other-path-to.db"
|
||||
)
|
||||
assert SQLiteConnectionString("some-path-to.db") == SQLiteConnectionString(
|
||||
"some-path-to.db"
|
||||
)
|
||||
|
||||
|
||||
def test_get_conn_string_no_credentials() -> None:
|
||||
"""Tests the mongo connection string generation."""
|
||||
conn = MongoConnection("localhost", "", 27017, None, None)
|
||||
assert conn.get_conn_string() == "mongodb://localhost:27017"
|
||||
|
||||
|
||||
def test_get_conn_string_no_port_but_credentials() -> None:
|
||||
"""Tests the mongo connection string generation."""
|
||||
conn = MongoConnection("localhost", "", None, "admin", "password")
|
||||
assert conn.get_conn_string() == "mongodb+srv://admin:password@localhost"
|
||||
|
||||
|
||||
def test_get_conn_simple() -> None:
|
||||
"""Tests the mongo connection string generation."""
|
||||
conn = MongoConnection("localhost", "", None, None, None)
|
||||
assert conn.get_conn_string() == "mongodb+srv://localhost"
|
@ -7,11 +7,15 @@ from typing import Any
|
||||
|
||||
import pytest
|
||||
from sqlalchemy.engine import Engine
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy.orm import Session, sessionmaker
|
||||
|
||||
from aki_prj23_transparenzregister.config.config_template import SQLiteConnectionString
|
||||
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_session, init_db
|
||||
from aki_prj23_transparenzregister.utils.sql.connector import (
|
||||
get_engine,
|
||||
init_db,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
@ -39,7 +43,11 @@ def empty_db() -> Generator[Session, None, None]:
|
||||
"""Generates a db Session to a sql_lite db."""
|
||||
if os.path.exists("test-db.db"):
|
||||
os.remove("test-db.db")
|
||||
db = get_session("sqlite:///test-db.db")
|
||||
db = sessionmaker(
|
||||
autocommit=False,
|
||||
autoflush=False,
|
||||
bind=get_engine(SQLiteConnectionString("test-db.db")),
|
||||
)()
|
||||
init_db(db)
|
||||
yield db
|
||||
db.close()
|
||||
|
@ -1,7 +1,11 @@
|
||||
"""Test for the main app dashboard."""
|
||||
import sys
|
||||
from collections.abc import Generator
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
from _pytest.monkeypatch import MonkeyPatch
|
||||
from pytest_mock import MockerFixture
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from aki_prj23_transparenzregister.ui import app
|
||||
@ -31,3 +35,27 @@ def test_go_to_company_page() -> None:
|
||||
"""Checks if the go_to_company_page callback yields a result."""
|
||||
output = app.go_to_company_page(1)
|
||||
assert output == "/Unternehmensdetails/1"
|
||||
|
||||
|
||||
def test_main_of_app(monkeypatch: MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(sys, "argv", [sys.argv[0]])
|
||||
with pytest.raises(SystemExit):
|
||||
app.main()
|
||||
|
||||
|
||||
def test_main_of_app_help(monkeypatch: MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(sys, "argv", [sys.argv[0], "-h"])
|
||||
with pytest.raises(SystemExit):
|
||||
app.main()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("upper", [True, False])
|
||||
def test_main_of_app_env(
|
||||
monkeypatch: MonkeyPatch, upper: bool, mocker: MockerFixture
|
||||
) -> None:
|
||||
MagicMock()
|
||||
monkeypatch.setattr(sys, "argv", [sys.argv[0], "ENV" if upper else "env"])
|
||||
mocked = mocker.patch("aki_prj23_transparenzregister.ui.app.Dash.run")
|
||||
mocked.return_value = None
|
||||
app.main()
|
||||
mocked.assert_called_once()
|
||||
|
@ -1,6 +1,7 @@
|
||||
"""Test the transfer functions from mongodb to sql."""
|
||||
import random
|
||||
import string
|
||||
import sys
|
||||
from datetime import date
|
||||
from typing import Any
|
||||
|
||||
@ -8,13 +9,13 @@ import numpy as np
|
||||
import pandas as pd
|
||||
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 CapitalTypeEnum, CurrencyEnum
|
||||
from aki_prj23_transparenzregister.utils import data_transfer
|
||||
from aki_prj23_transparenzregister.utils.data_transfer import norm_capital
|
||||
from aki_prj23_transparenzregister.utils.sql import entities
|
||||
|
||||
|
||||
@ -1025,7 +1026,9 @@ def test_add_annual_report_financial_key_error(full_db: Session) -> None:
|
||||
@pytest.mark.parametrize("currency", ["€", "EUR"])
|
||||
def test_norm_capital_eur(currency: str, capital_type: str) -> None:
|
||||
"""Tests if eur entries can be converted / normed correctly."""
|
||||
assert norm_capital({"value": 5, "currency": currency, "type": capital_type}) == {
|
||||
assert data_transfer.norm_capital(
|
||||
{"value": 5, "currency": currency, "type": capital_type}
|
||||
) == {
|
||||
"capital_value": 5.0,
|
||||
"capital_currency": CurrencyEnum("EUR"),
|
||||
"capital_type": CapitalTypeEnum(capital_type),
|
||||
@ -1036,7 +1039,7 @@ def test_norm_capital_eur(currency: str, capital_type: str) -> None:
|
||||
@pytest.mark.parametrize("currency", ["DM", "DEM"])
|
||||
def test_norm_capital_dm(currency: str, capital_type: CapitalTypeEnum) -> None:
|
||||
"""Tests if dm entries can be converted / normed correctly."""
|
||||
assert norm_capital(
|
||||
assert data_transfer.norm_capital(
|
||||
capital={"value": 5, "currency": currency, "type": capital_type}
|
||||
) == {
|
||||
"capital_value": 2.56,
|
||||
@ -1047,7 +1050,7 @@ def test_norm_capital_dm(currency: str, capital_type: CapitalTypeEnum) -> None:
|
||||
|
||||
def test_norm_capital_fail() -> None:
|
||||
"""Tests if the entry is dropped if it isn't complete."""
|
||||
assert norm_capital({"something": "something"}) == {} # type: ignore
|
||||
assert data_transfer.norm_capital({"something": "something"}) == {} # type: ignore
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
@ -1062,3 +1065,30 @@ def test_norm_capital_fail() -> None:
|
||||
)
|
||||
def test_get_geocodes(zip_code: str | None, results: dict) -> None:
|
||||
assert data_transfer.get_geocodes(zip_code) == results
|
||||
|
||||
|
||||
def test_transfer_data_cli(monkeypatch: MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(sys, "argv", [sys.argv[0]])
|
||||
with pytest.raises(SystemExit):
|
||||
data_transfer.transfer_data_cli()
|
||||
|
||||
|
||||
def test_transfer_data_cli_help(monkeypatch: MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(sys, "argv", [sys.argv[0], "-h"])
|
||||
with pytest.raises(SystemExit):
|
||||
data_transfer.transfer_data_cli()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("upper", [True, False])
|
||||
def test_transfer_data_cli_env(
|
||||
monkeypatch: MonkeyPatch, upper: bool, mocker: MockerFixture
|
||||
) -> None:
|
||||
monkeypatch.setattr(sys, "argv", [sys.argv[0], "ENV" if upper else "env"])
|
||||
mocker.patch(
|
||||
"aki_prj23_transparenzregister.utils.data_transfer.transfer_data", lambda _: _
|
||||
)
|
||||
spy = mocker.spy(data_transfer, "transfer_data")
|
||||
|
||||
# with pytest.raises(KeyError):
|
||||
data_transfer.transfer_data_cli()
|
||||
spy.assert_called_once()
|
||||
|
@ -1,15 +1,21 @@
|
||||
"""Smoke-test over the logger config."""
|
||||
from argparse import ArgumentParser
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from aki_prj23_transparenzregister.utils.logger_config import configer_logger
|
||||
from aki_prj23_transparenzregister.utils.logger_config import (
|
||||
add_logger_options_to_argparse,
|
||||
configer_logger,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("parser", [True, False])
|
||||
@pytest.mark.parametrize("path", [None, "test-log.log", ""])
|
||||
@pytest.mark.parametrize("upper", [True, False])
|
||||
@pytest.mark.parametrize("level", ["info", "debug", "error", "warning"])
|
||||
@pytest.mark.parametrize("level", ["info", "debug", "error"])
|
||||
def test_configer_logger(
|
||||
parser: bool,
|
||||
level: str,
|
||||
upper: bool,
|
||||
path: Path | str | None,
|
||||
@ -17,10 +23,32 @@ def test_configer_logger(
|
||||
"""Tests the configuration of the logger.
|
||||
|
||||
Args:
|
||||
parser: If the arguments should be given via the parser or not.
|
||||
level: The log-level to configure.
|
||||
upper: If the upper variant of the level should be used.
|
||||
path: The path where to save the log.
|
||||
"""
|
||||
if level.upper():
|
||||
level = level.upper()
|
||||
configer_logger(level, path) # type: ignore
|
||||
if parser:
|
||||
args_parser = ArgumentParser()
|
||||
add_logger_options_to_argparse(args_parser)
|
||||
configer_logger(
|
||||
namespace=args_parser.parse_args(
|
||||
[
|
||||
"--level",
|
||||
level if level else "",
|
||||
"--log-path",
|
||||
str(path) if path else "",
|
||||
]
|
||||
)
|
||||
)
|
||||
else:
|
||||
configer_logger(level=level, path=path) # type: ignore
|
||||
|
||||
|
||||
def test_add_logger_options_to_argparse() -> None:
|
||||
"""A test checking if the ArgumentParser is modified."""
|
||||
args_parser = ArgumentParser()
|
||||
|
||||
add_logger_options_to_argparse(args_parser)
|
||||
|
@ -1,30 +1,12 @@
|
||||
"""Tests for connecting to the mongodb."""
|
||||
from unittest.mock import patch
|
||||
|
||||
from aki_prj23_transparenzregister.config.config_template import MongoConnection
|
||||
from aki_prj23_transparenzregister.utils.mongo.connector import (
|
||||
MongoConnection,
|
||||
MongoConnector,
|
||||
)
|
||||
|
||||
|
||||
def test_get_conn_string_no_credentials() -> None:
|
||||
"""Tests the mongo connection string generation."""
|
||||
conn = MongoConnection("localhost", "", 27017, None, None)
|
||||
assert conn.get_conn_string() == "mongodb://localhost:27017"
|
||||
|
||||
|
||||
def test_get_conn_string_no_port_but_credentials() -> None:
|
||||
"""Tests the mongo connection string generation."""
|
||||
conn = MongoConnection("localhost", "", None, "admin", "password")
|
||||
assert conn.get_conn_string() == "mongodb+srv://admin:password@localhost"
|
||||
|
||||
|
||||
def test_get_conn_simple() -> None:
|
||||
"""Tests the mongo connection string generation."""
|
||||
conn = MongoConnection("localhost", "", None, None, None)
|
||||
assert conn.get_conn_string() == "mongodb+srv://localhost"
|
||||
|
||||
|
||||
def test_mongo_connector() -> None:
|
||||
"""Tests the MongoConnector."""
|
||||
with patch("pymongo.MongoClient") as mock_mongo_client:
|
||||
|
@ -1,16 +1,14 @@
|
||||
"""Tests the sql connector."""
|
||||
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_engine,
|
||||
get_session,
|
||||
init_db,
|
||||
)
|
||||
@ -24,7 +22,13 @@ def test_get_engine_pg() -> None:
|
||||
) as mock_create_engine:
|
||||
result = "someThing"
|
||||
mock_create_engine.return_value = result
|
||||
assert get_pg_engine(conn_args) == result
|
||||
assert get_engine(conn_args) == result
|
||||
|
||||
|
||||
def test_get_engine_fail() -> None:
|
||||
"""Tests what happens if the wrong type is given to the engine factory."""
|
||||
with pytest.raises(TypeError, match="The type of the configuration is invalid."):
|
||||
get_engine(None) # type: ignore
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
@ -44,29 +48,10 @@ def delete_sqlite_table() -> Generator[str, None, None]:
|
||||
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"
|
||||
"aki_prj23_transparenzregister.utils.sql.connector.get_engine"
|
||||
) as mock_get_engine, patch(
|
||||
"aki_prj23_transparenzregister.utils.sql.connector.declarative_base"
|
||||
) as mock_declarative_base:
|
||||
@ -77,6 +62,6 @@ def test_init_pd_db() -> None:
|
||||
mock_declarative_base.return_value = mock_value
|
||||
|
||||
mock_value = Mock(spec=JsonFileConfigProvider)
|
||||
mock_value.get_postgre_connection_string.return_value = ""
|
||||
mock_value.get_sql_connection_string.return_value = ""
|
||||
|
||||
init_db(get_session(mock_value))
|
||||
|
@ -1,13 +1,20 @@
|
||||
"""Test if the sql db can be copied."""
|
||||
import os
|
||||
import sys
|
||||
from collections.abc import Generator
|
||||
|
||||
import pandas as pd
|
||||
import pytest
|
||||
from _pytest.monkeypatch import MonkeyPatch
|
||||
from sqlalchemy.engine import Engine
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy.orm import Session, sessionmaker
|
||||
|
||||
from aki_prj23_transparenzregister.utils.sql.connector import Base, get_session, init_db
|
||||
from aki_prj23_transparenzregister.config.config_template import SQLiteConnectionString
|
||||
from aki_prj23_transparenzregister.utils.sql.connector import (
|
||||
Base,
|
||||
get_engine,
|
||||
init_db,
|
||||
)
|
||||
from aki_prj23_transparenzregister.utils.sql.copy_sql import (
|
||||
copy_db_cli,
|
||||
transfer_db_function,
|
||||
@ -19,7 +26,13 @@ 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")
|
||||
|
||||
db = sessionmaker(
|
||||
autocommit=False,
|
||||
autoflush=False,
|
||||
bind=get_engine(SQLiteConnectionString("secondary.db")),
|
||||
)()
|
||||
|
||||
init_db(db)
|
||||
yield db
|
||||
db.close()
|
||||
@ -44,13 +57,15 @@ def test_transfer_db(full_db: Session, destination_db: Session) -> None:
|
||||
)
|
||||
|
||||
|
||||
def test_copy_db_cli_help1() -> None:
|
||||
def test_copy_db_cli_help1(monkeypatch: MonkeyPatch) -> None:
|
||||
"""Tests if the help argument exits the software gracefully."""
|
||||
with pytest.raises(SystemExit):
|
||||
copy_db_cli(["-h"])
|
||||
with monkeypatch.context() as m, pytest.raises(SystemExit): # noqa: PT012
|
||||
m.setattr(sys, "argv", [sys.argv[0], "-h"])
|
||||
copy_db_cli()
|
||||
|
||||
|
||||
def test_copy_db_cli_help2() -> None:
|
||||
def test_copy_db_cli_help2(monkeypatch: MonkeyPatch) -> None:
|
||||
"""Tests if the help argument exits the software gracefully."""
|
||||
with pytest.raises(SystemExit):
|
||||
copy_db_cli(["eskse", "-h", "asdf"])
|
||||
with monkeypatch.context() as m, pytest.raises(SystemExit): # noqa: PT012
|
||||
m.setattr(sys, "argv", [sys.argv[0], "eskse", "-h", "asdf"])
|
||||
copy_db_cli()
|
||||
|
Reference in New Issue
Block a user