Anton Palgunov
d77f01e31a
Add /stats slash command showing model config, context engine (Semantic RLE status), skills usage (from curator telemetry), curator state, cron jobs, 24h activity, and system memory. Pure local compute — no LLM call, no prompt-cache impact. - COMMAND_REGISTRY: stats (Info category) - gateway/run.py: dispatch + _handle_stats_command handler - tests/gateway/test_stats_command.py: 7 tests
205 lines
6.9 KiB
Python
205 lines
6.9 KiB
Python
"""Tests for /stats command — comprehensive system report."""
|
|
|
|
from unittest.mock import MagicMock, AsyncMock, patch
|
|
import json, os, tempfile, time
|
|
from datetime import datetime, timezone, timedelta
|
|
|
|
import pytest
|
|
|
|
|
|
# ---- Helpers ----
|
|
|
|
def _mock_event(command="stats", args=""):
|
|
"""Build a minimal MessageEvent for slash-command tests."""
|
|
event = MagicMock()
|
|
event.source.platform.value = "telegram"
|
|
event.source.chat_id = "test-chat"
|
|
event.source.chat_type = "dm"
|
|
event.get_command.return_value = command
|
|
event.get_command_args.return_value = args
|
|
return event
|
|
|
|
|
|
def _make_runner():
|
|
"""Build a GatewayRunner with minimum wiring for _handle_stats_command."""
|
|
from gateway.run import GatewayRunner
|
|
runner = GatewayRunner.__new__(GatewayRunner)
|
|
# Session store mocks
|
|
runner.session_store = MagicMock()
|
|
session_entry = MagicMock()
|
|
session_entry.session_key = "agent:main:telegram:dm:test-chat"
|
|
session_entry.session_id = "20260529_test"
|
|
runner.session_store.get_or_create_session.return_value = session_entry
|
|
# Config mock
|
|
runner._read_user_config = MagicMock(return_value={
|
|
"model": {"default": "deepseek-v4-pro", "provider": "opencode_go"},
|
|
"fallback_providers": [
|
|
{"provider": "opencode_go", "model": "deepseek-v4-pro"},
|
|
],
|
|
})
|
|
# Agent cache
|
|
runner._agent_cache = {}
|
|
runner._agent_cache_lock = MagicMock()
|
|
runner._agent_cache_lock.__enter__ = lambda s: None
|
|
runner._agent_cache_lock.__exit__ = lambda s, *a: None
|
|
return runner
|
|
|
|
|
|
# ---- Tests ----
|
|
|
|
class TestStatsCommand:
|
|
"""Unit tests for _handle_stats_command."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_output_has_expected_sections(self):
|
|
"""Output contains all major dashboard sections."""
|
|
runner = _make_runner()
|
|
event = _mock_event()
|
|
|
|
with patch("hermes_state.SessionDB") as mock_db_cls:
|
|
mock_db = MagicMock()
|
|
mock_db.list_sessions.return_value = [
|
|
{"session_id": "s1", "updated_at": datetime.now(timezone.utc).isoformat(), "message_count": 5}
|
|
]
|
|
mock_db_cls.return_value = mock_db
|
|
|
|
result = await runner._handle_stats_command(event)
|
|
|
|
# All sections should be present for a working config
|
|
assert "Hermes Stats" in result
|
|
assert "Model" in result
|
|
assert "deepseek-v4-pro" in result
|
|
assert "opencode_go" in result
|
|
assert "Context Engine" in result
|
|
assert "Skills" in result
|
|
assert "Curator" in result
|
|
assert "Cron" in result
|
|
assert "Activity" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_model_section_shows_fallback(self):
|
|
"""Fallback chain is rendered when configured."""
|
|
runner = _make_runner()
|
|
runner._read_user_config.return_value["fallback_providers"] = [
|
|
{"provider": "custom", "model": "gemma-local"},
|
|
]
|
|
event = _mock_event()
|
|
|
|
with patch("hermes_state.SessionDB") as mock_db_cls:
|
|
mock_db = MagicMock()
|
|
mock_db.list_sessions.return_value = []
|
|
mock_db_cls.return_value = mock_db
|
|
|
|
result = await runner._handle_stats_command(event)
|
|
|
|
assert "gemma-local" in result
|
|
assert "custom" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_fallback_shown_when_empty(self):
|
|
"""No fallback line when chain is empty."""
|
|
runner = _make_runner()
|
|
runner._read_user_config.return_value["fallback_providers"] = []
|
|
event = _mock_event()
|
|
|
|
with patch("hermes_state.SessionDB") as mock_db_cls:
|
|
mock_db = MagicMock()
|
|
mock_db.list_sessions.return_value = []
|
|
mock_db_cls.return_value = mock_db
|
|
|
|
result = await runner._handle_stats_command(event)
|
|
|
|
assert "Fallback" not in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_curator_section_no_state_file(self):
|
|
"""Curator section shows defaults when no state file exists."""
|
|
runner = _make_runner()
|
|
event = _mock_event()
|
|
|
|
with (
|
|
patch("hermes_state.SessionDB") as mock_db_cls,
|
|
patch("os.path.isfile", return_value=False),
|
|
):
|
|
mock_db = MagicMock()
|
|
mock_db.list_sessions.return_value = []
|
|
mock_db_cls.return_value = mock_db
|
|
|
|
result = await runner._handle_stats_command(event)
|
|
|
|
assert "Runs: 0" in result
|
|
assert "Archived: 0" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_activity_section_with_recent_sessions(self):
|
|
"""Activity section shows 24h stats from session DB."""
|
|
runner = _make_runner()
|
|
event = _mock_event()
|
|
|
|
now = datetime.now(timezone.utc)
|
|
recent = [
|
|
{
|
|
"session_id": "s1", "title": "Debug session",
|
|
"updated_at": now.isoformat(),
|
|
"message_count": 42,
|
|
"preview": "Fixed the bug",
|
|
},
|
|
{
|
|
"session_id": "s2",
|
|
"updated_at": (now - timedelta(hours=48)).isoformat(),
|
|
"message_count": 10,
|
|
},
|
|
]
|
|
|
|
with patch("hermes_state.SessionDB") as mock_db_cls:
|
|
mock_db = MagicMock()
|
|
mock_db.list_sessions.return_value = recent
|
|
mock_db_cls.return_value = mock_db
|
|
|
|
result = await runner._handle_stats_command(event)
|
|
|
|
# First session is within 24h
|
|
assert "Sessions: 1" in result
|
|
assert "Messages: 42" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cron_section_handles_import_error(self):
|
|
"""Cron section is skipped gracefully when cron module unavailable."""
|
|
runner = _make_runner()
|
|
event = _mock_event()
|
|
|
|
with (
|
|
patch("hermes_state.SessionDB") as mock_db_cls,
|
|
patch("cron.jobs.load_jobs", side_effect=ImportError("no cron")),
|
|
):
|
|
mock_db = MagicMock()
|
|
mock_db.list_sessions.return_value = []
|
|
mock_db_cls.return_value = mock_db
|
|
|
|
result = await runner._handle_stats_command(event)
|
|
|
|
assert isinstance(result, str)
|
|
assert "Hermes Stats" in result
|
|
# Cron section should be absent (not crash)
|
|
assert "Cron" not in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rle_section_not_installed(self):
|
|
"""RLE section shows 'not installed' when plugin dir missing."""
|
|
runner = _make_runner()
|
|
event = _mock_event()
|
|
|
|
with (
|
|
patch("hermes_state.SessionDB") as mock_db_cls,
|
|
patch("os.path.isdir", return_value=False),
|
|
):
|
|
mock_db = MagicMock()
|
|
mock_db.list_sessions.return_value = [
|
|
{"session_id": "s1", "updated_at": datetime.now(timezone.utc).isoformat(), "message_count": 0}
|
|
]
|
|
mock_db_cls.return_value = mock_db
|
|
|
|
result = await runner._handle_stats_command(event)
|
|
|
|
assert "not installed" in result
|