code-server
c6b68f0b6b
Updated existing tests to work with cryptographic visibility markers: 1. test_agent_loop_metadata.py: - Updated test_suppress_mode_adds_hidden_prefix to verify [HIDDEN:signature] format - Added validation for 8-character hex signature - Updated test_normal_mode_no_hidden_prefix to check for [HIDDEN: prefix 2. test_idle_heartbeat_integration.py: - Updated test_idle_heartbeat_end_to_end to search for [HIDDEN: prefix - Added signature format validation (8-char hex) - Updated docstring to reflect signed markers All 86 tests now pass (excluding OAuth tests as specified). The changes maintain backwards compatibility while enforcing the new cryptographic signing requirement. Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
111 lines
3.8 KiB
Python
111 lines
3.8 KiB
Python
# tests/test_idle_heartbeat_integration.py
|
|
import pytest
|
|
import asyncio
|
|
from pathlib import Path
|
|
from datetime import datetime, timedelta
|
|
from nanobot.agent.loop import AgentLoop
|
|
from nanobot.bus.queue import MessageBus
|
|
from nanobot.heartbeat.service import HeartbeatService
|
|
from nanobot.session.manager import SessionManager
|
|
from nanobot.providers.base import LLMProvider, LLMResponse
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_idle_heartbeat_end_to_end(tmp_path):
|
|
"""
|
|
Integration test: heartbeat triggers when idle, runs in main session,
|
|
output is suppressed, session contains [HIDDEN:signature] content.
|
|
"""
|
|
workspace = tmp_path / "test-integration"
|
|
workspace.mkdir()
|
|
|
|
# Use test-specific session key
|
|
test_session_key = "telegram:test_integration"
|
|
|
|
# Create HEARTBEAT.md with content
|
|
heartbeat_file = workspace / "HEARTBEAT.md"
|
|
heartbeat_file.write_text("# Test Task\n- Check something")
|
|
|
|
# Create mock provider
|
|
provider = MagicMock(spec=LLMProvider)
|
|
provider.chat = AsyncMock(return_value=LLMResponse(
|
|
content="Heartbeat executed successfully",
|
|
tool_calls=[] # has_tool_calls is a property, not a parameter
|
|
))
|
|
provider.get_default_model = MagicMock(return_value="test-model")
|
|
provider.thinking_budget = 0
|
|
|
|
# Create components
|
|
bus = MessageBus()
|
|
sessions = SessionManager(workspace)
|
|
# Override sessions_dir to use tmp_path for test isolation
|
|
sessions.sessions_dir = tmp_path / "sessions"
|
|
sessions.sessions_dir.mkdir()
|
|
loop = AgentLoop(
|
|
bus=bus,
|
|
provider=provider,
|
|
workspace=workspace,
|
|
session_manager=sessions
|
|
)
|
|
|
|
# Create session with old user message
|
|
session = sessions.get_or_create(test_session_key)
|
|
old_timestamp = (datetime.now() - timedelta(minutes=31)).isoformat()
|
|
session.messages.append({
|
|
"role": "user",
|
|
"content": "Old user message",
|
|
"timestamp": old_timestamp
|
|
})
|
|
sessions.save(session)
|
|
|
|
# Create heartbeat callback
|
|
async def on_heartbeat(prompt: str, metadata=None):
|
|
return await loop.process_direct(
|
|
prompt,
|
|
session_key=test_session_key,
|
|
channel="telegram",
|
|
chat_id="test_integration",
|
|
metadata=metadata,
|
|
)
|
|
|
|
# Create heartbeat service
|
|
heartbeat = HeartbeatService(
|
|
workspace=workspace,
|
|
on_heartbeat=on_heartbeat,
|
|
interval_s=1,
|
|
enabled=True,
|
|
session_manager=sessions,
|
|
target_session_key=test_session_key,
|
|
idle_threshold_s=30 * 60,
|
|
)
|
|
|
|
# Trigger heartbeat
|
|
await heartbeat._tick()
|
|
|
|
# Reload session from disk
|
|
sessions._cache.clear() # Clear cache to force reload
|
|
session = sessions.get_or_create(test_session_key)
|
|
|
|
# Verify:
|
|
# 1. Session has new messages
|
|
assert len(session.messages) > 1
|
|
|
|
# 2. Find the heartbeat response (assistant message with signed visibility marker)
|
|
heartbeat_messages = [
|
|
m for m in session.messages
|
|
if m.get("role") == "assistant" and "[HIDDEN:" in m.get("content", "")
|
|
]
|
|
assert len(heartbeat_messages) == 1, "Expected exactly 1 [HIDDEN:signature] heartbeat message"
|
|
|
|
# 3. Verify content is prefixed with [HIDDEN:signature]
|
|
heartbeat_msg = heartbeat_messages[0]
|
|
assert heartbeat_msg["content"].startswith("[HIDDEN:")
|
|
# Verify signature format (8-char hex)
|
|
content = heartbeat_msg["content"]
|
|
prefix_end = content.index("]")
|
|
signature = content[8:prefix_end] # Skip "[HIDDEN:" to get signature
|
|
assert len(signature) == 8, f"Expected 8-char signature, got {len(signature)}"
|
|
assert all(c in "0123456789abcdef" for c in signature), "Signature should be hex"
|
|
assert "Heartbeat executed successfully" in heartbeat_msg["content"]
|