feat(mem0-plugin): plugin v0.2.1 — Tiers 1-8 + PostHog telemetry + review fixes (#5215)
This commit is contained in:
Kartik
@@ -0,0 +1,88 @@
|
||||
"""Tests for setup_coding_categories.py -- CODING_CATEGORIES list completeness."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import os
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
|
||||
SCRIPTS_DIR = os.path.join(os.path.dirname(__file__), "..", "scripts")
|
||||
|
||||
EXPECTED_KEYS = [
|
||||
"architecture_decisions",
|
||||
"anti_patterns",
|
||||
"task_learnings",
|
||||
"tooling_setup",
|
||||
"bug_fixes",
|
||||
"coding_conventions",
|
||||
"user_preferences",
|
||||
"dependency_decisions",
|
||||
"performance_findings",
|
||||
"security_constraints",
|
||||
"testing_patterns",
|
||||
"data_model",
|
||||
"api_contracts",
|
||||
"deployment_runbook",
|
||||
"team_norms",
|
||||
"domain_glossary",
|
||||
"experiment_results",
|
||||
]
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def coding_categories():
|
||||
"""Import CODING_CATEGORIES from setup_coding_categories, ensuring scripts/ is on path."""
|
||||
abs_scripts = os.path.abspath(SCRIPTS_DIR)
|
||||
inserted = False
|
||||
if abs_scripts not in sys.path:
|
||||
sys.path.insert(0, abs_scripts)
|
||||
inserted = True
|
||||
# Force re-import in case another test already loaded a stale version
|
||||
mod_name = "setup_coding_categories"
|
||||
if mod_name in sys.modules:
|
||||
del sys.modules[mod_name]
|
||||
mod = importlib.import_module(mod_name)
|
||||
yield mod.CODING_CATEGORIES
|
||||
if inserted and abs_scripts in sys.path:
|
||||
sys.path.remove(abs_scripts)
|
||||
|
||||
|
||||
def test_total_count(coding_categories):
|
||||
"""CODING_CATEGORIES must contain exactly 17 entries."""
|
||||
assert len(coding_categories) == 17, (
|
||||
f"Expected 17 categories, found {len(coding_categories)}: "
|
||||
f"{[list(c.keys())[0] for c in coding_categories]}"
|
||||
)
|
||||
|
||||
|
||||
def test_all_expected_keys_present(coding_categories):
|
||||
"""Every expected category key must appear exactly once."""
|
||||
actual_keys = [list(cat.keys())[0] for cat in coding_categories]
|
||||
for key in EXPECTED_KEYS:
|
||||
assert key in actual_keys, f"Missing expected category key: '{key}'"
|
||||
|
||||
|
||||
def test_no_duplicate_keys(coding_categories):
|
||||
"""No category key may appear more than once."""
|
||||
actual_keys = [list(cat.keys())[0] for cat in coding_categories]
|
||||
seen = set()
|
||||
duplicates = []
|
||||
for key in actual_keys:
|
||||
if key in seen:
|
||||
duplicates.append(key)
|
||||
seen.add(key)
|
||||
assert not duplicates, f"Duplicate category keys found: {duplicates}"
|
||||
|
||||
|
||||
def test_each_description_is_non_empty_string(coding_categories):
|
||||
"""Every category must have a non-empty string description."""
|
||||
for cat in coding_categories:
|
||||
assert len(cat) == 1, f"Category dict should have exactly one key, got: {cat}"
|
||||
key = list(cat.keys())[0]
|
||||
description = cat[key]
|
||||
assert isinstance(description, str), (
|
||||
f"Category '{key}' description is not a string: {type(description)}"
|
||||
)
|
||||
assert description.strip(), f"Category '{key}' has an empty description"
|
||||
@@ -0,0 +1,355 @@
|
||||
"""Tests for import_competing_tools.py — competing tool file importers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from unittest import mock
|
||||
|
||||
SCRIPTS_DIR = os.path.join(os.path.dirname(__file__), "..", "scripts")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# split_sections tests (unit tests on the splitter functions)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_split_by_headers_cursorrules():
|
||||
"""split_by_headers correctly splits .cursorrules content on ## headers."""
|
||||
from import_competing_tools import split_by_headers
|
||||
|
||||
content = """\
|
||||
# My Cursor Rules
|
||||
|
||||
Some preamble text that belongs to the first section.
|
||||
|
||||
## TypeScript Conventions
|
||||
|
||||
Always use strict mode. Prefer const over let.
|
||||
Never use var.
|
||||
|
||||
## React Patterns
|
||||
|
||||
Use functional components with hooks.
|
||||
Avoid class components.
|
||||
|
||||
## Testing
|
||||
|
||||
Write tests for all utility functions.
|
||||
"""
|
||||
chunks = split_by_headers(content, "## ")
|
||||
assert len(chunks) == 4 # preamble + 3 sections
|
||||
|
||||
# First chunk is the preamble (before any ## header)
|
||||
assert "preamble text" in chunks[0]
|
||||
|
||||
# Remaining chunks start with their header
|
||||
assert chunks[1].startswith("## TypeScript Conventions")
|
||||
assert "strict mode" in chunks[1]
|
||||
|
||||
assert chunks[2].startswith("## React Patterns")
|
||||
assert "functional components" in chunks[2]
|
||||
|
||||
assert chunks[3].startswith("## Testing")
|
||||
assert "utility functions" in chunks[3]
|
||||
|
||||
|
||||
def test_split_by_headers_copilot():
|
||||
"""split_by_headers correctly splits copilot-instructions.md on ## headers."""
|
||||
from import_competing_tools import split_by_headers
|
||||
|
||||
content = """\
|
||||
## Code Style
|
||||
|
||||
Use 2-space indentation. Always add trailing commas.
|
||||
|
||||
## Architecture
|
||||
|
||||
Follow clean architecture principles. Keep business logic in domain layer.
|
||||
"""
|
||||
chunks = split_by_headers(content, "## ")
|
||||
assert len(chunks) == 2
|
||||
assert chunks[0].startswith("## Code Style")
|
||||
assert "2-space indentation" in chunks[0]
|
||||
assert chunks[1].startswith("## Architecture")
|
||||
assert "clean architecture" in chunks[1]
|
||||
|
||||
|
||||
def test_split_by_headers_no_headers():
|
||||
"""split_by_headers returns entire content as one chunk if no headers found."""
|
||||
from import_competing_tools import split_by_headers
|
||||
|
||||
content = "This file has no headers at all. Just plain text."
|
||||
chunks = split_by_headers(content, "## ")
|
||||
assert len(chunks) == 1
|
||||
assert "Just plain text" in chunks[0]
|
||||
|
||||
|
||||
def test_split_cline_multiple_md_files(tmp_path):
|
||||
"""cmd_cline processes multiple .md files from memory-bank/ directory."""
|
||||
from import_competing_tools import filter_and_truncate
|
||||
|
||||
# Create a temporary memory-bank directory with .md files
|
||||
mb_dir = tmp_path / "memory-bank"
|
||||
mb_dir.mkdir()
|
||||
|
||||
(mb_dir / "architecture.md").write_text(
|
||||
"# Architecture Decisions\n\nUse microservices architecture with event sourcing."
|
||||
)
|
||||
(mb_dir / "conventions.md").write_text(
|
||||
"# Code Conventions\n\nAll functions must have type hints. Use black formatter."
|
||||
)
|
||||
(mb_dir / "empty.md").write_text("") # empty file should be skipped
|
||||
|
||||
# Read and verify we can split the files
|
||||
md_files = sorted(f for f in os.listdir(str(mb_dir)) if f.endswith(".md"))
|
||||
assert "architecture.md" in md_files
|
||||
assert "conventions.md" in md_files
|
||||
assert "empty.md" in md_files
|
||||
|
||||
non_empty = []
|
||||
for filename in md_files:
|
||||
filepath = os.path.join(str(mb_dir), filename)
|
||||
with open(filepath) as f:
|
||||
content = f.read().strip()
|
||||
if content:
|
||||
chunks = filter_and_truncate([content])
|
||||
non_empty.extend(chunks)
|
||||
|
||||
assert len(non_empty) == 2
|
||||
assert any("microservices" in c for c in non_empty)
|
||||
assert any("type hints" in c for c in non_empty)
|
||||
|
||||
|
||||
def test_split_by_hr_or_headers_continue():
|
||||
"""split_by_hr_or_headers correctly splits .continue/rules.md."""
|
||||
from import_competing_tools import split_by_hr_or_headers
|
||||
|
||||
content = """\
|
||||
## First Section
|
||||
|
||||
Content of first section.
|
||||
|
||||
---
|
||||
|
||||
## Second Section
|
||||
|
||||
Content of second section.
|
||||
|
||||
---
|
||||
|
||||
Third section without a header (just after HR).
|
||||
"""
|
||||
chunks = split_by_hr_or_headers(content)
|
||||
# Should split into meaningful chunks
|
||||
assert len(chunks) >= 2
|
||||
assert any("First Section" in c for c in chunks)
|
||||
assert any("Second Section" in c for c in chunks)
|
||||
|
||||
|
||||
def test_filter_and_truncate_skips_short():
|
||||
"""filter_and_truncate skips chunks shorter than MIN_CHUNK_CHARS (50)."""
|
||||
from import_competing_tools import filter_and_truncate
|
||||
|
||||
chunks = [
|
||||
"Short", # < 50 chars, should be filtered
|
||||
"A" * 49, # exactly 49 chars, should be filtered
|
||||
"A" * 50, # exactly 50 chars, should be kept
|
||||
"A long enough chunk that definitely passes the minimum length filter.",
|
||||
]
|
||||
result = filter_and_truncate(chunks)
|
||||
assert len(result) == 2
|
||||
assert all(len(c) >= 50 for c in result)
|
||||
|
||||
|
||||
def test_filter_and_truncate_truncates_long():
|
||||
"""filter_and_truncate truncates chunks over MAX_CHUNK_CHARS (10000)."""
|
||||
from import_competing_tools import MAX_CHUNK_CHARS, filter_and_truncate
|
||||
|
||||
long_chunk = "X" * (MAX_CHUNK_CHARS + 500)
|
||||
result = filter_and_truncate([long_chunk])
|
||||
assert len(result) == 1
|
||||
assert len(result[0]) == MAX_CHUNK_CHARS
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Mock API tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_mock_response(status: int = 201, body: dict | None = None) -> mock.MagicMock:
|
||||
"""Create a mock HTTP response object."""
|
||||
if body is None:
|
||||
body = {"id": "new-mem-id", "memory": "test"}
|
||||
resp = mock.MagicMock()
|
||||
resp.status = status
|
||||
resp.read.return_value = json.dumps(body).encode()
|
||||
resp.__enter__ = lambda s: s
|
||||
resp.__exit__ = mock.MagicMock(return_value=False)
|
||||
return resp
|
||||
|
||||
|
||||
def test_cursorrules_import_api_call(tmp_path):
|
||||
"""cmd_cursorrules calls the API with correct app_id (top-level), infer=False, and correct source."""
|
||||
from import_competing_tools import cmd_cursorrules
|
||||
|
||||
# Create a .cursorrules file with enough content
|
||||
cursorrules = tmp_path / ".cursorrules"
|
||||
cursorrules.write_text(
|
||||
"## TypeScript Rules\n\nAlways use strict TypeScript. Never use 'any' type. "
|
||||
"Prefer interfaces over type aliases for object shapes."
|
||||
)
|
||||
|
||||
captured_requests: list[dict] = []
|
||||
|
||||
def mock_urlopen(req, timeout=None):
|
||||
body = json.loads(req.data.decode())
|
||||
captured_requests.append(body)
|
||||
return _make_mock_response(201)
|
||||
|
||||
with mock.patch("import_competing_tools.resolve_api_key", return_value="m0-testkey"), \
|
||||
mock.patch("import_competing_tools.resolve_user_id", return_value="testuser"), \
|
||||
mock.patch("import_competing_tools.resolve_project_id", return_value="my-project"), \
|
||||
mock.patch("import_competing_tools.resolve_branch", return_value="main"), \
|
||||
mock.patch("urllib.request.urlopen", side_effect=mock_urlopen):
|
||||
|
||||
original_cwd = os.getcwd()
|
||||
os.chdir(str(tmp_path))
|
||||
try:
|
||||
cmd_cursorrules(["--path", str(cursorrules)])
|
||||
finally:
|
||||
os.chdir(original_cwd)
|
||||
|
||||
assert len(captured_requests) >= 1
|
||||
|
||||
req_body = captured_requests[0]
|
||||
|
||||
# app_id must be top-level (not inside metadata)
|
||||
assert req_body["app_id"] == "my-project", f"Expected app_id at top level, got: {req_body}"
|
||||
|
||||
# infer must be False (not "false", but the boolean False)
|
||||
assert req_body["infer"] is False, f"Expected infer=False, got: {req_body['infer']}"
|
||||
|
||||
# source must be cursor-import
|
||||
assert req_body["metadata"]["source"] == "cursor-import", (
|
||||
f"Expected source=cursor-import, got: {req_body['metadata'].get('source')}"
|
||||
)
|
||||
|
||||
# user_id must be set
|
||||
assert req_body["user_id"] == "testuser"
|
||||
|
||||
# messages must be a list with role/content
|
||||
assert isinstance(req_body["messages"], list)
|
||||
assert req_body["messages"][0]["role"] == "user"
|
||||
assert len(req_body["messages"][0]["content"]) > 0
|
||||
|
||||
|
||||
def test_copilot_import_api_call(tmp_path):
|
||||
"""cmd_copilot calls the API with source=copilot-import."""
|
||||
from import_competing_tools import cmd_copilot
|
||||
|
||||
copilot_dir = tmp_path / ".github"
|
||||
copilot_dir.mkdir()
|
||||
copilot_file = copilot_dir / "copilot-instructions.md"
|
||||
copilot_file.write_text(
|
||||
"## Code Style\n\nUse 2-space indentation. Always add trailing commas in multi-line structures. "
|
||||
"Prefer const over let. Never use var in JavaScript code."
|
||||
)
|
||||
|
||||
captured: list[dict] = []
|
||||
|
||||
def mock_urlopen(req, timeout=None):
|
||||
captured.append(json.loads(req.data.decode()))
|
||||
return _make_mock_response(201)
|
||||
|
||||
with mock.patch("import_competing_tools.resolve_api_key", return_value="m0-key"), \
|
||||
mock.patch("import_competing_tools.resolve_user_id", return_value="user1"), \
|
||||
mock.patch("import_competing_tools.resolve_project_id", return_value="proj1"), \
|
||||
mock.patch("import_competing_tools.resolve_branch", return_value="main"), \
|
||||
mock.patch("urllib.request.urlopen", side_effect=mock_urlopen):
|
||||
|
||||
cmd_copilot(["--path", str(copilot_file)])
|
||||
|
||||
assert len(captured) >= 1
|
||||
assert captured[0]["metadata"]["source"] == "copilot-import"
|
||||
assert captured[0]["app_id"] == "proj1"
|
||||
assert captured[0]["infer"] is False
|
||||
|
||||
|
||||
def test_cline_import_multiple_files(tmp_path):
|
||||
"""cmd_cline imports one memory per non-empty .md file."""
|
||||
from import_competing_tools import cmd_cline
|
||||
|
||||
mb = tmp_path / "memory-bank"
|
||||
mb.mkdir()
|
||||
(mb / "arch.md").write_text(
|
||||
"Architecture: microservices with event-sourcing. Each service owns its database. "
|
||||
"Communication via message bus only. No direct service-to-service HTTP calls."
|
||||
)
|
||||
(mb / "style.md").write_text(
|
||||
"Code style: PEP 8 for Python. Black formatter. isort for imports. "
|
||||
"Line length 120. Type hints required on all public functions and methods."
|
||||
)
|
||||
(mb / "empty.md").write_text("")
|
||||
|
||||
captured: list[dict] = []
|
||||
|
||||
def mock_urlopen(req, timeout=None):
|
||||
captured.append(json.loads(req.data.decode()))
|
||||
return _make_mock_response(201)
|
||||
|
||||
with mock.patch("import_competing_tools.resolve_api_key", return_value="m0-key"), \
|
||||
mock.patch("import_competing_tools.resolve_user_id", return_value="user1"), \
|
||||
mock.patch("import_competing_tools.resolve_project_id", return_value="proj1"), \
|
||||
mock.patch("import_competing_tools.resolve_branch", return_value="main"), \
|
||||
mock.patch("urllib.request.urlopen", side_effect=mock_urlopen):
|
||||
|
||||
cmd_cline(["--path", str(mb)])
|
||||
|
||||
# Should have exactly 2 imports (empty.md skipped)
|
||||
assert len(captured) == 2
|
||||
sources = {r["metadata"]["source"] for r in captured}
|
||||
assert sources == {"cline-import"}
|
||||
for r in captured:
|
||||
assert r["infer"] is False
|
||||
assert r["app_id"] == "proj1"
|
||||
|
||||
|
||||
def test_no_api_key_does_not_call_api(tmp_path):
|
||||
"""When no API key is set, no HTTP call is made."""
|
||||
from import_competing_tools import cmd_cursorrules
|
||||
|
||||
cursorrules = tmp_path / ".cursorrules"
|
||||
cursorrules.write_text("## Rules\n\n" + "x" * 100)
|
||||
|
||||
with mock.patch("import_competing_tools.resolve_api_key", return_value=""), \
|
||||
mock.patch("urllib.request.urlopen") as mock_url:
|
||||
|
||||
cmd_cursorrules(["--path", str(cursorrules)])
|
||||
|
||||
mock_url.assert_not_called()
|
||||
|
||||
|
||||
def test_missing_file_does_not_call_api(tmp_path):
|
||||
"""When the source file doesn't exist, no HTTP call is made."""
|
||||
from import_competing_tools import cmd_cursorrules
|
||||
|
||||
with mock.patch("import_competing_tools.resolve_api_key", return_value="m0-key"), \
|
||||
mock.patch("urllib.request.urlopen") as mock_url:
|
||||
|
||||
cmd_cursorrules(["--path", str(tmp_path / "nonexistent.cursorrules")])
|
||||
|
||||
mock_url.assert_not_called()
|
||||
|
||||
|
||||
def test_main_unknown_subcommand_exits_zero():
|
||||
"""Calling main() with an unknown subcommand exits 0."""
|
||||
import subprocess
|
||||
result = subprocess.run(
|
||||
[sys.executable, os.path.join(SCRIPTS_DIR, "import_competing_tools.py"), "unknown"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env={**os.environ, "MEM0_API_KEY": ""},
|
||||
)
|
||||
assert result.returncode == 0
|
||||
@@ -0,0 +1,298 @@
|
||||
"""Tests for parse_export_file.py — mem0 export file parser."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
SCRIPTS_DIR = os.path.join(os.path.dirname(__file__), "..", "scripts")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Direct function tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_parse_blocks_single_valid_block():
|
||||
"""parse_blocks returns one record for a single valid block."""
|
||||
from parse_export_file import parse_blocks
|
||||
|
||||
content = """\
|
||||
---
|
||||
id: abc123
|
||||
created_at: 2024-01-15T10:00:00Z
|
||||
type: task_learnings
|
||||
confidence: 0.85
|
||||
branch: main
|
||||
files: src/foo.py, src/bar.py
|
||||
categories: coding_conventions, task_learnings
|
||||
---
|
||||
Always use context managers when opening files.
|
||||
"""
|
||||
records = parse_blocks(content)
|
||||
assert len(records) == 1
|
||||
r = records[0]
|
||||
assert r["id"] == "abc123"
|
||||
assert r["type"] == "task_learnings"
|
||||
assert r["confidence"] == "0.85"
|
||||
assert r["branch"] == "main"
|
||||
assert r["files"] == ["src/foo.py", "src/bar.py"]
|
||||
assert r["categories"] == ["coding_conventions", "task_learnings"]
|
||||
assert "Always use context managers" in r["content"]
|
||||
|
||||
|
||||
def test_parse_blocks_multiple_blocks():
|
||||
"""parse_blocks returns the correct number of records for multiple blocks."""
|
||||
from parse_export_file import parse_blocks
|
||||
|
||||
content = """\
|
||||
---
|
||||
id: mem001
|
||||
type: architecture_decisions
|
||||
confidence: 0.9
|
||||
branch: main
|
||||
files:
|
||||
categories: architecture_decisions
|
||||
---
|
||||
Use hexagonal architecture for the core domain.
|
||||
|
||||
---
|
||||
id: mem002
|
||||
type: anti_patterns
|
||||
confidence: 0.75
|
||||
branch: feat/refactor
|
||||
files: src/legacy.py
|
||||
categories: anti_patterns
|
||||
---
|
||||
Avoid direct database calls from view layer.
|
||||
|
||||
---
|
||||
id: mem003
|
||||
type: coding_conventions
|
||||
confidence: 0.8
|
||||
branch: main
|
||||
files: src/utils.py, src/helpers.py
|
||||
categories:
|
||||
---
|
||||
Use snake_case for all Python identifiers.
|
||||
"""
|
||||
records = parse_blocks(content)
|
||||
assert len(records) == 3
|
||||
|
||||
assert records[0]["id"] == "mem001"
|
||||
assert records[0]["categories"] == ["architecture_decisions"]
|
||||
assert "hexagonal architecture" in records[0]["content"]
|
||||
|
||||
assert records[1]["id"] == "mem002"
|
||||
assert records[1]["files"] == ["src/legacy.py"]
|
||||
assert "direct database calls" in records[1]["content"]
|
||||
|
||||
assert records[2]["id"] == "mem003"
|
||||
assert records[2]["files"] == ["src/utils.py", "src/helpers.py"]
|
||||
assert records[2]["categories"] == []
|
||||
assert "snake_case" in records[2]["content"]
|
||||
|
||||
|
||||
def test_parse_blocks_missing_optional_fields():
|
||||
"""parse_blocks uses defaults when optional fields are absent."""
|
||||
from parse_export_file import parse_blocks
|
||||
|
||||
# confidence, branch, files, categories all absent
|
||||
content = """\
|
||||
---
|
||||
id: xyz789
|
||||
type: task_learnings
|
||||
---
|
||||
Run tests before committing.
|
||||
"""
|
||||
records = parse_blocks(content)
|
||||
assert len(records) == 1
|
||||
r = records[0]
|
||||
assert r["id"] == "xyz789"
|
||||
assert r["confidence"] == "" # default empty string
|
||||
assert r["branch"] == "" # default empty string
|
||||
assert r["files"] == [] # default empty list
|
||||
assert r["categories"] == [] # default empty list
|
||||
assert "Run tests" in r["content"]
|
||||
|
||||
|
||||
def test_parse_blocks_filters_empty_content():
|
||||
"""parse_blocks skips blocks whose content is empty or whitespace-only."""
|
||||
from parse_export_file import parse_blocks
|
||||
|
||||
content = """\
|
||||
---
|
||||
id: empty1
|
||||
type: task_learnings
|
||||
---
|
||||
|
||||
---
|
||||
id: real1
|
||||
type: task_learnings
|
||||
---
|
||||
This block has real content.
|
||||
|
||||
---
|
||||
id: empty2
|
||||
type: coding_conventions
|
||||
---
|
||||
|
||||
"""
|
||||
records = parse_blocks(content)
|
||||
# Only the block with actual content should be returned
|
||||
assert len(records) == 1
|
||||
assert records[0]["id"] == "real1"
|
||||
assert "real content" in records[0]["content"]
|
||||
|
||||
|
||||
def test_parse_blocks_round_trip():
|
||||
"""Content formatted by export matches what parse_blocks expects."""
|
||||
from parse_export_file import parse_blocks
|
||||
|
||||
# Simulate the exact format produced by the export skill
|
||||
memory_id = "test-id-001"
|
||||
created_at = "2024-06-01T12:00:00Z"
|
||||
mem_type = "architecture_decisions"
|
||||
confidence = "0.92"
|
||||
branch = "feat/new-feature"
|
||||
files = ["src/main.py", "tests/test_main.py"]
|
||||
categories = ["architecture_decisions", "coding_conventions"]
|
||||
memory_content = "Use dependency injection for all service classes."
|
||||
|
||||
# Format exactly as the export skill would
|
||||
block = (
|
||||
"---\n"
|
||||
f"id: {memory_id}\n"
|
||||
f"created_at: {created_at}\n"
|
||||
f"type: {mem_type}\n"
|
||||
f"confidence: {confidence}\n"
|
||||
f"branch: {branch}\n"
|
||||
f"files: {', '.join(files)}\n"
|
||||
f"categories: {', '.join(categories)}\n"
|
||||
"---\n"
|
||||
f"{memory_content}\n"
|
||||
"\n"
|
||||
)
|
||||
|
||||
records = parse_blocks(block)
|
||||
assert len(records) == 1
|
||||
r = records[0]
|
||||
assert r["id"] == memory_id
|
||||
assert r["type"] == mem_type
|
||||
assert r["confidence"] == confidence
|
||||
assert r["branch"] == branch
|
||||
assert r["files"] == files
|
||||
assert r["categories"] == categories
|
||||
assert r["content"] == memory_content
|
||||
|
||||
|
||||
def test_parse_blocks_multiline_content():
|
||||
"""parse_blocks correctly captures multi-line memory content."""
|
||||
from parse_export_file import parse_blocks
|
||||
|
||||
content = """\
|
||||
---
|
||||
id: multi001
|
||||
type: task_learnings
|
||||
---
|
||||
Line one of the memory.
|
||||
Line two of the memory.
|
||||
|
||||
Line four after blank line.
|
||||
"""
|
||||
records = parse_blocks(content)
|
||||
assert len(records) == 1
|
||||
assert "Line one" in records[0]["content"]
|
||||
assert "Line two" in records[0]["content"]
|
||||
assert "Line four" in records[0]["content"]
|
||||
|
||||
|
||||
def test_parse_blocks_empty_input():
|
||||
"""parse_blocks returns empty list for empty input."""
|
||||
from parse_export_file import parse_blocks
|
||||
|
||||
assert parse_blocks("") == []
|
||||
assert parse_blocks(" \n ") == []
|
||||
|
||||
|
||||
def test_parse_blocks_no_blocks():
|
||||
"""parse_blocks returns empty list for content without any --- delimiters."""
|
||||
from parse_export_file import parse_blocks
|
||||
|
||||
assert parse_blocks("Just some text without any delimiters.") == []
|
||||
|
||||
|
||||
def test_parse_blocks_value_with_colon():
|
||||
"""parse_blocks handles values that themselves contain colons."""
|
||||
from parse_export_file import parse_blocks
|
||||
|
||||
content = """\
|
||||
---
|
||||
id: colon-test
|
||||
type: task_learnings
|
||||
created_at: 2024-01-01T10:00:00Z
|
||||
---
|
||||
Timestamp values contain colons and should parse correctly.
|
||||
"""
|
||||
records = parse_blocks(content)
|
||||
assert len(records) == 1
|
||||
assert records[0]["id"] == "colon-test"
|
||||
# created_at field should be captured (it's in the record if present)
|
||||
assert "2024-01-01T10:00:00Z" in records[0].get("created_at", "")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CLI / subprocess tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_main_cli_outputs_json(tmp_path):
|
||||
"""Running parse_export_file.py as a script outputs valid JSON."""
|
||||
export_file = tmp_path / "mem0-export-test.md"
|
||||
export_file.write_text("""\
|
||||
---
|
||||
id: cli-test-001
|
||||
type: task_learnings
|
||||
confidence: 0.8
|
||||
branch: main
|
||||
files:
|
||||
categories: task_learnings
|
||||
---
|
||||
Prefer composition over inheritance.
|
||||
""")
|
||||
|
||||
result = subprocess.run(
|
||||
[sys.executable, os.path.join(SCRIPTS_DIR, "parse_export_file.py"), str(export_file)],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
assert result.returncode == 0
|
||||
records = json.loads(result.stdout)
|
||||
assert isinstance(records, list)
|
||||
assert len(records) == 1
|
||||
assert records[0]["id"] == "cli-test-001"
|
||||
|
||||
|
||||
def test_main_cli_no_args_exits_zero():
|
||||
"""Running parse_export_file.py with no arguments exits 0 and prints []."""
|
||||
result = subprocess.run(
|
||||
[sys.executable, os.path.join(SCRIPTS_DIR, "parse_export_file.py")],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
assert result.returncode == 0
|
||||
assert result.stdout.strip() == "[]"
|
||||
|
||||
|
||||
def test_main_cli_missing_file_exits_zero(tmp_path):
|
||||
"""Running parse_export_file.py with a missing file exits 0 and prints []."""
|
||||
result = subprocess.run(
|
||||
[sys.executable, os.path.join(SCRIPTS_DIR, "parse_export_file.py"),
|
||||
str(tmp_path / "nonexistent.md")],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
assert result.returncode == 0
|
||||
assert result.stdout.strip() == "[]"
|
||||
@@ -0,0 +1,424 @@
|
||||
"""Tests for parse_mem0_config.py — mem0.md retention policy parser."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
SCRIPTS_DIR = os.path.join(os.path.dirname(__file__), "..", "scripts")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# parse_retention — unit tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_parse_retention_valid_section():
|
||||
"""parse_retention extracts day-count policies correctly."""
|
||||
from parse_mem0_config import parse_retention
|
||||
|
||||
content = """\
|
||||
# Project Config
|
||||
|
||||
## Retention
|
||||
|
||||
session_state: 90d
|
||||
compact_summary: 60d
|
||||
decision: 180d
|
||||
"""
|
||||
result = parse_retention(content)
|
||||
assert result == {
|
||||
"session_state": 90,
|
||||
"compact_summary": 60,
|
||||
"decision": 180,
|
||||
}
|
||||
|
||||
|
||||
def test_parse_retention_forever_returns_none():
|
||||
"""parse_retention maps 'forever' to None."""
|
||||
from parse_mem0_config import parse_retention
|
||||
|
||||
content = """\
|
||||
## Retention
|
||||
|
||||
user_preference: forever
|
||||
anti_pattern: forever
|
||||
session_state: 30d
|
||||
"""
|
||||
result = parse_retention(content)
|
||||
assert result["user_preference"] is None
|
||||
assert result["anti_pattern"] is None
|
||||
assert result["session_state"] == 30
|
||||
|
||||
|
||||
def test_parse_retention_no_section_returns_empty():
|
||||
"""parse_retention returns {} when there is no ## Retention heading."""
|
||||
from parse_mem0_config import parse_retention
|
||||
|
||||
content = """\
|
||||
# Project Config
|
||||
|
||||
## Some Other Section
|
||||
|
||||
key: value
|
||||
"""
|
||||
result = parse_retention(content)
|
||||
assert result == {}
|
||||
|
||||
|
||||
def test_parse_retention_stops_at_next_heading():
|
||||
"""parse_retention stops reading at the next ## heading."""
|
||||
from parse_mem0_config import parse_retention
|
||||
|
||||
content = """\
|
||||
## Retention
|
||||
|
||||
session_state: 7d
|
||||
|
||||
## Other Section
|
||||
|
||||
other_key: 999d
|
||||
"""
|
||||
result = parse_retention(content)
|
||||
assert "session_state" in result
|
||||
assert "other_key" not in result
|
||||
|
||||
|
||||
def test_parse_retention_malformed_lines_skipped():
|
||||
"""Malformed lines (no colon, bad day format) are silently ignored."""
|
||||
from parse_mem0_config import parse_retention
|
||||
|
||||
content = """\
|
||||
## Retention
|
||||
|
||||
session_state: 90d
|
||||
bad_line_no_colon
|
||||
another: badvalue
|
||||
decision: 30d
|
||||
"""
|
||||
result = parse_retention(content)
|
||||
assert result == {"session_state": 90, "decision": 30}
|
||||
|
||||
|
||||
def test_parse_retention_comments_ignored():
|
||||
"""Inline # comments are stripped before parsing."""
|
||||
from parse_mem0_config import parse_retention
|
||||
|
||||
content = """\
|
||||
## Retention
|
||||
|
||||
session_state: 90d # rolling 90-day window
|
||||
user_preference: forever # never prune preferences
|
||||
"""
|
||||
result = parse_retention(content)
|
||||
assert result["session_state"] == 90
|
||||
assert result["user_preference"] is None
|
||||
|
||||
|
||||
def test_parse_retention_case_insensitive_heading():
|
||||
"""## retention (lowercase) is matched the same as ## Retention."""
|
||||
from parse_mem0_config import parse_retention
|
||||
|
||||
content = """\
|
||||
## retention
|
||||
|
||||
session_state: 14d
|
||||
"""
|
||||
result = parse_retention(content)
|
||||
assert result == {"session_state": 14}
|
||||
|
||||
|
||||
def test_parse_retention_empty_section_returns_empty():
|
||||
"""A ## Retention section with no valid lines returns {}."""
|
||||
from parse_mem0_config import parse_retention
|
||||
|
||||
content = """\
|
||||
## Retention
|
||||
|
||||
# only comments here
|
||||
|
||||
## Next Section
|
||||
"""
|
||||
result = parse_retention(content)
|
||||
assert result == {}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# load_retention_policies — integration tests with tmp files
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_load_retention_policies_with_tmp_file(tmp_path):
|
||||
"""load_retention_policies reads a real mem0.md from disk."""
|
||||
from parse_mem0_config import load_retention_policies
|
||||
|
||||
mem0_md = tmp_path / "mem0.md"
|
||||
mem0_md.write_text(
|
||||
"""\
|
||||
# My Project
|
||||
|
||||
## Retention
|
||||
|
||||
session_state: 90d
|
||||
compact_summary: 60d
|
||||
decision: forever
|
||||
""",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
result = load_retention_policies(str(tmp_path))
|
||||
assert result == {
|
||||
"session_state": 90,
|
||||
"compact_summary": 60,
|
||||
"decision": None,
|
||||
}
|
||||
|
||||
|
||||
def test_load_retention_policies_no_mem0_md_returns_empty(tmp_path):
|
||||
"""load_retention_policies returns {} when no mem0.md exists."""
|
||||
from parse_mem0_config import load_retention_policies
|
||||
|
||||
result = load_retention_policies(str(tmp_path))
|
||||
assert result == {}
|
||||
|
||||
|
||||
def test_load_retention_policies_no_retention_section_returns_empty(tmp_path):
|
||||
"""load_retention_policies returns {} when mem0.md has no ## Retention."""
|
||||
from parse_mem0_config import load_retention_policies
|
||||
|
||||
mem0_md = tmp_path / "mem0.md"
|
||||
mem0_md.write_text(
|
||||
"""\
|
||||
# My Project
|
||||
|
||||
Some general project notes here.
|
||||
No retention section.
|
||||
""",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
result = load_retention_policies(str(tmp_path))
|
||||
assert result == {}
|
||||
|
||||
|
||||
def test_load_retention_policies_defaults_to_cwd(tmp_path, monkeypatch):
|
||||
"""load_retention_policies uses os.getcwd() when cwd is None."""
|
||||
from parse_mem0_config import load_retention_policies
|
||||
|
||||
monkeypatch.chdir(tmp_path)
|
||||
mem0_md = tmp_path / "mem0.md"
|
||||
mem0_md.write_text("## Retention\nsession_state: 45d\n", encoding="utf-8")
|
||||
|
||||
result = load_retention_policies() # no cwd arg
|
||||
assert result == {"session_state": 45}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CLI / main() — subprocess test
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_cli_main_prints_json(tmp_path):
|
||||
"""CLI: python parse_mem0_config.py <cwd> prints valid JSON."""
|
||||
mem0_md = tmp_path / "mem0.md"
|
||||
mem0_md.write_text(
|
||||
"## Retention\nsession_state: 90d\nuser_preference: forever\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
result = subprocess.run(
|
||||
[sys.executable, os.path.join(SCRIPTS_DIR, "parse_mem0_config.py"), str(tmp_path)],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
assert result.returncode == 0
|
||||
data = json.loads(result.stdout)
|
||||
assert data["session_state"] == 90
|
||||
assert data["user_preference"] is None
|
||||
|
||||
|
||||
def test_cli_main_no_file_prints_empty_json(tmp_path):
|
||||
"""CLI: prints '{}' when no mem0.md exists."""
|
||||
result = subprocess.run(
|
||||
[sys.executable, os.path.join(SCRIPTS_DIR, "parse_mem0_config.py"), str(tmp_path)],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
assert result.returncode == 0
|
||||
assert json.loads(result.stdout) == {}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# parse_section_kv — unit tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_parse_section_kv_basic():
|
||||
"""parse_section_kv extracts key-value pairs from a named section."""
|
||||
from parse_mem0_config import parse_section_kv
|
||||
|
||||
content = """\
|
||||
## Search
|
||||
|
||||
default_limit: 10
|
||||
boost_recency: true
|
||||
"""
|
||||
result = parse_section_kv(content, "Search")
|
||||
assert result == {"default_limit": "10", "boost_recency": "true"}
|
||||
|
||||
|
||||
def test_parse_section_kv_missing_section():
|
||||
"""parse_section_kv returns {} when section doesn't exist."""
|
||||
from parse_mem0_config import parse_section_kv
|
||||
|
||||
result = parse_section_kv("## Other\nfoo: bar\n", "Search")
|
||||
assert result == {}
|
||||
|
||||
|
||||
def test_parse_section_kv_stops_at_next_heading():
|
||||
"""parse_section_kv stops at the next ## heading."""
|
||||
from parse_mem0_config import parse_section_kv
|
||||
|
||||
content = """\
|
||||
## Identity
|
||||
|
||||
user_id: kartik
|
||||
project_id: mem0
|
||||
|
||||
## Other
|
||||
|
||||
ignored: yes
|
||||
"""
|
||||
result = parse_section_kv(content, "Identity")
|
||||
assert result == {"user_id": "kartik", "project_id": "mem0"}
|
||||
assert "ignored" not in result
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# parse_section_list — unit tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_parse_section_list_basic():
|
||||
"""parse_section_list extracts list items from a named section."""
|
||||
from parse_mem0_config import parse_section_list
|
||||
|
||||
content = """\
|
||||
## Categories
|
||||
|
||||
- architecture_decisions
|
||||
- bug_fixes
|
||||
- coding_conventions
|
||||
"""
|
||||
result = parse_section_list(content, "Categories")
|
||||
assert result == ["architecture_decisions", "bug_fixes", "coding_conventions"]
|
||||
|
||||
|
||||
def test_parse_section_list_bare_lines():
|
||||
"""parse_section_list works with bare lines (no bullet prefix)."""
|
||||
from parse_mem0_config import parse_section_list
|
||||
|
||||
content = """\
|
||||
## Categories
|
||||
|
||||
architecture_decisions
|
||||
bug_fixes
|
||||
"""
|
||||
result = parse_section_list(content, "Categories")
|
||||
assert result == ["architecture_decisions", "bug_fixes"]
|
||||
|
||||
|
||||
def test_parse_section_list_missing_section():
|
||||
"""parse_section_list returns [] when section doesn't exist."""
|
||||
from parse_mem0_config import parse_section_list
|
||||
|
||||
result = parse_section_list("## Other\n- foo\n", "Categories")
|
||||
assert result == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# load_full_config — integration tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_load_full_config_all_sections(tmp_path):
|
||||
"""load_full_config extracts all sections from mem0.md."""
|
||||
from parse_mem0_config import load_full_config
|
||||
|
||||
mem0_md = tmp_path / "mem0.md"
|
||||
mem0_md.write_text(
|
||||
"""\
|
||||
# My Project
|
||||
|
||||
## Retention
|
||||
|
||||
session_state: 90d
|
||||
decision: forever
|
||||
|
||||
## Search
|
||||
|
||||
default_limit: 20
|
||||
boost_recency: true
|
||||
|
||||
## Categories
|
||||
|
||||
- architecture_decisions
|
||||
- bug_fixes
|
||||
- security_constraints
|
||||
|
||||
## Identity
|
||||
|
||||
user_id: kartik
|
||||
project_id: my-project
|
||||
""",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
config = load_full_config(str(tmp_path))
|
||||
assert config["retention"] == {"session_state": 90, "decision": None}
|
||||
assert config["search"] == {"default_limit": "20", "boost_recency": "true"}
|
||||
assert config["categories"] == ["architecture_decisions", "bug_fixes", "security_constraints"]
|
||||
assert config["identity"] == {"user_id": "kartik", "project_id": "my-project"}
|
||||
|
||||
|
||||
def test_load_full_config_partial_sections(tmp_path):
|
||||
"""load_full_config only includes sections that exist."""
|
||||
from parse_mem0_config import load_full_config
|
||||
|
||||
mem0_md = tmp_path / "mem0.md"
|
||||
mem0_md.write_text("## Retention\nsession_state: 30d\n", encoding="utf-8")
|
||||
|
||||
config = load_full_config(str(tmp_path))
|
||||
assert "retention" in config
|
||||
assert "search" not in config
|
||||
assert "categories" not in config
|
||||
assert "identity" not in config
|
||||
|
||||
|
||||
def test_load_full_config_no_file(tmp_path):
|
||||
"""load_full_config returns {} when no mem0.md exists."""
|
||||
from parse_mem0_config import load_full_config
|
||||
|
||||
config = load_full_config(str(tmp_path))
|
||||
assert config == {}
|
||||
|
||||
|
||||
def test_cli_full_flag(tmp_path):
|
||||
"""CLI: --full prints all sections as JSON."""
|
||||
mem0_md = tmp_path / "mem0.md"
|
||||
mem0_md.write_text(
|
||||
"## Retention\nsession_state: 90d\n\n## Search\nlimit: 10\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
result = subprocess.run(
|
||||
[sys.executable, os.path.join(SCRIPTS_DIR, "parse_mem0_config.py"), "--full", str(tmp_path)],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
assert result.returncode == 0
|
||||
data = json.loads(result.stdout)
|
||||
assert "retention" in data
|
||||
assert "search" in data
|
||||
@@ -0,0 +1,54 @@
|
||||
"""Tests for on_pre_commit.py — pre-commit memory capture."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
SCRIPTS_DIR = os.path.join(os.path.dirname(__file__), "..", "scripts")
|
||||
|
||||
|
||||
def test_import_succeeds():
|
||||
"""on_pre_commit module can be imported."""
|
||||
import on_pre_commit
|
||||
|
||||
assert hasattr(on_pre_commit, "main")
|
||||
|
||||
|
||||
def test_no_api_key_exits_zero(monkeypatch):
|
||||
"""main() exits 0 when no API key is set."""
|
||||
import on_pre_commit
|
||||
|
||||
monkeypatch.delenv("MEM0_API_KEY", raising=False)
|
||||
monkeypatch.delenv("CLAUDE_PLUGIN_OPTION_MEM0_API_KEY", raising=False)
|
||||
monkeypatch.setattr("sys.stdin", MagicMock(isatty=lambda: True))
|
||||
|
||||
assert on_pre_commit.main() == 0
|
||||
|
||||
|
||||
def test_empty_diff_exits_zero(monkeypatch):
|
||||
"""main() exits 0 when stdin diff is empty."""
|
||||
from io import StringIO
|
||||
|
||||
import on_pre_commit
|
||||
|
||||
monkeypatch.setenv("MEM0_API_KEY", "test-key")
|
||||
monkeypatch.setattr("sys.stdin", StringIO(""))
|
||||
|
||||
assert on_pre_commit.main() == 0
|
||||
|
||||
|
||||
def test_get_staged_summary_runs():
|
||||
"""get_staged_summary doesn't crash even outside a git repo."""
|
||||
import on_pre_commit
|
||||
|
||||
result = on_pre_commit.get_staged_summary()
|
||||
assert isinstance(result, str)
|
||||
|
||||
|
||||
def test_get_commit_message_runs():
|
||||
"""get_commit_message doesn't crash even outside a git repo."""
|
||||
import on_pre_commit
|
||||
|
||||
result = on_pre_commit.get_commit_message()
|
||||
assert isinstance(result, str)
|
||||
@@ -141,3 +141,108 @@ def test_cli_report_no_data(tmp_path):
|
||||
env=env,
|
||||
)
|
||||
assert result.returncode == 0
|
||||
|
||||
|
||||
def test_peek_returns_json_without_clearing(_isolate_stats_file):
|
||||
"""peek returns JSON stats without deleting the stats file."""
|
||||
import session_stats
|
||||
|
||||
session_stats.init()
|
||||
session_stats.record_add("decisions")
|
||||
session_stats.record_add("decisions")
|
||||
session_stats.record_search()
|
||||
|
||||
result = session_stats.peek()
|
||||
data = json.loads(result)
|
||||
assert data["adds"] == 2
|
||||
assert data["searches"] == 1
|
||||
|
||||
assert os.path.isfile(_isolate_stats_file)
|
||||
|
||||
|
||||
def test_category_counts_tracked(_isolate_stats_file):
|
||||
"""category_counts tracks per-category add counts."""
|
||||
import session_stats
|
||||
|
||||
session_stats.init()
|
||||
session_stats.record_add("bug_fixes")
|
||||
session_stats.record_add("bug_fixes")
|
||||
session_stats.record_add("bug_fixes")
|
||||
session_stats.record_add("decisions")
|
||||
|
||||
with open(_isolate_stats_file) as f:
|
||||
data = json.load(f)
|
||||
assert data["category_counts"]["bug_fixes"] == 3
|
||||
assert data["category_counts"]["decisions"] == 1
|
||||
|
||||
|
||||
def test_category_counts_empty_category_not_tracked(_isolate_stats_file):
|
||||
"""Empty category string doesn't appear in category_counts."""
|
||||
import session_stats
|
||||
|
||||
session_stats.init()
|
||||
session_stats.record_add("")
|
||||
session_stats.record_add()
|
||||
|
||||
with open(_isolate_stats_file) as f:
|
||||
data = json.load(f)
|
||||
assert data["category_counts"] == {}
|
||||
|
||||
|
||||
def test_recent_ids_tracked(_isolate_stats_file):
|
||||
"""record_add with memory_id stores ID in recent_ids."""
|
||||
import session_stats
|
||||
|
||||
session_stats.init()
|
||||
session_stats.record_add("decision", "abc-123")
|
||||
session_stats.record_add("convention", "def-456")
|
||||
|
||||
with open(_isolate_stats_file) as f:
|
||||
data = json.load(f)
|
||||
assert len(data["recent_ids"]) == 2
|
||||
assert data["recent_ids"][0]["id"] == "abc-123"
|
||||
assert data["recent_ids"][1]["id"] == "def-456"
|
||||
assert data["recent_ids"][0]["category"] == "decision"
|
||||
|
||||
|
||||
def test_recent_ids_capped(_isolate_stats_file):
|
||||
"""recent_ids list is capped at MAX_RECENT_IDS."""
|
||||
import session_stats
|
||||
|
||||
session_stats.init()
|
||||
for i in range(60):
|
||||
session_stats.record_add("test", f"id-{i}")
|
||||
|
||||
with open(_isolate_stats_file) as f:
|
||||
data = json.load(f)
|
||||
assert len(data["recent_ids"]) == session_stats.MAX_RECENT_IDS
|
||||
assert data["recent_ids"][0]["id"] == f"id-{60 - session_stats.MAX_RECENT_IDS}"
|
||||
|
||||
|
||||
def test_recent_ids_empty_without_memory_id(_isolate_stats_file):
|
||||
"""record_add without memory_id doesn't add to recent_ids."""
|
||||
import session_stats
|
||||
|
||||
session_stats.init()
|
||||
session_stats.record_add("decision")
|
||||
session_stats.record_add("convention", "")
|
||||
|
||||
with open(_isolate_stats_file) as f:
|
||||
data = json.load(f)
|
||||
assert data["recent_ids"] == []
|
||||
|
||||
|
||||
def test_cli_peek(tmp_path):
|
||||
"""Test CLI invocation: session_stats.py peek outputs JSON."""
|
||||
env = {**os.environ, "USER": "test"}
|
||||
subprocess.run(
|
||||
[sys.executable, os.path.join(SCRIPTS_DIR, "session_stats.py"), "init"],
|
||||
capture_output=True, text=True, env=env,
|
||||
)
|
||||
result = subprocess.run(
|
||||
[sys.executable, os.path.join(SCRIPTS_DIR, "session_stats.py"), "peek"],
|
||||
capture_output=True, text=True, env=env,
|
||||
)
|
||||
assert result.returncode == 0
|
||||
data = json.loads(result.stdout)
|
||||
assert "adds" in data
|
||||
|
||||
@@ -0,0 +1,163 @@
|
||||
"""Tests for telemetry.py — fire-and-forget PostHog plugin telemetry."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import urllib.error
|
||||
|
||||
SCRIPTS_DIR = os.path.join(os.path.dirname(__file__), "..", "scripts")
|
||||
sys.path.insert(0, os.path.abspath(SCRIPTS_DIR))
|
||||
|
||||
|
||||
def test_import_succeeds():
|
||||
import telemetry
|
||||
|
||||
assert hasattr(telemetry, "emit")
|
||||
assert hasattr(telemetry, "main")
|
||||
|
||||
|
||||
def test_opt_out_skips_send(monkeypatch):
|
||||
import telemetry
|
||||
|
||||
monkeypatch.setenv("MEM0_TELEMETRY", "false")
|
||||
sent = []
|
||||
monkeypatch.setattr(telemetry, "send", lambda p: sent.append(p))
|
||||
telemetry.emit("session_start")
|
||||
assert sent == []
|
||||
|
||||
|
||||
def test_opt_out_variants(monkeypatch):
|
||||
import telemetry
|
||||
|
||||
for val in ("0", "no", "off", "FALSE", "No"):
|
||||
monkeypatch.setenv("MEM0_TELEMETRY", val)
|
||||
assert not telemetry.is_enabled()
|
||||
|
||||
|
||||
def test_enabled_by_default(monkeypatch):
|
||||
import telemetry
|
||||
|
||||
monkeypatch.delenv("MEM0_TELEMETRY", raising=False)
|
||||
assert telemetry.is_enabled()
|
||||
|
||||
|
||||
def test_posthog_payload_structure(monkeypatch):
|
||||
import telemetry
|
||||
|
||||
monkeypatch.setenv("MEM0_RESOLVED_USER_ID", "testuser")
|
||||
monkeypatch.setenv("MEM0_PROJECT_ID", "test-project")
|
||||
monkeypatch.delenv("MEM0_API_KEY", raising=False)
|
||||
monkeypatch.delenv("CLAUDE_PLUGIN_OPTION_MEM0_API_KEY", raising=False)
|
||||
|
||||
payload = telemetry.build_posthog_payload("plugin.session_start", {"memory_count": 5})
|
||||
|
||||
assert payload["api_key"] == telemetry.POSTHOG_API_KEY
|
||||
assert payload["event"] == "plugin.session_start"
|
||||
assert "distinct_id" in payload
|
||||
assert payload["properties"]["source"] == "plugin"
|
||||
assert payload["properties"]["plugin_version"] == "0.2.1"
|
||||
assert payload["properties"]["memory_count"] == 5
|
||||
assert payload["properties"]["$process_person_profile"] is False
|
||||
|
||||
raw = json.dumps(payload)
|
||||
assert "testuser" not in raw
|
||||
assert "test-project" not in raw
|
||||
|
||||
|
||||
def test_distinct_id_from_api_key(monkeypatch):
|
||||
import hashlib
|
||||
|
||||
import telemetry
|
||||
|
||||
monkeypatch.setenv("MEM0_API_KEY", "m0-testkey123")
|
||||
expected = hashlib.md5(b"m0-testkey123").hexdigest()
|
||||
assert telemetry._distinct_id() == expected
|
||||
|
||||
|
||||
def test_distinct_id_fallback_no_key(monkeypatch):
|
||||
import telemetry
|
||||
|
||||
monkeypatch.delenv("MEM0_API_KEY", raising=False)
|
||||
monkeypatch.delenv("CLAUDE_PLUGIN_OPTION_MEM0_API_KEY", raising=False)
|
||||
monkeypatch.setenv("MEM0_RESOLVED_USER_ID", "kartik")
|
||||
assert telemetry._distinct_id() == telemetry._sha256("kartik")
|
||||
|
||||
|
||||
def test_hash_deterministic():
|
||||
import telemetry
|
||||
|
||||
h1 = telemetry._sha256("same-value")
|
||||
h2 = telemetry._sha256("same-value")
|
||||
assert h1 == h2
|
||||
assert h1 != telemetry._sha256("different-value")
|
||||
|
||||
|
||||
def test_platform_claude_code(monkeypatch):
|
||||
import telemetry
|
||||
|
||||
monkeypatch.setenv("CLAUDECODE", "1")
|
||||
monkeypatch.delenv("CURSOR_PLUGIN_ROOT", raising=False)
|
||||
monkeypatch.delenv("CODEX_PLUGIN_ROOT", raising=False)
|
||||
assert telemetry.detect_platform() == "claude-code"
|
||||
|
||||
|
||||
def test_platform_cursor(monkeypatch):
|
||||
import telemetry
|
||||
|
||||
monkeypatch.delenv("CLAUDECODE", raising=False)
|
||||
monkeypatch.delenv("CLAUDE_PLUGIN_ROOT", raising=False)
|
||||
monkeypatch.setenv("CURSOR_PLUGIN_ROOT", "/path")
|
||||
monkeypatch.delenv("CODEX_PLUGIN_ROOT", raising=False)
|
||||
assert telemetry.detect_platform() == "cursor"
|
||||
|
||||
|
||||
def test_platform_codex(monkeypatch):
|
||||
import telemetry
|
||||
|
||||
monkeypatch.delenv("CLAUDECODE", raising=False)
|
||||
monkeypatch.delenv("CLAUDE_PLUGIN_ROOT", raising=False)
|
||||
monkeypatch.delenv("CURSOR_PLUGIN_ROOT", raising=False)
|
||||
monkeypatch.setenv("CODEX_PLUGIN_ROOT", "/path")
|
||||
assert telemetry.detect_platform() == "codex"
|
||||
|
||||
|
||||
def test_sampling_drops_at_high_random(monkeypatch):
|
||||
import telemetry
|
||||
|
||||
monkeypatch.setattr(telemetry.random, "random", lambda: 0.5)
|
||||
assert telemetry._should_sample() is False
|
||||
|
||||
|
||||
def test_sampling_sends_at_low_random(monkeypatch):
|
||||
import telemetry
|
||||
|
||||
monkeypatch.setattr(telemetry.random, "random", lambda: 0.05)
|
||||
assert telemetry._should_sample() is True
|
||||
|
||||
|
||||
def test_send_fails_silently(monkeypatch):
|
||||
import telemetry
|
||||
|
||||
def raise_error(req, timeout):
|
||||
raise urllib.error.URLError("connection refused")
|
||||
|
||||
monkeypatch.setattr(telemetry.urllib.request, "urlopen", raise_error)
|
||||
telemetry.send({"event": "test"})
|
||||
|
||||
|
||||
def test_cli_exits_zero_when_disabled(monkeypatch):
|
||||
import telemetry
|
||||
|
||||
monkeypatch.setenv("MEM0_TELEMETRY", "false")
|
||||
monkeypatch.setattr(sys, "argv", ["telemetry.py", "session_start"])
|
||||
assert telemetry.main() == 0
|
||||
|
||||
|
||||
def test_cli_no_args_exits_nonzero(monkeypatch):
|
||||
import telemetry
|
||||
|
||||
monkeypatch.delenv("MEM0_TELEMETRY", raising=False)
|
||||
monkeypatch.setattr(sys, "argv", ["telemetry.py"])
|
||||
assert telemetry.main() == 1
|
||||
@@ -0,0 +1,194 @@
|
||||
"""Tests for write-path app_id migration and API key resolution.
|
||||
|
||||
Verifies that all scripts writing to the Mem0 API:
|
||||
1. Pass app_id as a top-level parameter (not in metadata)
|
||||
2. Do NOT include project_id in metadata
|
||||
3. Include branch in metadata when available
|
||||
4. Use resolve_api_key() for key resolution with userConfig fallback
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
|
||||
def test_auto_import_post_memory_uses_app_id():
|
||||
"""auto_import.post_memory sends app_id top-level, not metadata.project_id."""
|
||||
from auto_import import post_memory
|
||||
|
||||
captured = {}
|
||||
|
||||
def mock_urlopen(req, timeout=None):
|
||||
body = json.loads(req.data.decode("utf-8"))
|
||||
captured.update(body)
|
||||
resp = MagicMock()
|
||||
resp.status = 200
|
||||
resp.__enter__ = lambda s: s
|
||||
resp.__exit__ = MagicMock(return_value=False)
|
||||
return resp
|
||||
|
||||
with patch("urllib.request.urlopen", side_effect=mock_urlopen):
|
||||
result = post_memory(
|
||||
api_key="test-key",
|
||||
content="test content",
|
||||
user_id="testuser",
|
||||
filename="CLAUDE.md",
|
||||
project_id="my-project",
|
||||
branch="main",
|
||||
)
|
||||
|
||||
assert result is True
|
||||
assert captured["app_id"] == "my-project"
|
||||
assert captured["user_id"] == "testuser"
|
||||
assert "project_id" not in captured.get("metadata", {})
|
||||
assert captured["metadata"]["type"] == "project_profile"
|
||||
assert captured["metadata"]["branch"] == "main"
|
||||
assert captured["infer"] is False
|
||||
|
||||
|
||||
def test_auto_import_post_memory_omits_empty_branch():
|
||||
"""auto_import.post_memory skips branch in metadata when empty."""
|
||||
from auto_import import post_memory
|
||||
|
||||
captured = {}
|
||||
|
||||
def mock_urlopen(req, timeout=None):
|
||||
body = json.loads(req.data.decode("utf-8"))
|
||||
captured.update(body)
|
||||
resp = MagicMock()
|
||||
resp.status = 200
|
||||
resp.__enter__ = lambda s: s
|
||||
resp.__exit__ = MagicMock(return_value=False)
|
||||
return resp
|
||||
|
||||
with patch("urllib.request.urlopen", side_effect=mock_urlopen):
|
||||
post_memory("key", "content", "user", "FILE.md", "proj", branch="")
|
||||
|
||||
assert "branch" not in captured.get("metadata", {})
|
||||
|
||||
|
||||
def test_on_pre_compact_store_memory_uses_app_id():
|
||||
"""on_pre_compact.store_memory sends app_id top-level."""
|
||||
from on_pre_compact import store_memory
|
||||
|
||||
captured = {}
|
||||
|
||||
def mock_urlopen(req, timeout=None):
|
||||
body = json.loads(req.data.decode("utf-8"))
|
||||
captured.update(body)
|
||||
resp = MagicMock()
|
||||
resp.status = 200
|
||||
resp.__enter__ = lambda s: s
|
||||
resp.__exit__ = MagicMock(return_value=False)
|
||||
return resp
|
||||
|
||||
with patch("urllib.request.urlopen", side_effect=mock_urlopen):
|
||||
result = store_memory(
|
||||
api_key="test-key",
|
||||
content="session state content",
|
||||
user_id="testuser",
|
||||
source="pre-compaction",
|
||||
session_id="sess-123",
|
||||
project_id="my-project",
|
||||
branch="feat/auth",
|
||||
)
|
||||
|
||||
assert result is True
|
||||
assert captured["app_id"] == "my-project"
|
||||
assert captured["user_id"] == "testuser"
|
||||
assert "project_id" not in captured.get("metadata", {})
|
||||
assert captured["metadata"]["type"] == "session_state"
|
||||
assert captured["metadata"]["source"] == "pre-compaction"
|
||||
assert captured["metadata"]["branch"] == "feat/auth"
|
||||
assert "expiration_date" in captured
|
||||
|
||||
|
||||
def test_capture_compact_summary_store_uses_app_id():
|
||||
"""capture_compact_summary.store_summary sends app_id top-level."""
|
||||
from capture_compact_summary import store_summary
|
||||
|
||||
captured = {}
|
||||
|
||||
def mock_urlopen(req, timeout=None):
|
||||
body = json.loads(req.data.decode("utf-8"))
|
||||
captured.update(body)
|
||||
resp = MagicMock()
|
||||
resp.status = 200
|
||||
resp.__enter__ = lambda s: s
|
||||
resp.__exit__ = MagicMock(return_value=False)
|
||||
return resp
|
||||
|
||||
with patch("urllib.request.urlopen", side_effect=mock_urlopen):
|
||||
result = store_summary(
|
||||
api_key="test-key",
|
||||
summary="compact summary text",
|
||||
user_id="testuser",
|
||||
session_id="sess-456",
|
||||
project_id="my-project",
|
||||
branch="main",
|
||||
)
|
||||
|
||||
assert result is True
|
||||
assert captured["app_id"] == "my-project"
|
||||
assert captured["user_id"] == "testuser"
|
||||
assert "project_id" not in captured.get("metadata", {})
|
||||
assert captured["metadata"]["type"] == "compact_summary"
|
||||
assert captured["metadata"]["branch"] == "main"
|
||||
assert captured["infer"] is False
|
||||
assert "expiration_date" in captured
|
||||
|
||||
|
||||
def test_no_metadata_project_id_anywhere():
|
||||
"""Ensure none of the write functions put project_id in metadata."""
|
||||
from auto_import import post_memory
|
||||
from capture_compact_summary import store_summary
|
||||
from on_pre_compact import store_memory
|
||||
|
||||
bodies = []
|
||||
|
||||
def mock_urlopen(req, timeout=None):
|
||||
body = json.loads(req.data.decode("utf-8"))
|
||||
bodies.append(body)
|
||||
resp = MagicMock()
|
||||
resp.status = 200
|
||||
resp.__enter__ = lambda s: s
|
||||
resp.__exit__ = MagicMock(return_value=False)
|
||||
return resp
|
||||
|
||||
with patch("urllib.request.urlopen", side_effect=mock_urlopen):
|
||||
post_memory("k", "c", "u", "f", "proj", "br")
|
||||
store_memory("k", "c", "u", "src", "sid", "proj", "br")
|
||||
store_summary("k", "s", "u", "sid", "proj", "br")
|
||||
|
||||
for i, body in enumerate(bodies):
|
||||
metadata = body.get("metadata", {})
|
||||
assert "project_id" not in metadata, f"Write function #{i} still has metadata.project_id"
|
||||
assert body.get("app_id") == "proj", f"Write function #{i} missing app_id top-level"
|
||||
|
||||
|
||||
def test_resolve_api_key_prefers_env_var(monkeypatch):
|
||||
"""resolve_api_key returns MEM0_API_KEY when both are set."""
|
||||
from _identity import resolve_api_key
|
||||
|
||||
monkeypatch.setenv("MEM0_API_KEY", "direct-key")
|
||||
monkeypatch.setenv("CLAUDE_PLUGIN_OPTION_MEM0_API_KEY", "fallback-key")
|
||||
assert resolve_api_key() == "direct-key"
|
||||
|
||||
|
||||
def test_resolve_api_key_falls_back_to_plugin_option(monkeypatch):
|
||||
"""resolve_api_key falls back to CLAUDE_PLUGIN_OPTION_MEM0_API_KEY."""
|
||||
from _identity import resolve_api_key
|
||||
|
||||
monkeypatch.delenv("MEM0_API_KEY", raising=False)
|
||||
monkeypatch.setenv("CLAUDE_PLUGIN_OPTION_MEM0_API_KEY", "fallback-key")
|
||||
assert resolve_api_key() == "fallback-key"
|
||||
|
||||
|
||||
def test_resolve_api_key_returns_empty_when_neither_set(monkeypatch):
|
||||
"""resolve_api_key returns empty string when no key is available."""
|
||||
from _identity import resolve_api_key
|
||||
|
||||
monkeypatch.delenv("MEM0_API_KEY", raising=False)
|
||||
monkeypatch.delenv("CLAUDE_PLUGIN_OPTION_MEM0_API_KEY", raising=False)
|
||||
assert resolve_api_key() == ""
|
||||
Reference in New Issue
Block a user