tests/test_idle_heartbeat_integration.py

@@ -12,13 +12,16 @@ from unittest.mock import AsyncMock, MagicMock
 
 
 @pytest.mark.asyncio
-async def test_idle_heartbeat_end_to_end():
+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] content.
     """
-    workspace = Path("/tmp/test-integration")
-    workspace.mkdir(exist_ok=True)
+    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"
@@ -44,7 +47,7 @@ async def test_idle_heartbeat_end_to_end():
     )
 
     # Create session with old user message
-    session = sessions.get_or_create("telegram:239824268")
+    session = sessions.get_or_create(test_session_key)
     old_timestamp = (datetime.now() - timedelta(minutes=31)).isoformat()
     session.messages.append({
         "role": "user",
@@ -57,9 +60,9 @@ async def test_idle_heartbeat_end_to_end():
     async def on_heartbeat(prompt: str, metadata=None):
         return await loop.process_direct(
             prompt,
-            session_key="telegram:239824268",
+            session_key=test_session_key,
             channel="telegram",
-            chat_id="239824268",
+            chat_id="test_integration",
             metadata=metadata,
         )
 
@@ -70,7 +73,7 @@ async def test_idle_heartbeat_end_to_end():
         interval_s=1,
         enabled=True,
         session_manager=sessions,
-        target_session_key="telegram:239824268",
+        target_session_key=test_session_key,
         idle_threshold_s=30 * 60,
     )
 
@@ -79,7 +82,7 @@ async def test_idle_heartbeat_end_to_end():
 
     # Reload session from disk
     sessions._cache.clear()  # Clear cache to force reload
-    session = sessions.get_or_create("telegram:239824268")
+    session = sessions.get_or_create(test_session_key)
 
     # Verify:
     # 1. Session has new messages
@@ -90,12 +93,9 @@ async def test_idle_heartbeat_end_to_end():
         m for m in session.messages
         if m.get("role") == "assistant" and "[HIDDEN]" in m.get("content", "")
     ]
-    assert len(heartbeat_messages) > 0, "No [HIDDEN] heartbeat message found"
+    assert len(heartbeat_messages) == 1, "Expected exactly 1 [HIDDEN] heartbeat message"
 
     # 3. Verify content is prefixed with [HIDDEN]
     heartbeat_msg = heartbeat_messages[0]
     assert heartbeat_msg["content"].startswith("[HIDDEN]")
     assert "Heartbeat executed successfully" in heartbeat_msg["content"]
-
-    # Cleanup
-    heartbeat_file.unlink()