test(gateway): replace heavy AIAgent with lightweight stub in cache-capacity tests #26
@@ -70,6 +70,7 @@ AUTHOR_MAP = {
|
||||
# Internal molecule-ai Gitea bot identity used by Claude-Code agents
|
||||
# (post-2026-05-06 GitHub suspension; no upstream/GitHub equivalent).
|
||||
"claude-ceo-assistant@agents.moleculesai.app": "claude-ceo-assistant",
|
||||
"dev-engineer-a-kimi@agents.moleculesai.app": "agent-dev-a",
|
||||
# OpenViking viking_read salvage (April 2026)
|
||||
"hitesh@gmail.com": "htsh",
|
||||
"pty819@outlook.com": "pty819",
|
||||
|
||||
@@ -83,6 +83,24 @@ def _ensure_telegram_mock() -> None:
|
||||
sys.modules["telegram.error"] = mod.error
|
||||
|
||||
|
||||
def _ensure_openai_mock() -> None:
|
||||
"""Install a minimal openai mock in sys.modules.
|
||||
|
||||
Idempotent — skips when the real library is available.
|
||||
The mock provides ``openai.OpenAI`` as a MagicMock subclass so
|
||||
``AIAgent`` instantiation in cache tests succeeds without a real
|
||||
OpenAI package (CI/test containers may lack it).
|
||||
"""
|
||||
import importlib.util
|
||||
|
||||
if importlib.util.find_spec("openai") is not None:
|
||||
return # Real library is installed — nothing to mock
|
||||
|
||||
mod = MagicMock()
|
||||
mod.OpenAI = MagicMock
|
||||
sys.modules["openai"] = mod
|
||||
|
||||
|
||||
def _ensure_discord_mock() -> None:
|
||||
"""Install a comprehensive discord mock in sys.modules.
|
||||
|
||||
@@ -218,6 +236,7 @@ def _ensure_discord_mock() -> None:
|
||||
# Run at collection time — before any test file's module-level imports.
|
||||
_ensure_telegram_mock()
|
||||
_ensure_discord_mock()
|
||||
_ensure_openai_mock()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -857,16 +857,27 @@ class TestAgentCacheSpilloverLive:
|
||||
runner._running_agents = {}
|
||||
return runner
|
||||
|
||||
class _StubAgent:
|
||||
"""Lightweight stand-in for AIAgent in cache-capacity tests.
|
||||
|
||||
The cache enforcement logic only needs identity (``id()``) and a
|
||||
``.client`` attribute (verified by active-safety tests). Real
|
||||
``AIAgent`` instantiation does tool-registry scan, MCP discovery,
|
||||
config loading and file I/O — ~200 ms each — which makes the
|
||||
concurrent-insert test (160 agents) too heavy for CI under xdist.
|
||||
"""
|
||||
def __init__(self):
|
||||
self.client = object() # non-None so active-safety assertions pass
|
||||
|
||||
def close(self):
|
||||
pass
|
||||
|
||||
def release_clients(self):
|
||||
pass
|
||||
|
||||
def _real_agent(self):
|
||||
"""A genuine AIAgent; no API calls are made during these tests."""
|
||||
from run_agent import AIAgent
|
||||
return AIAgent(
|
||||
model="anthropic/claude-sonnet-4", api_key="test",
|
||||
base_url="https://openrouter.ai/api/v1", provider="openrouter",
|
||||
max_iterations=5, quiet_mode=True,
|
||||
skip_context_files=True, skip_memory=True,
|
||||
platform="telegram",
|
||||
)
|
||||
"""A lightweight stub; replaces the heavy AIAgent for cache tests."""
|
||||
return self._StubAgent()
|
||||
|
||||
def test_fill_to_cap_then_spillover(self, monkeypatch):
|
||||
"""Fill to cap with real agents, insert one more, oldest evicted."""
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
|
||||
import json
|
||||
import os
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -302,7 +303,10 @@ class TestTencentTokenhubContextLength:
|
||||
|
||||
def test_hy3_preview_context_length(self):
|
||||
from agent.model_metadata import get_model_context_length
|
||||
ctx = get_model_context_length("hy3-preview")
|
||||
# Mock out the live OpenRouter call so the test is deterministic
|
||||
# and falls through to the hardcoded default.
|
||||
with patch("agent.model_metadata.fetch_model_metadata", return_value={}):
|
||||
ctx = get_model_context_length("hy3-preview")
|
||||
assert ctx == 256000
|
||||
|
||||
|
||||
|
||||
@@ -12,7 +12,19 @@ import subprocess
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import patch
|
||||
|
||||
from hermes_cli.main import cmd_update
|
||||
def _cmd_update(args):
|
||||
"""Freshly resolve cmd_update so patches target the current module object.
|
||||
|
||||
``test_env_loader.py`` and ``test_skills_subparser.py`` remove
|
||||
``hermes_cli.main`` from ``sys.modules`` and re-import it. Tests that
|
||||
import ``cmd_update`` at module level end up with a stale function whose
|
||||
``__globals__`` reference the *old* module, so ``@patch("hermes_cli.main.sys")``
|
||||
and ``@patch("hermes_cli.main._restore_stashed_changes")`` silently miss.
|
||||
Importing locally inside each test fixes the mismatch.
|
||||
"""
|
||||
import hermes_cli.main as _main
|
||||
|
||||
return _main.cmd_update(args)
|
||||
|
||||
|
||||
def _make_run_side_effect(
|
||||
@@ -74,7 +86,7 @@ class TestUpdateYesConfigMigration:
|
||||
args = SimpleNamespace(yes=True)
|
||||
|
||||
with patch("builtins.input") as mock_input:
|
||||
cmd_update(args)
|
||||
_cmd_update(args)
|
||||
# Never prompted the user.
|
||||
mock_input.assert_not_called()
|
||||
|
||||
@@ -118,7 +130,7 @@ class TestUpdateYesConfigMigration:
|
||||
) as mock_sys:
|
||||
mock_sys.stdin.isatty.return_value = True
|
||||
mock_sys.stdout.isatty.return_value = True
|
||||
cmd_update(args)
|
||||
_cmd_update(args)
|
||||
# The user was actually prompted.
|
||||
assert mock_input.called
|
||||
prompts = [c.args[0] if c.args else "" for c in mock_input.call_args_list]
|
||||
@@ -156,7 +168,7 @@ class TestUpdateYesStashRestore:
|
||||
|
||||
args = SimpleNamespace(yes=True)
|
||||
|
||||
cmd_update(args)
|
||||
_cmd_update(args)
|
||||
|
||||
# _restore_stashed_changes was called, and called with prompt_user=False
|
||||
# every time (so the user never sees "Restore local changes now?").
|
||||
|
||||
@@ -1970,6 +1970,9 @@ class TestPtyWebSocket:
|
||||
self.ws_module = ws
|
||||
monkeypatch.setattr(ws, "_DASHBOARD_EMBEDDED_CHAT_ENABLED", True)
|
||||
self.token = ws._SESSION_TOKEN
|
||||
# Purge any stale subscriber state left behind by other test suites
|
||||
# that also drive the /api/events endpoint (e.g. gateway tests).
|
||||
ws._event_channels.clear()
|
||||
self.client = TestClient(ws.app)
|
||||
|
||||
def _url(self, token: str | None = None, **params: str) -> str:
|
||||
|
||||
Reference in New Issue
Block a user