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

test: Adding unit tests

Browse Source
This commit is contained in:
TrisNol
2023-07-16 11:25:21 +02:00
parent 9da3e4adb9
commit b788ee3659
10 changed files with 4336 additions and 4390 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7828
poetry.lock generated
View File

File diff suppressed because it is too large Load Diff

223
pyproject.toml
View File

@ -1,111 +1,112 @@
[build-system]
build-backend = "poetry.core.masonry.api"
requires = ["poetry-core"]
[tookl.mypy]
disallow_untyped_defs = true
follow_imports = "silent"
python_version = "3.11"
warn_redudant_casts = true
warn_unused_ignores = true
[tool.black]
target-version = ["py311"]
[tool.coverage.run]
branch = true
dynamic_context = "test_function"
relative_files = true
source = ["src"]
[tool.poetry]
authors = ["AKI Projektgruppe 23"]
description = "A project analysing the german transparenzregister and other data sources to find shared business interests and shared personal and other links for lots of companies."
name = "aki-prj23-transparenzregister"
packages = [{include = "aki_prj23_transparenzregister", from = "src"}]
readme = "README.md"
version = "0.1.0"
[tool.poetry.dependencies]
loguru = "^0.7.0"
matplotlib = "^3.7.1"
plotly = "^5.14.1"
python = "^3.11"
seaborn = "^0.12.2"
selenium = "^4.10.0"
tqdm = "^4.65.0"
[tool.poetry.group.develop.dependencies]
black = {extras = ["jupyter"], version = "^23.3.0"}
jupyterlab = "^4.0.0"
nbconvert = "^7.4.0"
pre-commit = "^3.3.2"
rise = "^5.7.1"
[tool.poetry.group.doc.dependencies]
jupyter = "^1.0.0"
myst-parser = "^1.0.0"
nbsphinx = "^0.9.2"
sphinx = "^6.0.0"
sphinx-copybutton = "^0.5.2"
sphinx-rtd-theme = "^1.2.1"
sphinx_autodoc_typehints = "*"
sphinxcontrib-mermaid = "^0.9.2"
sphinxcontrib-napoleon = "^0.7"
[tool.poetry.group.lint.dependencies]
black = "^23.3.0"
mypy = "^1.3.0"
pandas-stubs = "^2.0.1.230501"
ruff = "^0.0.270"
types-requests = "^2.31.0.1"
[tool.poetry.group.test.dependencies]
pytest = "^7.3.1"
pytest-clarity = "^1.0.1"
pytest-cov = "^4.1.0"
pytest-mock = "^3.10.0"
pytest-repeat = "^0.9.1"
[tool.ruff]
exclude = [
".bzr",
".direnv",
".eggs",
".git",
".git-rewrite",
".hg",
".mypy_cache",
".nox",
".pants.d",
".pytype",
".ruff_cache",
".svn",
".tox",
".venv",
"__pypackages__",
"_build",
"buck-out",
"build",
"dist",
"node_modules",
"venv"
]
# Never enforce `E501` (line length violations).
ignore = ["E501"]
line-length = 88
# Enable flake8-bugbear (`B`) rules.
select = ["E", "F", "B", "I", "S", "RSE", "RET", "SLF", "SIM", "TID", "PD", "PL", "PLE", "PLR", "PLW", "NPY", "UP", "D", "N", "A", "C4", "T20", "PT"]
src = ["src"]
target-version = "py311"
# Avoid trying to fix flake8-bugbear (`B`) violations.
unfixable = ["B"]
[tool.ruff.flake8-builtins]
builtins-ignorelist = ["id"]
[tool.ruff.per-file-ignores]
"tests/*.py" = ["S101"]
[tool.ruff.pydocstyle]
convention = "google"
[build-system]
build-backend = "poetry.core.masonry.api"
requires = ["poetry-core"]
[tookl.mypy]
disallow_untyped_defs = true
follow_imports = "silent"
python_version = "3.11"
warn_redudant_casts = true
warn_unused_ignores = true
[tool.black]
target-version = ["py311"]
[tool.coverage.run]
branch = true
dynamic_context = "test_function"
relative_files = true
source = ["src"]
[tool.poetry]
authors = ["AKI Projektgruppe 23"]
description = "A project analysing the german transparenzregister and other data sources to find shared business interests and shared personal and other links for lots of companies."
name = "aki-prj23-transparenzregister"
packages = [{include = "aki_prj23_transparenzregister", from = "src"}]
readme = "README.md"
version = "0.1.0"
[tool.poetry.dependencies]
loguru = "^0.7.0"
matplotlib = "^3.7.1"
plotly = "^5.14.1"
pymongo = "^4.4.1"
python = "^3.11"
seaborn = "^0.12.2"
selenium = "^4.10.0"
tqdm = "^4.65.0"
[tool.poetry.group.develop.dependencies]
black = {extras = ["jupyter"], version = "^23.3.0"}
jupyterlab = "^4.0.0"
nbconvert = "^7.4.0"
pre-commit = "^3.3.2"
rise = "^5.7.1"
[tool.poetry.group.doc.dependencies]
jupyter = "^1.0.0"
myst-parser = "^1.0.0"
nbsphinx = "^0.9.2"
sphinx = "^6.0.0"
sphinx-copybutton = "^0.5.2"
sphinx-rtd-theme = "^1.2.1"
sphinx_autodoc_typehints = "*"
sphinxcontrib-mermaid = "^0.9.2"
sphinxcontrib-napoleon = "^0.7"
[tool.poetry.group.lint.dependencies]
black = "^23.3.0"
mypy = "^1.3.0"
pandas-stubs = "^2.0.1.230501"
ruff = "^0.0.270"
types-requests = "^2.31.0.1"
[tool.poetry.group.test.dependencies]
pytest = "^7.3.1"
pytest-clarity = "^1.0.1"
pytest-cov = "^4.1.0"
pytest-mock = "^3.10.0"
pytest-repeat = "^0.9.1"
[tool.ruff]
exclude = [
".bzr",
".direnv",
".eggs",
".git",
".git-rewrite",
".hg",
".mypy_cache",
".nox",
".pants.d",
".pytype",
".ruff_cache",
".svn",
".tox",
".venv",
"__pypackages__",
"_build",
"buck-out",
"build",
"dist",
"node_modules",
"venv"
]
# Never enforce `E501` (line length violations).
ignore = ["E501"]
line-length = 88
# Enable flake8-bugbear (`B`) rules.
select = ["E", "F", "B", "I", "S", "RSE", "RET", "SLF", "SIM", "TID", "PD", "PL", "PLE", "PLR", "PLW", "NPY", "UP", "D", "N", "A", "C4", "T20", "PT"]
src = ["src"]
target-version = "py311"
# Avoid trying to fix flake8-bugbear (`B`) violations.
unfixable = ["B"]
[tool.ruff.flake8-builtins]
builtins-ignorelist = ["id"]
[tool.ruff.per-file-ignores]
"tests/*.py" = ["S101", "D100", "D101", "D107", "D103"]
[tool.ruff.pydocstyle]
convention = "google"

