repositories
loading repo index
repositories
loading repo index
repository
loading code, commits, and activity
public Clawd ADK gateway launch mirror
stars
latest
clone command
git clone gitlawb://did:key:z6Mkq5mY...iFZ5/my-project-publ...git clone gitlawb://did:key:z6Mkq5mY.../my-project-publ...2fa351d6docs: add automaton and perps launch sources15d ago| #1 | """Tests for the Hermes auxiliary LLM adapter.""" |
| #2 | |
| #3 | from __future__ import annotations |
| #4 | |
| #5 | import sys |
| #6 | import types |
| #7 | from typing import Any |
| #8 | from unittest.mock import MagicMock |
| #9 | |
| #10 | import pytest |
| #11 | |
| #12 | from mnemosyne.core import llm_backends |
| #13 | from mnemosyne.core.llm_backends import get_host_llm_backend |
| #14 | |
| #15 | |
| #16 | # --------------------------------------------------------------------------- |
| #17 | # Fake `agent` package — Hermes is not a test-time dependency. |
| #18 | # Patching `agent.auxiliary_client.call_llm` requires the dotted path to exist |
| #19 | # in sys.modules before unittest.mock can resolve it. |
| #20 | # --------------------------------------------------------------------------- |
| #21 | |
| #22 | @pytest.fixture |
| #23 | def fake_agent_module(monkeypatch): |
| #24 | """Inject a fake ``agent.auxiliary_client`` into sys.modules. |
| #25 | |
| #26 | Yields the auxiliary_client submodule so tests can attach call_llm / |
| #27 | extract_content_or_reasoning mocks. Module is scrubbed on teardown so |
| #28 | later tests see a clean slate. |
| #29 | """ |
| #30 | agent_pkg = types.ModuleType("agent") |
| #31 | aux_client = types.ModuleType("agent.auxiliary_client") |
| #32 | agent_pkg.auxiliary_client = aux_client # type: ignore[attr-defined] |
| #33 | |
| #34 | monkeypatch.setitem(sys.modules, "agent", agent_pkg) |
| #35 | monkeypatch.setitem(sys.modules, "agent.auxiliary_client", aux_client) |
| #36 | yield aux_client |
| #37 | |
| #38 | |
| #39 | def _import_adapter(): |
| #40 | # Lazy import so the module-level adapter never references a missing |
| #41 | # ``agent.*`` at collection time. |
| #42 | from hermes_memory_provider import hermes_llm_adapter |
| #43 | return hermes_llm_adapter |
| #44 | |
| #45 | |
| #46 | # --------------------------------------------------------------------------- |
| #47 | # HermesAuxLLMBackend.complete() |
| #48 | # --------------------------------------------------------------------------- |
| #49 | |
| #50 | def test_complete_calls_call_llm_with_compression_task(fake_agent_module): |
| #51 | """Adapter must invoke call_llm(task='compression', ...) with passed args.""" |
| #52 | captured = {} |
| #53 | |
| #54 | def fake_call_llm(**kwargs): |
| #55 | captured.update(kwargs) |
| #56 | return {"choices": [{"message": {"content": "Summary."}}]} |
| #57 | |
| #58 | fake_agent_module.call_llm = fake_call_llm |
| #59 | |
| #60 | adapter = _import_adapter() |
| #61 | backend = adapter.HermesAuxLLMBackend() |
| #62 | out = backend.complete( |
| #63 | "the prompt", |
| #64 | max_tokens=128, |
| #65 | temperature=0.2, |
| #66 | timeout=12.0, |
| #67 | ) |
| #68 | assert out == "Summary." |
| #69 | assert captured["task"] == "compression" |
| #70 | assert captured["temperature"] == 0.2 |
| #71 | assert captured["max_tokens"] == 128 |
| #72 | assert captured["timeout"] == 12.0 |
| #73 | # System prompt + user prompt structure. |
| #74 | msgs = captured["messages"] |
| #75 | assert msgs[0]["role"] == "system" |
| #76 | assert "memory consolidation engine" in msgs[0]["content"].lower() |
| #77 | assert msgs[1] == {"role": "user", "content": "the prompt"} |
| #78 | # Optional overrides not passed when None. |
| #79 | assert "provider" not in captured |
| #80 | assert "model" not in captured |
| #81 | |
| #82 | |
| #83 | def test_complete_passes_provider_and_model_overrides(fake_agent_module): |
| #84 | captured = {} |
| #85 | fake_agent_module.call_llm = lambda **kw: (captured.update(kw) or {"choices": [{"message": {"content": "ok"}}]}) |
| #86 | |
| #87 | adapter = _import_adapter() |
| #88 | backend = adapter.HermesAuxLLMBackend() |
| #89 | backend.complete( |
| #90 | "x", |
| #91 | max_tokens=64, |
| #92 | temperature=0.0, |
| #93 | timeout=10.0, |
| #94 | provider="openai-codex", |
| #95 | model="gpt-5.1-mini", |
| #96 | ) |
| #97 | assert captured["provider"] == "openai-codex" |
| #98 | assert captured["model"] == "gpt-5.1-mini" |
| #99 | |
| #100 | |
| #101 | def test_complete_returns_none_when_agent_import_unavailable(monkeypatch): |
| #102 | """No fake agent in sys.modules → adapter returns None, never raises.""" |
| #103 | monkeypatch.delitem(sys.modules, "agent", raising=False) |
| #104 | monkeypatch.delitem(sys.modules, "agent.auxiliary_client", raising=False) |
| #105 | |
| #106 | adapter = _import_adapter() |
| #107 | backend = adapter.HermesAuxLLMBackend() |
| #108 | assert backend.complete("x", max_tokens=64, temperature=0.0, timeout=5.0) is None |
| #109 | |
| #110 | |
| #111 | def test_complete_returns_none_when_call_llm_raises(fake_agent_module): |
| #112 | def boom(**kwargs): |
| #113 | raise RuntimeError("hermes is angry") |
| #114 | |
| #115 | fake_agent_module.call_llm = boom |
| #116 | |
| #117 | adapter = _import_adapter() |
| #118 | backend = adapter.HermesAuxLLMBackend() |
| #119 | assert backend.complete("x", max_tokens=64, temperature=0.0, timeout=5.0) is None |
| #120 | |
| #121 | |
| #122 | def test_complete_returns_none_when_response_has_no_content(fake_agent_module): |
| #123 | fake_agent_module.call_llm = lambda **kw: {"choices": [{"message": {"content": ""}}]} |
| #124 | adapter = _import_adapter() |
| #125 | backend = adapter.HermesAuxLLMBackend() |
| #126 | assert backend.complete("x", max_tokens=64, temperature=0.0, timeout=5.0) is None |
| #127 | |
| #128 | |
| #129 | # --------------------------------------------------------------------------- |
| #130 | # _extract_content() shape parsing |
| #131 | # --------------------------------------------------------------------------- |
| #132 | |
| #133 | def test_extract_content_prefers_hermes_canonical_helper(fake_agent_module): |
| #134 | """When Hermes' helper exists, use it (handles reasoning models).""" |
| #135 | fake_agent_module.extract_content_or_reasoning = lambda resp: "from-helper" |
| #136 | adapter = _import_adapter() |
| #137 | out = adapter._extract_content({"choices": [{"message": {"content": "from-shape"}}]}) |
| #138 | assert out == "from-helper" |
| #139 | |
| #140 | |
| #141 | def test_extract_content_falls_back_when_helper_returns_empty(fake_agent_module): |
| #142 | fake_agent_module.extract_content_or_reasoning = lambda resp: "" |
| #143 | adapter = _import_adapter() |
| #144 | out = adapter._extract_content({"choices": [{"message": {"content": "from-shape"}}]}) |
| #145 | assert out == "from-shape" |
| #146 | |
| #147 | |
| #148 | def test_extract_content_handles_object_response(fake_agent_module): |
| #149 | fake_agent_module.extract_content_or_reasoning = lambda resp: "" |
| #150 | |
| #151 | class Msg: |
| #152 | content = "object-content" |
| #153 | |
| #154 | class Choice: |
| #155 | message = Msg() |
| #156 | |
| #157 | class Resp: |
| #158 | choices = [Choice()] |
| #159 | |
| #160 | adapter = _import_adapter() |
| #161 | assert adapter._extract_content(Resp()) == "object-content" |
| #162 | |
| #163 | |
| #164 | def test_extract_content_handles_dict_response(fake_agent_module): |
| #165 | fake_agent_module.extract_content_or_reasoning = lambda resp: "" |
| #166 | adapter = _import_adapter() |
| #167 | out = adapter._extract_content({"choices": [{"message": {"content": "dict-content"}}]}) |
| #168 | assert out == "dict-content" |
| #169 | |
| #170 | |
| #171 | def test_extract_content_handles_object_with_content_attr(fake_agent_module): |
| #172 | fake_agent_module.extract_content_or_reasoning = lambda resp: "" |
| #173 | |
| #174 | class Resp: |
| #175 | content = "wrapped-content" |
| #176 | |
| #177 | adapter = _import_adapter() |
| #178 | assert adapter._extract_content(Resp()) == "wrapped-content" |
| #179 | |
| #180 | |
| #181 | def test_extract_content_returns_none_for_unrecognized_shape(fake_agent_module): |
| #182 | fake_agent_module.extract_content_or_reasoning = lambda resp: "" |
| #183 | adapter = _import_adapter() |
| #184 | assert adapter._extract_content(object()) is None |
| #185 | assert adapter._extract_content(None) is None |
| #186 | assert adapter._extract_content({"unrelated": "shape"}) is None |
| #187 | |
| #188 | |
| #189 | # --------------------------------------------------------------------------- |
| #190 | # register/unregister |
| #191 | # --------------------------------------------------------------------------- |
| #192 | |
| #193 | def test_register_hermes_host_llm_installs_backend(monkeypatch): |
| #194 | adapter = _import_adapter() |
| #195 | assert get_host_llm_backend() is None |
| #196 | assert adapter.register_hermes_host_llm() is True |
| #197 | backend = get_host_llm_backend() |
| #198 | assert backend is not None |
| #199 | assert backend.name == "hermes" |
| #200 | # Cleanup happens via autouse fixture, but be explicit. |
| #201 | adapter.unregister_hermes_host_llm() |
| #202 | |
| #203 | |
| #204 | def test_unregister_clears_backend(): |
| #205 | adapter = _import_adapter() |
| #206 | adapter.register_hermes_host_llm() |
| #207 | assert get_host_llm_backend() is not None |
| #208 | adapter.unregister_hermes_host_llm() |
| #209 | assert get_host_llm_backend() is None |
| #210 | |
| #211 | |
| #212 | def test_register_returns_false_when_mnemosyne_registry_missing(monkeypatch): |
| #213 | """If the registry import fails, register returns False instead of raising.""" |
| #214 | adapter = _import_adapter() |
| #215 | # Sabotage the import path used inside register_hermes_host_llm. |
| #216 | monkeypatch.setitem(sys.modules, "mnemosyne.core.llm_backends", None) |
| #217 | try: |
| #218 | assert adapter.register_hermes_host_llm() is False |
| #219 | finally: |
| #220 | # Restore the real module so later tests see a working registry. |
| #221 | monkeypatch.undo() |
| #222 |