dinos
690b903f85
* chore: add pytest-asyncio in auto mode * test: migrate channel and stream tests to native async Convert run_async() wrapper tests to plain 'async def test_*' under pytest-asyncio auto mode. collect_events() in stream_v3_fakes becomes a coroutine awaited at every call site. * test: migrate command and model/middleware tests to native async Convert run_async() wrappers (import, alias, and fixture forms) to plain 'async def test_*'. Multi-call tests merge onto one loop as sequential awaits; none asserted on loop identity. * test: migrate TUI, notifier, gateway, and session tests to native async TUI/notifier/gateway files convert run_async wrappers to plain async tests. test_sessions.py's unittest.TestCase classes move to unittest.IsolatedAsyncioTestCase (pytest-asyncio does not await async methods on plain TestCase; converting blindly would have made ~70 tests silently vacuous). Its setUpClass keeps a one-shot asyncio.run() since IsolatedAsyncioTestCase has no async class-level hook. TestLoadingWidget in test_tui_widgets.py drops its TestCase base for the same reason. * test: replace direct asyncio.run() calls with native async tests Convert tests that called asyncio.run() (directly or via a local _run helper) to plain 'async def test_*'; delete the local helpers. * test: drop undeclared anyio markers and delete run_async helper The @pytest.mark.anyio tests relied on anyio being a transitive dep of httpx; auto-mode pytest-asyncio collects them natively. run_async() and its fixture are unreferenced after the migration, so remove them — pytest-asyncio's per-test loop teardown covers the pending-task cancellation the helper existed for (verified: full suite runs with no 'Event loop is closed' errors or destroyed-task warnings).
102 lines
3.7 KiB
Python
102 lines
3.7 KiB
Python
"""Additional smoke tests for channels without dedicated suites."""
|
|
|
|
import pytest
|
|
|
|
from EvoScientist.channels.base import ChannelError, OutboundMessage
|
|
from EvoScientist.channels.email.channel import EmailChannel, EmailConfig
|
|
from EvoScientist.channels.imessage.channel_rpc import (
|
|
IMessageChannelRpc,
|
|
IMessageConfig,
|
|
)
|
|
from EvoScientist.channels.qq.channel import QQChannel, QQConfig
|
|
from EvoScientist.channels.signal.channel import SignalChannel, SignalConfig
|
|
|
|
|
|
class TestEmailChannelSmoke:
|
|
async def test_start_raises_without_required_imap_settings(self):
|
|
channel = EmailChannel(EmailConfig())
|
|
with pytest.raises(
|
|
ChannelError, match="imap_host and imap_username are required"
|
|
):
|
|
await channel.start()
|
|
|
|
async def test_send_returns_false_when_smtp_not_ready(self):
|
|
channel = EmailChannel(EmailConfig())
|
|
msg = OutboundMessage(
|
|
channel="email",
|
|
chat_id="user@example.com",
|
|
content="hello",
|
|
metadata={"chat_id": "user@example.com"},
|
|
)
|
|
assert await channel.send(msg) is False
|
|
|
|
|
|
class TestSignalChannelSmoke:
|
|
async def test_start_raises_without_phone_number(self):
|
|
channel = SignalChannel(SignalConfig())
|
|
with pytest.raises(ChannelError, match="phone_number is required"):
|
|
await channel.start()
|
|
|
|
async def test_send_returns_false_when_not_connected(self):
|
|
channel = SignalChannel(SignalConfig(phone_number="+123456789"))
|
|
msg = OutboundMessage(
|
|
channel="signal",
|
|
chat_id="+123456789",
|
|
content="hello",
|
|
metadata={"chat_id": "+123456789"},
|
|
)
|
|
assert await channel.send(msg) is False
|
|
|
|
|
|
class TestQQChannelSmoke:
|
|
async def test_start_raises_when_sdk_missing(self, monkeypatch):
|
|
from EvoScientist.channels.qq import channel as qq_module
|
|
|
|
monkeypatch.setattr(qq_module, "QQ_AVAILABLE", False)
|
|
channel = QQChannel(QQConfig(app_id="id", app_secret="secret"))
|
|
with pytest.raises(ChannelError, match="SDK not installed"):
|
|
await channel.start()
|
|
|
|
async def test_start_raises_without_credentials_when_sdk_available(
|
|
self, monkeypatch
|
|
):
|
|
from EvoScientist.channels.qq import channel as qq_module
|
|
|
|
monkeypatch.setattr(qq_module, "QQ_AVAILABLE", True)
|
|
channel = QQChannel(QQConfig(app_id="", app_secret=""))
|
|
with pytest.raises(ChannelError, match="app_id and app_secret are required"):
|
|
await channel.start()
|
|
|
|
async def test_send_returns_false_without_client(self):
|
|
channel = QQChannel(QQConfig(app_id="id", app_secret="secret"))
|
|
msg = OutboundMessage(
|
|
channel="qq",
|
|
chat_id="openid",
|
|
content="hello",
|
|
metadata={"chat_id": "openid"},
|
|
)
|
|
assert await channel.send(msg) is False
|
|
|
|
|
|
class TestIMessageChannelSmoke:
|
|
async def test_start_wraps_rpc_bootstrap_error(self, monkeypatch):
|
|
async def _broken_start(self):
|
|
raise RuntimeError("imsg not found")
|
|
|
|
from EvoScientist.channels.imessage import channel_rpc as imessage_module
|
|
|
|
monkeypatch.setattr(imessage_module.ImsgRpcClient, "start", _broken_start)
|
|
channel = IMessageChannelRpc(IMessageConfig())
|
|
with pytest.raises(ChannelError, match="Failed to start imsg"):
|
|
await channel.start()
|
|
|
|
async def test_send_returns_false_without_rpc_client(self):
|
|
channel = IMessageChannelRpc(IMessageConfig())
|
|
msg = OutboundMessage(
|
|
channel="imessage",
|
|
chat_id="+123456789",
|
|
content="hello",
|
|
metadata={"chat_id": "+123456789"},
|
|
)
|
|
assert await channel.send(msg) is False
|