103
src/aki_prj23_transparenzregister/utils/company_mongo_service.py
View File

@ -1,54 +1,49 @@
"""CompanyMongoService."""
from models.company import Company
from utils.company_service_interface import CompanyServiceInterface
from utils.mongo import MongoConnector
class CompanyMongoService(CompanyServiceInterface):
"""_summary_.
Args:
CompanyServiceInterface (_type_): _description_
"""
def __init__(self, connector: MongoConnector):
"""_summary_.
Args:
connector (MongoConnector): _description_
"""
self.collection = connector.database["companies"]
def get_all(self) -> list[Company]:
"""_summary_.
Returns:
list[Company]: _description_
"""
result = self.collection.find()
return list(result)
def get_by_id(self, id: str) -> Company | None:
"""_summary_.
Args:
id (str): _description_
Returns:
Company | None: _description_
"""
result = list(self.collection.find({"id": id}))
if len(result) == 1:
return result[0]
return None
def insert(self, company: Company):
"""_summary_.
Args:
company (Company): _description_
Returns:
_type_: _description_
"""
return self.collection.insert_one(company.to_dict())
"""CompanyMongoService."""
from aki_prj23_transparenzregister.models.company import Company, CompanyID
from aki_prj23_transparenzregister.utils.mongo import MongoConnector
class CompanyMongoService:
"""_summary_."""
def __init__(self, connector: MongoConnector):
"""_summary_.
Args:
connector (MongoConnector): _description_
"""
self.collection = connector.database["companies"]
def get_all(self) -> list[Company]:
"""_summary_.
Returns:
list[Company]: _description_
"""
result = self.collection.find()
return list(result)
def get_by_id(self, id: CompanyID) -> Company | None:
"""_summary_.
Args:
id (str): _description_
Returns:
Company | None: _description_
"""
result = list(self.collection.find({"id": id}))
if len(result) == 1:
return result[0]
return None
def insert(self, company: Company):
"""_summary_.
Args:
company (Company): _description_
Returns:
_type_: _description_
"""
return self.collection.insert_one(company.to_dict())

