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

test: Cover apps/find_missing_companies.py

Browse Source
This commit is contained in:
TrisNol 2023-11-05 13:47:06 +01:00
parent f7ec3eaf24
commit f9d3f0eb76
3 changed files with 32 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
src/aki_prj23_transparenzregister/apps/find_missing_companies.py
View File

@ -33,18 +33,18 @@ from aki_prj23_transparenzregister.utils.mongo.connector import MongoConnector
from aki_prj23_transparenzregister.utils.sql import connector, entities
def work(company: entities.Company, config_provider: ConfigProvider) -> None:
def work(company_name: str, config_provider: ConfigProvider) -> None:
"""Main method.
Args:
company (entities.Company): Company to be searched for
company_name (str): Name of the company to search for
config_provider (ConfigProvider): ConfigProvider
"""
with tempfile.TemporaryDirectory() as tmp_dir:
xml_dir = os.path.join(*[tmp_dir, "xml"])
os.makedirs(xml_dir, exist_ok=True)
try:
extract.scrape(company.name, xml_dir, True, True) # type: ignore
extract.scrape(company_name, xml_dir, True, True) # type: ignore
except Exception as e:
logger.error(e)
return
@ -90,12 +90,12 @@ def work(company: entities.Company, config_provider: ConfigProvider) -> None:
with connector.get_session(config_provider) as session:
company = (
session.query(entities.MissingCompany) # type: ignore
.where(entities.MissingCompany.name == company.name)
.where(entities.MissingCompany.name == company_name)
.first()
)
company.searched_for = True # type: ignore
session.commit()
logger.info(f"Processed {company.name}")
logger.info(f"Processed {company_name}")
except Exception as e:
logger.error(e)
return
@ -133,7 +133,7 @@ if __name__ == "__main__":
batch_size = 5
pool = multiprocessing.Pool(processes=batch_size)
# Scrape data from unternehmensregister
params = [(company, config_provider) for company in missing_companies]
params = [(company.name, config_provider) for company in missing_companies]
# Map the process_handler function to the parameter list using the Pool
pool.starmap(work, params)

2
src/aki_prj23_transparenzregister/utils/data_extraction/unternehmensregister/extract.py
View File

@ -117,7 +117,7 @@ def scrape(
try:
wait.until(
lambda: wait_for_download_condition(download_path, num_files) # type: ignore
lambda x: wait_for_download_condition(download_path, num_files) # type: ignore
)
file_name = "".join(e for e in company_name if e.isalnum()) + ".xml"
rename_latest_file(

25
tests/apps/find_missing_companies_test.py
View File

@ -1,6 +1,31 @@
"""Testing find_missing_companies.py."""
from unittest.mock import Mock, patch
from aki_prj23_transparenzregister.apps import find_missing_companies
def test_import_find_missing_companies() -> None:
assert find_missing_companies
@patch("aki_prj23_transparenzregister.apps.find_missing_companies.MongoConnector")
@patch("aki_prj23_transparenzregister.apps.find_missing_companies.CompanyMongoService")
@patch(
"aki_prj23_transparenzregister.apps.find_missing_companies.load.load_directory_to_mongo"
)
@patch("aki_prj23_transparenzregister.apps.find_missing_companies.connector")
def test_work(
connector_mock: Mock,
load_directory_to_mongo_mock: Mock,
company_mongo_service_mock: Mock,
mongo_connector_mock: Mock,
) -> None:
config_provider_mock = Mock()
config_provider_mock.session.return_value = Mock()
load_directory_to_mongo_mock.return_value = 42
find_missing_companies.work(
"Atos IT-Dienstleistung und Beratung GmbH", config_provider_mock
)
assert True