"""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