51
src/aki_prj23_transparenzregister/utils/company_service_interface.py
View File

@ -1,51 +0,0 @@
"""CompanyServiceInterface."""
import abc
from models import Company
class CompanyServiceInterface(abc.ABC):
"""Generic abstract interface for CRUD operations of a Company.
Args:
ABC (_type_): Abstract class
"""
@abc.abstractmethod
def get_all(self) -> list[Company.Company]:
"""_summary_.
Raises:
NotImplementedError: _description_
Returns:
list[Company.Company]: _description_
"""
raise NotImplementedError
@abc.abstractmethod
def get_by_id(self, id: Company.CompayID) -> Company.Company | None:
"""_summary_.
Args:
id (Company.CompayID): _description_
Raises:
NotImplementedError: _description_
Returns:
Company.Company | None: _description_
"""
raise NotImplementedError
@abc.abstractmethod
def insert(self, company: Company.Company):
"""_summary_.
Args:
company (Company.Company): _description_
Raises:
NotImplementedError: _description_
"""
raise NotImplementedError

215
src/aki_prj23_transparenzregister/utils/mongo.py
View File

@ -1,155 +1,60 @@
"""Mongo Wrapper."""
from dataclasses import dataclass
import pymongo
from models.news import News
from utils.news_service_interface import (
NewsServiceInterface,
)
@dataclass
class MongoConnection:
"""_summary_."""
hostname: str
database: str
port: int | None
username: str | None
password: str | None
class MongoConnector:
"""Wrapper for establishing a connection to a MongoDB instance."""
def __init__(self, connection: MongoConnection):
"""_summary_.
Args:
connection (MongoConnection): Wrapper for connection string
"""
self.client = self.connect(
connection.hostname,
connection.port,
connection.username,
connection.password,
)
self.database = self.client[connection.database]
def connect(
self,
hostname: str,
port: int | None,
username: str | None,
password: str | None,
) -> pymongo.MongoClient:
"""_summary_.
Args:
hostname (str): hostname
port (int): port
username (str): Username
password (str): Password
Returns:
pymongo.MongoClient: MongoClient connect to the DB
"""
if username is not None and password is not None:
connection_string = f"mongodb+srv://{username}:{password}@{hostname}"
else:
connection_string = f"mongodb+srv://{hostname}"
if port is not None:
connection_string += f":{port}"
connection_string = connection_string.replace("mongodb+srv", "mongodb")
return pymongo.MongoClient(connection_string)
class MongoNewsService(NewsServiceInterface):
"""_summary_.
Args:
NewsServiceInterface (_type_): _description_
"""
def __init__(self, connector: MongoConnector):
"""_summary_.
Args:
connector (MongoConnector): _description_
"""
self.collection = connector.database["news"]
def get_all(self) -> list[News]:
"""_summary_.
Returns:
list[News]: _description_
"""
result = self.collection.find()
return [MongoEntryTransformer.transform_outgoing(elem) for elem in result]
def get_by_id(self, id: str) -> News | None:
"""_summary_.
Args:
id (str): _description_
Returns:
News | None: _description_
"""
result = list(self.collection.find({"_id": id}))
if len(result) == 1:
return MongoEntryTransformer.transform_outgoing(list(result)[0])
return None
def insert(self, news: News):
"""_summary_.
Args:
news (News): _description_
Returns:
_type_: _description_
"""
return self.collection.insert_one(MongoEntryTransformer.transform_ingoing(news))
class MongoEntryTransformer:
"""_summary_.
Returns:
_type_: _description_
"""
@staticmethod
def transform_ingoing(news: News) -> dict:
"""Convert a News object to a dictionary compatible with a MongoDB entry.
Args:
news (News): News object to be transformed
Returns:
dict: Transformed data with added _id field
"""
transport_object = news.to_dict()
transport_object["_id"] = news.id
del transport_object["id"]
return transport_object
@staticmethod
def transform_outgoing(data: dict) -> News:
"""Reverse the transform_ingoing method.
Args:
data (dict): dict from the MongoDB to be transformed
Returns:
News: News entry based on MongoDB document
"""
return News(
id=data["_id"],
title=data["title"],
date=data["date"],
text=data["text"],
source_url=data["source_url"],
)
"""Mongo Wrapper."""
from dataclasses import dataclass
import pymongo
@dataclass
class MongoConnection:
"""_summary_."""
hostname: str
database: str
port: int | None
username: str | None
password: str | None
class MongoConnector:
"""Wrapper for establishing a connection to a MongoDB instance."""
def __init__(self, connection: MongoConnection):
"""_summary_.
Args:
connection (MongoConnection): Wrapper for connection string
"""
self.client = self.connect(
connection.hostname,
connection.port,
connection.username,
connection.password,
)
self.database = self.client[connection.database]
def connect(
self,
hostname: str,
port: int | None,
username: str | None,
password: str | None,
) -> pymongo.MongoClient:
"""_summary_.
Args:
hostname (str): hostname
port (int): port
username (str): Username
password (str): Password
Returns:
pymongo.MongoClient: MongoClient connect to the DB
"""
if username is not None and password is not None:
connection_string = f"mongodb+srv://{username}:{password}@{hostname}"
else:
connection_string = f"mongodb+srv://{hostname}"
if port is not None:
connection_string += f":{port}"
connection_string = connection_string.replace("mongodb+srv", "mongodb")
return pymongo.MongoClient(connection_string)

