feat: Mission Control v2 — swarm, L402, voice, marketplace, React dashboard
Major expansion of the Timmy Time Dashboard: Backend modules: - Swarm subsystem: registry, manager, bidder, coordinator, agent_runner, swarm_node, tasks, comms - L402/Lightning: payment_handler, l402_proxy with HMAC macaroons - Voice NLU: regex-based intent detection (chat, status, swarm, task, help, voice) - Notifications: push notifier for swarm events - Shortcuts: Siri Shortcuts iOS integration endpoints - WebSocket: live dashboard event manager - Inter-agent: agent-to-agent messaging layer Dashboard routes: - /swarm/* — swarm management and agent registry - /marketplace — agent catalog with sat pricing - /voice/* — voice command processing - /mobile — mobile status endpoint - /swarm/live — WebSocket live feed React web dashboard (dashboard-web/): - Sovereign Terminal design — dark theme with Bitcoin orange accents - Three-column layout: status sidebar, workspace tabs, context panel - Chat, Swarm, Tasks, Marketplace tab views - JetBrains Mono typography, terminal aesthetic - Framer Motion animations throughout Tests: 228 passing (expanded from 93) Includes Kimi's additional templates and QA work.
This commit is contained in:
68
tests/test_agent_runner.py
Normal file
68
tests/test_agent_runner.py
Normal file
@@ -0,0 +1,68 @@
|
||||
"""TDD tests for swarm/agent_runner.py — sub-agent entry point.
|
||||
|
||||
Written RED-first: define expected behaviour, then make it pass.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import signal
|
||||
import sys
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def tmp_swarm_db(tmp_path, monkeypatch):
|
||||
db_path = tmp_path / "swarm.db"
|
||||
monkeypatch.setattr("swarm.tasks.DB_PATH", db_path)
|
||||
monkeypatch.setattr("swarm.registry.DB_PATH", db_path)
|
||||
yield db_path
|
||||
|
||||
|
||||
def test_agent_runner_module_is_importable():
|
||||
"""The agent_runner module should import without errors."""
|
||||
import swarm.agent_runner
|
||||
assert hasattr(swarm.agent_runner, "main")
|
||||
|
||||
|
||||
def test_agent_runner_main_is_coroutine():
|
||||
"""main() should be an async function."""
|
||||
from swarm.agent_runner import main
|
||||
assert asyncio.iscoroutinefunction(main)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_agent_runner_creates_node_and_joins():
|
||||
"""main() should create a SwarmNode and call join()."""
|
||||
mock_node = MagicMock()
|
||||
mock_node.join = AsyncMock()
|
||||
mock_node.leave = AsyncMock()
|
||||
|
||||
with patch("sys.argv", ["agent_runner", "--agent-id", "test-1", "--name", "TestBot"]):
|
||||
with patch("swarm.swarm_node.SwarmNode", return_value=mock_node) as MockNodeClass:
|
||||
# We need to stop the event loop from waiting forever
|
||||
# Patch signal to immediately set the stop event
|
||||
original_signal = signal.signal
|
||||
|
||||
def fake_signal(sig, handler):
|
||||
if sig in (signal.SIGTERM, signal.SIGINT):
|
||||
# Immediately call the handler to stop the loop
|
||||
handler(sig, None)
|
||||
return original_signal(sig, handler)
|
||||
|
||||
with patch("signal.signal", side_effect=fake_signal):
|
||||
from swarm.agent_runner import main
|
||||
await main()
|
||||
|
||||
MockNodeClass.assert_called_once_with("test-1", "TestBot")
|
||||
mock_node.join.assert_awaited_once()
|
||||
mock_node.leave.assert_awaited_once()
|
||||
|
||||
|
||||
def test_agent_runner_has_dunder_main_guard():
|
||||
"""The module should have an if __name__ == '__main__' guard."""
|
||||
import inspect
|
||||
import swarm.agent_runner
|
||||
source = inspect.getsource(swarm.agent_runner)
|
||||
assert '__name__' in source
|
||||
assert '__main__' in source
|
||||
Reference in New Issue
Block a user