169 lines
5.1 KiB
Python
169 lines
5.1 KiB
Python
"""Tests for GET /api/world/state endpoint."""
|
|
|
|
import json
|
|
import time
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from dashboard.routes.world import (
|
|
_STALE_THRESHOLD,
|
|
_build_world_state,
|
|
_read_presence_file,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _build_world_state
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_build_world_state_maps_fields():
|
|
presence = {
|
|
"version": 1,
|
|
"liveness": "2026-03-19T02:00:00Z",
|
|
"mood": "exploring",
|
|
"current_focus": "reviewing PR",
|
|
"energy": 0.8,
|
|
"confidence": 0.9,
|
|
"active_threads": [{"type": "thinking", "ref": "test", "status": "active"}],
|
|
"recent_events": [],
|
|
"concerns": [],
|
|
}
|
|
result = _build_world_state(presence)
|
|
|
|
assert result["timmyState"]["mood"] == "exploring"
|
|
assert result["timmyState"]["activity"] == "reviewing PR"
|
|
assert result["timmyState"]["energy"] == 0.8
|
|
assert result["timmyState"]["confidence"] == 0.9
|
|
assert result["updatedAt"] == "2026-03-19T02:00:00Z"
|
|
assert result["version"] == 1
|
|
assert result["visitorPresent"] is False
|
|
assert len(result["activeThreads"]) == 1
|
|
|
|
|
|
def test_build_world_state_defaults():
|
|
"""Missing fields get safe defaults."""
|
|
result = _build_world_state({})
|
|
assert result["timmyState"]["mood"] == "focused"
|
|
assert result["timmyState"]["energy"] == 0.5
|
|
assert result["version"] == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _read_presence_file
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_read_presence_file_missing(tmp_path):
|
|
with patch("dashboard.routes.world._PRESENCE_FILE", tmp_path / "nope.json"):
|
|
assert _read_presence_file() is None
|
|
|
|
|
|
def test_read_presence_file_stale(tmp_path):
|
|
f = tmp_path / "presence.json"
|
|
f.write_text(json.dumps({"version": 1}))
|
|
# Backdate the file
|
|
stale_time = time.time() - _STALE_THRESHOLD - 10
|
|
import os
|
|
|
|
os.utime(f, (stale_time, stale_time))
|
|
with patch("dashboard.routes.world._PRESENCE_FILE", f):
|
|
assert _read_presence_file() is None
|
|
|
|
|
|
def test_read_presence_file_fresh(tmp_path):
|
|
f = tmp_path / "presence.json"
|
|
f.write_text(json.dumps({"version": 1, "mood": "focused"}))
|
|
with patch("dashboard.routes.world._PRESENCE_FILE", f):
|
|
result = _read_presence_file()
|
|
assert result is not None
|
|
assert result["version"] == 1
|
|
|
|
|
|
def test_read_presence_file_bad_json(tmp_path):
|
|
f = tmp_path / "presence.json"
|
|
f.write_text("not json {{{")
|
|
with patch("dashboard.routes.world._PRESENCE_FILE", f):
|
|
assert _read_presence_file() is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Full endpoint via TestClient
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
app = FastAPI()
|
|
from dashboard.routes.world import router
|
|
|
|
app.include_router(router)
|
|
return TestClient(app)
|
|
|
|
|
|
def test_world_state_endpoint_with_file(client, tmp_path):
|
|
"""Endpoint returns data from presence file when fresh."""
|
|
f = tmp_path / "presence.json"
|
|
f.write_text(
|
|
json.dumps(
|
|
{
|
|
"version": 1,
|
|
"liveness": "2026-03-19T02:00:00Z",
|
|
"mood": "exploring",
|
|
"current_focus": "testing",
|
|
"active_threads": [],
|
|
"recent_events": [],
|
|
"concerns": [],
|
|
}
|
|
)
|
|
)
|
|
with patch("dashboard.routes.world._PRESENCE_FILE", f):
|
|
resp = client.get("/api/world/state")
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["timmyState"]["mood"] == "exploring"
|
|
assert data["timmyState"]["activity"] == "testing"
|
|
assert resp.headers["cache-control"] == "no-cache, no-store"
|
|
|
|
|
|
def test_world_state_endpoint_fallback(client, tmp_path):
|
|
"""Endpoint falls back to live state when file missing."""
|
|
with (
|
|
patch("dashboard.routes.world._PRESENCE_FILE", tmp_path / "nope.json"),
|
|
patch("timmy.workshop_state.get_state_dict") as mock_get,
|
|
):
|
|
mock_get.return_value = {
|
|
"version": 1,
|
|
"liveness": "2026-03-19T02:00:00Z",
|
|
"mood": "idle",
|
|
"current_focus": "",
|
|
"active_threads": [],
|
|
"recent_events": [],
|
|
"concerns": [],
|
|
}
|
|
resp = client.get("/api/world/state")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json()["timmyState"]["mood"] == "idle"
|
|
|
|
|
|
def test_world_state_endpoint_full_fallback(client, tmp_path):
|
|
"""Endpoint returns safe defaults when everything fails."""
|
|
with (
|
|
patch("dashboard.routes.world._PRESENCE_FILE", tmp_path / "nope.json"),
|
|
patch(
|
|
"timmy.workshop_state.get_state_dict",
|
|
side_effect=RuntimeError("boom"),
|
|
),
|
|
):
|
|
resp = client.get("/api/world/state")
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["timmyState"]["mood"] == "idle"
|
|
assert data["version"] == 1
|