94
src/aki_prj23_transparenzregister/utils/news_mongo_service.py Normal file
View File

@ -0,0 +1,94 @@
"""MongoNewsService."""
from aki_prj23_transparenzregister.models.news import News
from aki_prj23_transparenzregister.utils.mongo import MongoConnector
class MongoNewsService:
"""_summary_.
Args:
NewsServiceInterface (_type_): _description_
"""
def __init__(self, connector: MongoConnector):
"""_summary_.
Args:
connector (MongoConnector): _description_
"""
self.collection = connector.database["news"]
def get_all(self) -> list[News]:
"""_summary_.
Returns:
list[News]: _description_
"""
result = self.collection.find()
return [MongoEntryTransformer.transform_outgoing(elem) for elem in result]
def get_by_id(self, id: str) -> News | None:
"""_summary_.
Args:
id (str): _description_
Returns:
News | None: _description_
"""
result = list(self.collection.find({"_id": id}))
if len(result) == 1:
return MongoEntryTransformer.transform_outgoing(list(result)[0])
return None
def insert(self, news: News):
"""_summary_.
Args:
news (News): _description_
Returns:
_type_: _description_
"""
return self.collection.insert_one(MongoEntryTransformer.transform_ingoing(news))
class MongoEntryTransformer:
"""_summary_.
Returns:
_type_: _description_
"""
@staticmethod
def transform_ingoing(news: News) -> dict:
"""Convert a News object to a dictionary compatible with a MongoDB entry.
Args:
news (News): News object to be transformed
Returns:
dict: Transformed data with added _id field
"""
transport_object = news.to_dict()
transport_object["_id"] = news.id
del transport_object["id"]
return transport_object
@staticmethod
def transform_outgoing(data: dict) -> News:
"""Reverse the transform_ingoing method.
Args:
data (dict): dict from the MongoDB to be transformed
Returns:
News: News entry based on MongoDB document
"""
return News(
id=data["_id"],
title=data["title"],
date=data["date"],
text=data["text"],
source_url=data["source_url"],
)

63
src/aki_prj23_transparenzregister/utils/news_service_interface.py
View File

@ -1,63 +0,0 @@
"""NewsServiceInterface."""
import abc
from models.news import News
class NewsServiceInterface(abc.ABC):
"""Generic abstract interface for a NewsService handling CRUD operations.
Args:
ABC (_type_): Abstract class
"""
@abc.abstractmethod
def get_all(self) -> list[News]:
"""Get a list of all News articles.
Raises:
NotImplementedError: To be defined by child classes
Returns:
list[News]: Results
"""
raise NotImplementedError
@abc.abstractmethod
def get_by_id(self, id: str) -> News | None:
"""Get an entry by an ID.
Args:
id (str): ID identifying the entry
Raises:
NotImplementedError: To be defined by child classes
Returns:
News | None: Found object or None if no entry with ID found
"""
raise NotImplementedError
@abc.abstractmethod
def insert(self, news: News):
"""Insert a News entry into the DB.
Args:
news (News): News object to be saved
Raises:
NotImplementedError: To be defined by child classes
"""
raise NotImplementedError
@abc.abstractmethod
def insert_many(self, news: list[News]):
"""Inserts many documents at once.
Args:
news (list[News]): List of News entries to be saved
Raises:
NotImplementedError: To be defined by child classes
"""
raise NotImplementedError

