starmen1265
c61296a24e
- Add TestUserContext: tests for UserContext dataclass (creation, to_dict, from_dict) - Add TestUserContextManager: tests for manager initialization and load - Add TestAtomicWrites: tests for atomic save with backup rotation - Add TestMultiUser: tests for multi-user isolation and convenience methods - Add tmp_users_root and user_context_manager fixtures to conftest.py - Export UserContext and UserContextManager from __init__.py Note: Implementation was already done in commit 9bd29a2 as part of 04-02 plan. This commit adds the tests that should have been part of 04-01 TDD workflow. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
390 lines
15 KiB
Python
390 lines
15 KiB
Python
"""Tests for UserContextManager - per-user project context tracking."""
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from gsd_openclaw.user_context import UserContext, UserContextManager
|
|
|
|
|
|
class TestUserContext:
|
|
"""Tests for UserContext dataclass."""
|
|
|
|
def test_create_with_all_fields(self) -> None:
|
|
"""UserContext can be created with all fields."""
|
|
context = UserContext(
|
|
schema_version="1.0",
|
|
user_id="user-123",
|
|
current_project="my-project",
|
|
recent_projects=["proj1", "proj2"],
|
|
last_active_at="2026-03-11T10:00:00Z",
|
|
preferences={"theme": "dark"}
|
|
)
|
|
assert context.schema_version == "1.0"
|
|
assert context.user_id == "user-123"
|
|
assert context.current_project == "my-project"
|
|
assert context.recent_projects == ["proj1", "proj2"]
|
|
assert context.last_active_at == "2026-03-11T10:00:00Z"
|
|
assert context.preferences == {"theme": "dark"}
|
|
|
|
def test_to_dict_returns_complete_dict(self) -> None:
|
|
"""UserContext.to_dict() returns complete dict with all fields."""
|
|
context = UserContext(
|
|
schema_version="1.0",
|
|
user_id="user-456",
|
|
current_project="test-project",
|
|
recent_projects=["a", "b", "c"],
|
|
last_active_at="2026-03-11T11:00:00Z",
|
|
preferences={"key": "value"}
|
|
)
|
|
result = context.to_dict()
|
|
assert isinstance(result, dict)
|
|
assert result["schema_version"] == "1.0"
|
|
assert result["user_id"] == "user-456"
|
|
assert result["current_project"] == "test-project"
|
|
assert result["recent_projects"] == ["a", "b", "c"]
|
|
assert result["last_active_at"] == "2026-03-11T11:00:00Z"
|
|
assert result["preferences"] == {"key": "value"}
|
|
|
|
def test_from_dict_reconstructs_from_dict(self) -> None:
|
|
"""UserContext.from_dict() reconstructs from dict, ignoring unknown fields."""
|
|
data = {
|
|
"schema_version": "1.0",
|
|
"user_id": "user-789",
|
|
"current_project": "reconstructed-project",
|
|
"recent_projects": ["x", "y"],
|
|
"last_active_at": "2026-03-11T12:00:00Z",
|
|
"preferences": {"setting": True},
|
|
"unknown_field": "should be ignored",
|
|
"another_unknown": 12345
|
|
}
|
|
context = UserContext.from_dict(data)
|
|
assert context.schema_version == "1.0"
|
|
assert context.user_id == "user-789"
|
|
assert context.current_project == "reconstructed-project"
|
|
assert context.recent_projects == ["x", "y"]
|
|
assert context.last_active_at == "2026-03-11T12:00:00Z"
|
|
assert context.preferences == {"setting": True}
|
|
# Unknown fields should not cause issues
|
|
assert not hasattr(context, "unknown_field")
|
|
|
|
|
|
class TestUserContextManager:
|
|
"""Tests for UserContextManager class."""
|
|
|
|
def test_init_sets_users_root_and_creates_context_path(self, tmp_users_root: Path) -> None:
|
|
"""UserContextManager.__init__ sets users_root and creates context_path correctly."""
|
|
manager = UserContextManager("user-abc", tmp_users_root)
|
|
assert manager.users_root == tmp_users_root
|
|
assert manager.context_path == tmp_users_root / "user-abc" / "context.json"
|
|
|
|
def test_init_with_default_users_root(self) -> None:
|
|
"""UserContextManager uses default USERS_ROOT when not specified."""
|
|
manager = UserContextManager("user-default")
|
|
expected_root = Path.home() / "gsd-workspaces" / ".users"
|
|
assert manager.users_root == expected_root
|
|
assert manager.context_path == expected_root / "user-default" / "context.json"
|
|
|
|
def test_load_returns_none_for_new_users(self, tmp_users_root: Path) -> None:
|
|
"""UserContextManager.load() returns None for new users (no existing context)."""
|
|
manager = UserContextManager("new-user-no-context", tmp_users_root)
|
|
result = manager.load()
|
|
assert result is None
|
|
|
|
def test_load_returns_usercontext_for_existing_file(self, tmp_users_root: Path) -> None:
|
|
"""UserContextManager.load() returns UserContext for existing context file."""
|
|
user_id = "existing-user"
|
|
context_path = tmp_users_root / user_id / "context.json"
|
|
context_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Write a valid context file
|
|
data = {
|
|
"schema_version": "1.0",
|
|
"user_id": user_id,
|
|
"current_project": "loaded-project",
|
|
"recent_projects": ["proj-a"],
|
|
"last_active_at": "2026-03-11T09:00:00Z",
|
|
"preferences": {}
|
|
}
|
|
with open(context_path, "w", encoding="utf-8") as f:
|
|
json.dump(data, f)
|
|
|
|
manager = UserContextManager(user_id, tmp_users_root)
|
|
result = manager.load()
|
|
|
|
assert result is not None
|
|
assert isinstance(result, UserContext)
|
|
assert result.user_id == user_id
|
|
assert result.current_project == "loaded-project"
|
|
|
|
|
|
class TestAtomicWrites:
|
|
"""Tests for atomic write pattern in UserContextManager.save()."""
|
|
|
|
def test_save_creates_file_atomically(self, tmp_users_root: Path) -> None:
|
|
"""save() creates file atomically (no temp files remain after save)."""
|
|
manager = UserContextManager("atomic-user", tmp_users_root)
|
|
context = UserContext(
|
|
user_id="atomic-user",
|
|
current_project="atomic-test",
|
|
last_active_at="2026-03-11T10:00:00Z"
|
|
)
|
|
manager.save(context)
|
|
|
|
# Main file should exist
|
|
assert manager.context_path.exists()
|
|
|
|
# No temp files should remain
|
|
temp_files = list(manager.context_path.parent.glob("*.tmp"))
|
|
assert len(temp_files) == 0
|
|
|
|
def test_save_creates_parent_directories(self, tmp_users_root: Path) -> None:
|
|
"""save() creates parent directories if missing."""
|
|
manager = UserContextManager("nested-user", tmp_users_root)
|
|
assert not manager.context_path.parent.exists()
|
|
|
|
context = UserContext(user_id="nested-user", last_active_at="2026-03-11T10:00:00Z")
|
|
manager.save(context)
|
|
|
|
assert manager.context_path.parent.exists()
|
|
assert manager.context_path.exists()
|
|
|
|
def test_save_rotates_backup(self, tmp_users_root: Path) -> None:
|
|
"""save() rotates backup (context.json -> context.json.bak)."""
|
|
manager = UserContextManager("backup-user", tmp_users_root)
|
|
|
|
# First save
|
|
context1 = UserContext(
|
|
user_id="backup-user",
|
|
current_project="first-project",
|
|
last_active_at="2026-03-11T10:00:00Z"
|
|
)
|
|
manager.save(context1)
|
|
|
|
# No backup yet
|
|
backup_path = manager.context_path.with_suffix(".json.bak")
|
|
assert not backup_path.exists()
|
|
|
|
# Second save
|
|
context2 = UserContext(
|
|
user_id="backup-user",
|
|
current_project="second-project",
|
|
last_active_at="2026-03-11T11:00:00Z"
|
|
)
|
|
manager.save(context2)
|
|
|
|
# Backup should now exist
|
|
assert backup_path.exists()
|
|
|
|
# Backup should contain first project
|
|
with open(backup_path, "r", encoding="utf-8") as f:
|
|
backup_data = json.load(f)
|
|
assert backup_data["current_project"] == "first-project"
|
|
|
|
def test_save_overwrites_old_backup(self, tmp_users_root: Path) -> None:
|
|
"""save() overwrites old backup on subsequent saves."""
|
|
manager = UserContextManager("overwrite-user", tmp_users_root)
|
|
backup_path = manager.context_path.with_suffix(".json.bak")
|
|
|
|
# Save three times
|
|
for i, project in enumerate(["proj1", "proj2", "proj3"]):
|
|
context = UserContext(
|
|
user_id="overwrite-user",
|
|
current_project=project,
|
|
last_active_at=f"2026-03-11T{10+i}:00:00Z"
|
|
)
|
|
manager.save(context)
|
|
|
|
# Backup should contain proj2 (the one before last)
|
|
with open(backup_path, "r", encoding="utf-8") as f:
|
|
backup_data = json.load(f)
|
|
assert backup_data["current_project"] == "proj2"
|
|
|
|
# Main file should contain proj3
|
|
with open(manager.context_path, "r", encoding="utf-8") as f:
|
|
main_data = json.load(f)
|
|
assert main_data["current_project"] == "proj3"
|
|
|
|
def test_save_load_roundtrip_preserves_data(self, tmp_users_root: Path) -> None:
|
|
"""save/load roundtrip preserves all data."""
|
|
manager = UserContextManager("roundtrip-user", tmp_users_root)
|
|
|
|
original = UserContext(
|
|
schema_version="1.0",
|
|
user_id="roundtrip-user",
|
|
current_project="roundtrip-project",
|
|
recent_projects=["r1", "r2", "r3"],
|
|
last_active_at="2026-03-11T12:30:45Z",
|
|
preferences={"setting1": "value1", "setting2": 42}
|
|
)
|
|
manager.save(original)
|
|
|
|
loaded = manager.load()
|
|
assert loaded is not None
|
|
assert loaded.schema_version == original.schema_version
|
|
assert loaded.user_id == original.user_id
|
|
assert loaded.current_project == original.current_project
|
|
assert loaded.recent_projects == original.recent_projects
|
|
assert loaded.last_active_at == original.last_active_at
|
|
assert loaded.preferences == original.preferences
|
|
|
|
def test_load_falls_back_to_backup_when_main_corrupted(self, tmp_users_root: Path) -> None:
|
|
"""load() falls back to backup when main file is corrupted."""
|
|
manager = UserContextManager("fallback-user", tmp_users_root)
|
|
|
|
# Create valid context and save twice to create backup
|
|
context = UserContext(
|
|
user_id="fallback-user",
|
|
current_project="first-project",
|
|
last_active_at="2026-03-11T09:00:00Z"
|
|
)
|
|
manager.save(context)
|
|
|
|
# Save again to create backup (backup only exists after second save)
|
|
context.current_project = "valid-project"
|
|
context.last_active_at = "2026-03-11T10:00:00Z"
|
|
manager.save(context)
|
|
|
|
# Backup should now exist with "first-project"
|
|
backup_path = manager.context_path.with_suffix(".json.bak")
|
|
assert backup_path.exists()
|
|
|
|
# Corrupt the main file
|
|
with open(manager.context_path, "w", encoding="utf-8") as f:
|
|
f.write("not valid json {{{")
|
|
|
|
# Load should fall back to backup
|
|
loaded = manager.load()
|
|
assert loaded is not None
|
|
assert loaded.current_project == "first-project"
|
|
|
|
|
|
class TestMultiUser:
|
|
"""Tests for multi-user isolation."""
|
|
|
|
def test_two_users_have_isolated_context_files(self, tmp_users_root: Path) -> None:
|
|
"""Two users have completely isolated context files."""
|
|
manager1 = UserContextManager("user-alice", tmp_users_root)
|
|
manager2 = UserContextManager("user-bob", tmp_users_root)
|
|
|
|
# Different context paths
|
|
assert manager1.context_path != manager2.context_path
|
|
assert "user-alice" in str(manager1.context_path)
|
|
assert "user-bob" in str(manager2.context_path)
|
|
|
|
def test_changes_to_one_user_do_not_affect_another(self, tmp_users_root: Path) -> None:
|
|
"""Changes to one user don't affect another."""
|
|
manager1 = UserContextManager("user-isolated-1", tmp_users_root)
|
|
manager2 = UserContextManager("user-isolated-2", tmp_users_root)
|
|
|
|
# Set context for user 1
|
|
context1 = UserContext(
|
|
user_id="user-isolated-1",
|
|
current_project="alice-project",
|
|
recent_projects=["alice-recent"],
|
|
last_active_at="2026-03-11T10:00:00Z"
|
|
)
|
|
manager1.save(context1)
|
|
|
|
# Set context for user 2
|
|
context2 = UserContext(
|
|
user_id="user-isolated-2",
|
|
current_project="bob-project",
|
|
recent_projects=["bob-recent"],
|
|
last_active_at="2026-03-11T11:00:00Z"
|
|
)
|
|
manager2.save(context2)
|
|
|
|
# Load both and verify isolation
|
|
loaded1 = manager1.load()
|
|
loaded2 = manager2.load()
|
|
|
|
assert loaded1 is not None
|
|
assert loaded2 is not None
|
|
assert loaded1.current_project == "alice-project"
|
|
assert loaded2.current_project == "bob-project"
|
|
assert loaded1.recent_projects == ["alice-recent"]
|
|
assert loaded2.recent_projects == ["bob-recent"]
|
|
|
|
def test_get_current_project_returns_current_or_none(self, tmp_users_root: Path) -> None:
|
|
"""get_current_project() returns current_project or None."""
|
|
manager = UserContextManager("get-project-user", tmp_users_root)
|
|
|
|
# New user has no context
|
|
assert manager.get_current_project() is None
|
|
|
|
# Set a project
|
|
context = UserContext(
|
|
user_id="get-project-user",
|
|
current_project="my-current-project",
|
|
last_active_at="2026-03-11T10:00:00Z"
|
|
)
|
|
manager.save(context)
|
|
|
|
assert manager.get_current_project() == "my-current-project"
|
|
|
|
def test_set_current_project_updates_project_and_recent(self, tmp_users_root: Path) -> None:
|
|
"""set_current_project() updates current_project and adds to recent_projects."""
|
|
manager = UserContextManager("set-project-user", tmp_users_root)
|
|
|
|
manager.set_current_project("new-project")
|
|
|
|
loaded = manager.load()
|
|
assert loaded is not None
|
|
assert loaded.current_project == "new-project"
|
|
assert "new-project" in loaded.recent_projects
|
|
|
|
def test_set_current_project_maintains_max_five_recent(self, tmp_users_root: Path) -> None:
|
|
"""set_current_project() maintains only last 5 recent projects."""
|
|
manager = UserContextManager("max-five-user", tmp_users_root)
|
|
|
|
# Add 7 projects
|
|
projects = [f"project-{i}" for i in range(7)]
|
|
for project in projects:
|
|
manager.set_current_project(project)
|
|
|
|
loaded = manager.load()
|
|
assert loaded is not None
|
|
|
|
# Should have exactly 5 recent projects
|
|
assert len(loaded.recent_projects) == 5
|
|
|
|
# Should be the most recent 5 (2-6), excluding project-0 and project-1
|
|
expected = ["project-6", "project-5", "project-4", "project-3", "project-2"]
|
|
assert loaded.recent_projects == expected
|
|
|
|
def test_get_recent_projects_returns_list(self, tmp_users_root: Path) -> None:
|
|
"""get_recent_projects() returns list of recent projects."""
|
|
manager = UserContextManager("recent-list-user", tmp_users_root)
|
|
|
|
# New user has empty list
|
|
assert manager.get_recent_projects() == []
|
|
|
|
# Add projects
|
|
manager.set_current_project("first")
|
|
manager.set_current_project("second")
|
|
|
|
recent = manager.get_recent_projects()
|
|
assert "second" in recent
|
|
assert "first" in recent
|
|
|
|
def test_new_user_gets_default_context_on_first_access(self, tmp_users_root: Path) -> None:
|
|
"""New user gets default context created on first access."""
|
|
manager = UserContextManager("default-context-user", tmp_users_root)
|
|
|
|
# No context file exists yet
|
|
assert not manager.context_path.exists()
|
|
|
|
# set_current_project should create default context
|
|
manager.set_current_project("first-project")
|
|
|
|
# Now context should exist with proper defaults
|
|
loaded = manager.load()
|
|
assert loaded is not None
|
|
assert loaded.schema_version == "1.0"
|
|
assert loaded.user_id == "default-context-user"
|
|
assert loaded.current_project == "first-project"
|
|
assert loaded.recent_projects == ["first-project"]
|
|
assert loaded.preferences == {}
|
|
assert loaded.last_active_at != "" # Should be set
|