mirror of
https://github.com/open-jarvis/OpenJarvis.git
synced 2026-07-30 19:02:16 +00:00
Prathap PandGitHub
c2756964a7
* fix(channels): wire channel→agent handler and fix Telegram send pipeline * format code * add supported tests
66 lines
2.4 KiB
Python
66 lines
2.4 KiB
Python
"""Tests for MonitorOperativeAgent."""
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
from openjarvis.agents.monitor_operative import MonitorOperativeAgent
|
|
from openjarvis.core.registry import AgentRegistry
|
|
|
|
|
|
def _make_engine(content: str = "Hello") -> MagicMock:
|
|
engine = MagicMock()
|
|
engine.engine_id = "mock"
|
|
engine.generate.return_value = {
|
|
"content": content,
|
|
"usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
|
|
"model": "test-model",
|
|
"finish_reason": "stop",
|
|
}
|
|
return engine
|
|
|
|
|
|
class TestMonitorOperativeAgent:
|
|
def test_registration(self) -> None:
|
|
# Import triggers registration; re-register after autouse fixture
|
|
# clears the registry (same pattern as test_monitor.py)
|
|
import openjarvis.agents.monitor_operative # noqa: F401
|
|
|
|
if not AgentRegistry.contains("monitor_operative"):
|
|
AgentRegistry.register_value("monitor_operative", MonitorOperativeAgent)
|
|
assert AgentRegistry.contains("monitor_operative")
|
|
cls = AgentRegistry.get("monitor_operative")
|
|
assert cls is MonitorOperativeAgent
|
|
|
|
def test_instantiation(self) -> None:
|
|
engine = _make_engine()
|
|
agent = MonitorOperativeAgent(engine, "test-model")
|
|
assert agent.agent_id == "monitor_operative"
|
|
assert agent.accepts_tools is True
|
|
|
|
def test_default_strategies(self) -> None:
|
|
engine = _make_engine()
|
|
agent = MonitorOperativeAgent(engine, "test-model")
|
|
assert agent._memory_extraction == "causality_graph"
|
|
assert agent._observation_compression == "summarize"
|
|
assert agent._retrieval_strategy == "hybrid_with_self_eval"
|
|
assert agent._task_decomposition == "phased"
|
|
|
|
def test_custom_strategies(self) -> None:
|
|
engine = _make_engine()
|
|
agent = MonitorOperativeAgent(
|
|
engine,
|
|
"test-model",
|
|
memory_extraction="scratchpad",
|
|
observation_compression="none",
|
|
retrieval_strategy="keyword",
|
|
task_decomposition="monolithic",
|
|
)
|
|
assert agent._memory_extraction == "scratchpad"
|
|
assert agent._observation_compression == "none"
|
|
|
|
def test_simple_run(self) -> None:
|
|
engine = _make_engine("The answer is 42.")
|
|
agent = MonitorOperativeAgent(engine, "test-model")
|
|
result = agent.run("What is the answer?")
|
|
assert result.content == "The answer is 42."
|
|
assert result.turns >= 1
|