2
tests/models/news_test.py
View File

@ -1,4 +1,4 @@
"""Test Models.nes."""
"""Test Models.nesws."""
from aki_prj23_transparenzregister.models.news import News

103
tests/utils/company_mongo_service_test.py Normal file
View File

@ -0,0 +1,103 @@
"""Test utils.company_mongo_service."""
from unittest.mock import Mock
import pytest
from aki_prj23_transparenzregister.models.company import Company
from aki_prj23_transparenzregister.utils.company_mongo_service import (
CompanyMongoService,
)
@pytest.fixture()
def mock_mongo_connector(mocker) -> Mock:
"""Mock MongoConnector class.
Args:
mocker (any): Library mocker
Returns:
Mock: Mocked MongoConnector
"""
mock = Mock()
mocker.patch(
"aki_prj23_transparenzregister.utils.mongo.MongoConnector", return_value=mock
)
return mock
@pytest.fixture()
def mock_collection() -> Mock:
"""Mock mongo collection.
Returns:
Mock: Mock object
"""
return Mock()
def test_init(mock_mongo_connector, mock_collection):
"""Test CompanyMongoService constructor.
Args:
mock_mongo_connector (Mock): Mocked MongoConnector library
mock_collection (Mock): Mocked pymongo collection
"""
mock_mongo_connector.database = {"companies": mock_collection}
service = CompanyMongoService(mock_mongo_connector)
assert service.collection == mock_collection
def test_get_all(mock_mongo_connector, mock_collection):
"""Test CompanyMongoService get_all method.
Args:
mock_mongo_connector (Mock): Mocked MongoConnector library
mock_collection (Mock): Mocked pymongo collection
"""
mock_mongo_connector.database = {"companies": mock_collection}
service = CompanyMongoService(mock_mongo_connector)
mock_result = [{"id": "42"}]
mock_collection.find.return_value = mock_result
assert service.get_all() == mock_result
def test_by_id_no_result(mock_mongo_connector, mock_collection):
"""Test CompanyMongoService get_by_id with no result.
Args:
mock_mongo_connector (Mock): Mocked MongoConnector library
mock_collection (Mock): Mocked pymongo collection
"""
mock_mongo_connector.database = {"companies": mock_collection}
service = CompanyMongoService(mock_mongo_connector)
mock_collection.find.return_value = []
assert service.get_by_id("Does not exist") is None
def test_by_id_result(mock_mongo_connector, mock_collection):
"""Test CompanyMongoService get_by_id with result.
Args:
mock_mongo_connector (Mock): Mocked MongoConnector library
mock_collection (Mock): Mocked pymongo collection
"""
mock_mongo_connector.database = {"companies": mock_collection}
service = CompanyMongoService(mock_mongo_connector)
mock_entry = {"id": "Does exist", "vaue": 42}
mock_collection.find.return_value = [mock_entry]
assert service.get_by_id("Does exist") == mock_entry
def test_insert(mock_mongo_connector, mock_collection):
"""Test CompanyMongoService insert method.
Args:
mock_mongo_connector (Mock): Mocked MongoConnector library
mock_collection (Mock): Mocked pymongo collection
"""
mock_mongo_connector.database = {"companies": mock_collection}
service = CompanyMongoService(mock_mongo_connector)
mock_result = 42
mock_collection.insert_one.return_value = mock_result
assert service.insert(Company(None, None, "", "", [])) == mock_result

44
tests/utils/news_mongo_service_test.py Normal file
View File

@ -0,0 +1,44 @@
from unittest.mock import Mock
import pytest
from aki_prj23_transparenzregister.utils.news_mongo_service import MongoNewsService
@pytest.fixture()
def mock_mongo_connector(mocker) -> Mock:
"""Mock MongoConnector class.
Args:
mocker (any): Library mocker
Returns:
Mock: Mocked MongoConnector
"""
mock = Mock()
mocker.patch(
"aki_prj23_transparenzregister.utils.mongo.MongoConnector", return_value=mock
)
return mock
@pytest.fixture()
def mock_collection() -> Mock:
"""Mock mongo collection.
Returns:
Mock: Mock object
"""
return Mock()
def test_init(mock_mongo_connector, mock_collection):
"""Test CompanyMongoService constructor.
Args:
mock_mongo_connector (Mock): Mocked MongoConnector library
mock_collection (Mock): Mocked pymongo collection
"""
mock_mongo_connector.database = {"news": mock_collection}
service = MongoNewsService(mock_mongo_connector)
assert service.collection == mock_collection