4020b5222f
Add infrastructure for running swarm agents as isolated Docker containers with HTTP-based coordination, startup recovery, and enhanced dashboard UI for agent management. - Dockerfile and docker-compose.yml for multi-service orchestration - DockerAgentRunner for programmatic container lifecycle management - Internal HTTP API for container agents to poll tasks and submit bids - Startup recovery system to reconcile orphaned tasks and stale agents - Enhanced UI partials for agent panels, chat, and task assignment - Timmy docker entry point with heartbeat and task polling - New Makefile targets for Docker workflows - Tests for swarm recovery Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
59 lines
1.7 KiB
Python
59 lines
1.7 KiB
Python
import sys
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
# ── Stub heavy optional dependencies so tests run without them installed ──────
|
|
# Uses setdefault: real module is used if already installed, mock otherwise.
|
|
for _mod in [
|
|
"agno",
|
|
"agno.agent",
|
|
"agno.models",
|
|
"agno.models.ollama",
|
|
"agno.db",
|
|
"agno.db.sqlite",
|
|
# AirLLM is optional (bigbrain extra) — stub it so backend tests can
|
|
# import timmy.backends and instantiate TimmyAirLLMAgent without a GPU.
|
|
"airllm",
|
|
# python-telegram-bot is optional (telegram extra) — stub so tests run
|
|
# without the package installed.
|
|
"telegram",
|
|
"telegram.ext",
|
|
]:
|
|
sys.modules.setdefault(_mod, MagicMock())
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_message_log():
|
|
"""Clear the in-memory chat log before and after every test."""
|
|
from dashboard.store import message_log
|
|
message_log.clear()
|
|
yield
|
|
message_log.clear()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_coordinator_state():
|
|
"""Clear the coordinator's in-memory state between tests.
|
|
|
|
The coordinator singleton is created at import time and persists across
|
|
the test session. Without this fixture, agents spawned in one test bleed
|
|
into the next through the auctions dict, comms listeners, and the
|
|
in-process node list.
|
|
"""
|
|
yield
|
|
from swarm.coordinator import coordinator
|
|
coordinator.auctions._auctions.clear()
|
|
coordinator.comms._listeners.clear()
|
|
coordinator._in_process_nodes.clear()
|
|
coordinator.manager.stop_all()
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
from dashboard.app import app
|
|
with TestClient(app) as c:
|
|
yield c
|