from datetime import datetime
from worker_watcher.db.repositories import WatcherRepository
from worker_watcher.enums import HealthStatus, ObservationStatus, OperationalState, Severity
from worker_watcher.models import (
EvidenceItem,
InstanceDescriptor,
JobDescriptor,
JobObservation,
RunDescriptor,
)
def make_observation(now: datetime, *, exit_code: int | None = None, run_id: str | None = None) -> JobObservation:
job = JobDescriptor("test.worker.main", "Test worker", "test", "worker", stale_after_seconds=60)
instance = InstanceDescriptor(job, "test-host", "process", "test-worker", process_name="test-worker")
run = RunDescriptor(run_id, now, None, exit_code=exit_code, error_message="boom" if exit_code else None) if run_id or exit_code is not None else None
return JobObservation(job, instance, run, OperationalState.RUNNING if not exit_code else OperationalState.STOPPED,
HealthStatus.HEALTHY if not exit_code else HealthStatus.FAILED, ObservationStatus.FRESH,
Severity.INFO if not exit_code else Severity.CRITICAL, "test", now,
evidence=(EvidenceItem("fixture", True, "test", now),))
def test_migration_foreign_keys_and_wal(connection) -> None:
assert connection.execute("PRAGMA foreign_keys").fetchone()[0] == 1
assert connection.execute("PRAGMA journal_mode").fetchone()[0].lower() == "wal"
assert connection.execute("SELECT COUNT(*) FROM schema_migrations").fetchone()[0] == 2
def test_idempotent_migration(connection, tmp_path) -> None:
from worker_watcher.db.migrations import apply_migrations
assert apply_migrations(connection) == 2
def test_identical_scan_has_no_duplicate_status_event(connection, now) -> None:
repository = WatcherRepository(connection)
observation = make_observation(now)
repository.persist_observation(observation, None, now)
repository.persist_observation(observation, None, now)
assert connection.execute("SELECT COUNT(*) FROM status_events").fetchone()[0] == 1
def test_status_change_creates_one_event(connection, now) -> None:
repository = WatcherRepository(connection)
repository.persist_observation(make_observation(now), None, now)
repository.persist_observation(make_observation(now, exit_code=2), None, now)
assert connection.execute("SELECT COUNT(*) FROM status_events").fetchone()[0] == 2
assert connection.execute("SELECT event_type FROM status_events ORDER BY id DESC").fetchone()[0] == "error_raised"
def test_native_run_is_not_duplicated(connection, now) -> None:
repository = WatcherRepository(connection)
observation = make_observation(now, run_id="native-1")
repository.persist_observation(observation, None, now)
repository.persist_observation(observation, None, now)
assert connection.execute("SELECT COUNT(*) FROM job_runs").fetchone()[0] == 1
def test_unavailable_retains_previous_job_health(connection, now) -> None:
repository = WatcherRepository(connection)
repository.persist_observation(make_observation(now), None, now)
unavailable = make_observation(now.replace(minute=1))
unavailable = JobObservation(unavailable.job, unavailable.instance, unavailable.run, OperationalState.UNKNOWN,
HealthStatus.UNKNOWN, ObservationStatus.UNAVAILABLE, Severity.WARNING, "permission denied", unavailable.observed_at)
repository.persist_observation(unavailable, None, unavailable.observed_at)
row = connection.execute("SELECT operational_state, health_status, observation_status FROM job_status_current").fetchone()
assert tuple(row) == ("running", "healthy", "unavailable")