Saket Aryan
1027 lines
34 KiB
Python
1027 lines
34 KiB
Python
"""Tests for CLI commands using mock backend."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from io import StringIO
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from click.exceptions import Exit as ClickExit
|
|
from rich.console import Console
|
|
|
|
from mem0_cli.commands.config_cmd import (
|
|
cmd_config_get,
|
|
cmd_config_set,
|
|
cmd_config_show,
|
|
)
|
|
from mem0_cli.commands.entities import cmd_entities_delete, cmd_entities_list
|
|
from mem0_cli.commands.memory import (
|
|
cmd_add,
|
|
cmd_delete,
|
|
cmd_delete_all,
|
|
cmd_get,
|
|
cmd_list,
|
|
cmd_search,
|
|
cmd_update,
|
|
)
|
|
from mem0_cli.commands.utils import (
|
|
cmd_import,
|
|
cmd_status,
|
|
cmd_version,
|
|
)
|
|
|
|
|
|
def _make_console():
|
|
buf = StringIO()
|
|
return Console(file=buf, force_terminal=False, no_color=True, width=120), buf
|
|
|
|
|
|
def _make_err_console():
|
|
buf = StringIO()
|
|
return Console(file=buf, force_terminal=False, no_color=True, width=120), buf
|
|
|
|
|
|
class TestAddCommand:
|
|
def test_add_text(self, mock_backend):
|
|
console, _buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_add(
|
|
mock_backend,
|
|
"I prefer dark mode",
|
|
user_id="alice",
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
messages=None,
|
|
file=None,
|
|
metadata=None,
|
|
immutable=False,
|
|
no_infer=False,
|
|
expires=None,
|
|
categories=None,
|
|
output="text",
|
|
)
|
|
mock_backend.add.assert_called_once()
|
|
|
|
def test_add_with_messages(self, mock_backend):
|
|
console, _buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
msgs = json.dumps([{"role": "user", "content": "I love Python"}])
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_add(
|
|
mock_backend,
|
|
None,
|
|
user_id="alice",
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
messages=msgs,
|
|
file=None,
|
|
metadata=None,
|
|
immutable=False,
|
|
no_infer=False,
|
|
expires=None,
|
|
categories=None,
|
|
output="text",
|
|
)
|
|
mock_backend.add.assert_called_once()
|
|
|
|
def test_add_with_metadata(self, mock_backend):
|
|
console, _buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_add(
|
|
mock_backend,
|
|
"test memory",
|
|
user_id="alice",
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
messages=None,
|
|
file=None,
|
|
metadata='{"source": "test"}',
|
|
immutable=False,
|
|
no_infer=False,
|
|
expires=None,
|
|
categories=None,
|
|
output="text",
|
|
)
|
|
call_kwargs = mock_backend.add.call_args
|
|
assert "metadata" in str(call_kwargs)
|
|
|
|
def test_add_json_output(self, mock_backend):
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_add(
|
|
mock_backend,
|
|
"test",
|
|
user_id="alice",
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
messages=None,
|
|
file=None,
|
|
metadata=None,
|
|
immutable=False,
|
|
no_infer=False,
|
|
expires=None,
|
|
categories=None,
|
|
output="json",
|
|
)
|
|
output = buf.getvalue()
|
|
assert '"results"' in output or '"memory"' in output
|
|
|
|
def test_add_quiet_output(self, mock_backend):
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_add(
|
|
mock_backend,
|
|
"test",
|
|
user_id="alice",
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
messages=None,
|
|
file=None,
|
|
metadata=None,
|
|
immutable=False,
|
|
no_infer=False,
|
|
expires=None,
|
|
categories=None,
|
|
output="quiet",
|
|
)
|
|
output = buf.getvalue()
|
|
# In quiet mode, no memory content should be printed (spinner may appear)
|
|
assert "dark mode" not in output
|
|
|
|
def test_add_no_content_exits(self, mock_backend):
|
|
console, _buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
# Patch stdin.isatty to return True so it doesn't try to read stdin
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
patch("mem0_cli.commands.memory.sys") as mock_sys,
|
|
):
|
|
mock_sys.stdin.isatty.return_value = True
|
|
with pytest.raises((SystemExit, ClickExit)):
|
|
cmd_add(
|
|
mock_backend,
|
|
None,
|
|
user_id="alice",
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
messages=None,
|
|
file=None,
|
|
metadata=None,
|
|
immutable=False,
|
|
no_infer=False,
|
|
expires=None,
|
|
categories=None,
|
|
output="text",
|
|
)
|
|
|
|
def test_add_invalid_metadata_json(self, mock_backend):
|
|
console, _buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
pytest.raises((SystemExit, ClickExit)),
|
|
):
|
|
cmd_add(
|
|
mock_backend,
|
|
"test",
|
|
user_id="alice",
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
messages=None,
|
|
file=None,
|
|
metadata="not-json",
|
|
immutable=False,
|
|
no_infer=False,
|
|
expires=None,
|
|
categories=None,
|
|
output="text",
|
|
)
|
|
|
|
def test_add_from_file(self, mock_backend, tmp_path):
|
|
file_path = tmp_path / "messages.json"
|
|
file_path.write_text(json.dumps([{"role": "user", "content": "hello"}]))
|
|
console, _buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_add(
|
|
mock_backend,
|
|
None,
|
|
user_id="alice",
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
messages=None,
|
|
file=file_path,
|
|
metadata=None,
|
|
immutable=False,
|
|
no_infer=False,
|
|
expires=None,
|
|
categories=None,
|
|
output="text",
|
|
)
|
|
mock_backend.add.assert_called_once()
|
|
|
|
def test_add_categories_csv(self, mock_backend):
|
|
console, _buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_add(
|
|
mock_backend,
|
|
"test",
|
|
user_id="alice",
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
messages=None,
|
|
file=None,
|
|
metadata=None,
|
|
immutable=False,
|
|
no_infer=False,
|
|
expires=None,
|
|
categories="health,prefs",
|
|
output="text",
|
|
)
|
|
mock_backend.add.assert_called_once()
|
|
|
|
|
|
class TestSearchCommand:
|
|
def test_search_text(self, mock_backend):
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_search(
|
|
mock_backend,
|
|
"preferences",
|
|
user_id="alice",
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
top_k=10,
|
|
threshold=0.3,
|
|
rerank=False,
|
|
keyword=False,
|
|
filter_json=None,
|
|
fields=None,
|
|
output="text",
|
|
)
|
|
output = buf.getvalue()
|
|
assert "Found 2" in output
|
|
|
|
def test_search_json(self, mock_backend):
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_search(
|
|
mock_backend,
|
|
"preferences",
|
|
user_id="alice",
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
top_k=10,
|
|
threshold=0.3,
|
|
rerank=False,
|
|
keyword=False,
|
|
filter_json=None,
|
|
fields=None,
|
|
output="json",
|
|
)
|
|
output = buf.getvalue()
|
|
assert '"memory"' in output
|
|
|
|
def test_search_table(self, mock_backend):
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_search(
|
|
mock_backend,
|
|
"preferences",
|
|
user_id="alice",
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
top_k=10,
|
|
threshold=0.3,
|
|
rerank=False,
|
|
keyword=False,
|
|
filter_json=None,
|
|
fields=None,
|
|
output="table",
|
|
)
|
|
output = buf.getvalue()
|
|
assert "dark mode" in output
|
|
|
|
def test_search_no_results(self, mock_backend):
|
|
mock_backend.search.return_value = []
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_search(
|
|
mock_backend,
|
|
"nonexistent",
|
|
user_id="alice",
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
top_k=10,
|
|
threshold=0.3,
|
|
rerank=False,
|
|
keyword=False,
|
|
filter_json=None,
|
|
fields=None,
|
|
output="text",
|
|
)
|
|
output = buf.getvalue()
|
|
assert "No memories found" in output
|
|
|
|
def test_search_with_filter(self, mock_backend):
|
|
console, _buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_search(
|
|
mock_backend,
|
|
"test",
|
|
user_id="alice",
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
top_k=5,
|
|
threshold=0.5,
|
|
rerank=False,
|
|
keyword=False,
|
|
filter_json='{"category": "prefs"}',
|
|
fields="memory,score",
|
|
output="text",
|
|
)
|
|
mock_backend.search.assert_called_once()
|
|
|
|
|
|
class TestGetCommand:
|
|
def test_get_text(self, mock_backend):
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_get(mock_backend, "abc-123-def-456", output="text")
|
|
output = buf.getvalue()
|
|
assert "dark mode" in output
|
|
assert "abc-123-def-456" in output
|
|
|
|
def test_get_json(self, mock_backend):
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_get(mock_backend, "abc-123-def-456", output="json")
|
|
output = buf.getvalue()
|
|
assert '"memory"' in output
|
|
|
|
|
|
class TestListCommand:
|
|
def test_list_table(self, mock_backend):
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_list(
|
|
mock_backend,
|
|
user_id="alice",
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
page=1,
|
|
page_size=100,
|
|
category=None,
|
|
after=None,
|
|
before=None,
|
|
output="table",
|
|
)
|
|
output = buf.getvalue()
|
|
assert "dark mode" in output
|
|
|
|
def test_list_json(self, mock_backend):
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_list(
|
|
mock_backend,
|
|
user_id="alice",
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
page=1,
|
|
page_size=100,
|
|
category=None,
|
|
after=None,
|
|
before=None,
|
|
output="json",
|
|
)
|
|
output = buf.getvalue()
|
|
assert '"memory"' in output
|
|
|
|
def test_list_empty(self, mock_backend):
|
|
mock_backend.list_memories.return_value = []
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_list(
|
|
mock_backend,
|
|
user_id="alice",
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
page=1,
|
|
page_size=100,
|
|
category=None,
|
|
after=None,
|
|
before=None,
|
|
output="text",
|
|
)
|
|
output = buf.getvalue()
|
|
assert "No memories found" in output
|
|
|
|
|
|
class TestUpdateCommand:
|
|
def test_update(self, mock_backend):
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_update(mock_backend, "abc-123", "New text", metadata=None, output="text")
|
|
output = buf.getvalue()
|
|
assert "updated" in output.lower()
|
|
|
|
def test_update_json(self, mock_backend):
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_update(mock_backend, "abc-123", "New text", metadata=None, output="json")
|
|
output = buf.getvalue()
|
|
assert '"memory"' in output
|
|
|
|
|
|
class TestDeleteCommand:
|
|
def test_delete_single(self, mock_backend):
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_delete(mock_backend, "abc-123", output="text")
|
|
output = buf.getvalue()
|
|
assert "deleted" in output.lower()
|
|
|
|
def test_delete_dry_run(self, mock_backend):
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_delete(mock_backend, "abc-123-def-456", dry_run=True, output="text")
|
|
output = buf.getvalue()
|
|
assert "dry run" in output.lower()
|
|
mock_backend.delete.assert_not_called()
|
|
|
|
|
|
class TestDeleteAllCommand:
|
|
def test_delete_all_force(self, mock_backend):
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_delete_all(
|
|
mock_backend,
|
|
force=True,
|
|
user_id="alice",
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
output="text",
|
|
)
|
|
output = buf.getvalue()
|
|
assert "deleted" in output.lower()
|
|
|
|
def test_delete_all_dry_run(self, mock_backend):
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_delete_all(
|
|
mock_backend,
|
|
force=True,
|
|
dry_run=True,
|
|
user_id="alice",
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
output="text",
|
|
)
|
|
output = buf.getvalue()
|
|
assert "dry run" in output.lower()
|
|
mock_backend.delete.assert_not_called()
|
|
|
|
def test_delete_all_project_wide(self, mock_backend):
|
|
console, _buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_delete_all(
|
|
mock_backend,
|
|
force=True,
|
|
all_=True,
|
|
user_id=None,
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
output="text",
|
|
)
|
|
mock_backend.delete.assert_called_once_with(
|
|
all=True,
|
|
user_id="*",
|
|
agent_id="*",
|
|
app_id="*",
|
|
run_id="*",
|
|
)
|
|
|
|
def test_delete_all_project_wide_dry_run(self, mock_backend):
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_delete_all(
|
|
mock_backend,
|
|
force=True,
|
|
all_=True,
|
|
dry_run=True,
|
|
user_id=None,
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
output="text",
|
|
)
|
|
output = buf.getvalue()
|
|
assert "project-wide" in output.lower()
|
|
mock_backend.delete.assert_not_called()
|
|
|
|
def test_delete_all_project_wide_async_response(self, mock_backend):
|
|
mock_backend.delete.return_value = {"message": "Memories deletion started..."}
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_delete_all(
|
|
mock_backend,
|
|
force=True,
|
|
all_=True,
|
|
user_id=None,
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
output="text",
|
|
)
|
|
output = buf.getvalue()
|
|
assert "background" in output.lower()
|
|
|
|
|
|
class TestStatusCommand:
|
|
def test_status_connected(self, mock_backend):
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.utils.console", console),
|
|
patch("mem0_cli.commands.utils.err_console", err_console),
|
|
):
|
|
cmd_status(mock_backend)
|
|
output = buf.getvalue()
|
|
assert "Connected" in output
|
|
|
|
def test_status_disconnected(self, mock_backend):
|
|
mock_backend.status.return_value = {
|
|
"connected": False,
|
|
"backend": "platform",
|
|
"error": "Connection refused",
|
|
}
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.utils.console", console),
|
|
patch("mem0_cli.commands.utils.err_console", err_console),
|
|
):
|
|
cmd_status(mock_backend)
|
|
output = buf.getvalue()
|
|
assert "Disconnected" in output
|
|
assert "Connection refused" in output
|
|
|
|
def test_status_json(self, mock_backend):
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.utils.console", console),
|
|
patch("mem0_cli.commands.utils.err_console", err_console),
|
|
):
|
|
cmd_status(mock_backend, output="json")
|
|
output = buf.getvalue()
|
|
assert '"connected"' in output
|
|
assert '"status"' in output
|
|
|
|
|
|
class TestVersionCommand:
|
|
def test_version(self):
|
|
console, buf = _make_console()
|
|
with patch("mem0_cli.commands.utils.console", console):
|
|
cmd_version()
|
|
output = buf.getvalue()
|
|
assert "0.1.0" in output
|
|
|
|
|
|
class TestImportCommand:
|
|
def test_import_json(self, mock_backend, tmp_path):
|
|
file_path = tmp_path / "import.json"
|
|
data = [
|
|
{"memory": "Test memory 1"},
|
|
{"memory": "Test memory 2"},
|
|
]
|
|
file_path.write_text(json.dumps(data))
|
|
console, _buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.utils.console", console),
|
|
patch("mem0_cli.commands.utils.err_console", err_console),
|
|
):
|
|
cmd_import(mock_backend, str(file_path), user_id="alice", agent_id=None)
|
|
assert mock_backend.add.call_count == 2
|
|
|
|
def test_import_invalid_file(self, mock_backend):
|
|
console, _buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.utils.console", console),
|
|
patch("mem0_cli.commands.utils.err_console", err_console),
|
|
pytest.raises((SystemExit, ClickExit)),
|
|
):
|
|
cmd_import(mock_backend, "/nonexistent/file.json", user_id=None, agent_id=None)
|
|
|
|
def test_import_json_output(self, mock_backend, tmp_path):
|
|
file_path = tmp_path / "import.json"
|
|
data = [{"memory": "Test memory 1"}]
|
|
file_path.write_text(json.dumps(data))
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.utils.console", console),
|
|
patch("mem0_cli.commands.utils.err_console", err_console),
|
|
):
|
|
cmd_import(mock_backend, str(file_path), user_id="alice", agent_id=None, output="json")
|
|
output = buf.getvalue()
|
|
assert '"added"' in output
|
|
|
|
|
|
class TestEntitiesListCommand:
|
|
def test_list_users(self, mock_backend):
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.entities.console", console),
|
|
patch("mem0_cli.commands.entities.err_console", err_console),
|
|
):
|
|
cmd_entities_list(mock_backend, "users", output="table")
|
|
output = buf.getvalue()
|
|
assert "alice" in output
|
|
|
|
def test_list_invalid_type(self, mock_backend):
|
|
console, _buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.entities.console", console),
|
|
patch("mem0_cli.commands.entities.err_console", err_console),
|
|
pytest.raises((SystemExit, ClickExit)),
|
|
):
|
|
cmd_entities_list(mock_backend, "invalid", output="table")
|
|
|
|
def test_list_json(self, mock_backend):
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.entities.console", console),
|
|
patch("mem0_cli.commands.entities.err_console", err_console),
|
|
):
|
|
cmd_entities_list(mock_backend, "users", output="json")
|
|
output = buf.getvalue()
|
|
assert '"alice"' in output
|
|
|
|
|
|
class TestConfigCommands:
|
|
def test_config_show(self, isolate_config):
|
|
from mem0_cli.config import Mem0Config, save_config
|
|
|
|
config = Mem0Config()
|
|
config.platform.api_key = "m0-test12345678"
|
|
save_config(config)
|
|
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.config_cmd.console", console),
|
|
patch("mem0_cli.commands.config_cmd.err_console", err_console),
|
|
):
|
|
cmd_config_show()
|
|
output = buf.getvalue()
|
|
assert "Configuration" in output
|
|
assert "m0-test12345678" not in output
|
|
|
|
def test_config_show_json(self, isolate_config):
|
|
from mem0_cli.config import Mem0Config, save_config
|
|
|
|
config = Mem0Config()
|
|
config.platform.api_key = "m0-test12345678"
|
|
config.defaults.user_id = "alice"
|
|
save_config(config)
|
|
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.config_cmd.console", console),
|
|
patch("mem0_cli.commands.config_cmd.err_console", err_console),
|
|
):
|
|
cmd_config_show(output="json")
|
|
output = buf.getvalue()
|
|
assert '"status"' in output
|
|
assert '"config show"' in output
|
|
|
|
def test_config_set_and_get(self, isolate_config):
|
|
console1, _buf1 = _make_console()
|
|
err_console1, _err_buf1 = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.config_cmd.console", console1),
|
|
patch("mem0_cli.commands.config_cmd.err_console", err_console1),
|
|
):
|
|
cmd_config_set("platform.base_url", "https://custom.api.mem0.ai")
|
|
|
|
console2, buf2 = _make_console()
|
|
err_console2, _err_buf2 = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.config_cmd.console", console2),
|
|
patch("mem0_cli.commands.config_cmd.err_console", err_console2),
|
|
):
|
|
cmd_config_get("platform.base_url")
|
|
output = buf2.getvalue()
|
|
assert "custom.api.mem0.ai" in output
|
|
|
|
def test_config_show_displays_defaults(self, isolate_config):
|
|
from mem0_cli.config import Mem0Config, save_config
|
|
|
|
config = Mem0Config()
|
|
config.defaults.user_id = "alice"
|
|
save_config(config)
|
|
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.config_cmd.console", console),
|
|
patch("mem0_cli.commands.config_cmd.err_console", err_console),
|
|
):
|
|
cmd_config_show()
|
|
output = buf.getvalue()
|
|
assert "defaults.user_id" in output
|
|
assert "alice" in output
|
|
|
|
|
|
class TestEntitiesDeleteCommand:
|
|
def test_delete_entity_with_force(self, mock_backend):
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.entities.console", console),
|
|
patch("mem0_cli.commands.entities.err_console", err_console),
|
|
):
|
|
cmd_entities_delete(
|
|
mock_backend,
|
|
user_id="alice",
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
force=True,
|
|
output="text",
|
|
)
|
|
mock_backend.delete_entities.assert_called_once_with(
|
|
user_id="alice", agent_id=None, app_id=None, run_id=None
|
|
)
|
|
output = buf.getvalue()
|
|
assert "deleted" in output.lower()
|
|
|
|
def test_delete_entity_no_id_exits(self, mock_backend):
|
|
console, _buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.entities.console", console),
|
|
patch("mem0_cli.commands.entities.err_console", err_console),
|
|
pytest.raises((SystemExit, ClickExit)),
|
|
):
|
|
cmd_entities_delete(
|
|
mock_backend,
|
|
user_id=None,
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
force=True,
|
|
output="text",
|
|
)
|
|
|
|
def test_delete_entity_json_output(self, mock_backend):
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.entities.console", console),
|
|
patch("mem0_cli.commands.entities.err_console", err_console),
|
|
):
|
|
cmd_entities_delete(
|
|
mock_backend,
|
|
user_id="alice",
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
force=True,
|
|
output="json",
|
|
)
|
|
output = buf.getvalue()
|
|
assert '"message"' in output
|
|
|
|
def test_delete_entity_dry_run(self, mock_backend):
|
|
console, buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.entities.console", console),
|
|
patch("mem0_cli.commands.entities.err_console", err_console),
|
|
):
|
|
cmd_entities_delete(
|
|
mock_backend,
|
|
user_id="alice",
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
force=True,
|
|
dry_run=True,
|
|
output="text",
|
|
)
|
|
output = buf.getvalue()
|
|
assert "dry run" in output.lower()
|
|
mock_backend.delete_entities.assert_not_called()
|
|
|
|
|
|
class TestEnableGraph:
|
|
def test_add_with_graph(self, mock_backend):
|
|
console, _buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_add(
|
|
mock_backend,
|
|
"test",
|
|
user_id="alice",
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
messages=None,
|
|
file=None,
|
|
metadata=None,
|
|
immutable=False,
|
|
no_infer=False,
|
|
expires=None,
|
|
categories=None,
|
|
enable_graph=True,
|
|
output="text",
|
|
)
|
|
call_kwargs = mock_backend.add.call_args
|
|
assert call_kwargs.kwargs.get("enable_graph") is True
|
|
|
|
def test_search_with_graph(self, mock_backend):
|
|
console, _buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_search(
|
|
mock_backend,
|
|
"test",
|
|
user_id="alice",
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
top_k=10,
|
|
threshold=0.3,
|
|
rerank=False,
|
|
keyword=False,
|
|
filter_json=None,
|
|
fields=None,
|
|
enable_graph=True,
|
|
output="text",
|
|
)
|
|
call_kwargs = mock_backend.search.call_args
|
|
assert call_kwargs.kwargs.get("enable_graph") is True
|
|
|
|
def test_list_with_graph(self, mock_backend):
|
|
console, _buf = _make_console()
|
|
err_console, _err_buf = _make_err_console()
|
|
with (
|
|
patch("mem0_cli.commands.memory.console", console),
|
|
patch("mem0_cli.commands.memory.err_console", err_console),
|
|
):
|
|
cmd_list(
|
|
mock_backend,
|
|
user_id="alice",
|
|
agent_id=None,
|
|
app_id=None,
|
|
run_id=None,
|
|
page=1,
|
|
page_size=100,
|
|
category=None,
|
|
after=None,
|
|
before=None,
|
|
enable_graph=True,
|
|
output="table",
|
|
)
|
|
call_kwargs = mock_backend.list_memories.call_args
|
|
assert call_kwargs.kwargs.get("enable_graph") is True
|