0c3253a485
asyncio.run() closes the event loop after execution, which breaks subsequent tests using asyncio.get_event_loop() (test_send_image_file).
100 lines
3.6 KiB
Python
100 lines
3.6 KiB
Python
"""Tests for cron/scheduler.py — origin resolution, delivery routing, and error logging."""
|
|
|
|
import asyncio
|
|
import logging
|
|
from unittest.mock import patch, MagicMock, AsyncMock
|
|
|
|
import pytest
|
|
|
|
from cron.scheduler import _resolve_origin, _deliver_result, run_job
|
|
|
|
|
|
class TestResolveOrigin:
|
|
def test_full_origin(self):
|
|
job = {
|
|
"origin": {
|
|
"platform": "telegram",
|
|
"chat_id": "123456",
|
|
"chat_name": "Test Chat",
|
|
}
|
|
}
|
|
result = _resolve_origin(job)
|
|
assert isinstance(result, dict)
|
|
assert result == job["origin"]
|
|
assert result["platform"] == "telegram"
|
|
assert result["chat_id"] == "123456"
|
|
assert result["chat_name"] == "Test Chat"
|
|
|
|
def test_no_origin(self):
|
|
assert _resolve_origin({}) is None
|
|
assert _resolve_origin({"origin": None}) is None
|
|
|
|
def test_missing_platform(self):
|
|
job = {"origin": {"chat_id": "123"}}
|
|
assert _resolve_origin(job) is None
|
|
|
|
def test_missing_chat_id(self):
|
|
job = {"origin": {"platform": "telegram"}}
|
|
assert _resolve_origin(job) is None
|
|
|
|
def test_empty_origin(self):
|
|
job = {"origin": {}}
|
|
assert _resolve_origin(job) is None
|
|
|
|
|
|
class TestDeliverResultMirrorLogging:
|
|
"""Verify that mirror_to_session failures are logged, not silently swallowed."""
|
|
|
|
def test_mirror_failure_is_logged(self, caplog):
|
|
"""When mirror_to_session raises, a warning should be logged."""
|
|
from gateway.config import Platform
|
|
|
|
pconfig = MagicMock()
|
|
pconfig.enabled = True
|
|
mock_cfg = MagicMock()
|
|
mock_cfg.platforms = {Platform.TELEGRAM: pconfig}
|
|
|
|
with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \
|
|
patch("asyncio.run", return_value=None), \
|
|
patch("gateway.mirror.mirror_to_session", side_effect=ConnectionError("network down")):
|
|
job = {
|
|
"id": "test-job",
|
|
"deliver": "origin",
|
|
"origin": {"platform": "telegram", "chat_id": "123"},
|
|
}
|
|
with caplog.at_level(logging.WARNING, logger="cron.scheduler"):
|
|
_deliver_result(job, "Hello!")
|
|
|
|
assert any("mirror_to_session failed" in r.message for r in caplog.records), \
|
|
f"Expected 'mirror_to_session failed' warning in logs, got: {[r.message for r in caplog.records]}"
|
|
|
|
|
|
class TestRunJobConfigLogging:
|
|
"""Verify that config.yaml parse failures are logged, not silently swallowed."""
|
|
|
|
def test_bad_config_yaml_is_logged(self, caplog, tmp_path):
|
|
"""When config.yaml is malformed, a warning should be logged."""
|
|
# Create a bad config.yaml
|
|
bad_yaml = tmp_path / "config.yaml"
|
|
bad_yaml.write_text("invalid: yaml: [[[bad")
|
|
|
|
job = {
|
|
"id": "test-job",
|
|
"name": "test",
|
|
"prompt": "hello",
|
|
}
|
|
|
|
with patch("cron.scheduler._hermes_home", tmp_path), \
|
|
patch("cron.scheduler._resolve_origin", return_value=None), \
|
|
patch("dotenv.load_dotenv"), \
|
|
patch("run_agent.AIAgent") as mock_agent_cls:
|
|
mock_agent = MagicMock()
|
|
mock_agent.run.return_value = ("output doc", "final response")
|
|
mock_agent_cls.return_value = mock_agent
|
|
|
|
with caplog.at_level(logging.WARNING, logger="cron.scheduler"):
|
|
run_job(job)
|
|
|
|
assert any("failed to load config.yaml" in r.message for r in caplog.records), \
|
|
f"Expected 'failed to load config.yaml' warning in logs, got: {[r.message for r in caplog.records]}"
|