forked from Rockachopa/Timmy-time-dashboard
4e11dd2490
Move 97 test files from flat tests/ into 13 subdirectories: tests/dashboard/ (8 files — routes, mobile, mission control) tests/swarm/ (17 files — coordinator, docker, routing, tasks) tests/timmy/ (12 files — agent, backends, CLI, tools) tests/self_coding/ (14 files — git safety, indexer, self-modify) tests/lightning/ (3 files — L402, LND, interface) tests/creative/ (8 files — assembler, director, image/music/video) tests/integrations/ (10 files — chat bridge, telegram, voice, websocket) tests/mcp/ (4 files — bootstrap, discovery, executor) tests/spark/ (3 files — engine, tools, events) tests/hands/ (3 files — registry, oracle, phase5) tests/scripture/ (1 file) tests/infrastructure/ (3 files — router cascade, API) tests/security/ (3 files — XSS, regression) Fix Path(__file__) reference in test_mobile_scenarios.py for new depth. Add __init__.py to all test subdirectories. Tests: 1503 passed, 9 failed (pre-existing), 53 errors (pre-existing) https://claude.ai/code/session_019oMFNvD8uSGSSmBMGkBfQN
277 lines
8.3 KiB
Python
277 lines
8.3 KiB
Python
"""Tests for Hands Infrastructure.
|
|
|
|
Tests HandRegistry, HandScheduler, and HandRunner.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from hands import HandRegistry, HandRunner, HandScheduler
|
|
from hands.models import HandConfig, HandStatus, ScheduleConfig
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_hands_dir():
|
|
"""Create a temporary hands directory with test Hands."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
hands_dir = Path(tmpdir)
|
|
|
|
# Create Oracle Hand
|
|
oracle_dir = hands_dir / "oracle"
|
|
oracle_dir.mkdir()
|
|
(oracle_dir / "HAND.toml").write_text('''
|
|
[hand]
|
|
name = "oracle"
|
|
description = "Bitcoin intelligence"
|
|
schedule = "0 7,19 * * *"
|
|
|
|
[tools]
|
|
required = ["mempool_fetch", "fee_estimate"]
|
|
|
|
[output]
|
|
dashboard = true
|
|
''')
|
|
(oracle_dir / "SYSTEM.md").write_text("# Oracle System Prompt\nYou are Oracle.")
|
|
|
|
# Create Sentinel Hand
|
|
sentinel_dir = hands_dir / "sentinel"
|
|
sentinel_dir.mkdir()
|
|
(sentinel_dir / "HAND.toml").write_text('''
|
|
[hand]
|
|
name = "sentinel"
|
|
description = "System health monitoring"
|
|
schedule = "*/15 * * * *"
|
|
enabled = true
|
|
''')
|
|
|
|
yield hands_dir
|
|
|
|
|
|
@pytest.fixture
|
|
def registry(temp_hands_dir):
|
|
"""Create HandRegistry with test Hands."""
|
|
db_path = temp_hands_dir / "test_hands.db"
|
|
reg = HandRegistry(hands_dir=temp_hands_dir, db_path=db_path)
|
|
return reg
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestHandRegistry:
|
|
"""HandRegistry tests."""
|
|
|
|
async def test_load_all_hands(self, registry, temp_hands_dir):
|
|
"""Should load all Hands from directory."""
|
|
hands = await registry.load_all()
|
|
|
|
assert len(hands) == 2
|
|
assert "oracle" in hands
|
|
assert "sentinel" in hands
|
|
|
|
async def test_get_hand(self, registry, temp_hands_dir):
|
|
"""Should get Hand by name."""
|
|
await registry.load_all()
|
|
|
|
hand = registry.get_hand("oracle")
|
|
assert hand.name == "oracle"
|
|
assert "Bitcoin" in hand.description
|
|
|
|
async def test_get_hand_not_found(self, registry):
|
|
"""Should raise for unknown Hand."""
|
|
from hands.registry import HandNotFoundError
|
|
|
|
with pytest.raises(HandNotFoundError):
|
|
registry.get_hand("nonexistent")
|
|
|
|
async def test_get_scheduled_hands(self, registry, temp_hands_dir):
|
|
"""Should return only Hands with schedules."""
|
|
await registry.load_all()
|
|
|
|
scheduled = registry.get_scheduled_hands()
|
|
|
|
assert len(scheduled) == 2
|
|
assert all(h.schedule is not None for h in scheduled)
|
|
|
|
async def test_state_management(self, registry, temp_hands_dir):
|
|
"""Should track Hand state."""
|
|
await registry.load_all()
|
|
|
|
state = registry.get_state("oracle")
|
|
assert state.name == "oracle"
|
|
assert state.status == HandStatus.IDLE
|
|
|
|
registry.update_state("oracle", status=HandStatus.RUNNING)
|
|
state = registry.get_state("oracle")
|
|
assert state.status == HandStatus.RUNNING
|
|
|
|
async def test_approval_queue(self, registry, temp_hands_dir):
|
|
"""Should manage approval queue."""
|
|
await registry.load_all()
|
|
|
|
# Create approval
|
|
request = await registry.create_approval(
|
|
hand_name="oracle",
|
|
action="post_tweet",
|
|
description="Post Bitcoin update",
|
|
context={"price": 50000},
|
|
)
|
|
|
|
assert request.id is not None
|
|
assert request.hand_name == "oracle"
|
|
|
|
# Get pending
|
|
pending = await registry.get_pending_approvals()
|
|
assert len(pending) == 1
|
|
|
|
# Resolve
|
|
result = await registry.resolve_approval(request.id, approved=True)
|
|
assert result is True
|
|
|
|
# Should be empty now
|
|
pending = await registry.get_pending_approvals()
|
|
assert len(pending) == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestHandScheduler:
|
|
"""HandScheduler tests."""
|
|
|
|
async def test_scheduler_initialization(self, registry):
|
|
"""Should initialize scheduler."""
|
|
scheduler = HandScheduler(registry)
|
|
assert scheduler.registry == registry
|
|
assert not scheduler._running
|
|
|
|
async def test_schedule_hand(self, registry, temp_hands_dir):
|
|
"""Should schedule a Hand."""
|
|
await registry.load_all()
|
|
scheduler = HandScheduler(registry)
|
|
|
|
hand = registry.get_hand("oracle")
|
|
job_id = await scheduler.schedule_hand(hand)
|
|
|
|
# Note: Job ID may be None if APScheduler not available
|
|
# But should not raise an exception
|
|
|
|
async def test_get_scheduled_jobs(self, registry, temp_hands_dir):
|
|
"""Should list scheduled jobs."""
|
|
await registry.load_all()
|
|
scheduler = HandScheduler(registry)
|
|
|
|
jobs = scheduler.get_scheduled_jobs()
|
|
assert isinstance(jobs, list)
|
|
|
|
async def test_trigger_hand_now(self, registry, temp_hands_dir):
|
|
"""Should manually trigger a Hand."""
|
|
await registry.load_all()
|
|
scheduler = HandScheduler(registry)
|
|
|
|
# This will fail because Hand isn't fully implemented
|
|
# But should not raise
|
|
result = await scheduler.trigger_hand_now("oracle")
|
|
# Result may be True or False depending on implementation
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestHandRunner:
|
|
"""HandRunner tests."""
|
|
|
|
async def test_load_system_prompt(self, registry, temp_hands_dir):
|
|
"""Should load SYSTEM.md."""
|
|
await registry.load_all()
|
|
runner = HandRunner(registry)
|
|
|
|
hand = registry.get_hand("oracle")
|
|
prompt = runner._load_system_prompt(hand)
|
|
|
|
assert "Oracle" in prompt
|
|
|
|
async def test_load_skills(self, registry, temp_hands_dir):
|
|
"""Should load SKILL.md files."""
|
|
# Create a skill file
|
|
skills_dir = temp_hands_dir / "oracle" / "skills"
|
|
skills_dir.mkdir()
|
|
(skills_dir / "bitcoin.md").write_text("# Bitcoin Expertise")
|
|
|
|
await registry.load_all()
|
|
runner = HandRunner(registry)
|
|
|
|
hand = registry.get_hand("oracle")
|
|
skills = runner._load_skills(hand)
|
|
|
|
assert len(skills) == 1
|
|
assert "Bitcoin" in skills[0]
|
|
|
|
async def test_build_prompt(self, registry, temp_hands_dir):
|
|
"""Should build execution prompt."""
|
|
await registry.load_all()
|
|
runner = HandRunner(registry)
|
|
|
|
hand = registry.get_hand("oracle")
|
|
system = "System prompt"
|
|
skills = ["Skill 1", "Skill 2"]
|
|
context = {"key": "value"}
|
|
|
|
prompt = runner._build_prompt(hand, system, skills, context)
|
|
|
|
assert "System Instructions" in prompt
|
|
assert "System prompt" in prompt
|
|
assert "Skill 1" in prompt
|
|
assert "key" in prompt
|
|
|
|
|
|
class TestHandConfig:
|
|
"""HandConfig model tests."""
|
|
|
|
def test_hand_config_creation(self):
|
|
"""Should create HandConfig."""
|
|
config = HandConfig(
|
|
name="test",
|
|
description="Test hand",
|
|
schedule=ScheduleConfig(cron="0 * * * *"),
|
|
)
|
|
|
|
assert config.name == "test"
|
|
assert config.schedule.cron == "0 * * * *"
|
|
|
|
def test_schedule_validation(self):
|
|
"""Should validate cron expression."""
|
|
# Valid cron
|
|
config = HandConfig(
|
|
name="test",
|
|
description="Test",
|
|
schedule=ScheduleConfig(cron="0 7 * * *"),
|
|
)
|
|
assert config.schedule.cron == "0 7 * * *"
|
|
|
|
|
|
class TestHandModels:
|
|
"""Hand model tests."""
|
|
|
|
def test_hand_status_enum(self):
|
|
"""HandStatus should have expected values."""
|
|
from hands.models import HandStatus
|
|
|
|
assert HandStatus.IDLE.value == "idle"
|
|
assert HandStatus.RUNNING.value == "running"
|
|
assert HandStatus.SCHEDULED.value == "scheduled"
|
|
|
|
def test_hand_state_to_dict(self):
|
|
"""HandState should serialize to dict."""
|
|
from hands.models import HandState
|
|
from datetime import datetime
|
|
|
|
state = HandState(
|
|
name="test",
|
|
status=HandStatus.RUNNING,
|
|
run_count=5,
|
|
)
|
|
|
|
data = state.to_dict()
|
|
assert data["name"] == "test"
|
|
assert data["status"] == "running"
|
|
assert data["run_count"] == 5
|