85 lines
2.5 KiB
Python
85 lines
2.5 KiB
Python
from unittest.mock import patch
|
|
|
|
from deerflow.agents.memory.thread_updater import ThreadMemoryUpdater
|
|
|
|
|
|
def test_scrub_sensitive_tolerates_non_numeric_confidence():
|
|
updater = ThreadMemoryUpdater()
|
|
cleaned = updater._scrub_sensitive(
|
|
{
|
|
"user": {},
|
|
"history": {},
|
|
"facts": [
|
|
{"content": "Uses React", "category": "knowledge", "confidence": "high"},
|
|
{"content": "Uses TypeScript", "category": "knowledge", "confidence": None},
|
|
],
|
|
},
|
|
"thread-test",
|
|
)
|
|
|
|
assert len(cleaned["facts"]) == 2
|
|
assert cleaned["facts"][0]["confidence"] == 0.5
|
|
assert cleaned["facts"][1]["confidence"] == 0.5
|
|
|
|
|
|
def test_update_memory_repairs_model_json_with_unescaped_inner_quotes():
|
|
class _Storage:
|
|
def __init__(self):
|
|
self.saved = None
|
|
|
|
def load(self, _thread_id):
|
|
return None
|
|
|
|
def save(self, _thread_id, data, expected_version=None):
|
|
self.saved = {
|
|
"thread_id": _thread_id,
|
|
"data": data,
|
|
"expected_version": expected_version,
|
|
}
|
|
return True
|
|
|
|
fake_storage = _Storage()
|
|
fake_model = type(
|
|
"M",
|
|
(),
|
|
{
|
|
"invoke": lambda self, prompt: type(
|
|
"R",
|
|
(),
|
|
{
|
|
"content": """
|
|
{
|
|
"user": {
|
|
"topOfMind": {
|
|
"summary": "反感“作为 AI"这种句式，认为回答不用寒暄直接说重点。",
|
|
"updatedAt": "2026-06-11T07:13:11Z"
|
|
}
|
|
},
|
|
"history": {},
|
|
"facts": [
|
|
{
|
|
"content": "偏好直接回答，不喜欢“作为 AI"式开场",
|
|
"category": "preference",
|
|
"confidence": 0.92
|
|
}
|
|
]
|
|
}
|
|
""".strip()
|
|
},
|
|
)()
|
|
},
|
|
)()
|
|
messages = [type("Msg", (), {"type": "human", "content": "请直接回答重点，不要寒暄。"})()]
|
|
|
|
with (
|
|
patch("deerflow.agents.memory.thread_updater.get_thread_memory_storage", return_value=fake_storage),
|
|
patch.object(ThreadMemoryUpdater, "_get_model", return_value=fake_model),
|
|
):
|
|
result = ThreadMemoryUpdater().update_memory(messages, "thread-test")
|
|
|
|
assert result is True
|
|
assert fake_storage.saved is not None
|
|
assert fake_storage.saved["expected_version"] == 0
|
|
assert fake_storage.saved["data"]["user"]["topOfMind"]["summary"].startswith("反感“作为 AI")
|
|
assert fake_storage.saved["data"]["facts"][0]["content"].startswith("偏好直接回答")
|