ca2ca5b9e1
Three shims caught UnscopedSecretError and degraded to "" (mem0._scoped_env) or to os.environ (langfuse._secret, azure_identity_adapter._scoped_env). Under multiplex os.environ holds the DEFAULT profile's .env, so the langfuse/azure fallback could ship another profile's keys, and the mem0 fallback silently routed a mis-spawned turn's memories into the default profile's account. The exception exists to surface exactly that spawn-site bug (agent/AGENTS.md: never add environ fallthrough, never swallow it). All three now call agent.secret_scope.get_secret directly: with a scope installed a miss returns the default; single-profile deployments (multiplex off) still read the process env inside get_secret; a scope-less multiplex caller raises. Behavior change: a mis-spawned child under gateway.multiplex_profiles now fails loud with UnscopedSecretError instead of running silently unauthenticated / on the default profile's identity. The #99121 contract (OSS mode needs no MEM0_API_KEY in scope) is unchanged and its test now installs an empty profile scope, which is the situation the issue described; a genuinely scope-less caller is asserted to raise in a new test. Tests: tests/plugins/test_scoped_secret_readers_fail_closed.py (scope wins over environ; scope-less multiplex raises) for langfuse + azure, sabotage red when the langfuse fallthrough is restored; tests/plugins/memory/test_mem0_v3.py:: test_load_config_fails_closed_without_scope_even_for_identity_settings, sabotage red with a swallowing wrapper reinstated.
554 lines
23 KiB
Python
554 lines
23 KiB
Python
"""Tests for Mem0 v3 API — new tool names, paginated responses, update/delete tools."""
|
||
|
||
import json
|
||
import threading
|
||
import time
|
||
import pytest
|
||
|
||
from agent import secret_scope
|
||
import plugins.memory.mem0 as mem0_plugin
|
||
from plugins.memory.mem0 import Mem0MemoryProvider
|
||
|
||
|
||
class FakeBackend:
|
||
"""Fake Mem0Backend for provider-level tests."""
|
||
|
||
def __init__(self, search_results=None, all_results=None):
|
||
self._search_results = search_results or []
|
||
self._all_results = all_results or {"results": [], "count": 0}
|
||
self.captured = []
|
||
|
||
def search(self, query, *, filters, top_k=10, rerank=True):
|
||
self.captured.append(("search", query, {"filters": filters, "top_k": top_k, "rerank": rerank}))
|
||
return self._search_results
|
||
|
||
def get_all(self, *, filters, page=1, page_size=100):
|
||
self.captured.append(("get_all", {"filters": filters, "page": page, "page_size": page_size}))
|
||
return self._all_results
|
||
|
||
def add(self, messages, *, user_id, agent_id, infer=False, metadata=None):
|
||
self.captured.append((
|
||
"add",
|
||
messages,
|
||
{"user_id": user_id, "agent_id": agent_id, "infer": infer, "metadata": metadata},
|
||
))
|
||
return {"status": "PENDING", "event_id": "evt-test-123"}
|
||
|
||
def update(self, memory_id, text):
|
||
self.captured.append(("update", memory_id, text))
|
||
return {"result": "Memory updated.", "memory_id": memory_id}
|
||
|
||
def delete(self, memory_id):
|
||
self.captured.append(("delete", memory_id))
|
||
return {"result": "Memory deleted.", "memory_id": memory_id}
|
||
|
||
|
||
class TestMem0V3Tools:
|
||
"""Test v3 tool names and response handling."""
|
||
|
||
def _make_provider(self, monkeypatch, backend):
|
||
provider = Mem0MemoryProvider()
|
||
provider.initialize("test-session")
|
||
provider._user_id = "u123"
|
||
provider._agent_id = "hermes"
|
||
provider._backend = backend
|
||
return provider
|
||
|
||
def test_search_returns_ids(self, monkeypatch):
|
||
backend = FakeBackend(search_results=[{"id": "mem-1", "memory": "foo", "score": 0.9}])
|
||
provider = self._make_provider(monkeypatch, backend)
|
||
result = json.loads(provider.handle_tool_call("mem0_search", {"query": "test"}))
|
||
assert result["results"][0]["id"] == "mem-1"
|
||
|
||
|
||
def test_add_uses_content_param(self, monkeypatch):
|
||
backend = FakeBackend()
|
||
provider = self._make_provider(monkeypatch, backend)
|
||
result = json.loads(provider.handle_tool_call("mem0_add", {"content": "user likes dark mode"}))
|
||
assert len(backend.captured) == 1
|
||
call = backend.captured[0]
|
||
assert call[2]["infer"] is False
|
||
assert call[2]["user_id"] == "u123"
|
||
assert call[2]["agent_id"] == "hermes"
|
||
assert "event_id" in result
|
||
|
||
|
||
def test_old_tool_names_return_unknown(self, monkeypatch):
|
||
backend = FakeBackend()
|
||
provider = self._make_provider(monkeypatch, backend)
|
||
result = json.loads(provider.handle_tool_call("mem0_profile", {}))
|
||
assert "error" in result
|
||
result = json.loads(provider.handle_tool_call("mem0_conclude", {}))
|
||
assert "error" in result
|
||
|
||
|
||
class TestMem0UpdateDelete:
|
||
|
||
def _make_provider(self, monkeypatch, backend):
|
||
provider = Mem0MemoryProvider()
|
||
provider.initialize("test-session")
|
||
provider._user_id = "u123"
|
||
provider._agent_id = "hermes"
|
||
provider._backend = backend
|
||
return provider
|
||
|
||
def test_update_calls_sdk(self, monkeypatch):
|
||
backend = FakeBackend()
|
||
provider = self._make_provider(monkeypatch, backend)
|
||
result = json.loads(provider.handle_tool_call(
|
||
"mem0_update", {"memory_id": "mem-1", "text": "updated fact"}
|
||
))
|
||
assert backend.captured[0][1] == "mem-1"
|
||
assert backend.captured[0][2] == "updated fact"
|
||
assert result["result"] == "Memory updated."
|
||
assert result["memory_id"] == "mem-1"
|
||
|
||
|
||
def test_delete_calls_sdk(self, monkeypatch):
|
||
backend = FakeBackend()
|
||
provider = self._make_provider(monkeypatch, backend)
|
||
result = json.loads(provider.handle_tool_call(
|
||
"mem0_delete", {"memory_id": "mem-1"}
|
||
))
|
||
assert backend.captured[0][1] == "mem-1"
|
||
assert result["result"] == "Memory deleted."
|
||
|
||
|
||
class TestMem0ErrorHandling:
|
||
|
||
def _make_provider(self, monkeypatch, backend):
|
||
provider = Mem0MemoryProvider()
|
||
provider.initialize("test-session")
|
||
provider._user_id = "u123"
|
||
provider._agent_id = "hermes"
|
||
provider._backend = backend
|
||
return provider
|
||
|
||
|
||
class TestMem0V3Internal:
|
||
|
||
def _make_provider(self, monkeypatch, backend):
|
||
provider = Mem0MemoryProvider()
|
||
provider.initialize("test-session")
|
||
provider._user_id = "u123"
|
||
provider._agent_id = "hermes"
|
||
provider._backend = backend
|
||
return provider
|
||
|
||
def test_sync_turn_explicit_kwargs(self, monkeypatch):
|
||
backend = FakeBackend()
|
||
provider = self._make_provider(monkeypatch, backend)
|
||
provider.sync_turn("user said", "assistant replied", session_id="s1")
|
||
provider._sync_thread.join(timeout=2)
|
||
assert len(backend.captured) == 1
|
||
call = backend.captured[0]
|
||
assert call[2]["user_id"] == "u123"
|
||
assert call[2]["agent_id"] == "hermes"
|
||
assert call[2]["infer"] is True
|
||
|
||
|
||
class TestSyncTurnTruncation:
|
||
"""sync_turn must cap messages before ingestion so small-context embedding
|
||
backends (OSS Ollama bge-small-zh-v1.5: 512 tokens; jina-embeddings-v3 token
|
||
limits) don't fail the whole extraction — a failure _try only logs."""
|
||
|
||
def _make_provider(self, monkeypatch, backend):
|
||
provider = Mem0MemoryProvider()
|
||
provider.initialize("test-session")
|
||
provider._user_id = "u123"
|
||
provider._agent_id = "hermes"
|
||
provider._backend = backend
|
||
return provider
|
||
|
||
def test_small_context_backend_never_sees_oversized_input(self, monkeypatch):
|
||
"""Regression for #106235/#37421: an OSS embedding backend with a small
|
||
context window raises on oversized input; truncation up front keeps the
|
||
extraction from being silently dropped (no breaker failures)."""
|
||
|
||
class SmallContextBackend(FakeBackend):
|
||
def add(self, messages, **kwargs):
|
||
if any(len(m["content"]) > mem0_plugin._SYNC_MSG_MAX_CHARS for m in messages):
|
||
raise RuntimeError("HTTP 500: embedding input exceeds model context")
|
||
return super().add(messages, **kwargs)
|
||
|
||
backend = SmallContextBackend()
|
||
provider = self._make_provider(monkeypatch, backend)
|
||
provider.sync_turn("Short question?", "".join(f"Fact {i}. " for i in range(200)), session_id="s1")
|
||
provider._sync_thread.join(timeout=2)
|
||
assert len(backend.captured) == 1
|
||
sent = backend.captured[0][1]
|
||
assert sent[0]["content"] == "Short question?" # under the cap: untouched
|
||
assert len(sent[1]["content"]) <= mem0_plugin._SYNC_MSG_MAX_CHARS and sent[1]["content"].endswith(".")
|
||
assert provider._consecutive_failures == 0
|
||
|
||
def test_the_boundary_kept_is_the_last_one_in_the_window_whatever_its_script(self):
|
||
"""A mixed-script turn must not be cut back to an early CJK stop.
|
||
|
||
The trim exists to keep as much of the turn as the embedder can take; picking the
|
||
first separator KIND that qualifies instead of the last boundary threw away most of
|
||
the allowed window whenever two kinds appeared — an early ``。`` (or ``.``, which
|
||
outranks ``!``/``?``) beat a boundary 240 characters later, so the facts stated in
|
||
the rest of the message never reached extraction.
|
||
"""
|
||
cap = mem0_plugin._SYNC_MSG_MAX_CHARS
|
||
early, late = cap // 2, cap - 9
|
||
|
||
for early_sep, late_sep in (("。", "."), (".", "!"), ("？", "?"), ("！", ".")):
|
||
text = "a" * early + early_sep + "b" * (late - early - 1) + late_sep + "c" * cap
|
||
assert text[late] == late_sep and len(text) > cap # both boundaries inside the window
|
||
kept = mem0_plugin._truncate_for_sync(text)
|
||
assert kept == text[:late + 1], f"{early_sep!r} before {late_sep!r} cut back to {len(kept)} chars"
|
||
assert kept.endswith(late_sep)
|
||
|
||
def test_a_boundary_only_in_the_first_third_still_falls_back_to_a_hard_cut(self):
|
||
"""Unsegmented input keeps the whole window rather than a sliver of a sentence."""
|
||
cap = mem0_plugin._SYNC_MSG_MAX_CHARS
|
||
text = "a" * 10 + "." + "b" * (cap * 2)
|
||
assert mem0_plugin._truncate_for_sync(text) == text[:cap]
|
||
|
||
def test_sync_max_chars_config_raises_cap(self, monkeypatch, tmp_path):
|
||
"""8k-token embedders should not be stuck at the 512-token default (#106235)."""
|
||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||
monkeypatch.setenv("MEM0_API_KEY", "test-key")
|
||
(tmp_path / "mem0.json").write_text('{"sync_max_chars": 3000}')
|
||
backend = FakeBackend()
|
||
provider = self._make_provider(monkeypatch, backend)
|
||
provider.sync_turn("hi", "Long answer. " * 200, session_id="s1") # 2600 chars
|
||
provider._sync_thread.join(timeout=2)
|
||
assert backend.captured[0][1][1]["content"] == "Long answer. " * 200
|
||
|
||
|
||
class TestMem0Prefetch:
|
||
"""prefetch() must recall on the CURRENT question, synchronously.
|
||
|
||
The old implementation ignored its ``query`` and returned whatever a
|
||
background ``queue_prefetch`` had warmed from the PREVIOUS turn — so the
|
||
first turn injected nothing and later turns injected stale, off-topic
|
||
memories. These lock the corrected behaviour.
|
||
"""
|
||
|
||
def _make_provider(self, backend):
|
||
provider = Mem0MemoryProvider()
|
||
provider.initialize("test-session")
|
||
provider._user_id = "u123"
|
||
provider._agent_id = "hermes"
|
||
provider._backend = backend
|
||
return provider
|
||
|
||
def test_prefetch_searches_current_query(self):
|
||
backend = FakeBackend(search_results=[{"id": "m1", "memory": "user prefers dark mode"}])
|
||
provider = self._make_provider(backend)
|
||
result = provider.prefetch("what theme do I like?")
|
||
kind, query, opts = backend.captured[0]
|
||
assert kind == "search"
|
||
assert query == "what theme do I like?"
|
||
assert opts["filters"] == {"user_id": "u123"}
|
||
assert opts["top_k"] == 10
|
||
assert opts["rerank"] is False
|
||
assert "## Mem0 Memory" in result
|
||
assert "user prefers dark mode" in result
|
||
|
||
|
||
def test_on_turn_start_queues_current_query(self):
|
||
backend = FakeBackend(search_results=[{"id": "m1", "memory": "lives in Berlin"}])
|
||
provider = self._make_provider(backend)
|
||
provider.on_turn_start(1, "where do I live?")
|
||
provider._prefetch_thread.join(timeout=1)
|
||
result = provider.prefetch("where do I live?")
|
||
assert "lives in Berlin" in result
|
||
assert len([c for c in backend.captured if c[0] == "search"]) == 1
|
||
|
||
def test_slow_prefetch_returns_quickly(self, monkeypatch):
|
||
entered = threading.Event()
|
||
release = threading.Event()
|
||
search_returned = threading.Event()
|
||
|
||
class SlowBackend(FakeBackend):
|
||
def search(self, query, *, filters, top_k=10, rerank=True):
|
||
entered.set()
|
||
try:
|
||
release.wait(30)
|
||
return super().search(
|
||
query, filters=filters, top_k=top_k, rerank=rerank
|
||
)
|
||
finally:
|
||
search_returned.set()
|
||
|
||
monkeypatch.setattr(mem0_plugin, "_PREFETCH_WAIT_SECS", 0.01)
|
||
provider = self._make_provider(
|
||
SlowBackend(search_results=[{"id": "m1", "memory": "lives in Berlin"}])
|
||
)
|
||
# DETERMINISTIC non-blocking witness — replaces `assert elapsed < 0.1`.
|
||
#
|
||
# The old form slept 0.2s in the backend and asserted prefetch returned
|
||
# in under 0.1s. That makes the OS scheduler part of the assertion: on
|
||
# a loaded box thread startup alone can eat the 100ms budget, so the
|
||
# inequality flips with nothing wrong in the code under test. Observed
|
||
# failing in a full-directory run of tests/plugins/memory.
|
||
#
|
||
# The real contract is that prefetch gives up on the slow backend
|
||
# instead of waiting for it. Assert it directly: the backend search is
|
||
# STILL PARKED (release unset, so `search_returned` cannot be set). If
|
||
# prefetch ever waited for the backend, the search would have returned
|
||
# first and this fails. No wall-clock constant.
|
||
assert provider.prefetch("where do I live?") == ""
|
||
assert entered.wait(30), "prefetch never reached the backend"
|
||
assert not search_returned.is_set(), (
|
||
"prefetch blocked on the slow backend: the backend search had "
|
||
"already returned by the time prefetch did"
|
||
)
|
||
|
||
release.set()
|
||
provider._prefetch_thread.join(timeout=30)
|
||
assert "lives in Berlin" in provider.prefetch("where do I live?")
|
||
|
||
|
||
def test_queue_prefetch_fires_no_search(self):
|
||
# prefetch is synchronous now, so the post-turn warm is redundant and
|
||
# must not fire a wasted backend search.
|
||
backend = FakeBackend(search_results=[{"id": "m1", "memory": "x"}])
|
||
provider = self._make_provider(backend)
|
||
provider.queue_prefetch("previous turn text")
|
||
assert backend.captured == []
|
||
|
||
|
||
class TestMem0V3Config:
|
||
|
||
def test_tool_schemas_four_tools(self):
|
||
provider = Mem0MemoryProvider()
|
||
schemas = provider.get_tool_schemas()
|
||
names = [s["name"] for s in schemas]
|
||
assert names == ["mem0_search", "mem0_add", "mem0_update", "mem0_delete"]
|
||
|
||
def test_system_prompt_new_tool_names(self):
|
||
provider = Mem0MemoryProvider()
|
||
provider._user_id = "test"
|
||
block = provider.system_prompt_block()
|
||
assert "mem0_search" in block
|
||
assert "mem0_add" in block
|
||
assert "mem0_update" in block
|
||
assert "mem0_delete" in block
|
||
assert "mem0_list" not in block
|
||
assert "mem0_profile" not in block
|
||
assert "mem0_conclude" not in block
|
||
|
||
|
||
class TestMem0ModeSwitch:
|
||
|
||
def test_oss_mode_initializes_without_platform_key_in_scope(
|
||
self, monkeypatch, tmp_path
|
||
):
|
||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||
monkeypatch.delenv("MEM0_API_KEY", raising=False)
|
||
(tmp_path / "mem0.json").write_text(
|
||
json.dumps(
|
||
{
|
||
"mode": "oss",
|
||
"oss": {"vector_store": {"provider": "qdrant"}},
|
||
}
|
||
)
|
||
)
|
||
|
||
# A profile scope WITHOUT the platform key: OSS mode must not demand MEM0_API_KEY. (A
|
||
# scope-less caller is a spawn-site bug and raises; see test_load_config_fails_closed_without_scope.)
|
||
token = secret_scope.set_secret_scope({})
|
||
secret_scope.set_multiplex_active(True)
|
||
try:
|
||
provider = Mem0MemoryProvider()
|
||
provider._create_backend = lambda: None # type: ignore[method-assign]
|
||
provider.initialize("test")
|
||
available = provider.is_available()
|
||
finally:
|
||
secret_scope.set_multiplex_active(False)
|
||
secret_scope.reset_secret_scope(token)
|
||
|
||
assert provider._mode == "oss"
|
||
assert provider._api_key == ""
|
||
assert available is True
|
||
|
||
def test_platform_config_still_fails_closed_without_profile_scope(
|
||
self, monkeypatch, tmp_path
|
||
):
|
||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||
monkeypatch.delenv("MEM0_API_KEY", raising=False)
|
||
|
||
token = secret_scope.set_secret_scope(None)
|
||
secret_scope.set_multiplex_active(True)
|
||
try:
|
||
with pytest.raises(secret_scope.UnscopedSecretError):
|
||
Mem0MemoryProvider().is_available()
|
||
finally:
|
||
secret_scope.set_multiplex_active(False)
|
||
secret_scope.reset_secret_scope(token)
|
||
|
||
def test_load_config_fails_closed_without_scope_even_for_identity_settings(
|
||
self, monkeypatch, tmp_path
|
||
):
|
||
"""A scope-less multiplex caller is a spawn-site bug: identity/mode reads must surface it,
|
||
not degrade to '' and route the turn's memories into the default profile's account."""
|
||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||
(tmp_path / "mem0.json").write_text(json.dumps({"mode": "oss", "oss": {"vector_store": {"provider": "qdrant"}}}))
|
||
|
||
token = secret_scope.set_secret_scope(None)
|
||
secret_scope.set_multiplex_active(True)
|
||
try:
|
||
with pytest.raises(secret_scope.UnscopedSecretError):
|
||
mem0_plugin._load_config()
|
||
finally:
|
||
secret_scope.set_multiplex_active(False)
|
||
secret_scope.reset_secret_scope(token)
|
||
|
||
def test_file_api_key_still_overrides_environment(self, monkeypatch, tmp_path):
|
||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||
monkeypatch.setenv("MEM0_API_KEY", "env-key")
|
||
(tmp_path / "mem0.json").write_text(
|
||
json.dumps({"api_key": "file-key"})
|
||
)
|
||
|
||
assert mem0_plugin._load_config()["api_key"] == "file-key"
|
||
|
||
def test_default_mode_is_platform(self, monkeypatch, tmp_path):
|
||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||
monkeypatch.setenv("MEM0_API_KEY", "test-key")
|
||
provider = Mem0MemoryProvider()
|
||
provider.initialize("test")
|
||
assert provider._mode == "platform"
|
||
|
||
def test_missing_mode_key_defaults_platform(self, monkeypatch, tmp_path):
|
||
"""Backward compat: old mem0.json without mode key works."""
|
||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||
config_path = tmp_path / "mem0.json"
|
||
config_path.write_text('{"user_id": "old-user"}')
|
||
monkeypatch.setenv("MEM0_API_KEY", "test-key")
|
||
provider = Mem0MemoryProvider()
|
||
provider.initialize("test")
|
||
assert provider._mode == "platform"
|
||
assert provider._user_id == "old-user"
|
||
|
||
def test_is_available_platform_needs_key(self, monkeypatch, tmp_path):
|
||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||
monkeypatch.delenv("MEM0_API_KEY", raising=False)
|
||
provider = Mem0MemoryProvider()
|
||
assert provider.is_available() is False
|
||
|
||
|
||
class TestMem0UserIdResolution:
|
||
"""user_id resolution: configured override > gateway-native id > placeholder.
|
||
|
||
Same human across CLI / Telegram / Discord / Slack / etc. should map to
|
||
the same memory store when MEM0_USER_ID is set, and only fall back to the
|
||
gateway-native id when it isn't.
|
||
"""
|
||
|
||
def _provider(self, monkeypatch, tmp_path):
|
||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||
monkeypatch.setenv("MEM0_API_KEY", "test-key")
|
||
provider = Mem0MemoryProvider()
|
||
# Skip backend instantiation — we only care about identity resolution.
|
||
provider._create_backend = lambda: None # type: ignore[method-assign]
|
||
return provider
|
||
|
||
def test_env_override_beats_gateway_native_id(self, monkeypatch, tmp_path):
|
||
monkeypatch.setenv("MEM0_USER_ID", "ryan@example.com")
|
||
provider = self._provider(monkeypatch, tmp_path)
|
||
provider.initialize("test", user_id="123456789", platform="telegram")
|
||
assert provider._user_id == "ryan@example.com"
|
||
|
||
def test_file_override_beats_gateway_native_id(self, monkeypatch, tmp_path):
|
||
monkeypatch.delenv("MEM0_USER_ID", raising=False)
|
||
(tmp_path / "mem0.json").write_text('{"user_id": "ryan@example.com"}')
|
||
provider = self._provider(monkeypatch, tmp_path)
|
||
provider.initialize("test", user_id="123456789", platform="telegram")
|
||
assert provider._user_id == "ryan@example.com"
|
||
|
||
def test_unset_falls_back_to_gateway_native_id(self, monkeypatch, tmp_path):
|
||
monkeypatch.delenv("MEM0_USER_ID", raising=False)
|
||
provider = self._provider(monkeypatch, tmp_path)
|
||
provider.initialize("test", user_id="123456789", platform="telegram")
|
||
assert provider._user_id == "123456789"
|
||
|
||
|
||
def test_legacy_placeholder_in_config_does_not_override_kwargs(self, monkeypatch, tmp_path):
|
||
# Setup wizard historically wrote {"user_id": "hermes-user"} as the
|
||
# suggested default. Treat that placeholder as unset so users on
|
||
# gateways still get gateway-native ids — not silent collisions.
|
||
monkeypatch.delenv("MEM0_USER_ID", raising=False)
|
||
(tmp_path / "mem0.json").write_text('{"user_id": "hermes-user"}')
|
||
provider = self._provider(monkeypatch, tmp_path)
|
||
provider.initialize("test", user_id="123456789", platform="telegram")
|
||
assert provider._user_id == "123456789"
|
||
|
||
|
||
class TestMem0WriteMetadata:
|
||
"""Writes carry metadata.channel so per-channel filtered views are possible
|
||
without coupling identity to the channel.
|
||
"""
|
||
|
||
def _make_provider(self, channel: str = "cli"):
|
||
provider = Mem0MemoryProvider()
|
||
provider._user_id = "u123"
|
||
provider._agent_id = "hermes"
|
||
provider._channel = channel
|
||
provider._backend = FakeBackend()
|
||
return provider
|
||
|
||
|
||
class _SentinelBackend:
|
||
def __init__(self, *args):
|
||
self.args = args
|
||
|
||
|
||
class TestCreateBackendRouting:
|
||
"""_create_backend() must pick the backend matching the configured mode/host."""
|
||
|
||
def _provider(self, monkeypatch, *, mode="platform", api_key="k", host=""):
|
||
# Neutralize lazy-install so the routing decision is all we exercise.
|
||
monkeypatch.setattr("tools.lazy_deps.ensure", lambda *a, **k: None, raising=False)
|
||
provider = Mem0MemoryProvider()
|
||
provider._mode = mode
|
||
provider._api_key = api_key
|
||
provider._host = host
|
||
provider._config = {"oss": {"vector_store": {"provider": "qdrant"}}}
|
||
return provider
|
||
|
||
def test_routes_to_selfhosted_when_host_set(self, monkeypatch):
|
||
captured = {}
|
||
|
||
class SH(_SentinelBackend):
|
||
def __init__(self, api_key, host):
|
||
captured["args"] = (api_key, host)
|
||
|
||
monkeypatch.setattr("plugins.memory.mem0._backend.SelfHostedBackend", SH)
|
||
provider = self._provider(monkeypatch, host="http://sh:8888", api_key="adminkey")
|
||
backend = provider._create_backend()
|
||
assert isinstance(backend, SH)
|
||
assert captured["args"] == ("adminkey", "http://sh:8888")
|
||
|
||
|
||
def test_oss_mode_takes_precedence_over_host(self, monkeypatch):
|
||
class OB(_SentinelBackend):
|
||
def __init__(self, cfg):
|
||
pass
|
||
|
||
monkeypatch.setattr("plugins.memory.mem0._backend.OSSBackend", OB)
|
||
provider = self._provider(monkeypatch, mode="oss", host="http://sh:8888")
|
||
assert isinstance(provider._create_backend(), OB)
|
||
|
||
def test_prompt_label_matches_routing_when_oss_and_host_both_set(self, monkeypatch):
|
||
# system_prompt_block must mirror _create_backend precedence: with both
|
||
# mode=oss and host set, OSS wins the routing, so the prompt must label
|
||
# OSS — not "self-hosted (HTTP API)". Guards the prompt-vs-routing lie.
|
||
provider = self._provider(monkeypatch, mode="oss", host="http://sh:8888")
|
||
provider._user_id = "test"
|
||
block = provider.system_prompt_block()
|
||
assert "OSS" in block
|
||
assert "HTTP API" not in block
|
||
|
||
|
||
class TestSelfHostedConfig:
|
||
"""Config plumbing for self-hosted (MEM0_HOST env + is_available)."""
|
||
|
||
def test_load_config_reads_mem0_host_env(self, monkeypatch):
|
||
monkeypatch.setenv("MEM0_HOST", "http://localhost:8888")
|
||
assert mem0_plugin._load_config()["host"] == "http://localhost:8888"
|