mirror of
https://github.com/supermemoryai/supermemory.git
synced 2026-09-15 23:31:05 +00:00
a2328e8ace
The pipecat, cartesia, and agent-framework packages all pass search results from the Supermemory SDK (pydantic models, attribute access, snake_case fields) into dedup/format helpers written for plain dicts. The consequences differed by package but all killed the feature's primary path: - pipecat: deduplicate_memories called r.get() on a Result model -> AttributeError -> the outer handler logs "Error processing frame" and forwards the frame unchanged, so memories are never injected for any user whose profile lookup returns search results - cartesia: identical crash, swallowed by _enrich_event_with_memories' generic except -> memory_context silently comes back empty and the previous memory block is stripped from the system prompt - agent-framework: extract_memory_text only handled dict/str, so model objects fell through to `return None` and every search-result memory was silently dropped from the injected context (The OpenAI package shares the same helper shape but works because it fetches profile data over raw aiohttp and receives dicts — that asymmetry is what hid this.) pipecat/cartesia gain an extract_search_result_fields helper that reads memory/updatedAt off dicts (camelCase keys) or SDK models (snake_case attributes, datetime-tolerant), used by both deduplicate_memories and format_memories_to_text; the formatter also no longer prints raw object reprs for non-dict items. agent-framework's extract_memory_text gains a model-object branch. Verified against the real SDK: built a supermemory.types.search_memories_response.Result from an API-shaped payload and ran it through dedup -> extract -> format (crashed with AttributeError before, renders "- [1 Jan] User prefers async" after). Test suites: pipecat 7 passed, cartesia 7 passed, agent-framework 56 passed (agent-framework-core pinned to 1.0.0rc3 — newer releases have dropped BaseContextProvider, which breaks the package import independently of this change).
133 lines
4.6 KiB
Python
133 lines
4.6 KiB
Python
"""Tests for utility functions."""
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from supermemory_agent_framework.utils import (
|
|
DeduplicatedMemories,
|
|
SimpleLogger,
|
|
convert_profile_to_markdown,
|
|
create_logger,
|
|
deduplicate_memories,
|
|
)
|
|
|
|
|
|
class TestDeduplicateMemories:
|
|
def test_empty_inputs(self) -> None:
|
|
result = deduplicate_memories()
|
|
assert result.static == []
|
|
assert result.dynamic == []
|
|
assert result.search_results == []
|
|
|
|
def test_static_only(self) -> None:
|
|
result = deduplicate_memories(
|
|
static=[{"memory": "User likes Python"}],
|
|
)
|
|
assert result.static == ["User likes Python"]
|
|
assert result.dynamic == []
|
|
assert result.search_results == []
|
|
|
|
def test_deduplication_priority(self) -> None:
|
|
result = deduplicate_memories(
|
|
static=[{"memory": "User likes Python"}],
|
|
dynamic=[{"memory": "User likes Python"}, {"memory": "User works remotely"}],
|
|
search_results=[{"memory": "User likes Python"}, {"memory": "User prefers async"}],
|
|
)
|
|
assert result.static == ["User likes Python"]
|
|
assert result.dynamic == ["User works remotely"]
|
|
assert result.search_results == ["User prefers async"]
|
|
|
|
def test_string_format(self) -> None:
|
|
result = deduplicate_memories(
|
|
static=["User likes Python"],
|
|
dynamic=["User works remotely"],
|
|
)
|
|
assert result.static == ["User likes Python"]
|
|
assert result.dynamic == ["User works remotely"]
|
|
|
|
def test_empty_strings_filtered(self) -> None:
|
|
result = deduplicate_memories(
|
|
static=["", " ", "User likes Python"],
|
|
)
|
|
assert result.static == ["User likes Python"]
|
|
|
|
def test_none_items_filtered(self) -> None:
|
|
result = deduplicate_memories(
|
|
static=[None, {"memory": "valid"}],
|
|
)
|
|
assert result.static == ["valid"]
|
|
|
|
def test_model_object_search_results(self) -> None:
|
|
# SDK search results are pydantic models (attribute access, no
|
|
# .get()); they used to fall through extract_memory_text and be
|
|
# silently dropped.
|
|
results = [
|
|
SimpleNamespace(memory="User prefers async"),
|
|
SimpleNamespace(memory="User prefers async"),
|
|
SimpleNamespace(memory=" "),
|
|
SimpleNamespace(memory=None),
|
|
]
|
|
result = deduplicate_memories(search_results=results)
|
|
assert result.search_results == ["User prefers async"]
|
|
|
|
def test_model_objects_deduplicate_against_profile(self) -> None:
|
|
result = deduplicate_memories(
|
|
static=["User likes Python"],
|
|
search_results=[
|
|
SimpleNamespace(memory="User likes Python"),
|
|
SimpleNamespace(memory="User prefers async"),
|
|
],
|
|
)
|
|
assert result.static == ["User likes Python"]
|
|
assert result.search_results == ["User prefers async"]
|
|
|
|
|
|
class TestConvertProfileToMarkdown:
|
|
def test_empty_profile(self) -> None:
|
|
result = convert_profile_to_markdown({"profile": {}})
|
|
assert result == ""
|
|
|
|
def test_static_only(self) -> None:
|
|
result = convert_profile_to_markdown(
|
|
{"profile": {"static": ["Likes Python", "Lives in SF"]}}
|
|
)
|
|
assert "## Static Profile" in result
|
|
assert "- Likes Python" in result
|
|
assert "- Lives in SF" in result
|
|
|
|
def test_both_sections(self) -> None:
|
|
result = convert_profile_to_markdown(
|
|
{
|
|
"profile": {
|
|
"static": ["Likes Python"],
|
|
"dynamic": ["Asked about AI"],
|
|
}
|
|
}
|
|
)
|
|
assert "## Static Profile" in result
|
|
assert "## Dynamic Profile" in result
|
|
|
|
|
|
class TestLogger:
|
|
def test_verbose_logger(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
logger = SimpleLogger(verbose=True)
|
|
logger.info("test message")
|
|
captured = capsys.readouterr()
|
|
assert "[supermemory] test message" in captured.out
|
|
|
|
def test_silent_logger(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
logger = SimpleLogger(verbose=False)
|
|
logger.info("test message")
|
|
captured = capsys.readouterr()
|
|
assert captured.out == ""
|
|
|
|
def test_error_prefix(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
logger = SimpleLogger(verbose=True)
|
|
logger.error("something failed")
|
|
captured = capsys.readouterr()
|
|
assert "ERROR:" in captured.out
|
|
|
|
def test_create_logger(self) -> None:
|
|
logger = create_logger(True)
|
|
assert isinstance(logger, SimpleLogger)
|