df98fc8bc5
- Add on_change async callback to WorkshopHeartbeat - Add /api/world/ws WebSocket endpoint for Workshop clients - broadcast_world_state() pushes timmy_state messages on change - Wire heartbeat to broadcaster at startup in app.py Fixes #375
222 lines
7.2 KiB
Python
222 lines
7.2 KiB
Python
"""Tests for Workshop presence heartbeat."""
|
|
|
|
import json
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from timmy.workshop_state import (
|
|
WorkshopHeartbeat,
|
|
_state_hash,
|
|
get_state_dict,
|
|
write_state,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# get_state_dict
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_get_state_dict_returns_v1_schema():
|
|
state = get_state_dict()
|
|
assert state["version"] == 1
|
|
assert "liveness" in state
|
|
assert "current_focus" in state
|
|
assert "mood" in state
|
|
assert isinstance(state["active_threads"], list)
|
|
assert isinstance(state["recent_events"], list)
|
|
assert isinstance(state["concerns"], list)
|
|
|
|
|
|
def test_get_state_dict_idle_mood():
|
|
"""Idle engagement + settled mood → 'idle' presence mood."""
|
|
from timmy.cognitive_state import CognitiveState, CognitiveTracker
|
|
|
|
tracker = CognitiveTracker.__new__(CognitiveTracker)
|
|
tracker.state = CognitiveState(engagement="idle", mood="settled")
|
|
with patch("timmy.cognitive_state.cognitive_tracker", tracker):
|
|
state = get_state_dict()
|
|
assert state["mood"] == "idle"
|
|
|
|
|
|
def test_get_state_dict_maps_mood():
|
|
"""Cognitive moods map to presence moods."""
|
|
from timmy.cognitive_state import CognitiveState, CognitiveTracker
|
|
|
|
for cog_mood, expected in [
|
|
("curious", "exploring"),
|
|
("hesitant", "uncertain"),
|
|
("energized", "excited"),
|
|
]:
|
|
tracker = CognitiveTracker.__new__(CognitiveTracker)
|
|
tracker.state = CognitiveState(engagement="deep", mood=cog_mood)
|
|
with patch("timmy.cognitive_state.cognitive_tracker", tracker):
|
|
state = get_state_dict()
|
|
assert state["mood"] == expected, f"Expected {expected} for {cog_mood}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# write_state
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_write_state_creates_file(tmp_path):
|
|
target = tmp_path / "presence.json"
|
|
state = {"version": 1, "liveness": "2026-01-01T00:00:00Z", "current_focus": ""}
|
|
write_state(state, path=target)
|
|
|
|
assert target.exists()
|
|
data = json.loads(target.read_text())
|
|
assert data["version"] == 1
|
|
|
|
|
|
def test_write_state_creates_parent_dirs(tmp_path):
|
|
target = tmp_path / "deep" / "nested" / "presence.json"
|
|
write_state({"version": 1}, path=target)
|
|
assert target.exists()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _state_hash
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_state_hash_ignores_liveness():
|
|
a = {"version": 1, "mood": "focused", "liveness": "2026-01-01T00:00:00Z"}
|
|
b = {"version": 1, "mood": "focused", "liveness": "2026-12-31T23:59:59Z"}
|
|
assert _state_hash(a) == _state_hash(b)
|
|
|
|
|
|
def test_state_hash_detects_mood_change():
|
|
a = {"version": 1, "mood": "focused", "liveness": "t1"}
|
|
b = {"version": 1, "mood": "idle", "liveness": "t1"}
|
|
assert _state_hash(a) != _state_hash(b)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# WorkshopHeartbeat — _write_if_changed
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_if_changed_writes_on_first_call(tmp_path):
|
|
target = tmp_path / "presence.json"
|
|
hb = WorkshopHeartbeat(interval=60, path=target)
|
|
|
|
with patch("timmy.workshop_state.get_state_dict") as mock_state:
|
|
mock_state.return_value = {
|
|
"version": 1,
|
|
"liveness": "t1",
|
|
"current_focus": "testing",
|
|
"mood": "focused",
|
|
}
|
|
await hb._write_if_changed()
|
|
|
|
assert target.exists()
|
|
data = json.loads(target.read_text())
|
|
assert data["version"] == 1
|
|
assert data["current_focus"] == "testing"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_if_changed_skips_when_unchanged(tmp_path):
|
|
target = tmp_path / "presence.json"
|
|
hb = WorkshopHeartbeat(interval=60, path=target)
|
|
|
|
fixed_state = {"version": 1, "liveness": "t1", "mood": "idle"}
|
|
|
|
with patch("timmy.workshop_state.get_state_dict", return_value=fixed_state):
|
|
await hb._write_if_changed() # First write
|
|
target.write_text("") # Clear to detect if second write happens
|
|
await hb._write_if_changed() # Should skip — state unchanged
|
|
|
|
# File should still be empty (second write was skipped)
|
|
assert target.read_text() == ""
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_if_changed_writes_on_state_change(tmp_path):
|
|
target = tmp_path / "presence.json"
|
|
hb = WorkshopHeartbeat(interval=60, path=target)
|
|
|
|
state_a = {"version": 1, "liveness": "t1", "mood": "idle"}
|
|
state_b = {"version": 1, "liveness": "t2", "mood": "focused"}
|
|
|
|
with patch("timmy.workshop_state.get_state_dict", return_value=state_a):
|
|
await hb._write_if_changed()
|
|
|
|
with patch("timmy.workshop_state.get_state_dict", return_value=state_b):
|
|
await hb._write_if_changed()
|
|
|
|
data = json.loads(target.read_text())
|
|
assert data["mood"] == "focused"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_if_changed_calls_on_change(tmp_path):
|
|
"""on_change callback is invoked with state dict when state changes."""
|
|
target = tmp_path / "presence.json"
|
|
received = []
|
|
|
|
async def capture(state_dict):
|
|
received.append(state_dict)
|
|
|
|
hb = WorkshopHeartbeat(interval=60, path=target, on_change=capture)
|
|
|
|
state = {"version": 1, "liveness": "t1", "mood": "focused"}
|
|
with patch("timmy.workshop_state.get_state_dict", return_value=state):
|
|
await hb._write_if_changed()
|
|
|
|
assert len(received) == 1
|
|
assert received[0]["mood"] == "focused"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_if_changed_skips_on_change_when_unchanged(tmp_path):
|
|
"""on_change is NOT called when state hash is unchanged."""
|
|
target = tmp_path / "presence.json"
|
|
call_count = 0
|
|
|
|
async def counter(_):
|
|
nonlocal call_count
|
|
call_count += 1
|
|
|
|
hb = WorkshopHeartbeat(interval=60, path=target, on_change=counter)
|
|
|
|
state = {"version": 1, "liveness": "t1", "mood": "idle"}
|
|
with patch("timmy.workshop_state.get_state_dict", return_value=state):
|
|
await hb._write_if_changed()
|
|
await hb._write_if_changed()
|
|
|
|
assert call_count == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# WorkshopHeartbeat — notify
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_notify_sets_trigger():
|
|
hb = WorkshopHeartbeat(interval=60)
|
|
assert not hb._trigger.is_set()
|
|
hb.notify()
|
|
assert hb._trigger.is_set()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# WorkshopHeartbeat — start/stop lifecycle
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_heartbeat_start_stop_lifecycle(tmp_path):
|
|
target = tmp_path / "presence.json"
|
|
hb = WorkshopHeartbeat(interval=60, path=target)
|
|
|
|
with patch("timmy.workshop_state.get_state_dict", return_value={"version": 1}):
|
|
await hb.start()
|
|
assert hb._task is not None
|
|
assert not hb._task.done()
|
|
await hb.stop()
|
|
assert hb._task is None
|