2002f03e42
Change-detectors, tautologies, source-reading tests, redundant duplicates, mock-echo tests and dead/unrunnable tests. Per-test rationale in the lane ledger (category + reason for every removal).
161 lines
5.2 KiB
Python
161 lines
5.2 KiB
Python
"""Tests for the ContextEngine ABC and plugin slot."""
|
|
|
|
import json
|
|
from typing import Any, Dict, List
|
|
|
|
from agent.context_engine import ContextEngine
|
|
from agent.context_compressor import ContextCompressor
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# A minimal concrete engine for testing the ABC
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class StubEngine(ContextEngine):
|
|
"""Minimal engine that satisfies the ABC without doing real work."""
|
|
|
|
def __init__(self, context_length=200000, threshold_pct=0.50):
|
|
self.context_length = context_length
|
|
self.threshold_tokens = int(context_length * threshold_pct)
|
|
self._compress_called = False
|
|
self._tools_called = []
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return "stub"
|
|
|
|
def update_model(self, model="", context_length=0, base_url="", api_key="",
|
|
provider="", api_mode="", **kwargs) -> None:
|
|
"""Mirror ContextCompressor.update_model — recompute threshold from the
|
|
new context_length. This is the mutation that corrupted the shared
|
|
singleton in #42449."""
|
|
self.context_length = context_length
|
|
self.threshold_tokens = int(context_length * 0.20)
|
|
|
|
def update_from_response(self, usage: Dict[str, Any]) -> None:
|
|
self.last_prompt_tokens = usage.get("prompt_tokens", 0)
|
|
self.last_completion_tokens = usage.get("completion_tokens", 0)
|
|
self.last_total_tokens = usage.get("total_tokens", 0)
|
|
|
|
def should_compress(self, prompt_tokens: int = None) -> bool:
|
|
tokens = prompt_tokens if prompt_tokens is not None else self.last_prompt_tokens
|
|
return tokens >= self.threshold_tokens
|
|
|
|
def compress(self, messages: List[Dict[str, Any]], current_tokens: int = None) -> List[Dict[str, Any]]:
|
|
self._compress_called = True
|
|
self.compression_count += 1
|
|
# Trivial: just return as-is
|
|
return messages
|
|
|
|
def get_tool_schemas(self) -> List[Dict[str, Any]]:
|
|
return [
|
|
{
|
|
"name": "stub_search",
|
|
"description": "Search the stub engine",
|
|
"parameters": {"type": "object", "properties": {}},
|
|
}
|
|
]
|
|
|
|
def handle_tool_call(self, name: str, args: Dict[str, Any]) -> str:
|
|
self._tools_called.append(name)
|
|
return json.dumps({"ok": True, "tool": name})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ABC contract tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Default method behavior
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestDefaults:
|
|
"""Verify ABC default implementations work correctly."""
|
|
|
|
|
|
|
|
def test_default_get_status(self):
|
|
engine = StubEngine()
|
|
engine.last_prompt_tokens = 50000
|
|
status = engine.get_status()
|
|
assert status["last_prompt_tokens"] == 50000
|
|
assert status["context_length"] == 200000
|
|
assert status["threshold_tokens"] == 100000
|
|
assert 0 < status["usage_percent"] <= 100
|
|
|
|
|
|
def test_on_session_reset(self):
|
|
engine = StubEngine()
|
|
engine.last_prompt_tokens = 999
|
|
engine.compression_count = 3
|
|
engine.on_session_reset()
|
|
assert engine.last_prompt_tokens == 0
|
|
assert engine.compression_count == 0
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# StubEngine behavior
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ContextCompressor session reset via ABC
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCompressorSessionReset:
|
|
"""Verify ContextCompressor.on_session_reset() clears all state."""
|
|
|
|
def test_reset_clears_state(self):
|
|
c = ContextCompressor(model="test", quiet_mode=True, config_context_length=200000)
|
|
c.last_prompt_tokens = 50000
|
|
c.compression_count = 3
|
|
c._previous_summary = "some old summary"
|
|
c._context_probed = True
|
|
c._context_probe_persistable = True
|
|
|
|
c.on_session_reset()
|
|
|
|
assert c.last_prompt_tokens == 0
|
|
assert c.last_completion_tokens == 0
|
|
assert c.last_total_tokens == 0
|
|
assert c.compression_count == 0
|
|
assert c._context_probed is False
|
|
assert c._context_probe_persistable is False
|
|
assert c._previous_summary is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Plugin slot (PluginManager integration)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestPluginContextEngineSlot:
|
|
"""Test register_context_engine on PluginContext."""
|
|
|
|
def test_register_engine(self):
|
|
from hermes_cli.plugins import PluginManager, PluginContext, PluginManifest
|
|
mgr = PluginManager()
|
|
manifest = PluginManifest(name="test-lcm")
|
|
ctx = PluginContext(manifest, mgr)
|
|
|
|
engine = StubEngine()
|
|
ctx.register_context_engine(engine)
|
|
|
|
assert mgr._context_engine is engine
|
|
assert mgr._context_engine.name == "stub"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|