170 lines
5.2 KiB
Python
170 lines
5.2 KiB
Python
from datetime import timedelta
|
|
|
|
import pytest
|
|
from django.utils import timezone
|
|
|
|
from apps.sources.models import RawDocument, Source, SourceRun
|
|
from apps.sources.services.health import collect_source_health
|
|
from apps.sources.tasks import cleanup_raw_documents, enforce_source_health, recover_source_health
|
|
|
|
|
|
def _create_run(
|
|
source: Source,
|
|
*,
|
|
status: str,
|
|
started_at=None,
|
|
error_category: str = "",
|
|
extracted_count: int = 0,
|
|
created_count: int = 0,
|
|
updated_count: int = 0,
|
|
duplicate_count: int = 0,
|
|
http_status: int = 200,
|
|
metrics: dict[str, object] | None = None,
|
|
):
|
|
started_at = started_at or timezone.now()
|
|
finished_at = started_at + timedelta(seconds=1)
|
|
return SourceRun.objects.create(
|
|
source=source,
|
|
started_at=started_at,
|
|
finished_at=finished_at,
|
|
status=status,
|
|
error_category=error_category,
|
|
http_status=http_status,
|
|
extracted_count=extracted_count,
|
|
created_count=created_count,
|
|
updated_count=updated_count,
|
|
duplicate_count=duplicate_count,
|
|
metrics=metrics or {},
|
|
)
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_enforce_source_health_quarantines_on_temporary_failures(source):
|
|
for index in range(6):
|
|
_create_run(
|
|
source,
|
|
status=SourceRun.Status.FAILED,
|
|
started_at=timezone.now() - timedelta(minutes=index),
|
|
error_category="timeout",
|
|
)
|
|
|
|
result = enforce_source_health()
|
|
assert result["quarantined"] == 1
|
|
source.refresh_from_db()
|
|
assert source.status == Source.Status.QUARANTINED
|
|
assert source.policy == Source.Policy.DENY
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_enforce_source_health_quarantines_on_policy_category(source):
|
|
for index in range(3):
|
|
_create_run(
|
|
source,
|
|
status=SourceRun.Status.FAILED,
|
|
started_at=timezone.now() - timedelta(minutes=index),
|
|
error_category="policy",
|
|
)
|
|
result = enforce_source_health()
|
|
assert result["quarantined"] == 1
|
|
source.refresh_from_db()
|
|
assert source.status == Source.Status.QUARANTINED
|
|
assert "beleid" in source.policy_reason
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_enforce_source_health_quarantines_on_parser_drift(source):
|
|
for index in range(3):
|
|
_create_run(
|
|
source,
|
|
status=SourceRun.Status.SUCCESS,
|
|
started_at=timezone.now() - timedelta(minutes=index),
|
|
extracted_count=0,
|
|
created_count=0,
|
|
updated_count=0,
|
|
duplicate_count=0,
|
|
http_status=200,
|
|
metrics={"parser": "generic-html", "warnings": ["veld ontbreekt", "datum ontbreekt"]},
|
|
)
|
|
assert enforce_source_health()["quarantined"] == 1
|
|
source.refresh_from_db()
|
|
assert source.status == Source.Status.QUARANTINED
|
|
assert source.policy == Source.Policy.DENY
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_collect_source_health_exposes_parser_and_counts(source):
|
|
_create_run(
|
|
source,
|
|
status=SourceRun.Status.SUCCESS,
|
|
error_category="",
|
|
extracted_count=2,
|
|
created_count=1,
|
|
updated_count=1,
|
|
duplicate_count=0,
|
|
metrics={"parser": "jsonld", "warnings": ["warn"]},
|
|
)
|
|
row = collect_source_health()[0]
|
|
assert row.source_id == source.pk
|
|
assert row.last_parser == "jsonld"
|
|
assert row.last_parser_warnings == 1
|
|
assert row.extracted_count == 2
|
|
assert row.updated_count == 1
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_recover_source_health_starts_canary_once(source, monkeypatch):
|
|
source.status = Source.Status.QUARANTINED
|
|
source.policy = Source.Policy.DENY
|
|
source.metadata = {
|
|
"source_health": {
|
|
"state": "quarantined",
|
|
"quarantine_reason": "tijdelijke fout",
|
|
"quarantined_at": timezone.now().isoformat(),
|
|
"recovery_due_at": timezone.now().isoformat(),
|
|
"canary_started": False,
|
|
}
|
|
}
|
|
source.save(update_fields=["status", "policy", "metadata"])
|
|
|
|
calls: list[int] = []
|
|
|
|
def fake_delay(source_id: int, force: bool = False):
|
|
calls.append(source_id)
|
|
return source_id
|
|
|
|
monkeypatch.setattr("apps.sources.tasks.fetch_source.delay", fake_delay)
|
|
|
|
result = recover_source_health()
|
|
assert result["canary_started"] == 1
|
|
assert calls == [source.pk]
|
|
|
|
source.refresh_from_db()
|
|
assert source.status == Source.Status.TRIAL
|
|
assert source.policy == Source.Policy.REVIEW
|
|
|
|
# Canary is eenmalig, niet continu herstartbaar.
|
|
source.refresh_from_db()
|
|
assert recover_source_health()["canary_started"] == 0
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_cleanup_retains_quarantined_raw_documents(source):
|
|
old = RawDocument.objects.create(
|
|
source=source,
|
|
kind=RawDocument.Kind.TEXT,
|
|
content_hash="a" * 64,
|
|
retain_until=timezone.now() - timedelta(days=1),
|
|
)
|
|
keep = RawDocument.objects.create(
|
|
source=source,
|
|
kind=RawDocument.Kind.TEXT,
|
|
content_hash="b" * 64,
|
|
retain_until=timezone.now() - timedelta(days=1),
|
|
quarantined=True,
|
|
)
|
|
|
|
result = cleanup_raw_documents()
|
|
assert result["deleted"] == 1
|
|
assert RawDocument.objects.filter(pk=old.pk).exists() is False
|
|
assert RawDocument.objects.filter(pk=keep.pk).exists() is True
|