1502 lines
47 KiB
Python
1502 lines
47 KiB
Python
"""Tests for channel plugin discovery, merging, and config compatibility."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
from nanobot.bus.events import OutboundMessage
|
|
from nanobot.bus.outbound_events import (
|
|
ProgressEvent,
|
|
StreamDeltaEvent,
|
|
StreamedResponseEvent,
|
|
StreamEndEvent,
|
|
outbound_message_for_event,
|
|
)
|
|
from nanobot.bus.queue import MessageBus
|
|
from nanobot.channels.base import BaseChannel
|
|
from nanobot.channels.manager import ChannelManager
|
|
from nanobot.config.loader import save_config
|
|
from nanobot.config.schema import ChannelsConfig, Config
|
|
from nanobot.providers.transcription import GroqTranscriptionProvider as _GroqProvider
|
|
from nanobot.providers.transcription import OpenAITranscriptionProvider as _OpenAIProvider
|
|
from nanobot.utils.restart import RestartNotice
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class _FakePlugin(BaseChannel):
|
|
name = "fakeplugin"
|
|
display_name = "Fake Plugin"
|
|
|
|
def __init__(self, config, bus):
|
|
super().__init__(config, bus)
|
|
self.login_calls: list[bool] = []
|
|
|
|
async def start(self) -> None:
|
|
pass
|
|
|
|
async def stop(self) -> None:
|
|
pass
|
|
|
|
async def send(self, msg: OutboundMessage) -> None:
|
|
pass
|
|
|
|
async def login(self, force: bool = False) -> bool:
|
|
self.login_calls.append(force)
|
|
return True
|
|
|
|
|
|
class _FakeTelegram(BaseChannel):
|
|
"""Plugin that tries to shadow built-in telegram."""
|
|
name = "telegram"
|
|
display_name = "Fake Telegram"
|
|
|
|
async def start(self) -> None:
|
|
pass
|
|
|
|
async def stop(self) -> None:
|
|
pass
|
|
|
|
async def send(self, msg: OutboundMessage) -> None:
|
|
pass
|
|
|
|
|
|
def _make_entry_point(name: str, cls: type):
|
|
"""Create a mock entry point that returns *cls* on load()."""
|
|
ep = SimpleNamespace(name=name, load=lambda _cls=cls: _cls)
|
|
return ep
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ChannelsConfig extra="allow"
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_channels_config_accepts_unknown_keys():
|
|
cfg = ChannelsConfig.model_validate({
|
|
"myplugin": {"enabled": True, "token": "abc"},
|
|
})
|
|
extra = cfg.model_extra
|
|
assert extra is not None
|
|
assert extra["myplugin"]["enabled"] is True
|
|
assert extra["myplugin"]["token"] == "abc"
|
|
|
|
|
|
def test_channels_config_getattr_returns_extra():
|
|
cfg = ChannelsConfig.model_validate({"myplugin": {"enabled": True}})
|
|
section = getattr(cfg, "myplugin", None)
|
|
assert isinstance(section, dict)
|
|
assert section["enabled"] is True
|
|
|
|
|
|
def test_channels_config_builtin_fields_removed():
|
|
"""After decoupling, ChannelsConfig has no explicit channel fields."""
|
|
cfg = ChannelsConfig()
|
|
assert not hasattr(cfg, "telegram")
|
|
assert cfg.send_progress is True
|
|
assert cfg.send_tool_hints is False
|
|
assert cfg.extract_document_text is True
|
|
|
|
|
|
def test_channels_config_extract_document_text_accepts_camel_alias():
|
|
cfg = ChannelsConfig.model_validate({"extractDocumentText": False})
|
|
|
|
assert cfg.extract_document_text is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# discover_plugins
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_EP_TARGET = "importlib.metadata.entry_points"
|
|
|
|
|
|
def test_discover_plugins_loads_entry_points():
|
|
from nanobot.channels.registry import discover_plugins
|
|
|
|
ep = _make_entry_point("line", _FakePlugin)
|
|
with patch(_EP_TARGET, return_value=[ep]):
|
|
result = discover_plugins()
|
|
|
|
assert "line" in result
|
|
assert result["line"] is _FakePlugin
|
|
|
|
|
|
def test_discover_plugins_skips_names_outside_enabled_set():
|
|
from nanobot.channels.registry import discover_plugins
|
|
|
|
loaded: list[str] = []
|
|
|
|
def _load_disabled():
|
|
loaded.append("disabled")
|
|
return _FakePlugin
|
|
|
|
ep = SimpleNamespace(name="disabled", load=_load_disabled)
|
|
with patch(_EP_TARGET, return_value=[ep]):
|
|
result = discover_plugins({"enabled"})
|
|
|
|
assert result == {}
|
|
assert loaded == []
|
|
|
|
|
|
def test_discover_plugins_handles_load_error():
|
|
from nanobot.channels.registry import discover_plugins
|
|
|
|
def _boom():
|
|
raise RuntimeError("broken")
|
|
|
|
ep = SimpleNamespace(name="broken", load=_boom)
|
|
with patch(_EP_TARGET, return_value=[ep]):
|
|
result = discover_plugins()
|
|
|
|
assert "broken" not in result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# discover_all — merge & priority
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_discover_all_includes_builtins():
|
|
from nanobot.channels.registry import discover_all, discover_channel_names
|
|
|
|
with patch(_EP_TARGET, return_value=[]):
|
|
result = discover_all()
|
|
|
|
# discover_all() only returns channels that are actually available (dependencies installed)
|
|
# discover_channel_names() returns all built-in channel names
|
|
# So we check that all actually loaded channels are in the result
|
|
for name in result:
|
|
assert name in discover_channel_names()
|
|
|
|
|
|
def test_discover_all_includes_external_plugin():
|
|
from nanobot.channels.registry import discover_all
|
|
|
|
ep = _make_entry_point("line", _FakePlugin)
|
|
with patch(_EP_TARGET, return_value=[ep]):
|
|
result = discover_all()
|
|
|
|
assert "line" in result
|
|
assert result["line"] is _FakePlugin
|
|
|
|
|
|
def test_discover_enabled_imports_only_enabled_builtins():
|
|
from nanobot.channels.registry import discover_enabled
|
|
|
|
loaded: list[str] = []
|
|
|
|
def _load_channel(name: str):
|
|
loaded.append(name)
|
|
return _FakePlugin
|
|
|
|
with (
|
|
patch("nanobot.channels.registry.load_channel_class", side_effect=_load_channel),
|
|
patch(_EP_TARGET, return_value=[]),
|
|
):
|
|
result = discover_enabled({"enabled"}, _names=["enabled", "disabled"])
|
|
|
|
assert result == {"enabled": _FakePlugin}
|
|
assert loaded == ["enabled"]
|
|
|
|
|
|
def test_discover_all_builtin_shadows_plugin():
|
|
from nanobot.channels.registry import discover_all
|
|
|
|
ep = _make_entry_point("telegram", _FakeTelegram)
|
|
with patch(_EP_TARGET, return_value=[ep]):
|
|
result = discover_all()
|
|
|
|
assert "telegram" in result
|
|
assert result["telegram"] is not _FakeTelegram
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Manager _init_channels with dict config (plugin scenario)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_manager_loads_plugin_from_dict_config():
|
|
"""ChannelManager should instantiate a plugin channel from a raw dict config."""
|
|
from nanobot.channels.manager import ChannelManager
|
|
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig.model_validate({
|
|
"fakeplugin": {"enabled": True, "allowFrom": ["*"]},
|
|
}),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="", api_base="")),
|
|
)
|
|
|
|
with patch(
|
|
"nanobot.channels.registry.discover_enabled",
|
|
return_value={"fakeplugin": _FakePlugin},
|
|
):
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.bus = MessageBus()
|
|
mgr.channels = {}
|
|
mgr._dispatch_task = None
|
|
mgr._init_channels()
|
|
|
|
assert "fakeplugin" in mgr.channels
|
|
assert isinstance(mgr.channels["fakeplugin"], _FakePlugin)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_base_channel_reads_current_transcription_config_each_call(
|
|
tmp_path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
):
|
|
"""BaseChannel.transcribe_audio resolves config at call time, not manager init time."""
|
|
from nanobot.providers import transcription as transcription_mod
|
|
|
|
config_path = tmp_path / "config.json"
|
|
config = Config()
|
|
config.transcription.provider = "openai"
|
|
config.transcription.model = "whisper-custom"
|
|
config.transcription.language = "en"
|
|
config.providers.openai.api_key = "openai-key"
|
|
config.providers.openai.api_base = "http://openai.local/v1/audio/transcriptions"
|
|
save_config(config, config_path)
|
|
monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
|
|
|
|
channel = _FakePlugin({"enabled": True, "allowFrom": ["*"]}, MessageBus())
|
|
|
|
calls: list[dict[str, object]] = []
|
|
|
|
class _StubOpenAI:
|
|
def __init__(self, api_key=None, api_base=None, language=None, model=None):
|
|
calls.append({
|
|
"provider": "openai",
|
|
"api_key": api_key,
|
|
"api_base": api_base,
|
|
"language": language,
|
|
"model": model,
|
|
})
|
|
|
|
async def transcribe(self, file_path):
|
|
return "openai-ok"
|
|
|
|
class _StubGroq:
|
|
def __init__(self, api_key=None, api_base=None, language=None, model=None):
|
|
calls.append({
|
|
"provider": "groq",
|
|
"api_key": api_key,
|
|
"api_base": api_base,
|
|
"language": language,
|
|
"model": model,
|
|
})
|
|
|
|
async def transcribe(self, file_path):
|
|
return "groq-ok"
|
|
|
|
with (
|
|
patch.object(transcription_mod, "OpenAITranscriptionProvider", _StubOpenAI),
|
|
patch.object(transcription_mod, "GroqTranscriptionProvider", _StubGroq),
|
|
):
|
|
assert await channel.transcribe_audio("/tmp/does-not-matter.wav") == "openai-ok"
|
|
|
|
config.transcription.provider = "groq"
|
|
config.transcription.model = "whisper-large-v3-turbo"
|
|
config.transcription.language = "ko"
|
|
config.providers.groq.api_key = "groq-key"
|
|
config.providers.groq.api_base = "http://groq.local/v1/audio/transcriptions"
|
|
save_config(config, config_path)
|
|
|
|
assert await channel.transcribe_audio("/tmp/does-not-matter.wav") == "groq-ok"
|
|
|
|
assert calls == [
|
|
{
|
|
"provider": "openai",
|
|
"api_key": "openai-key",
|
|
"api_base": "http://openai.local/v1/audio/transcriptions",
|
|
"language": "en",
|
|
"model": "whisper-custom",
|
|
},
|
|
{
|
|
"provider": "groq",
|
|
"api_key": "groq-key",
|
|
"api_base": "http://groq.local/v1/audio/transcriptions",
|
|
"language": "ko",
|
|
"model": "whisper-large-v3-turbo",
|
|
},
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_base_channel_respects_disabled_transcription_config(
|
|
tmp_path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
):
|
|
config_path = tmp_path / "config.json"
|
|
config = Config()
|
|
config.transcription.enabled = False
|
|
config.providers.groq.api_key = "groq-key"
|
|
save_config(config, config_path)
|
|
monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
|
|
|
|
channel = _FakePlugin({"enabled": True, "allowFrom": ["*"]}, MessageBus())
|
|
|
|
with patch("nanobot.providers.transcription.GroqTranscriptionProvider") as provider:
|
|
assert await channel.transcribe_audio("/tmp/does-not-matter.wav") == ""
|
|
provider.assert_not_called()
|
|
|
|
|
|
def test_openai_transcription_provider_honors_api_base_argument():
|
|
from nanobot.providers.transcription import OpenAITranscriptionProvider
|
|
|
|
default = OpenAITranscriptionProvider(api_key="k")
|
|
assert default.api_url == "https://api.openai.com/v1/audio/transcriptions"
|
|
|
|
custom = OpenAITranscriptionProvider(
|
|
api_key="k", api_base="http://override/v1/audio/transcriptions"
|
|
)
|
|
assert custom.api_url == "http://override/v1/audio/transcriptions"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Transcription provider HTTP tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _StubResponse:
|
|
status_code = 200
|
|
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
def json(self):
|
|
return {"text": "hello"}
|
|
|
|
|
|
def _stub_async_client(captured: dict[str, object]):
|
|
"""Return an httpx.AsyncClient stub that records POST calls into *captured*."""
|
|
class _AsyncClient:
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
async def post(self, url, headers=None, files=None, timeout=None):
|
|
captured["files"] = files
|
|
return _StubResponse()
|
|
|
|
return _AsyncClient()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"provider_cls,language",
|
|
[(_GroqProvider, "ko"), (_OpenAIProvider, "en")],
|
|
ids=["groq", "openai"],
|
|
)
|
|
@pytest.mark.asyncio
|
|
async def test_transcription_provider_includes_language(tmp_path, provider_cls, language):
|
|
"""Provider must include the 'language' field in multipart body when set."""
|
|
audio = tmp_path / "sample.wav"
|
|
audio.write_bytes(b"audio")
|
|
captured: dict[str, object] = {}
|
|
|
|
with patch("nanobot.providers.transcription.httpx.AsyncClient", return_value=_stub_async_client(captured)):
|
|
provider = provider_cls(api_key="k", language=language)
|
|
result = await provider.transcribe(audio)
|
|
|
|
assert result == "hello"
|
|
assert captured["files"]["language"] == (None, language)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"provider_cls",
|
|
[_GroqProvider, _OpenAIProvider],
|
|
ids=["groq", "openai"],
|
|
)
|
|
@pytest.mark.asyncio
|
|
async def test_transcription_provider_omits_language_when_none(tmp_path, provider_cls):
|
|
"""When language is not set, the 'language' key must be absent from the multipart body."""
|
|
audio = tmp_path / "sample.wav"
|
|
audio.write_bytes(b"audio")
|
|
captured: dict[str, object] = {}
|
|
|
|
with patch("nanobot.providers.transcription.httpx.AsyncClient", return_value=_stub_async_client(captured)):
|
|
provider = provider_cls(api_key="k")
|
|
result = await provider.transcribe(audio)
|
|
|
|
assert result == "hello"
|
|
assert "language" not in captured["files"]
|
|
|
|
|
|
def test_channels_login_uses_discovered_plugin_class(monkeypatch):
|
|
from typer.testing import CliRunner
|
|
|
|
from nanobot.cli.commands import app
|
|
from nanobot.config.schema import Config
|
|
|
|
runner = CliRunner()
|
|
seen: dict[str, object] = {}
|
|
|
|
class _LoginPlugin(_FakePlugin):
|
|
display_name = "Login Plugin"
|
|
|
|
async def login(self, force: bool = False) -> bool:
|
|
seen["force"] = force
|
|
seen["config"] = self.config
|
|
return True
|
|
|
|
monkeypatch.setattr("nanobot.config.loader.load_config", lambda config_path=None: Config())
|
|
monkeypatch.setattr(
|
|
"nanobot.channels.registry.discover_all",
|
|
lambda: {"fakeplugin": _LoginPlugin},
|
|
)
|
|
|
|
result = runner.invoke(app, ["channels", "login", "fakeplugin", "--force"])
|
|
|
|
assert result.exit_code == 0
|
|
assert seen["force"] is True
|
|
|
|
|
|
def test_channels_login_sets_custom_config_path(monkeypatch, tmp_path):
|
|
from typer.testing import CliRunner
|
|
|
|
from nanobot.cli.commands import app
|
|
from nanobot.config.schema import Config
|
|
|
|
runner = CliRunner()
|
|
seen: dict[str, object] = {}
|
|
config_path = tmp_path / "custom-config.json"
|
|
|
|
class _LoginPlugin(_FakePlugin):
|
|
async def login(self, force: bool = False) -> bool:
|
|
return True
|
|
|
|
monkeypatch.setattr("nanobot.config.loader.load_config", lambda config_path=None: Config())
|
|
monkeypatch.setattr(
|
|
"nanobot.config.loader.set_config_path",
|
|
lambda path: seen.__setitem__("config_path", path),
|
|
)
|
|
monkeypatch.setattr(
|
|
"nanobot.channels.registry.discover_all",
|
|
lambda: {"fakeplugin": _LoginPlugin},
|
|
)
|
|
|
|
result = runner.invoke(app, ["channels", "login", "fakeplugin", "--config", str(config_path)])
|
|
|
|
assert result.exit_code == 0
|
|
assert seen["config_path"] == config_path.resolve()
|
|
|
|
|
|
def test_channels_status_sets_custom_config_path(monkeypatch, tmp_path):
|
|
from typer.testing import CliRunner
|
|
|
|
from nanobot.cli.commands import app
|
|
from nanobot.config.schema import Config
|
|
|
|
runner = CliRunner()
|
|
seen: dict[str, object] = {}
|
|
config_path = tmp_path / "custom-config.json"
|
|
|
|
monkeypatch.setattr("nanobot.config.loader.load_config", lambda config_path=None: Config())
|
|
monkeypatch.setattr(
|
|
"nanobot.config.loader.set_config_path",
|
|
lambda path: seen.__setitem__("config_path", path),
|
|
)
|
|
monkeypatch.setattr("nanobot.channels.registry.discover_all", lambda: {})
|
|
|
|
result = runner.invoke(app, ["channels", "status", "--config", str(config_path)])
|
|
|
|
assert result.exit_code == 0
|
|
assert seen["config_path"] == config_path.resolve()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_manager_skips_disabled_plugin():
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig.model_validate({
|
|
"fakeplugin": {"enabled": False},
|
|
}),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
|
|
)
|
|
|
|
ep = _make_entry_point("fakeplugin", _FakePlugin)
|
|
with patch(_EP_TARGET, return_value=[ep]):
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.bus = MessageBus()
|
|
mgr.channels = {}
|
|
mgr._dispatch_task = None
|
|
mgr._init_channels()
|
|
|
|
assert "fakeplugin" not in mgr.channels
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Built-in channel default_config() and dict->Pydantic conversion
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_builtin_channel_default_config():
|
|
"""Built-in channels expose default_config() returning a dict with 'enabled': False."""
|
|
from nanobot.channels.telegram import TelegramChannel
|
|
cfg = TelegramChannel.default_config()
|
|
assert isinstance(cfg, dict)
|
|
assert cfg["enabled"] is False
|
|
assert "token" in cfg
|
|
|
|
|
|
def test_builtin_channel_init_from_dict():
|
|
"""Built-in channels accept a raw dict and convert to Pydantic internally."""
|
|
from nanobot.channels.telegram import TelegramChannel
|
|
bus = MessageBus()
|
|
ch = TelegramChannel({"enabled": False, "token": "test-tok", "allowFrom": ["*"]}, bus)
|
|
assert ch.config.token == "test-tok"
|
|
assert ch.config.allow_from == ["*"]
|
|
|
|
|
|
def test_channels_config_send_max_retries_default():
|
|
"""ChannelsConfig should have send_max_retries with default value of 3."""
|
|
cfg = ChannelsConfig()
|
|
assert hasattr(cfg, 'send_max_retries')
|
|
assert cfg.send_max_retries == 3
|
|
|
|
|
|
def test_channels_config_send_max_retries_upper_bound():
|
|
"""send_max_retries should be bounded to prevent resource exhaustion."""
|
|
from pydantic import ValidationError
|
|
|
|
# Value too high should be rejected
|
|
with pytest.raises(ValidationError):
|
|
ChannelsConfig(send_max_retries=100)
|
|
|
|
# Negative should be rejected
|
|
with pytest.raises(ValidationError):
|
|
ChannelsConfig(send_max_retries=-1)
|
|
|
|
# Boundary values should be allowed
|
|
cfg_min = ChannelsConfig(send_max_retries=0)
|
|
assert cfg_min.send_max_retries == 0
|
|
|
|
cfg_max = ChannelsConfig(send_max_retries=10)
|
|
assert cfg_max.send_max_retries == 10
|
|
|
|
# Value above upper bound should be rejected
|
|
with pytest.raises(ValidationError):
|
|
ChannelsConfig(send_max_retries=11)
|
|
|
|
|
|
def test_channels_config_transcription_language_pattern():
|
|
"""transcription_language must match ISO-639 format (2-3 lowercase letters) or be None."""
|
|
from pydantic import ValidationError
|
|
|
|
# Valid values
|
|
assert ChannelsConfig(transcription_language="en").transcription_language == "en"
|
|
assert ChannelsConfig(transcription_language="kor").transcription_language == "kor"
|
|
assert ChannelsConfig(transcription_language=None).transcription_language is None
|
|
|
|
# Invalid values
|
|
with pytest.raises(ValidationError):
|
|
ChannelsConfig(transcription_language="EN") # uppercase
|
|
with pytest.raises(ValidationError):
|
|
ChannelsConfig(transcription_language="english") # full word
|
|
with pytest.raises(ValidationError):
|
|
ChannelsConfig(transcription_language="en-US") # BCP 47 tag
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _send_with_retry
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_with_retry_succeeds_first_try():
|
|
"""_send_with_retry should succeed on first try and not retry."""
|
|
call_count = 0
|
|
|
|
class _FailingChannel(BaseChannel):
|
|
name = "failing"
|
|
display_name = "Failing"
|
|
|
|
async def start(self) -> None:
|
|
pass
|
|
|
|
async def stop(self) -> None:
|
|
pass
|
|
|
|
async def send(self, msg: OutboundMessage) -> None:
|
|
nonlocal call_count
|
|
call_count += 1
|
|
# Succeeds on first try
|
|
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig(send_max_retries=3),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
|
|
)
|
|
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.bus = MessageBus()
|
|
mgr.channels = {"failing": _FailingChannel(fake_config, mgr.bus)}
|
|
mgr._dispatch_task = None
|
|
|
|
msg = OutboundMessage(channel="failing", chat_id="123", content="test")
|
|
await mgr._send_with_retry(mgr.channels["failing"], msg)
|
|
|
|
assert call_count == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_with_retry_retries_on_failure():
|
|
"""_send_with_retry should retry on failure up to max_retries times."""
|
|
call_count = 0
|
|
|
|
class _FailingChannel(BaseChannel):
|
|
name = "failing"
|
|
display_name = "Failing"
|
|
|
|
async def start(self) -> None:
|
|
pass
|
|
|
|
async def stop(self) -> None:
|
|
pass
|
|
|
|
async def send(self, msg: OutboundMessage) -> None:
|
|
nonlocal call_count
|
|
call_count += 1
|
|
raise RuntimeError("simulated failure")
|
|
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig(send_max_retries=3),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
|
|
)
|
|
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.bus = MessageBus()
|
|
mgr.channels = {"failing": _FailingChannel(fake_config, mgr.bus)}
|
|
mgr._dispatch_task = None
|
|
|
|
msg = OutboundMessage(channel="failing", chat_id="123", content="test")
|
|
|
|
# Patch asyncio.sleep to avoid actual delays
|
|
with patch("nanobot.channels.manager.asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
|
|
await mgr._send_with_retry(mgr.channels["failing"], msg)
|
|
|
|
assert call_count == 3 # 3 total attempts (initial + 2 retries)
|
|
assert mock_sleep.call_count == 2 # 2 sleeps between retries
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_with_retry_no_retry_when_max_is_zero():
|
|
"""_send_with_retry should not retry when send_max_retries is 0."""
|
|
call_count = 0
|
|
|
|
class _FailingChannel(BaseChannel):
|
|
name = "failing"
|
|
display_name = "Failing"
|
|
|
|
async def start(self) -> None:
|
|
pass
|
|
|
|
async def stop(self) -> None:
|
|
pass
|
|
|
|
async def send(self, msg: OutboundMessage) -> None:
|
|
nonlocal call_count
|
|
call_count += 1
|
|
raise RuntimeError("simulated failure")
|
|
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig(send_max_retries=0),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
|
|
)
|
|
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.bus = MessageBus()
|
|
mgr.channels = {"failing": _FailingChannel(fake_config, mgr.bus)}
|
|
mgr._dispatch_task = None
|
|
|
|
msg = OutboundMessage(channel="failing", chat_id="123", content="test")
|
|
|
|
with patch("nanobot.channels.manager.asyncio.sleep", new_callable=AsyncMock):
|
|
await mgr._send_with_retry(mgr.channels["failing"], msg)
|
|
|
|
assert call_count == 1 # Called once but no retry (max(0, 1) = 1)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_with_retry_calls_send_delta():
|
|
"""_send_with_retry should call send_delta for stream delta events."""
|
|
send_delta_called = False
|
|
|
|
class _StreamingChannel(BaseChannel):
|
|
name = "streaming"
|
|
display_name = "Streaming"
|
|
|
|
async def start(self) -> None:
|
|
pass
|
|
|
|
async def stop(self) -> None:
|
|
pass
|
|
|
|
async def send(self, msg: OutboundMessage) -> None:
|
|
pass # Should not be called
|
|
|
|
async def send_delta(
|
|
self,
|
|
chat_id: str,
|
|
delta: str,
|
|
metadata: dict | None = None,
|
|
*,
|
|
stream_id: str | None = None,
|
|
stream_end: bool = False,
|
|
resuming: bool = False,
|
|
) -> None:
|
|
nonlocal send_delta_called
|
|
send_delta_called = True
|
|
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig(send_max_retries=3),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
|
|
)
|
|
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.bus = MessageBus()
|
|
mgr.channels = {"streaming": _StreamingChannel(fake_config, mgr.bus)}
|
|
mgr._dispatch_task = None
|
|
|
|
msg = outbound_message_for_event(
|
|
channel="streaming",
|
|
chat_id="123",
|
|
event=StreamDeltaEvent(content="test delta"),
|
|
)
|
|
await mgr._send_with_retry(mgr.channels["streaming"], msg)
|
|
|
|
assert send_delta_called is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_with_retry_supports_legacy_stream_delta_signature():
|
|
"""External plugins with the old send_delta signature should keep working."""
|
|
calls: list[tuple[str, str, dict]] = []
|
|
|
|
class _LegacyStreamingChannel(BaseChannel):
|
|
name = "legacy_streaming"
|
|
display_name = "Legacy Streaming"
|
|
|
|
async def start(self) -> None:
|
|
pass
|
|
|
|
async def stop(self) -> None:
|
|
pass
|
|
|
|
async def send(self, msg: OutboundMessage) -> None:
|
|
pass
|
|
|
|
async def send_delta(
|
|
self,
|
|
chat_id: str,
|
|
delta: str,
|
|
metadata: dict | None = None,
|
|
) -> None:
|
|
calls.append((chat_id, delta, dict(metadata or {})))
|
|
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig(send_max_retries=3),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
|
|
)
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.bus = MessageBus()
|
|
mgr.channels = {"legacy_streaming": _LegacyStreamingChannel(fake_config, mgr.bus)}
|
|
mgr._dispatch_task = None
|
|
|
|
await mgr._send_with_retry(
|
|
mgr.channels["legacy_streaming"],
|
|
outbound_message_for_event(
|
|
channel="legacy_streaming",
|
|
chat_id="123",
|
|
event=StreamDeltaEvent(content="hello", stream_id="s1"),
|
|
),
|
|
)
|
|
await mgr._send_with_retry(
|
|
mgr.channels["legacy_streaming"],
|
|
outbound_message_for_event(
|
|
channel="legacy_streaming",
|
|
chat_id="123",
|
|
event=StreamEndEvent(content="", stream_id="s1", resuming=True),
|
|
),
|
|
)
|
|
|
|
assert calls == [
|
|
("123", "hello", {"_stream_id": "s1", "_stream_delta": True}),
|
|
("123", "", {"_stream_id": "s1", "_stream_end": True}),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_with_retry_supports_legacy_reasoning_signature():
|
|
"""External plugins with the old reasoning hook signature should keep working."""
|
|
deltas: list[tuple[str, str, dict]] = []
|
|
ends: list[tuple[str, dict]] = []
|
|
|
|
class _LegacyReasoningChannel(BaseChannel):
|
|
name = "legacy_reasoning"
|
|
display_name = "Legacy Reasoning"
|
|
|
|
async def start(self) -> None:
|
|
pass
|
|
|
|
async def stop(self) -> None:
|
|
pass
|
|
|
|
async def send(self, msg: OutboundMessage) -> None:
|
|
pass
|
|
|
|
async def send_reasoning_delta(
|
|
self,
|
|
chat_id: str,
|
|
delta: str,
|
|
metadata: dict | None = None,
|
|
) -> None:
|
|
deltas.append((chat_id, delta, dict(metadata or {})))
|
|
|
|
async def send_reasoning_end(
|
|
self,
|
|
chat_id: str,
|
|
metadata: dict | None = None,
|
|
) -> None:
|
|
ends.append((chat_id, dict(metadata or {})))
|
|
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig(send_max_retries=3),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
|
|
)
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.bus = MessageBus()
|
|
mgr.channels = {"legacy_reasoning": _LegacyReasoningChannel(fake_config, mgr.bus)}
|
|
mgr._dispatch_task = None
|
|
|
|
await mgr._send_with_retry(
|
|
mgr.channels["legacy_reasoning"],
|
|
outbound_message_for_event(
|
|
channel="legacy_reasoning",
|
|
chat_id="123",
|
|
event=ProgressEvent(content="thinking", reasoning_delta=True, stream_id="r1"),
|
|
),
|
|
)
|
|
await mgr._send_with_retry(
|
|
mgr.channels["legacy_reasoning"],
|
|
outbound_message_for_event(
|
|
channel="legacy_reasoning",
|
|
chat_id="123",
|
|
event=ProgressEvent(reasoning_end=True, stream_id="r1"),
|
|
),
|
|
)
|
|
|
|
assert deltas == [
|
|
("123", "thinking", {"_reasoning_delta": True, "_stream_id": "r1"}),
|
|
]
|
|
assert ends == [
|
|
("123", {"_reasoning_end": True, "_stream_id": "r1"}),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_with_retry_skips_send_when_streamed():
|
|
"""_send_with_retry should not call send for streamed response events."""
|
|
send_called = False
|
|
send_delta_called = False
|
|
|
|
class _StreamedChannel(BaseChannel):
|
|
name = "streamed"
|
|
display_name = "Streamed"
|
|
|
|
async def start(self) -> None:
|
|
pass
|
|
|
|
async def stop(self) -> None:
|
|
pass
|
|
|
|
async def send(self, msg: OutboundMessage) -> None:
|
|
nonlocal send_called
|
|
send_called = True
|
|
|
|
async def send_delta(
|
|
self,
|
|
chat_id: str,
|
|
delta: str,
|
|
metadata: dict | None = None,
|
|
*,
|
|
stream_id: str | None = None,
|
|
stream_end: bool = False,
|
|
resuming: bool = False,
|
|
) -> None:
|
|
nonlocal send_delta_called
|
|
send_delta_called = True
|
|
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig(send_max_retries=3),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
|
|
)
|
|
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.bus = MessageBus()
|
|
mgr.channels = {"streamed": _StreamedChannel(fake_config, mgr.bus)}
|
|
mgr._dispatch_task = None
|
|
|
|
msg = outbound_message_for_event(
|
|
channel="streamed",
|
|
chat_id="123",
|
|
event=StreamedResponseEvent(),
|
|
content="test",
|
|
)
|
|
await mgr._send_with_retry(mgr.channels["streamed"], msg)
|
|
|
|
assert send_called is False
|
|
assert send_delta_called is False
|
|
|
|
|
|
def test_outbound_duplicate_suppression_is_scoped_to_origin_message() -> None:
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig(send_max_retries=3),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
|
|
)
|
|
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.bus = MessageBus()
|
|
mgr.channels = {}
|
|
mgr._dispatch_task = None
|
|
mgr._origin_reply_fingerprints = {}
|
|
|
|
first = OutboundMessage(
|
|
channel="feishu",
|
|
chat_id="chat123",
|
|
content="Done",
|
|
metadata={"message_id": "msg-1"},
|
|
)
|
|
duplicate = OutboundMessage(
|
|
channel="feishu",
|
|
chat_id="chat123",
|
|
content=" Done ",
|
|
metadata={"origin_message_id": "msg-1"},
|
|
)
|
|
separate_turn = OutboundMessage(
|
|
channel="feishu",
|
|
chat_id="chat123",
|
|
content="Done",
|
|
metadata={"message_id": "msg-2"},
|
|
)
|
|
new_origin_content = OutboundMessage(
|
|
channel="feishu",
|
|
chat_id="chat123",
|
|
content="Done with extra details",
|
|
metadata={"origin_message_id": "msg-1"},
|
|
)
|
|
|
|
assert mgr._should_suppress_outbound(first) is False
|
|
assert mgr._should_suppress_outbound(duplicate) is True
|
|
assert mgr._should_suppress_outbound(separate_turn) is False
|
|
assert mgr._should_suppress_outbound(new_origin_content) is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_with_retry_propagates_cancelled_error():
|
|
"""_send_with_retry should re-raise CancelledError for graceful shutdown."""
|
|
class _CancellingChannel(BaseChannel):
|
|
name = "cancelling"
|
|
display_name = "Cancelling"
|
|
|
|
async def start(self) -> None:
|
|
pass
|
|
|
|
async def stop(self) -> None:
|
|
pass
|
|
|
|
async def send(self, msg: OutboundMessage) -> None:
|
|
raise asyncio.CancelledError("simulated cancellation")
|
|
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig(send_max_retries=3),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
|
|
)
|
|
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.bus = MessageBus()
|
|
mgr.channels = {"cancelling": _CancellingChannel(fake_config, mgr.bus)}
|
|
mgr._dispatch_task = None
|
|
|
|
msg = OutboundMessage(channel="cancelling", chat_id="123", content="test")
|
|
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await mgr._send_with_retry(mgr.channels["cancelling"], msg)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_with_retry_propagates_cancelled_error_during_sleep():
|
|
"""_send_with_retry should re-raise CancelledError during sleep."""
|
|
call_count = 0
|
|
|
|
class _FailingChannel(BaseChannel):
|
|
name = "failing"
|
|
display_name = "Failing"
|
|
|
|
async def start(self) -> None:
|
|
pass
|
|
|
|
async def stop(self) -> None:
|
|
pass
|
|
|
|
async def send(self, msg: OutboundMessage) -> None:
|
|
nonlocal call_count
|
|
call_count += 1
|
|
raise RuntimeError("simulated failure")
|
|
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig(send_max_retries=3),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
|
|
)
|
|
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.bus = MessageBus()
|
|
mgr.channels = {"failing": _FailingChannel(fake_config, mgr.bus)}
|
|
mgr._dispatch_task = None
|
|
|
|
msg = OutboundMessage(channel="failing", chat_id="123", content="test")
|
|
|
|
# Mock sleep to raise CancelledError
|
|
async def cancel_during_sleep(_):
|
|
raise asyncio.CancelledError("cancelled during sleep")
|
|
|
|
with patch("nanobot.channels.manager.asyncio.sleep", side_effect=cancel_during_sleep):
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await mgr._send_with_retry(mgr.channels["failing"], msg)
|
|
|
|
# Should have attempted once before sleep was cancelled
|
|
assert call_count == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ChannelManager - lifecycle and getters
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class _ChannelWithAllowFrom(BaseChannel):
|
|
"""Channel with configurable allow_from."""
|
|
name = "withallow"
|
|
display_name = "With Allow"
|
|
|
|
def __init__(self, config, bus, allow_from):
|
|
super().__init__(config, bus)
|
|
if isinstance(self.config, dict):
|
|
self.config["allow_from"] = allow_from
|
|
else:
|
|
self.config.allow_from = allow_from
|
|
|
|
async def start(self) -> None:
|
|
pass
|
|
|
|
async def stop(self) -> None:
|
|
pass
|
|
|
|
async def send(self, msg: OutboundMessage) -> None:
|
|
pass
|
|
|
|
|
|
class _StartableChannel(BaseChannel):
|
|
"""Channel that tracks start/stop calls."""
|
|
name = "startable"
|
|
display_name = "Startable"
|
|
|
|
def __init__(self, config, bus):
|
|
super().__init__(config, bus)
|
|
self.started = False
|
|
self.stopped = False
|
|
|
|
async def start(self) -> None:
|
|
self.started = True
|
|
|
|
async def stop(self) -> None:
|
|
self.stopped = True
|
|
|
|
async def send(self, msg: OutboundMessage) -> None:
|
|
pass
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validate_allow_from_allows_empty_list():
|
|
"""Empty allow_from is valid now — pairing store handles unapproved senders."""
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig(),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
|
|
)
|
|
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.channels = {"test": _ChannelWithAllowFrom(fake_config, None, [])}
|
|
mgr._dispatch_task = None
|
|
|
|
# Should not raise — empty list defers to pairing store
|
|
mgr._validate_allow_from()
|
|
assert list(mgr.channels) == ["test"]
|
|
assert mgr.channels["test"].config.allow_from == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validate_allow_from_passes_with_asterisk():
|
|
"""_validate_allow_from should not raise when allow_from contains '*'."""
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig(),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
|
|
)
|
|
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.channels = {"test": _ChannelWithAllowFrom(fake_config, None, ["*"])}
|
|
mgr._dispatch_task = None
|
|
|
|
# Should not raise
|
|
mgr._validate_allow_from()
|
|
assert list(mgr.channels) == ["test"]
|
|
assert mgr.channels["test"].config.allow_from == ["*"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validate_allow_from_allows_empty_dict_allow_from():
|
|
"""Empty dict-backed allow_from is valid — pairing store handles approval."""
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig(),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
|
|
)
|
|
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.channels = {"test": _ChannelWithAllowFrom({"enabled": True}, None, [])}
|
|
mgr._dispatch_task = None
|
|
|
|
mgr._validate_allow_from()
|
|
assert list(mgr.channels) == ["test"]
|
|
assert mgr.channels["test"].config["allow_from"] == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validate_allow_from_allows_missing_allow_from():
|
|
"""Omitted allowFrom is valid — channel operates in pairing-only mode."""
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig(),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
|
|
)
|
|
|
|
class _NoAllowFromChannel(BaseChannel):
|
|
name = "noallow"
|
|
display_name = "No Allow"
|
|
|
|
async def start(self) -> None:
|
|
pass
|
|
|
|
async def stop(self) -> None:
|
|
pass
|
|
|
|
async def send(self, msg: OutboundMessage) -> None:
|
|
pass
|
|
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.channels = {"test": _NoAllowFromChannel({"enabled": True}, None)}
|
|
mgr._dispatch_task = None
|
|
|
|
# Should not raise — pairing-only mode
|
|
mgr._validate_allow_from()
|
|
assert list(mgr.channels) == ["test"]
|
|
assert "allow_from" not in mgr.channels["test"].config
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_channel_returns_channel_if_exists():
|
|
"""get_channel should return the channel if it exists."""
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig(),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
|
|
)
|
|
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.bus = MessageBus()
|
|
mgr.channels = {"telegram": _StartableChannel(fake_config, mgr.bus)}
|
|
mgr._dispatch_task = None
|
|
|
|
assert mgr.get_channel("telegram") is not None
|
|
assert mgr.get_channel("nonexistent") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_status_returns_running_state():
|
|
"""get_status should return enabled and running state for each channel."""
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig(),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
|
|
)
|
|
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.bus = MessageBus()
|
|
ch = _StartableChannel(fake_config, mgr.bus)
|
|
mgr.channels = {"startable": ch}
|
|
mgr._dispatch_task = None
|
|
|
|
status = mgr.get_status()
|
|
|
|
assert status["startable"]["enabled"] is True
|
|
assert status["startable"]["running"] is False # Not started yet
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enabled_channels_returns_channel_names():
|
|
"""enabled_channels should return list of enabled channel names."""
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig(),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
|
|
)
|
|
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.bus = MessageBus()
|
|
mgr.channels = {
|
|
"telegram": _StartableChannel(fake_config, mgr.bus),
|
|
"slack": _StartableChannel(fake_config, mgr.bus),
|
|
}
|
|
mgr._dispatch_task = None
|
|
|
|
enabled = mgr.enabled_channels
|
|
|
|
assert "telegram" in enabled
|
|
assert "slack" in enabled
|
|
assert len(enabled) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_all_cancels_dispatcher_and_stops_channels():
|
|
"""stop_all should cancel the dispatch task and stop all channels."""
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig(),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
|
|
)
|
|
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.bus = MessageBus()
|
|
|
|
ch = _StartableChannel(fake_config, mgr.bus)
|
|
mgr.channels = {"startable": ch}
|
|
|
|
# Create a real cancelled task
|
|
async def dummy_task():
|
|
while True:
|
|
await asyncio.sleep(1)
|
|
|
|
dispatch_task = asyncio.create_task(dummy_task())
|
|
mgr._dispatch_task = dispatch_task
|
|
|
|
await mgr.stop_all()
|
|
|
|
# Task should be cancelled
|
|
assert dispatch_task.cancelled()
|
|
# Channel should be stopped
|
|
assert ch.stopped is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_channel_logs_error_on_failure():
|
|
"""_start_channel should log error when channel start fails."""
|
|
class _FailingChannel(BaseChannel):
|
|
name = "failing"
|
|
display_name = "Failing"
|
|
|
|
async def start(self) -> None:
|
|
raise RuntimeError("connection failed")
|
|
|
|
async def stop(self) -> None:
|
|
pass
|
|
|
|
async def send(self, msg: OutboundMessage) -> None:
|
|
pass
|
|
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig(),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
|
|
)
|
|
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.bus = MessageBus()
|
|
mgr.channels = {}
|
|
mgr._dispatch_task = None
|
|
|
|
ch = _FailingChannel(fake_config, mgr.bus)
|
|
|
|
# Should not raise, just log error
|
|
await mgr._start_channel("failing", ch)
|
|
assert mgr.channels == {}
|
|
assert mgr._dispatch_task is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_all_handles_channel_exception():
|
|
"""stop_all should handle exceptions when stopping channels gracefully."""
|
|
class _StopFailingChannel(BaseChannel):
|
|
name = "stopfailing"
|
|
display_name = "Stop Failing"
|
|
|
|
async def start(self) -> None:
|
|
pass
|
|
|
|
async def stop(self) -> None:
|
|
raise RuntimeError("stop failed")
|
|
|
|
async def send(self, msg: OutboundMessage) -> None:
|
|
pass
|
|
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig(),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
|
|
)
|
|
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.bus = MessageBus()
|
|
mgr.channels = {"stopfailing": _StopFailingChannel(fake_config, mgr.bus)}
|
|
mgr._dispatch_task = None
|
|
|
|
# Should not raise even if channel.stop() raises
|
|
await mgr.stop_all()
|
|
assert list(mgr.channels) == ["stopfailing"]
|
|
assert mgr._dispatch_task is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_all_handles_channel_stop_cancelled_task():
|
|
"""stop_all should treat a channel's already-cancelled internals as stopped."""
|
|
|
|
class _StopCancelledChannel(BaseChannel):
|
|
name = "stopcancelled"
|
|
display_name = "Stop Cancelled"
|
|
|
|
async def start(self) -> None:
|
|
pass
|
|
|
|
async def stop(self) -> None:
|
|
raise asyncio.CancelledError("server task cancelled")
|
|
|
|
async def send(self, msg: OutboundMessage) -> None:
|
|
pass
|
|
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig(),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
|
|
)
|
|
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.bus = MessageBus()
|
|
next_channel = _StartableChannel(fake_config, mgr.bus)
|
|
mgr.channels = {
|
|
"stopcancelled": _StopCancelledChannel(fake_config, mgr.bus),
|
|
"next": next_channel,
|
|
}
|
|
mgr._dispatch_task = None
|
|
|
|
await mgr.stop_all()
|
|
|
|
assert next_channel.stopped is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_all_no_channels_logs_warning():
|
|
"""start_all should log warning when no channels are enabled."""
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig(),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
|
|
)
|
|
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.bus = MessageBus()
|
|
mgr.channels = {} # No channels
|
|
mgr._dispatch_task = None
|
|
|
|
# Should return early without creating dispatch task
|
|
await mgr.start_all()
|
|
|
|
assert mgr._dispatch_task is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_all_creates_dispatch_task():
|
|
"""start_all should create the dispatch task when channels exist."""
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig(),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
|
|
)
|
|
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.bus = MessageBus()
|
|
|
|
ch = _StartableChannel(fake_config, mgr.bus)
|
|
mgr.channels = {"startable": ch}
|
|
mgr._dispatch_task = None
|
|
|
|
# Cancel immediately after start to avoid running forever
|
|
async def cancel_after_start():
|
|
await asyncio.sleep(0.01)
|
|
if mgr._dispatch_task:
|
|
mgr._dispatch_task.cancel()
|
|
|
|
cancel_task = asyncio.create_task(cancel_after_start())
|
|
|
|
try:
|
|
await mgr.start_all()
|
|
except asyncio.CancelledError:
|
|
pass
|
|
finally:
|
|
cancel_task.cancel()
|
|
try:
|
|
await cancel_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
# Dispatch task should have been created
|
|
assert mgr._dispatch_task is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_notify_restart_done_enqueues_outbound_message():
|
|
"""Restart notice should schedule send_with_retry for target channel."""
|
|
fake_config = SimpleNamespace(
|
|
channels=ChannelsConfig(),
|
|
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
|
|
)
|
|
|
|
mgr = ChannelManager.__new__(ChannelManager)
|
|
mgr.config = fake_config
|
|
mgr.bus = MessageBus()
|
|
mgr.channels = {"feishu": _StartableChannel(fake_config, mgr.bus)}
|
|
mgr._dispatch_task = None
|
|
mgr._send_with_retry = AsyncMock()
|
|
|
|
notice = RestartNotice(channel="feishu", chat_id="oc_123", started_at_raw="100.0")
|
|
with patch("nanobot.channels.manager.consume_restart_notice_from_env", return_value=notice):
|
|
mgr._notify_restart_done_if_needed()
|
|
|
|
await asyncio.sleep(0)
|
|
mgr._send_with_retry.assert_awaited_once()
|
|
sent_channel, sent_msg = mgr._send_with_retry.await_args.args
|
|
assert sent_channel is mgr.channels["feishu"]
|
|
assert sent_msg.channel == "feishu"
|
|
assert sent_msg.chat_id == "oc_123"
|
|
assert sent_msg.content.startswith("Restart completed")
|