mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-08-01 19:06:01 +00:00
352f247a81
* feat(memory): add mem0 HTTP memory backend * fix(memory): address mem0 review feedback --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
631 lines
25 KiB
Python
631 lines
25 KiB
Python
"""Unit tests for the mem0 HTTP memory backend (backends/mem0/)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import threading
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from deerflow.agents.memory.backends.mem0.client import Mem0APIError, Mem0AuthError, Mem0Client
|
|
from deerflow.agents.memory.backends.mem0.config import Mem0Config
|
|
|
|
|
|
class TestMem0Config:
|
|
def test_defaults(self) -> None:
|
|
cfg = Mem0Config.from_backend_config({})
|
|
assert cfg.api_key_env == "MEM0_API_KEY"
|
|
assert cfg.base_url == "https://api.mem0.ai"
|
|
assert cfg.allow_insecure_http is False
|
|
assert cfg.top_k == 8
|
|
assert cfg.score_threshold == 0.1
|
|
assert cfg.max_injection_chars == 12000
|
|
assert cfg.timeout_seconds == 10.0
|
|
assert cfg.startup_policy == "fail_fast"
|
|
assert cfg.read_policy == "fail_open"
|
|
assert cfg.write_policy == "log_and_drop"
|
|
|
|
def test_custom_values_and_nested_failure_policy(self) -> None:
|
|
cfg = Mem0Config.from_backend_config(
|
|
{
|
|
"api_key_env": "MY_MEM0_KEY",
|
|
"base_url": "http://mem0.local:8888/",
|
|
"allow_insecure_http": True,
|
|
"top_k": 5,
|
|
"score_threshold": 0.3,
|
|
"max_injection_chars": 4000,
|
|
"timeout_seconds": 3,
|
|
"startup_policy": "tolerate",
|
|
"failure_policy": {"read": "fail_closed", "write": "raise"},
|
|
}
|
|
)
|
|
assert cfg.api_key_env == "MY_MEM0_KEY"
|
|
assert cfg.base_url == "http://mem0.local:8888" # trailing slash stripped
|
|
assert cfg.allow_insecure_http is True
|
|
assert cfg.top_k == 5
|
|
assert cfg.score_threshold == 0.3
|
|
assert cfg.max_injection_chars == 4000
|
|
assert cfg.timeout_seconds == 3.0
|
|
assert cfg.startup_policy == "tolerate"
|
|
assert cfg.read_policy == "fail_closed"
|
|
assert cfg.write_policy == "raise"
|
|
|
|
def test_unknown_keys_rejected_except_host_injected(self) -> None:
|
|
with pytest.raises(ValueError, match="unknown"):
|
|
Mem0Config.from_backend_config({"typo_knob": 1})
|
|
# Host-injected keys must be tolerated (factory injects storage_path
|
|
# into every backend's backend_config).
|
|
cfg = Mem0Config.from_backend_config({"storage_path": "/tmp/x", "should_keep_hidden_message": None})
|
|
assert cfg.base_url == "https://api.mem0.ai"
|
|
|
|
def test_insecure_http_requires_explicit_opt_in(self) -> None:
|
|
with pytest.raises(ValueError, match="allow_insecure_http"):
|
|
Mem0Config.from_backend_config({"base_url": "http://mem0.local:8888"})
|
|
|
|
@pytest.mark.parametrize("base_url", ["mem0.local:8888", "ftp://mem0.local", "https:///missing-host"])
|
|
def test_invalid_base_url_rejected(self, base_url: str) -> None:
|
|
with pytest.raises(ValueError, match="base_url"):
|
|
Mem0Config.from_backend_config({"base_url": base_url, "allow_insecure_http": True})
|
|
|
|
@pytest.mark.parametrize(
|
|
("key", "value"),
|
|
[
|
|
("startup_policy", "sometimes"),
|
|
("top_k", 0),
|
|
("top_k", 1001),
|
|
("score_threshold", 1.5),
|
|
("max_injection_chars", 0),
|
|
("timeout_seconds", 0),
|
|
],
|
|
)
|
|
def test_invalid_values_rejected(self, key: str, value: object) -> None:
|
|
with pytest.raises(ValueError):
|
|
Mem0Config.from_backend_config({key: value})
|
|
|
|
@pytest.mark.parametrize("policy", ["read", "write"])
|
|
def test_invalid_failure_policy_rejected(self, policy: str) -> None:
|
|
with pytest.raises(ValueError, match=policy):
|
|
Mem0Config.from_backend_config({"failure_policy": {policy: "bogus"}})
|
|
|
|
def test_resolve_api_key(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
cfg = Mem0Config.from_backend_config({})
|
|
monkeypatch.delenv("MEM0_API_KEY", raising=False)
|
|
with pytest.raises(ValueError, match="MEM0_API_KEY"):
|
|
cfg.resolve_api_key()
|
|
monkeypatch.setenv("MEM0_API_KEY", " ")
|
|
with pytest.raises(ValueError, match="MEM0_API_KEY"):
|
|
cfg.resolve_api_key()
|
|
monkeypatch.setenv("MEM0_API_KEY", "secret-key")
|
|
assert cfg.resolve_api_key() == "secret-key"
|
|
|
|
|
|
def _client(handler) -> Mem0Client:
|
|
return Mem0Client(
|
|
base_url="https://api.mem0.ai",
|
|
api_key="test-key",
|
|
transport=httpx.MockTransport(handler),
|
|
)
|
|
|
|
|
|
class TestMem0Client:
|
|
def test_add_memories_payload(self) -> None:
|
|
seen = {}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
seen["path"] = request.url.path
|
|
seen["auth"] = request.headers["authorization"]
|
|
seen["body"] = httpx.QueryParams # placeholder, replaced below
|
|
import json
|
|
|
|
seen["body"] = json.loads(request.content)
|
|
return httpx.Response(200, json={"status": "PENDING", "event_id": "evt-1"})
|
|
|
|
client = _client(handler)
|
|
result = client.add_memories(
|
|
messages=[{"role": "user", "content": "hi"}],
|
|
user_id="u1",
|
|
agent_id="lead_agent",
|
|
run_id="t-1",
|
|
)
|
|
assert result["event_id"] == "evt-1"
|
|
assert seen["path"] == "/v3/memories/add/"
|
|
assert seen["auth"] == "Token test-key"
|
|
assert seen["body"] == {
|
|
"messages": [{"role": "user", "content": "hi"}],
|
|
"user_id": "u1",
|
|
"agent_id": "lead_agent",
|
|
"run_id": "t-1",
|
|
}
|
|
|
|
def test_search_memories_returns_results(self) -> None:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
import json
|
|
|
|
body = json.loads(request.content)
|
|
assert request.url.path == "/v3/memories/search/"
|
|
assert body == {
|
|
"query": "hobbies",
|
|
"filters": {"user_id": "u1"},
|
|
"top_k": 5,
|
|
"threshold": 0.2,
|
|
}
|
|
return httpx.Response(200, json={"results": [{"id": "m1", "memory": "likes cricket", "score": 0.9}]})
|
|
|
|
results = _client(handler).search_memories(query="hobbies", filters={"user_id": "u1"}, top_k=5, threshold=0.2)
|
|
assert results == [{"id": "m1", "memory": "likes cricket", "score": 0.9}]
|
|
|
|
def test_list_memories_paginates_and_respects_max_items(self) -> None:
|
|
pages = {
|
|
1: {"results": [{"id": "a"}, {"id": "b"}], "next": "https://x/?page=2"},
|
|
2: {"results": [{"id": "c"}], "next": None},
|
|
}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
page = int(request.url.params["page"])
|
|
return httpx.Response(200, json=pages[page])
|
|
|
|
client = _client(handler)
|
|
assert client.list_memories(filters={"user_id": "u1"}) == [{"id": "a"}, {"id": "b"}, {"id": "c"}]
|
|
assert client.list_memories(filters={"user_id": "u1"}, max_items=2) == [{"id": "a"}, {"id": "b"}]
|
|
|
|
def test_delete_all_memories_uses_query_params(self) -> None:
|
|
seen = {}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
seen["method"] = request.method
|
|
seen["path"] = request.url.path
|
|
seen["params"] = dict(request.url.params)
|
|
return httpx.Response(200, json={"message": "deleted"})
|
|
|
|
_client(handler).delete_all_memories(user_id="u1", agent_id="lead_agent", run_id=None)
|
|
assert seen == {
|
|
"method": "DELETE",
|
|
"path": "/v1/memories/",
|
|
"params": {"user_id": "u1", "agent_id": "lead_agent"},
|
|
}
|
|
|
|
def test_401_raises_auth_error(self) -> None:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(401, json={"detail": "invalid key"})
|
|
|
|
with pytest.raises(Mem0AuthError):
|
|
_client(handler).ping()
|
|
|
|
def test_other_4xx_raises_api_error(self) -> None:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(400, json={"error": "bad request"})
|
|
|
|
with pytest.raises(Mem0APIError, match="400"):
|
|
_client(handler).list_memories(filters={"user_id": "u1"})
|
|
|
|
def test_transport_error_raises_api_error(self) -> None:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
raise httpx.ConnectError("boom")
|
|
|
|
with pytest.raises(Mem0APIError, match="boom"):
|
|
_client(handler).ping()
|
|
|
|
def test_malformed_json_raises_api_error(self) -> None:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(200, content=b"not json{")
|
|
|
|
with pytest.raises(Mem0APIError, match="malformed JSON"):
|
|
_client(handler).list_memories(filters={"user_id": "u1"})
|
|
|
|
|
|
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage # noqa: E402
|
|
|
|
from deerflow.agents.memory.backends.mem0.message_filtering import ( # noqa: E402
|
|
extract_message_text,
|
|
filter_messages_for_memory,
|
|
)
|
|
|
|
|
|
def _clarification_kwargs() -> dict:
|
|
return {
|
|
"hide_from_ui": True,
|
|
"human_input_response": {
|
|
"version": 1,
|
|
"kind": "human_input_response",
|
|
"source": "clarification",
|
|
"request_id": "req-1",
|
|
"response_kind": "text",
|
|
"value": "the user answered this",
|
|
},
|
|
}
|
|
|
|
|
|
class TestMessageFiltering:
|
|
def test_keeps_user_and_final_assistant(self) -> None:
|
|
msgs = [HumanMessage(content="hello"), AIMessage(content="hi there")]
|
|
assert filter_messages_for_memory(msgs) == msgs
|
|
|
|
def test_drops_tool_messages_and_tool_call_ai(self) -> None:
|
|
tool_ai = AIMessage(content="", tool_calls=[{"name": "t", "args": {}, "id": "1"}])
|
|
msgs = [HumanMessage(content="q"), tool_ai, ToolMessage(content="out", tool_call_id="1"), AIMessage(content="a")]
|
|
assert filter_messages_for_memory(msgs) == [msgs[0], msgs[3]]
|
|
|
|
def test_drops_hidden_framework_messages_keeps_clarification(self) -> None:
|
|
hidden = HumanMessage(content="todo reminder", additional_kwargs={"hide_from_ui": True})
|
|
clarification = HumanMessage(content="the user answered this", additional_kwargs=_clarification_kwargs())
|
|
assert filter_messages_for_memory([hidden, clarification]) == [clarification]
|
|
|
|
def test_upload_only_human_drops_it_and_following_ai(self) -> None:
|
|
upload_only = HumanMessage(content="<uploaded_files>\nfile.pdf\n</uploaded_files>")
|
|
ack = AIMessage(content="I see your file")
|
|
followup = HumanMessage(content="what is in it?")
|
|
assert filter_messages_for_memory([upload_only, ack, followup]) == [followup]
|
|
|
|
def test_upload_block_stripped_from_mixed_message(self) -> None:
|
|
mixed = HumanMessage(content="<current_uploads>\nf.txt\n</current_uploads>\nsummarize this")
|
|
(kept,) = filter_messages_for_memory([mixed])
|
|
assert extract_message_text(kept) == "summarize this"
|
|
|
|
def test_extract_message_text_handles_list_content(self) -> None:
|
|
msg = AIMessage(content=[{"type": "text", "text": "part one"}, "part two"])
|
|
assert extract_message_text(msg) == "part one part two"
|
|
|
|
def test_extract_message_text_treats_none_as_empty(self) -> None:
|
|
msg = AIMessage(content="")
|
|
msg.content = None
|
|
assert extract_message_text(msg) == ""
|
|
|
|
|
|
from typing import Any # noqa: E402
|
|
|
|
from deerflow.agents.memory.backends.mem0.mem0_manager import Mem0Manager # noqa: E402
|
|
|
|
|
|
class FakeMem0Client:
|
|
"""Test double injected as manager._client (records calls, returns fixtures)."""
|
|
|
|
def __init__(self) -> None:
|
|
self.added: list[dict[str, Any]] = []
|
|
self.deleted: list[dict[str, Any]] = []
|
|
self.search_calls: list[dict[str, Any]] = []
|
|
self.list_calls: list[dict[str, Any]] = []
|
|
self.pings = 0
|
|
self.search_results: list[dict[str, Any]] = []
|
|
self.list_results: list[dict[str, Any]] = []
|
|
self.error: Exception | None = None
|
|
self.closed = False
|
|
|
|
def _maybe_raise(self) -> None:
|
|
if self.error is not None:
|
|
raise self.error
|
|
|
|
def add_memories(self, **kwargs: Any) -> dict[str, Any]:
|
|
self._maybe_raise()
|
|
self.added.append(kwargs)
|
|
return {"status": "PENDING", "event_id": "evt-fake"}
|
|
|
|
def search_memories(self, **kwargs: Any) -> list[dict[str, Any]]:
|
|
self._maybe_raise()
|
|
self.search_calls.append(kwargs)
|
|
return self.search_results
|
|
|
|
def list_memories(self, **kwargs: Any) -> list[dict[str, Any]]:
|
|
self._maybe_raise()
|
|
self.list_calls.append(kwargs)
|
|
return self.list_results
|
|
|
|
def delete_all_memories(self, **kwargs: Any) -> None:
|
|
self._maybe_raise()
|
|
self.deleted.append(kwargs)
|
|
|
|
def ping(self) -> None:
|
|
self._maybe_raise()
|
|
self.pings += 1
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _mem0_api_key(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Mem0Manager resolves the API key eagerly at construction; provide a dummy
|
|
so the suite is hermetic (tests that need it missing delete it themselves)."""
|
|
monkeypatch.setenv("MEM0_API_KEY", "test-key")
|
|
|
|
|
|
def _manager(backend_config: dict | None = None, *, mode: str = "middleware") -> tuple[Mem0Manager, FakeMem0Client]:
|
|
mgr = Mem0Manager(backend_config=backend_config or {}, mode=mode)
|
|
fake = FakeMem0Client()
|
|
mgr._client = fake
|
|
return mgr, fake
|
|
|
|
|
|
class TestMem0ManagerConstruction:
|
|
def test_from_config_fail_fast_pings(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("MEM0_API_KEY", "k")
|
|
fake = FakeMem0Client()
|
|
monkeypatch.setattr(
|
|
"deerflow.agents.memory.backends.mem0.mem0_manager.Mem0Client",
|
|
lambda **kwargs: fake,
|
|
)
|
|
Mem0Manager.from_config({}, mode="middleware")
|
|
assert fake.pings == 1
|
|
|
|
def test_from_config_tolerate_skips_ping(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("MEM0_API_KEY", "k")
|
|
fake = FakeMem0Client()
|
|
monkeypatch.setattr(
|
|
"deerflow.agents.memory.backends.mem0.mem0_manager.Mem0Client",
|
|
lambda **kwargs: fake,
|
|
)
|
|
mgr = Mem0Manager.from_config({"startup_policy": "tolerate"}, mode="tool")
|
|
assert fake.pings == 0
|
|
assert mgr.mode == "tool"
|
|
|
|
def test_from_config_missing_key_raises(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.delenv("MEM0_API_KEY", raising=False)
|
|
with pytest.raises(ValueError, match="MEM0_API_KEY"):
|
|
Mem0Manager.from_config({})
|
|
|
|
def test_supports_search_enables_tool_mode(self) -> None:
|
|
mgr, _fake = _manager(mode="tool")
|
|
assert mgr.supports_search is True
|
|
|
|
def test_close_releases_http_client(self) -> None:
|
|
mgr, fake = _manager()
|
|
mgr.close()
|
|
assert fake.closed is True
|
|
|
|
|
|
class TestMem0ManagerAdd:
|
|
def test_add_maps_filtered_messages_and_identity(self) -> None:
|
|
mgr, fake = _manager()
|
|
tool_ai = AIMessage(content="", tool_calls=[{"name": "t", "args": {}, "id": "1"}])
|
|
mgr.add(
|
|
"thread-1",
|
|
[HumanMessage(content="I prefer dark mode"), tool_ai, AIMessage(content="Noted.")],
|
|
agent_name="lead_agent",
|
|
user_id="u1",
|
|
)
|
|
assert len(fake.added) == 1
|
|
call = fake.added[0]
|
|
assert call["user_id"] == "u1"
|
|
assert call["agent_id"] == "lead_agent"
|
|
assert call["run_id"] == "thread-1"
|
|
assert call["messages"] == [
|
|
{"role": "user", "content": "I prefer dark mode"},
|
|
{"role": "assistant", "content": "Noted."},
|
|
]
|
|
|
|
def test_add_without_optional_ids_uses_run_id_only(self) -> None:
|
|
mgr, fake = _manager()
|
|
mgr.add("thread-9", [HumanMessage(content="hello")])
|
|
call = fake.added[0]
|
|
assert call["user_id"] is None
|
|
assert call["agent_id"] is None
|
|
assert call["run_id"] == "thread-9"
|
|
|
|
def test_add_empty_after_filter_is_noop(self) -> None:
|
|
mgr, fake = _manager()
|
|
hidden = HumanMessage(content="internal", additional_kwargs={"hide_from_ui": True})
|
|
mgr.add("thread-1", [hidden], user_id="u1")
|
|
assert fake.added == []
|
|
|
|
def test_add_write_error_log_and_drop(self, caplog: pytest.LogCaptureFixture) -> None:
|
|
mgr, fake = _manager()
|
|
fake.error = Mem0APIError("server down")
|
|
mgr.add("thread-1", [HumanMessage(content="hi")], user_id="u1") # must not raise
|
|
assert any("mem0" in r.message for r in caplog.records)
|
|
|
|
def test_add_write_error_raise_policy(self) -> None:
|
|
from deerflow.agents.memory.manager import MemoryManagerError
|
|
|
|
mgr, fake = _manager({"failure_policy": {"write": "raise"}})
|
|
fake.error = Mem0APIError("server down")
|
|
with pytest.raises(MemoryManagerError):
|
|
mgr.add("thread-1", [HumanMessage(content="hi")], user_id="u1")
|
|
|
|
def test_async_add_offloads_sync_http_client(self) -> None:
|
|
mgr, fake = _manager(mode="tool")
|
|
event_loop_thread = threading.get_ident()
|
|
called_from: list[int] = []
|
|
original_add = fake.add_memories
|
|
|
|
def recording_add(**kwargs: Any) -> dict[str, Any]:
|
|
called_from.append(threading.get_ident())
|
|
return original_add(**kwargs)
|
|
|
|
fake.add_memories = recording_add
|
|
asyncio.run(mgr.aadd("thread-1", [HumanMessage(content="hi")], user_id="u1"))
|
|
|
|
assert called_from and called_from[0] != event_loop_thread
|
|
|
|
|
|
class TestMem0ManagerGetContext:
|
|
def test_formats_dedupes_and_scopes(self) -> None:
|
|
mgr, fake = _manager()
|
|
fake.list_results = [
|
|
{"id": "m1", "memory": "likes cricket"},
|
|
{"id": "m1", "memory": "likes cricket"}, # dup by id
|
|
{"id": "m2", "memory": ""}, # empty dropped
|
|
{"id": "m3", "memory": "lives in Austin"},
|
|
]
|
|
ctx = mgr.get_context("u1", agent_name="lead_agent", thread_id="t-1")
|
|
assert ctx == "- likes cricket\n- lives in Austin"
|
|
call = fake.list_calls[0]
|
|
assert call["filters"] == {"AND": [{"user_id": "u1"}, {"agent_id": "lead_agent"}, {"run_id": "t-1"}]}
|
|
assert call["max_items"] == 8 # default top_k
|
|
|
|
def test_no_identity_returns_empty(self) -> None:
|
|
mgr, fake = _manager()
|
|
assert mgr.get_context(None) == ""
|
|
assert fake.list_calls == []
|
|
|
|
def test_read_error_fail_open_returns_empty(self) -> None:
|
|
mgr, fake = _manager()
|
|
fake.error = Mem0APIError("down")
|
|
assert mgr.get_context("u1") == ""
|
|
|
|
def test_read_error_fail_closed_raises(self) -> None:
|
|
from deerflow.agents.memory.manager import MemoryManagerError
|
|
|
|
mgr, fake = _manager({"failure_policy": {"read": "fail_closed"}})
|
|
fake.error = Mem0APIError("down")
|
|
with pytest.raises(MemoryManagerError):
|
|
mgr.get_context("u1")
|
|
|
|
def test_truncates_to_max_injection_chars(self) -> None:
|
|
mgr, fake = _manager({"max_injection_chars": 20})
|
|
fake.list_results = [{"id": f"m{i}", "memory": "x" * 30} for i in range(3)]
|
|
ctx = mgr.get_context("u1")
|
|
assert len(ctx) <= 20
|
|
|
|
def test_async_get_context_offloads_sync_http_client(self) -> None:
|
|
mgr, fake = _manager()
|
|
event_loop_thread = threading.get_ident()
|
|
called_from: list[int] = []
|
|
original_list = fake.list_memories
|
|
|
|
def recording_list(**kwargs: Any) -> list[dict[str, Any]]:
|
|
called_from.append(threading.get_ident())
|
|
return original_list(**kwargs)
|
|
|
|
fake.list_memories = recording_list
|
|
asyncio.run(mgr.aget_context("u1"))
|
|
|
|
assert called_from and called_from[0] != event_loop_thread
|
|
|
|
|
|
class TestMem0ManagerSearch:
|
|
def test_maps_results_to_backend_neutral_shape(self) -> None:
|
|
mgr, fake = _manager()
|
|
fake.search_results = [
|
|
{
|
|
"id": "m1",
|
|
"memory": "likes cricket",
|
|
"score": 0.9,
|
|
"categories": ["hobbies"],
|
|
"created_at": "2026-01-15T10:30:00Z",
|
|
"metadata": {"source": "chat"},
|
|
}
|
|
]
|
|
results = mgr.search("sports", top_k=5, user_id="u1")
|
|
assert results == [
|
|
{
|
|
"id": "m1",
|
|
"content": "likes cricket",
|
|
"category": "hobbies",
|
|
"confidence": 0.9,
|
|
"createdAt": "2026-01-15T10:30:00Z",
|
|
"source": "chat",
|
|
}
|
|
]
|
|
call = fake.search_calls[0]
|
|
assert call["filters"] == {"user_id": "u1"}
|
|
assert call["threshold"] == 0.1 # default score_threshold
|
|
|
|
def test_category_filter_anded_in(self) -> None:
|
|
mgr, fake = _manager()
|
|
mgr.search("q", user_id="u1", agent_name="lead_agent", category="preference")
|
|
assert fake.search_calls[0]["filters"] == {"AND": [{"user_id": "u1"}, {"agent_id": "lead_agent"}, {"categories": {"contains": "preference"}}]}
|
|
|
|
def test_no_identity_returns_empty(self) -> None:
|
|
mgr, _fake = _manager()
|
|
assert mgr.search("q") == []
|
|
|
|
def test_async_search_offloads_sync_http_client(self) -> None:
|
|
mgr, fake = _manager(mode="tool")
|
|
event_loop_thread = threading.get_ident()
|
|
called_from: list[int] = []
|
|
original_search = fake.search_memories
|
|
|
|
def recording_search(**kwargs: Any) -> list[dict[str, Any]]:
|
|
called_from.append(threading.get_ident())
|
|
return original_search(**kwargs)
|
|
|
|
fake.search_memories = recording_search
|
|
asyncio.run(mgr.asearch("q", user_id="u1"))
|
|
|
|
assert called_from and called_from[0] != event_loop_thread
|
|
|
|
|
|
class TestMem0ManagerManage:
|
|
def test_get_memory_maps_full_bucket(self) -> None:
|
|
mgr, fake = _manager()
|
|
fake.list_results = [{"id": "m1", "memory": "likes cricket", "created_at": "2026-01-15T10:30:00Z"}]
|
|
doc = mgr.get_memory(user_id="u1")
|
|
assert doc["facts"][0]["id"] == "m1"
|
|
assert doc["facts"][0]["content"] == "likes cricket"
|
|
assert fake.list_calls[0].get("max_items") is None # full listing
|
|
|
|
def test_get_memory_no_identity_returns_empty_doc(self) -> None:
|
|
mgr, _fake = _manager()
|
|
assert mgr.get_memory() == {"facts": []}
|
|
|
|
def test_clear_memory_deletes_bucket_and_returns_empty(self) -> None:
|
|
mgr, fake = _manager()
|
|
assert mgr.clear_memory(user_id="u1", agent_name="lead_agent") == {"facts": []}
|
|
assert fake.deleted == [{"user_id": "u1", "agent_id": "lead_agent", "run_id": None}]
|
|
|
|
def test_clear_memory_user_wide_when_agent_none(self) -> None:
|
|
mgr, fake = _manager()
|
|
mgr.clear_memory(user_id="u1")
|
|
assert fake.deleted[0]["agent_id"] is None
|
|
|
|
def test_delete_memory_returns_none(self) -> None:
|
|
mgr, fake = _manager()
|
|
assert mgr.delete_memory(user_id="u1") is None
|
|
assert len(fake.deleted) == 1
|
|
|
|
def test_clear_memory_no_identity_is_noop(self) -> None:
|
|
mgr, fake = _manager()
|
|
assert mgr.clear_memory() == {"facts": []}
|
|
assert fake.deleted == []
|
|
|
|
def test_delete_memory_no_identity_is_noop(self) -> None:
|
|
mgr, fake = _manager()
|
|
assert mgr.delete_memory() is None
|
|
assert fake.deleted == []
|
|
|
|
def test_clear_memory_empty_string_identity_is_noop(self) -> None:
|
|
mgr, fake = _manager()
|
|
assert mgr.clear_memory(user_id="", agent_name="") == {"facts": []}
|
|
assert fake.deleted == []
|
|
|
|
def test_export_delegates_to_get_memory(self) -> None:
|
|
mgr, fake = _manager()
|
|
fake.list_results = [{"id": "m1", "memory": "x"}]
|
|
assert mgr.export_memory(user_id="u1")["facts"][0]["id"] == "m1"
|
|
|
|
def test_tier3_defaults_raise_not_implemented(self) -> None:
|
|
mgr, _fake = _manager()
|
|
with pytest.raises(NotImplementedError):
|
|
mgr.create_fact("x", user_id="u1")
|
|
with pytest.raises(NotImplementedError):
|
|
mgr.import_memory({"facts": []}, user_id="u1")
|
|
|
|
|
|
class TestMem0Discovery:
|
|
def test_scan_backends_registers_mem0(self) -> None:
|
|
import deerflow.agents.memory.manager as manager_module
|
|
|
|
manager_module._backends_cache = None
|
|
try:
|
|
registry = manager_module._scan_backends()
|
|
finally:
|
|
manager_module._backends_cache = None
|
|
assert registry["mem0"] is Mem0Manager
|
|
|
|
def test_factory_resolves_mem0(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from deerflow.agents.memory.manager import get_memory_manager, reset_memory_manager
|
|
from deerflow.config.memory_config import MemoryConfig, get_memory_config, set_memory_config
|
|
|
|
monkeypatch.setenv("MEM0_API_KEY", "k")
|
|
monkeypatch.setattr(Mem0Client, "ping", lambda self: None)
|
|
|
|
original_config = get_memory_config()
|
|
set_memory_config(MemoryConfig(manager_class="mem0"))
|
|
reset_memory_manager()
|
|
try:
|
|
mgr = get_memory_manager()
|
|
assert isinstance(mgr, Mem0Manager)
|
|
finally:
|
|
reset_memory_manager()
|
|
set_memory_config(original_config)
|