diff --git a/gsd-openclaw/src/gsd_openclaw/__init__.py b/gsd-openclaw/src/gsd_openclaw/__init__.py index 36e8f2d..a99faec 100644 --- a/gsd-openclaw/src/gsd_openclaw/__init__.py +++ b/gsd-openclaw/src/gsd_openclaw/__init__.py @@ -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", ] diff --git a/gsd-openclaw/tests/conftest.py b/gsd-openclaw/tests/conftest.py index 271c17d..e7300d8 100644 --- a/gsd-openclaw/tests/conftest.py +++ b/gsd-openclaw/tests/conftest.py @@ -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) diff --git a/gsd-openclaw/tests/test_user_context.py b/gsd-openclaw/tests/test_user_context.py new file mode 100644 index 0000000..34a7ad3 --- /dev/null +++ b/gsd-openclaw/tests/test_user_context.py @@ -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