mirror of
https://github.com/fhswf/aki_prj23_transparenzregister.git
synced 2025-04-22 16:12:55 +02:00
92d14b5824
Includes a new "app" running the ingestion jobs (aka fetch_news and find_missing_companies + enrich_company_financials) on a schedule. This also fixes an issue with the previous schedule implementation by persisting the schedule in a file that survives new deployment and continues where it left off.
42 lines
1.2 KiB
Python
42 lines
1.2 KiB
Python
"""Testing apps/ingest.py."""
|
|
import json
|
|
import tempfile
|
|
from unittest.mock import Mock, patch
|
|
|
|
from aki_prj23_transparenzregister.apps import ingest
|
|
|
|
|
|
def test_import() -> None:
|
|
assert ingest
|
|
|
|
|
|
def test_load_schedule() -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
data = {"test": "test"}
|
|
path = f"{temp_dir}/schedule.json"
|
|
with open(path, "w") as file:
|
|
json.dump({"test": "test"}, file)
|
|
assert ingest.load_schedule(path) == data
|
|
|
|
|
|
def test_load_scheduler_no_result() -> None:
|
|
assert ingest.load_schedule("./hello_there.json") == {}
|
|
|
|
|
|
def test_save_schedule() -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
data = {"test": "test"}
|
|
path = f"{temp_dir}/schedule.json"
|
|
|
|
ingest.save_schedule(data, path)
|
|
with open(path) as file:
|
|
assert json.load(file) == data
|
|
|
|
|
|
@patch("aki_prj23_transparenzregister.apps.ingest.find_missing_companies_main")
|
|
@patch("aki_prj23_transparenzregister.apps.ingest.enrich_company_financials_main")
|
|
def test_main(mock_financials: Mock, mock_find_missing: Mock) -> None:
|
|
ingest.main(Mock())
|
|
assert mock_financials.called
|
|
assert mock_find_missing.called
|