96 lines
2.8 KiB
Python
96 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import date
|
|
|
|
import pytest
|
|
from django.core.management import call_command
|
|
|
|
from apps.ingestion.services.daily_orchestration import parse_enabled_extractors, run_daily_orchestration
|
|
|
|
|
|
@dataclass
|
|
class _FakeExtractorResult:
|
|
records_count: int
|
|
output_path: str
|
|
|
|
|
|
class _FakeExtractor:
|
|
def __init__(self, name: str):
|
|
self.name = name
|
|
|
|
def run(self, *, snapshot_date=None):
|
|
if snapshot_date:
|
|
return _FakeExtractorResult(records_count=3, output_path=f"/tmp/{self.name}-{snapshot_date}.json")
|
|
return _FakeExtractorResult(records_count=3, output_path=f"/tmp/{self.name}.json")
|
|
|
|
|
|
@dataclass
|
|
class _FakeImportRun:
|
|
id: int = 11
|
|
status: str = "success"
|
|
files_processed: int = 2
|
|
files_total: int = 2
|
|
rows_upserted: int = 20
|
|
rows_failed: int = 0
|
|
|
|
|
|
class _FakeImporter:
|
|
def __init__(self, **_kwargs):
|
|
pass
|
|
|
|
def run(self):
|
|
return _FakeImportRun()
|
|
|
|
|
|
def test_parse_enabled_extractors():
|
|
assert parse_enabled_extractors("lba,bcl") == ["lba", "bcl"]
|
|
assert parse_enabled_extractors(" lba , , bcl ") == ["lba", "bcl"]
|
|
assert parse_enabled_extractors("") == []
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_daily_orchestration_runs_extractors_then_import(settings, monkeypatch):
|
|
settings.DAILY_ORCHESTRATION_EXTRACTORS = "lba,bcl"
|
|
|
|
monkeypatch.setattr(
|
|
"apps.ingestion.services.daily_orchestration.create_extractor",
|
|
lambda name: _FakeExtractor(name),
|
|
)
|
|
monkeypatch.setattr(
|
|
"apps.ingestion.services.daily_orchestration.SnapshotImporter",
|
|
_FakeImporter,
|
|
)
|
|
|
|
result = run_daily_orchestration(snapshot_date=date(2026, 3, 13))
|
|
assert [row.extractor_name for row in result.extractors_run] == ["lba", "bcl"]
|
|
assert result.import_run_id == 11
|
|
assert result.import_status == "success"
|
|
assert result.rows_upserted == 20
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_daily_orchestration_raises_when_no_extractors_configured(settings):
|
|
settings.DAILY_ORCHESTRATION_EXTRACTORS = ""
|
|
with pytest.raises(ValueError, match="cannot be empty"):
|
|
run_daily_orchestration()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_run_daily_orchestration_command(settings, monkeypatch, capsys):
|
|
settings.DAILY_ORCHESTRATION_EXTRACTORS = "lba,bcl"
|
|
|
|
monkeypatch.setattr(
|
|
"apps.ingestion.services.daily_orchestration.create_extractor",
|
|
lambda name: _FakeExtractor(name),
|
|
)
|
|
monkeypatch.setattr(
|
|
"apps.ingestion.services.daily_orchestration.SnapshotImporter",
|
|
_FakeImporter,
|
|
)
|
|
|
|
call_command("run_daily_orchestration", "--snapshot-date", "2026-03-13")
|
|
captured = capsys.readouterr()
|
|
assert "Daily orchestration completed" in captured.out
|
|
assert "extractors=[lba:3, bcl:3]" in captured.out
|