mirror of
https://github.com/supermemoryai/supermemory.git
synced 2026-08-28 05:25:33 +00:00
The pipecat, cartesia, and agent-framework packages passed search results into dedup/format helpers written for plain dicts, but typed SDK results are pydantic models (snake_case attributes, no dict interface). pipecat and cartesia crashed with AttributeError — swallowed upstream, so no memories were ever injected for users with search results — while agent-framework silently dropped every search-result memory. Extract memory fields through a tolerant accessor that accepts both the camelCase dicts returned by the profile endpoint and SDK result models, so the helpers work regardless of installed SDK version. The pipecat/cartesia dependency stubs move from test_empty_profile.py into a shared tests/conftest.py so the new utils tests can reuse them. Fixes #1266
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_items(self) -> None:
|
|
# SDK search results are pydantic models with a `memory` attribute,
|
|
# not dicts — they must not be silently dropped (#1266).
|
|
result = deduplicate_memories(
|
|
search_results=[SimpleNamespace(memory="User prefers async")],
|
|
)
|
|
assert result.search_results == ["User prefers async"]
|
|
|
|
def test_model_items_deduplicate_against_static(self) -> None:
|
|
result = deduplicate_memories(
|
|
static=[{"memory": "User likes Python"}],
|
|
search_results=[
|
|
SimpleNamespace(memory="User likes Python"),
|
|
SimpleNamespace(memory="User prefers async"),
|
|
],
|
|
)
|
|
assert result.search_results == ["User prefers async"]
|
|
|
|
def test_model_items_without_memory_text_filtered(self) -> None:
|
|
result = deduplicate_memories(
|
|
search_results=[SimpleNamespace(memory=None), SimpleNamespace(memory=" ")],
|
|
)
|
|
assert result.search_results == []
|
|
|
|
|
|
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)
|