Add produce_agent_state() to infrastructure/presence that converts ADR-023 presence dicts into Matrix-compatible agent_state messages with type, agent_id, data (display_name, role, status, mood, energy, bark), and Unix timestamp. Includes status derivation from current_focus (thinking/speaking/idle/online) and comprehensive tests covering all fields, defaults, and status mapping. Fixes #669 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
166 lines
6.4 KiB
Python
166 lines
6.4 KiB
Python
"""Tests for infrastructure.presence — presence state serializer."""
|
|
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from infrastructure.presence import produce_agent_state, serialize_presence
|
|
|
|
|
|
class TestSerializePresence:
|
|
"""Round-trip and edge-case tests for serialize_presence()."""
|
|
|
|
@pytest.fixture()
|
|
def full_presence(self):
|
|
"""A complete ADR-023 presence dict."""
|
|
return {
|
|
"version": 1,
|
|
"liveness": "2026-03-21T12:00:00Z",
|
|
"current_focus": "writing tests",
|
|
"mood": "focused",
|
|
"energy": 0.9,
|
|
"confidence": 0.85,
|
|
"active_threads": [
|
|
{"type": "thinking", "ref": "refactor presence", "status": "active"}
|
|
],
|
|
"recent_events": ["committed code"],
|
|
"concerns": ["test coverage"],
|
|
"familiar": {"name": "Pip", "state": "alert"},
|
|
}
|
|
|
|
def test_full_round_trip(self, full_presence):
|
|
"""All ADR-023 fields map to the expected camelCase keys."""
|
|
result = serialize_presence(full_presence)
|
|
|
|
assert result["timmyState"]["mood"] == "focused"
|
|
assert result["timmyState"]["activity"] == "writing tests"
|
|
assert result["timmyState"]["energy"] == 0.9
|
|
assert result["timmyState"]["confidence"] == 0.85
|
|
assert result["familiar"] == {"name": "Pip", "state": "alert"}
|
|
assert result["activeThreads"] == full_presence["active_threads"]
|
|
assert result["recentEvents"] == ["committed code"]
|
|
assert result["concerns"] == ["test coverage"]
|
|
assert result["visitorPresent"] is False
|
|
assert result["updatedAt"] == "2026-03-21T12:00:00Z"
|
|
assert result["version"] == 1
|
|
|
|
def test_defaults_on_empty_dict(self):
|
|
"""Missing fields fall back to safe defaults."""
|
|
result = serialize_presence({})
|
|
|
|
assert result["timmyState"]["mood"] == "calm"
|
|
assert result["timmyState"]["activity"] == "idle"
|
|
assert result["timmyState"]["energy"] == 0.5
|
|
assert result["timmyState"]["confidence"] == 0.7
|
|
assert result["familiar"] is None
|
|
assert result["activeThreads"] == []
|
|
assert result["recentEvents"] == []
|
|
assert result["concerns"] == []
|
|
assert result["visitorPresent"] is False
|
|
assert result["version"] == 1
|
|
# updatedAt should be an ISO timestamp string
|
|
assert "T" in result["updatedAt"]
|
|
|
|
def test_partial_presence(self):
|
|
"""Only some fields provided — others get defaults."""
|
|
result = serialize_presence({"mood": "excited", "energy": 0.3})
|
|
|
|
assert result["timmyState"]["mood"] == "excited"
|
|
assert result["timmyState"]["energy"] == 0.3
|
|
assert result["timmyState"]["confidence"] == 0.7 # default
|
|
assert result["activeThreads"] == [] # default
|
|
|
|
def test_return_type_is_dict(self, full_presence):
|
|
"""serialize_presence always returns a plain dict."""
|
|
result = serialize_presence(full_presence)
|
|
assert isinstance(result, dict)
|
|
assert isinstance(result["timmyState"], dict)
|
|
|
|
def test_visitor_present_always_false(self, full_presence):
|
|
"""visitorPresent is always False — set by the WS layer, not here."""
|
|
assert serialize_presence(full_presence)["visitorPresent"] is False
|
|
assert serialize_presence({})["visitorPresent"] is False
|
|
|
|
|
|
class TestProduceAgentState:
|
|
"""Tests for produce_agent_state() — Matrix agent_state message producer."""
|
|
|
|
@pytest.fixture()
|
|
def full_presence(self):
|
|
"""A presence dict with all agent_state-relevant fields."""
|
|
return {
|
|
"display_name": "Timmy",
|
|
"role": "companion",
|
|
"current_focus": "thinking about tests",
|
|
"mood": "focused",
|
|
"energy": 0.9,
|
|
"bark": "Running test suite...",
|
|
}
|
|
|
|
@patch("infrastructure.presence.time")
|
|
def test_full_message_structure(self, mock_time, full_presence):
|
|
"""Returns dict with type, agent_id, data, and ts keys."""
|
|
mock_time.time.return_value = 1742529600
|
|
result = produce_agent_state("timmy", full_presence)
|
|
|
|
assert result["type"] == "agent_state"
|
|
assert result["agent_id"] == "timmy"
|
|
assert result["ts"] == 1742529600
|
|
assert isinstance(result["data"], dict)
|
|
|
|
def test_data_fields(self, full_presence):
|
|
"""data dict contains all required presence fields."""
|
|
data = produce_agent_state("timmy", full_presence)["data"]
|
|
|
|
assert data["display_name"] == "Timmy"
|
|
assert data["role"] == "companion"
|
|
assert data["status"] == "thinking"
|
|
assert data["mood"] == "focused"
|
|
assert data["energy"] == 0.9
|
|
assert data["bark"] == "Running test suite..."
|
|
|
|
def test_defaults_on_empty_presence(self):
|
|
"""Missing fields get sensible defaults."""
|
|
result = produce_agent_state("timmy", {})
|
|
data = result["data"]
|
|
|
|
assert data["display_name"] == "Timmy" # agent_id.title()
|
|
assert data["role"] == "assistant"
|
|
assert data["status"] == "idle"
|
|
assert data["mood"] == "calm"
|
|
assert data["energy"] == 0.5
|
|
assert data["bark"] == ""
|
|
|
|
def test_ts_is_unix_timestamp(self):
|
|
"""ts should be an integer Unix timestamp."""
|
|
result = produce_agent_state("timmy", {})
|
|
assert isinstance(result["ts"], int)
|
|
assert result["ts"] > 0
|
|
|
|
@pytest.mark.parametrize(
|
|
("focus", "expected_status"),
|
|
[
|
|
("thinking about code", "thinking"),
|
|
("speaking to user", "speaking"),
|
|
("talking with agent", "speaking"),
|
|
("idle", "idle"),
|
|
("", "idle"),
|
|
("writing tests", "online"),
|
|
("reviewing PR", "online"),
|
|
],
|
|
)
|
|
def test_status_derivation(self, focus, expected_status):
|
|
"""current_focus maps to the correct Matrix status."""
|
|
data = produce_agent_state("t", {"current_focus": focus})["data"]
|
|
assert data["status"] == expected_status
|
|
|
|
def test_agent_id_passed_through(self):
|
|
"""agent_id appears in the top-level message."""
|
|
result = produce_agent_state("spark", {})
|
|
assert result["agent_id"] == "spark"
|
|
|
|
def test_display_name_from_agent_id(self):
|
|
"""When display_name is missing, it's derived from agent_id.title()."""
|
|
data = produce_agent_state("spark", {})["data"]
|
|
assert data["display_name"] == "Spark"
|