test(04-01): add comprehensive tests for UserContextManager
- 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>
This commit is contained in:
@@ -14,6 +14,7 @@ from gsd_openclaw.command_executor import CommandResult, OutputCapture, CommandE
|
||||
from gsd_openclaw.approval_handler import ApprovalHandler, ApprovalState
|
||||
from gsd_openclaw.progress_monitor import ProgressMonitor, ProgressInfo
|
||||
from gsd_openclaw.orchestrator import Orchestrator
|
||||
from gsd_openclaw.user_context import UserContext, UserContextManager
|
||||
|
||||
__all__ = [
|
||||
"SessionState",
|
||||
@@ -33,4 +34,6 @@ __all__ = [
|
||||
"ProgressMonitor",
|
||||
"ProgressInfo",
|
||||
"Orchestrator",
|
||||
"UserContext",
|
||||
"UserContextManager",
|
||||
]
|
||||
|
||||
@@ -90,3 +90,32 @@ def sample_state() -> SessionState:
|
||||
last_output_snapshot="Test output snapshot",
|
||||
display_name="Test Project Display Name"
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tmp_users_root(tmp_path: Path) -> Path:
|
||||
"""Temporary root for user context testing.
|
||||
|
||||
This ensures tests don't pollute the real ~/gsd-workspaces/.users directory.
|
||||
|
||||
Args:
|
||||
tmp_path: Pytest's built-in temporary directory fixture.
|
||||
|
||||
Returns:
|
||||
Path to a temporary .users directory.
|
||||
"""
|
||||
return tmp_path / "gsd-workspaces" / ".users"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def user_context_manager(tmp_users_root: Path) -> "UserContextManager":
|
||||
"""UserContextManager with temporary root for isolation.
|
||||
|
||||
Args:
|
||||
tmp_users_root: Fixture providing temporary user context root.
|
||||
|
||||
Returns:
|
||||
UserContextManager instance configured for testing.
|
||||
"""
|
||||
from gsd_openclaw.user_context import UserContextManager
|
||||
return UserContextManager("test-user", tmp_users_root)
|
||||
|
||||
@@ -0,0 +1,389 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user