ik
f2ad5ecfa3
Bug: community_scheduler always had status='ready' even when individual sources were failed or stale. Success of one source masked failure of another. Fix: - Overall status is 'degraded' if any enabled source has failed - Overall status is 'stale' if all sources are stale but none failed - Overall status is 'ready' only when at least one source is healthy - Overall status is 'not_started' when no sources are enabled - Readiness (ready flag) still NOT blocked by import health (A01 requirement) Verification: - 7/7 readiness tests pass - 121/121 Python tests pass (1 skipped) - Failed/stale sources are now visible in JSON without blocking scheduler
196 lines
8.6 KiB
Python
196 lines
8.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 CommunityImportRun, DataSource, 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"
|
|
assert components["minio"]["status"] == "ready"
|
|
assert components["official_import"]["status"] == "optional"
|
|
assert components["official_import"]["last_run_status"] is None
|
|
assert "community_scheduler" in components
|
|
|
|
|
|
def test_required_import_success_shows_ready_status() -> None:
|
|
"""A01: Successful import is diagnostic, not blocking."""
|
|
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"
|
|
assert components["official_import"]["blocking"] is False
|
|
|
|
|
|
def test_unavailable_storage_blocks_readiness_but_stale_import_does_not() -> None:
|
|
"""A01: Infrastructure failures block, stale imports are diagnostic only."""
|
|
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"
|
|
assert components["official_import"]["blocking"] is False
|
|
|
|
|
|
def test_community_scheduler_success_shows_ready_status() -> None:
|
|
"""A01: Successful scheduler is diagnostic, not blocking."""
|
|
engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(engine)
|
|
now = datetime.now(timezone.utc)
|
|
with Session(engine) as session:
|
|
session.add(DataSource(key="rf4db", name="RF4DB", base_url="https://rf4db.com", default_confidence=70, enabled=True))
|
|
session.add(CommunityImportRun(
|
|
source_system="rf4db",
|
|
started_at=now - timedelta(minutes=30),
|
|
status="success",
|
|
source_url="fixture://rf4db",
|
|
rows_seen=5, rows_created=5, rows_updated=0, error_summary=None,
|
|
))
|
|
session.commit()
|
|
ready, components = readiness_report(
|
|
session, AvailableStorage(), import_required=False,
|
|
import_interval_seconds=3600, community_import_interval_seconds=1800, now=now,
|
|
)
|
|
assert ready is True
|
|
assert components["community_scheduler"]["status"] == "ready"
|
|
assert "rf4db" in components["community_scheduler"]["sources"]
|
|
assert components["community_scheduler"]["sources"]["rf4db"]["blocking"] is False
|
|
|
|
|
|
def test_community_scheduler_stale_does_not_block_readiness() -> None:
|
|
"""A01: Stale scheduler is diagnostic, never blocks readiness."""
|
|
engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(engine)
|
|
now = datetime.now(timezone.utc)
|
|
with Session(engine) as session:
|
|
session.add(DataSource(key="rf4db", name="RF4DB", base_url="https://rf4db.com", default_confidence=70, enabled=True))
|
|
session.add(CommunityImportRun(
|
|
source_system="rf4db",
|
|
started_at=now - timedelta(hours=2),
|
|
status="success",
|
|
source_url="fixture://rf4db",
|
|
rows_seen=5, rows_created=5, rows_updated=0, error_summary=None,
|
|
))
|
|
session.commit()
|
|
ready, components = readiness_report(
|
|
session, AvailableStorage(), import_required=False,
|
|
import_interval_seconds=3600, community_import_interval_seconds=1800, now=now,
|
|
)
|
|
assert ready is True # A01: stale does NOT block readiness
|
|
assert components["community_scheduler"]["status"] == "stale" # Overall reflects stale source
|
|
assert components["community_scheduler"]["sources"]["rf4db"]["status"] == "stale"
|
|
assert components["community_scheduler"]["sources"]["rf4db"]["blocking"] is False
|
|
|
|
|
|
def test_community_scheduler_failed_does_not_block_readiness() -> None:
|
|
"""A01: Failed scheduler is diagnostic, never blocks readiness."""
|
|
engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(engine)
|
|
now = datetime.now(timezone.utc)
|
|
with Session(engine) as session:
|
|
session.add(DataSource(key="rf4db", name="RF4DB", base_url="https://rf4db.com", default_confidence=70, enabled=True))
|
|
session.add(CommunityImportRun(
|
|
source_system="rf4db",
|
|
started_at=now - timedelta(minutes=30),
|
|
status="failed",
|
|
source_url="fixture://rf4db",
|
|
rows_seen=0, rows_created=0, rows_updated=0,
|
|
error_summary="ConnectionError",
|
|
))
|
|
session.commit()
|
|
ready, components = readiness_report(
|
|
session, AvailableStorage(), import_required=False,
|
|
import_interval_seconds=3600, community_import_interval_seconds=1800, now=now,
|
|
)
|
|
assert ready is True # A01: failed does NOT block readiness
|
|
assert components["community_scheduler"]["status"] == "degraded" # Overall reflects failed source
|
|
assert components["community_scheduler"]["sources"]["rf4db"]["status"] == "failed"
|
|
assert components["community_scheduler"]["sources"]["rf4db"]["blocking"] is False
|
|
|
|
|
|
def test_community_scheduler_tracked_per_source_with_backoff() -> None:
|
|
"""A01: Per-source health tracking with backoff detection."""
|
|
engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(engine)
|
|
now = datetime.now(timezone.utc)
|
|
with Session(engine) as session:
|
|
session.add(DataSource(key="rf4db", name="RF4DB", base_url="https://rf4db.com", default_confidence=70, enabled=True))
|
|
session.add(DataSource(key="rf4stat-fishing", name="RF4-STAT", base_url="https://rf4-stat.ru", default_confidence=65, enabled=True))
|
|
# rf4db: healthy
|
|
session.add(CommunityImportRun(
|
|
source_system="rf4db",
|
|
started_at=now - timedelta(minutes=30),
|
|
status="success",
|
|
source_url="fixture://rf4db",
|
|
rows_seen=5, rows_created=5, rows_updated=0, error_summary=None,
|
|
))
|
|
# rf4stat-fishing: multiple recent failures → backoff recommended
|
|
for i in range(6):
|
|
session.add(CommunityImportRun(
|
|
source_system="rf4stat-fishing",
|
|
started_at=now - timedelta(hours=i),
|
|
status="failed",
|
|
source_url="fixture://rf4stat",
|
|
rows_seen=0, rows_created=0, rows_updated=0,
|
|
error_summary="TimeoutError",
|
|
))
|
|
session.commit()
|
|
ready, components = readiness_report(
|
|
session, AvailableStorage(), import_required=False,
|
|
import_interval_seconds=3600, community_import_interval_seconds=1800, now=now,
|
|
)
|
|
assert ready is True
|
|
sources = components["community_scheduler"]["sources"]
|
|
assert sources["rf4db"]["status"] == "ready"
|
|
assert sources["rf4db"]["recent_failures_24h"] == 0
|
|
assert sources["rf4db"]["backoff_recommended"] is False
|
|
assert sources["rf4stat-fishing"]["status"] == "failed"
|
|
assert sources["rf4stat-fishing"]["recent_failures_24h"] == 6
|
|
assert sources["rf4stat-fishing"]["backoff_recommended"] is True
|