feat(mem0-plugin): onboarding, project scoping, identity banner (#5207)

mem0-plugin/tests/conftest.py

@@ -0,0 +1,54 @@
+"""Shared fixtures for mem0-plugin tests."""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+
+import pytest
+
+SCRIPTS_DIR = os.path.join(os.path.dirname(__file__), "..", "scripts")
+
+
+@pytest.fixture(autouse=True)
+def _scripts_on_path():
+    """Ensure scripts/ is on sys.path so we can import _project, session_stats, etc."""
+    abs_scripts = os.path.abspath(SCRIPTS_DIR)
+    if abs_scripts not in sys.path:
+        sys.path.insert(0, abs_scripts)
+    yield
+    if abs_scripts in sys.path:
+        sys.path.remove(abs_scripts)
+
+
+@pytest.fixture()
+def tmp_git_repo(tmp_path):
+    """Create a temp dir with a git repo and HTTPS remote."""
+    subprocess.run(["git", "init"], cwd=tmp_path, capture_output=True, check=True)
+    subprocess.run(
+        ["git", "remote", "add", "origin", "https://github.com/mem0ai/mem0.git"],
+        cwd=tmp_path,
+        capture_output=True,
+        check=True,
+    )
+    return tmp_path
+
+
+@pytest.fixture()
+def tmp_git_repo_ssh(tmp_path):
+    """Create a temp dir with a git repo and SSH remote."""
+    subprocess.run(["git", "init"], cwd=tmp_path, capture_output=True, check=True)
+    subprocess.run(
+        ["git", "remote", "add", "origin", "git@github.com:acme/cool-project.git"],
+        cwd=tmp_path,
+        capture_output=True,
+        check=True,
+    )
+    return tmp_path
+
+
+@pytest.fixture()
+def tmp_no_git(tmp_path):
+    """Temp dir with no git repo."""
+    return tmp_path

mem0-plugin/tests/test_project.py

@@ -0,0 +1,107 @@
+"""Tests for _project.py — project_id + branch resolver."""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+
+
+def test_resolve_project_id_from_https_remote(tmp_git_repo):
+    from _project import resolve_project_id
+
+    pid = resolve_project_id(str(tmp_git_repo))
+    assert pid == "mem0ai-mem0"
+
+
+def test_resolve_project_id_from_ssh_remote(tmp_git_repo_ssh):
+    from _project import resolve_project_id
+
+    pid = resolve_project_id(str(tmp_git_repo_ssh))
+    assert pid == "acme-cool-project"
+
+
+def test_resolve_project_id_fallback_basename(tmp_no_git):
+    from _project import resolve_project_id
+
+    pid = resolve_project_id(str(tmp_no_git))
+    assert pid == os.path.basename(str(tmp_no_git))
+
+
+def test_resolve_project_id_from_env(tmp_no_git, monkeypatch):
+    from _project import resolve_project_id
+
+    monkeypatch.setenv("MEM0_PROJECT_ID", "my-override")
+    pid = resolve_project_id(str(tmp_no_git))
+    assert pid == "my-override"
+
+
+def test_resolve_project_id_from_project_map(tmp_no_git):
+    from _project import resolve_project_id, save_project_mapping
+
+    save_project_mapping(str(tmp_no_git), "custom-project")
+    pid = resolve_project_id(str(tmp_no_git))
+    assert pid == "custom-project"
+
+
+def test_save_project_mapping_creates_file(tmp_no_git):
+    from _project import save_project_mapping
+
+    save_project_mapping(str(tmp_no_git), "test-proj")
+    map_path = os.path.expanduser("~/.mem0/project_map.json")
+    assert os.path.isfile(map_path)
+    with open(map_path) as f:
+        data = json.load(f)
+    assert data[str(tmp_no_git)] == "test-proj"
+
+
+def test_resolve_branch_in_git_repo(tmp_git_repo):
+    from _project import resolve_branch
+
+    subprocess.run(
+        ["git", "checkout", "-b", "feat/test-branch"],
+        cwd=tmp_git_repo,
+        capture_output=True,
+        check=True,
+    )
+    branch = resolve_branch(str(tmp_git_repo))
+    assert branch == "feat/test-branch"
+
+
+def test_resolve_branch_no_git(tmp_no_git):
+    from _project import resolve_branch
+
+    branch = resolve_branch(str(tmp_no_git))
+    assert branch == "unknown"
+
+
+def test_remote_url_to_slug_various_formats():
+    from _project import _remote_url_to_slug
+
+    assert _remote_url_to_slug("https://github.com/mem0ai/mem0.git") == "mem0ai-mem0"
+    assert _remote_url_to_slug("git@github.com:mem0ai/mem0.git") == "mem0ai-mem0"
+    assert _remote_url_to_slug("ssh://git@github.com/acme/app.git") == "acme-app"
+    assert _remote_url_to_slug("https://gitlab.com/org/sub/repo.git") == "sub-repo"
+    assert _remote_url_to_slug("git@bitbucket.org:team/project.git") == "team-project"
+
+
+def test_remote_url_to_slug_no_git_suffix():
+    from _project import _remote_url_to_slug
+
+    assert _remote_url_to_slug("https://github.com/foo/bar") == "foo-bar"
+
+
+def test_resolve_project_id_priority_order(tmp_git_repo, monkeypatch):
+    """Env var > project_map > git remote > basename."""
+    from _project import resolve_project_id, save_project_mapping
+
+    # Git remote gives "mem0ai-mem0"
+    assert resolve_project_id(str(tmp_git_repo)) == "mem0ai-mem0"
+
+    # project_map overrides git remote
+    save_project_mapping(str(tmp_git_repo), "from-map")
+    assert resolve_project_id(str(tmp_git_repo)) == "from-map"
+
+    # Env var overrides everything
+    monkeypatch.setenv("MEM0_PROJECT_ID", "from-env")
+    assert resolve_project_id(str(tmp_git_repo)) == "from-env"

mem0-plugin/tests/test_session_stats.py

@@ -0,0 +1,143 @@
+"""Tests for session_stats.py — session-level memory operation tracker."""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+
+import pytest
+
+SCRIPTS_DIR = os.path.join(os.path.dirname(__file__), "..", "scripts")
+
+
+@pytest.fixture(autouse=True)
+def _isolate_stats_file(tmp_path, monkeypatch):
+    """Point STATS_FILE to a temp location so tests don't interfere."""
+    stats_file = str(tmp_path / "test_stats.json")
+    monkeypatch.setattr("session_stats.STATS_FILE", stats_file)
+    yield stats_file
+
+
+def test_init_creates_file(_isolate_stats_file):
+    import session_stats
+
+    session_stats.init()
+    assert os.path.isfile(_isolate_stats_file)
+    with open(_isolate_stats_file) as f:
+        data = json.load(f)
+    assert data["adds"] == 0
+    assert data["searches"] == 0
+    assert data["categories"] == []
+
+
+def test_record_add_increments(_isolate_stats_file):
+    import session_stats
+
+    session_stats.init()
+    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["adds"] == 3
+    assert set(data["categories"]) == {"bug_fixes", "decisions"}
+
+
+def test_record_search_increments(_isolate_stats_file):
+    import session_stats
+
+    session_stats.init()
+    session_stats.record_search()
+    session_stats.record_search()
+
+    with open(_isolate_stats_file) as f:
+        data = json.load(f)
+    assert data["searches"] == 2
+
+
+def test_report_returns_summary(_isolate_stats_file):
+    import session_stats
+
+    session_stats.init()
+    session_stats.record_add("architecture_decisions")
+    session_stats.record_add("task_learnings")
+    session_stats.record_search()
+    session_stats.record_search()
+    session_stats.record_search()
+
+    result = session_stats.report()
+    assert "wrote 2 memories" in result
+    assert "retrieved 3" in result
+    assert "architecture_decisions" in result
+    assert "task_learnings" in result
+
+
+def test_report_empty_session(_isolate_stats_file):
+    import session_stats
+
+    session_stats.init()
+    result = session_stats.report()
+    assert result == ""
+
+
+def test_report_cleans_up_file(_isolate_stats_file):
+    import session_stats
+
+    session_stats.init()
+    session_stats.record_add()
+    session_stats.report()
+    assert not os.path.isfile(_isolate_stats_file)
+
+
+def test_record_add_no_category(_isolate_stats_file):
+    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["adds"] == 2
+    assert data["categories"] == []
+
+
+def test_duplicate_categories_not_added(_isolate_stats_file):
+    import session_stats
+
+    session_stats.init()
+    session_stats.record_add("bug_fixes")
+    session_stats.record_add("bug_fixes")
+    session_stats.record_add("bug_fixes")
+
+    with open(_isolate_stats_file) as f:
+        data = json.load(f)
+    assert data["adds"] == 3
+    assert data["categories"] == ["bug_fixes"]
+
+
+def test_cli_init(tmp_path):
+    """Test CLI invocation: session_stats.py init."""
+    env = {**os.environ, "USER": "test"}
+    result = subprocess.run(
+        [sys.executable, os.path.join(SCRIPTS_DIR, "session_stats.py"), "init"],
+        capture_output=True,
+        text=True,
+        env=env,
+    )
+    assert result.returncode == 0
+
+
+def test_cli_report_no_data(tmp_path):
+    """Test CLI invocation: report with no prior init prints fallback."""
+    env = {**os.environ, "USER": f"test_{os.getpid()}"}
+    result = subprocess.run(
+        [sys.executable, os.path.join(SCRIPTS_DIR, "session_stats.py"), "report"],
+        capture_output=True,
+        text=True,
+        env=env,
+    )
+    assert result.returncode == 0