75 lines
2.6 KiB
Python
75 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.database import Base
|
|
from app.models import ImportStatus, OfficialRecordImport
|
|
from app.readiness import readiness_report
|
|
|
|
|
|
class AvailableStorage:
|
|
def list_buckets(self) -> dict[str, list[object]]:
|
|
return {"Buckets": []}
|
|
|
|
|
|
class UnavailableStorage:
|
|
def list_buckets(self) -> None:
|
|
raise ConnectionError("fixture unavailable")
|
|
|
|
|
|
def test_optional_import_does_not_block_dependencies() -> None:
|
|
engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
ready, components = readiness_report(
|
|
session, AvailableStorage(), import_required=False, import_interval_seconds=3600,
|
|
)
|
|
assert ready is True
|
|
assert components == {
|
|
"postgresql": {"status": "ready"},
|
|
"minio": {"status": "ready"},
|
|
"official_import": {"status": "optional", "last_run_status": None},
|
|
}
|
|
|
|
|
|
def test_required_import_must_be_recent_and_successful() -> None:
|
|
engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(engine)
|
|
now = datetime.now(timezone.utc)
|
|
with Session(engine) as session:
|
|
session.add(OfficialRecordImport(
|
|
started_at=now - timedelta(minutes=30), finished_at=now - timedelta(minutes=29),
|
|
status=ImportStatus.success, source_url="fixture://records", rows_seen=1,
|
|
rows_created=1, rows_updated=0,
|
|
))
|
|
session.commit()
|
|
ready, components = readiness_report(
|
|
session, AvailableStorage(), import_required=True,
|
|
import_interval_seconds=3600, now=now,
|
|
)
|
|
assert ready is True
|
|
assert components["official_import"]["status"] == "ready"
|
|
|
|
|
|
def test_unavailable_storage_and_stale_import_fail_readiness() -> None:
|
|
engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(engine)
|
|
now = datetime.now(timezone.utc)
|
|
with Session(engine) as session:
|
|
session.add(OfficialRecordImport(
|
|
started_at=now - timedelta(hours=3), finished_at=now - timedelta(hours=3),
|
|
status=ImportStatus.success, source_url="fixture://records", rows_seen=1,
|
|
rows_created=1, rows_updated=0,
|
|
))
|
|
session.commit()
|
|
ready, components = readiness_report(
|
|
session, UnavailableStorage(), import_required=True,
|
|
import_interval_seconds=3600, now=now,
|
|
)
|
|
assert ready is False
|
|
assert components["minio"]["status"] == "unavailable"
|
|
assert components["official_import"]["status"] == "stale"
|