107 lines
3.1 KiB
Python
107 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
from memory_gateway_plugin.tools import memory_append_episode, memory_feedback, memory_search, memory_upsert
|
|
|
|
|
|
class FakeClient:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
def search_memory(self, payload):
|
|
self.calls.append(("search", payload))
|
|
return {"ok": True, "data": {"results": []}}
|
|
|
|
def append_episode(self, payload):
|
|
self.calls.append(("append", payload))
|
|
return {"ok": True, "data": payload}
|
|
|
|
def upsert_memory(self, payload):
|
|
self.calls.append(("upsert", payload))
|
|
return {"ok": True, "data": payload}
|
|
|
|
def send_feedback(self, memory_id, payload):
|
|
self.calls.append(("feedback", memory_id, payload))
|
|
return {"ok": True, "data": payload}
|
|
|
|
|
|
def test_memory_search_empty_query_rejected():
|
|
client = FakeClient()
|
|
result = memory_search(query="", user_id="u", agent_id="a", client=client)
|
|
|
|
assert result["ok"] is False
|
|
assert client.calls == []
|
|
|
|
|
|
def test_append_episode_rejects_api_key():
|
|
result = memory_append_episode(
|
|
user_id="u",
|
|
agent_id="a",
|
|
session_id="s",
|
|
episode_summary="api_key=sk-secret",
|
|
client=FakeClient(),
|
|
)
|
|
|
|
assert result["ok"] is False
|
|
assert result["reason"] == "secret_like_content"
|
|
|
|
|
|
def test_append_episode_rejects_password():
|
|
result = memory_append_episode(
|
|
user_id="u",
|
|
agent_id="a",
|
|
session_id="s",
|
|
episode_summary="password=hunter2",
|
|
client=FakeClient(),
|
|
)
|
|
|
|
assert result["ok"] is False
|
|
assert result["reason"] == "secret_like_content"
|
|
|
|
|
|
def test_append_episode_rejects_raw_transcript():
|
|
content = "\n".join(["User: hi", "Assistant: hello", "User: remember this", "Assistant: ok"])
|
|
result = memory_append_episode(user_id="u", agent_id="a", session_id="s", episode_summary=content, client=FakeClient())
|
|
|
|
assert result["ok"] is False
|
|
assert result["reason"] == "raw_chat_transcript"
|
|
|
|
|
|
def test_append_episode_accepts_stable_preference():
|
|
client = FakeClient()
|
|
result = memory_append_episode(
|
|
user_id="u",
|
|
agent_id="a",
|
|
session_id="s",
|
|
episode_summary="用户稳定偏好：以后所有技术方案都使用中文输出。",
|
|
tags=["preference"],
|
|
client=client,
|
|
)
|
|
|
|
assert result["ok"] is True
|
|
assert client.calls[0][0] == "append"
|
|
|
|
|
|
def test_upsert_uses_long_term_namespace_when_provided():
|
|
client = FakeClient()
|
|
namespace = "user/u/long_term"
|
|
result = memory_upsert(
|
|
user_id="u",
|
|
agent_id="a",
|
|
namespace=namespace,
|
|
memory_type="preference",
|
|
content="用户稳定偏好：使用中文输出。",
|
|
client=client,
|
|
)
|
|
|
|
assert result["ok"] is True
|
|
assert client.calls[0][1]["namespace"] == namespace
|
|
|
|
|
|
def test_feedback_calls_correct_endpoint():
|
|
client = FakeClient()
|
|
result = memory_feedback(user_id="u", agent_id="a", memory_id="mem_1", feedback="reject", client=client)
|
|
|
|
assert result["ok"] is True
|
|
assert client.calls[0] == ("feedback", "mem_1", {"user_id": "u", "agent_id": "a", "feedback": "incorrect", "comment": None})
|
|
|