47 lines
1.5 KiB
Python
47 lines
1.5 KiB
Python
"""Tests for _shared.heartbeat — Task A1."""
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
# Make `_shared` importable (same pattern as test_reliable_queue.py)
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))
|
|
|
|
from _shared.heartbeat import WorkerStats, build_payload, render_state
|
|
|
|
|
|
def test_build_payload_has_contract_fields():
|
|
s = WorkerStats(); s.jobs_done = 3; s.last_job_at = "2026-06-29T00:00:00Z"
|
|
payload = json.loads(build_payload("image-render", "render", "idle", s))
|
|
assert payload["name"] == "image-render"
|
|
assert payload["kind"] == "render"
|
|
assert payload["state"] == "idle"
|
|
assert payload["jobs_done"] == 3
|
|
assert payload["last_job_at"] == "2026-06-29T00:00:00Z"
|
|
assert payload["ts"].endswith("Z")
|
|
|
|
|
|
def test_build_payload_merges_extra():
|
|
payload = json.loads(build_payload("task-watcher", "watcher", "free", WorkerStats(), extra={"mode": "free"}))
|
|
assert payload["mode"] == "free"
|
|
|
|
|
|
class _FakeRedis:
|
|
def __init__(self, paused): self._paused = paused
|
|
async def get(self, key): return b"1" if self._paused else None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_render_state_paused_overrides_busy():
|
|
s = WorkerStats(); s.busy = True
|
|
assert await render_state(_FakeRedis(paused=True), s) == "paused"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_render_state_busy_then_idle():
|
|
s = WorkerStats(); s.busy = True
|
|
assert await render_state(_FakeRedis(paused=False), s) == "busy"
|
|
s.busy = False
|
|
assert await render_state(_FakeRedis(paused=False), s) == "idle"
|