| 1 | from __future__ import annotations |
| 2 | |
| 3 | import os |
| 4 | import shutil |
| 5 | import subprocess |
| 6 | import sys |
| 7 | import threading |
| 8 | import uuid |
| 9 | from pathlib import Path |
| 10 | from types import ModuleType, SimpleNamespace |
| 11 | |
| 12 | import pytest |
| 13 | |
| 14 | PROJECT_ROOT = Path(__file__).resolve().parents[1] |
| 15 | if str(PROJECT_ROOT) not in sys.path: |
| 16 | sys.path.insert(0, str(PROJECT_ROOT)) |
| 17 | |
| 18 | from plugins._time_travel.helpers import time_travel as tt |
| 19 | from plugins._time_travel.helpers.time_travel import ( |
| 20 | TimeTravelConflictError, |
| 21 | TimeTravelError, |
| 22 | TimeTravelService, |
| 23 | WorkspaceRejectedError, |
| 24 | _workspace_from_display, |
| 25 | resolve_workspace, |
| 26 | ) |
| 27 | |
| 28 | |
| 29 | def run_git(repo_dir: Path, *args: str, check: bool = True) -> str: |
| 30 | completed = subprocess.run( |
| 31 | ["git", "-C", str(repo_dir), *args], |
| 32 | check=check, |
| 33 | text=True, |
| 34 | capture_output=True, |
| 35 | ) |
| 36 | return completed.stdout.strip() |
| 37 | |
| 38 | |
| 39 | @pytest.fixture |
| 40 | def workspace(): |
| 41 | name = f"tt-{uuid.uuid4().hex}" |
| 42 | root = PROJECT_ROOT / "usr" / "time-travel-tests" / name |
| 43 | root.mkdir(parents=True) |
| 44 | service = TimeTravelService(_workspace_from_display(f"/a0/usr/time-travel-tests/{name}")) |
| 45 | try: |
| 46 | yield root, service |
| 47 | finally: |
| 48 | shutil.rmtree(root, ignore_errors=True) |
| 49 | shutil.rmtree(service.workspace.shadow_path, ignore_errors=True) |
| 50 | |
| 51 | |
| 52 | def tracked_paths(service: TimeTravelService, commit_hash: str = "HEAD") -> set[str]: |
| 53 | output = service._git("ls-tree", "-r", "--name-only", commit_hash).stdout |
| 54 | return {line.strip() for line in output.splitlines() if line.strip()} |
| 55 | |
| 56 | |
| 57 | def test_shadow_history_snapshot_diff_travel_preserve_refs_and_root_revert(workspace): |
| 58 | root, service = workspace |
| 59 | (root / "a.txt").write_text("one\n", encoding="utf-8") |
| 60 | |
| 61 | initial = service.snapshot(trigger="manual", metadata={"context_id": "ctx"}) |
| 62 | duplicate = service.snapshot(trigger="manual") |
| 63 | |
| 64 | assert initial.created is True |
| 65 | assert duplicate.created is False |
| 66 | assert duplicate.hash == initial.hash |
| 67 | assert (service.workspace.repo_git_path / "objects").is_dir() |
| 68 | |
| 69 | (root / "a.txt").write_text("one\ntwo\n", encoding="utf-8") |
| 70 | present = service.present_summary() |
| 71 | assert present["dirty"] is True |
| 72 | assert present["files"][0]["path"] == "a.txt" |
| 73 | assert "+two" in service.history_diff(commit_hash=initial.hash, path="a.txt", mode="present")["patch"] |
| 74 | |
| 75 | second = service.snapshot(trigger="tool", metadata={"tool_name": "code_execution_tool"}) |
| 76 | history = service.history_list(limit=10) |
| 77 | assert [commit["hash"] for commit in history["commits"][:2]] == [second.hash, initial.hash] |
| 78 | assert history["commits"][0]["metadata"]["tool_name"] == "code_execution_tool" |
| 79 | assert "+two" in service.history_diff(commit_hash=second.hash, path="a.txt", mode="commit")["patch"] |
| 80 | |
| 81 | service.travel(commit_hash=initial.hash) |
| 82 | assert (root / "a.txt").read_text(encoding="utf-8") == "one\n" |
| 83 | preserved = service._git( |
| 84 | "for-each-ref", |
| 85 | "--format=%(objectname)", |
| 86 | "refs/a0-time-travel/preserved", |
| 87 | ).stdout |
| 88 | assert second.hash in preserved |
| 89 | assert second.hash in [commit["hash"] for commit in service.history_list(limit=10)["commits"]] |
| 90 | |
| 91 | reverted = service.revert(commit_hash=initial.hash) |
| 92 | assert reverted["ok"] is True |
| 93 | assert not (root / "a.txt").exists() |
| 94 | assert reverted["snapshot"]["created"] is True |
| 95 | |
| 96 | |
| 97 | def test_revert_conflict_auto_snapshots_present_without_losing_changes(workspace): |
| 98 | root, service = workspace |
| 99 | (root / "a.txt").write_text("one\n", encoding="utf-8") |
| 100 | first = service.snapshot(trigger="manual") |
| 101 | (root / "a.txt").write_text("one\ntwo\n", encoding="utf-8") |
| 102 | second = service.snapshot(trigger="manual") |
| 103 | (root / "a.txt").write_text("custom\n", encoding="utf-8") |
| 104 | |
| 105 | with pytest.raises(TimeTravelConflictError): |
| 106 | service.revert(commit_hash=second.hash) |
| 107 | |
| 108 | assert (root / "a.txt").read_text(encoding="utf-8") == "custom\n" |
| 109 | assert service.current_hash() not in {first.hash, second.hash} |
| 110 | assert "custom" in service.history_diff(commit_hash=service.current_hash(), path="a.txt", mode="commit")["patch"] |
| 111 | |
| 112 | |
| 113 | def test_kernel_boundary_real_git_repo_and_git_dir_exclusion(workspace): |
| 114 | root, service = workspace |
| 115 | with pytest.raises(WorkspaceRejectedError): |
| 116 | _workspace_from_display("/tmp/outside") |
| 117 | |
| 118 | run_git(root, "init") |
| 119 | run_git(root, "config", "user.name", "Test User") |
| 120 | run_git(root, "config", "user.email", "test@example.com") |
| 121 | (root / "tracked.txt").write_text("tracked\n", encoding="utf-8") |
| 122 | run_git(root, "add", "tracked.txt") |
| 123 | run_git(root, "commit", "-m", "real initial") |
| 124 | real_head = run_git(root, "rev-parse", "HEAD") |
| 125 | (root / "untracked.txt").write_text("shadow only\n", encoding="utf-8") |
| 126 | real_status_before = run_git(root, "status", "--short") |
| 127 | |
| 128 | snapshot = service.snapshot(trigger="manual") |
| 129 | |
| 130 | assert snapshot.created is True |
| 131 | assert run_git(root, "rev-parse", "HEAD") == real_head |
| 132 | assert run_git(root, "status", "--short") == real_status_before |
| 133 | assert all(not path.startswith(".git/") and path != ".git" for path in tracked_paths(service)) |
| 134 | assert "tracked.txt" in tracked_paths(service, snapshot.hash) |
| 135 | assert "untracked.txt" in tracked_paths(service, snapshot.hash) |
| 136 | |
| 137 | |
| 138 | def test_snapshot_force_adds_curated_paths_ignored_by_workspace_gitignore(workspace): |
| 139 | root, service = workspace |
| 140 | (root / ".gitignore").write_text("ignored.txt\nignored-dir/\n.env\n", encoding="utf-8") |
| 141 | (root / "ignored.txt").write_text("still important\n", encoding="utf-8") |
| 142 | (root / "ignored-dir").mkdir() |
| 143 | (root / "ignored-dir" / "note.txt").write_text("nested\n", encoding="utf-8") |
| 144 | (root / ".env").write_text("SECRET=still excluded\n", encoding="utf-8") |
| 145 | |
| 146 | snapshot = service.snapshot(trigger="manual") |
| 147 | paths = tracked_paths(service, snapshot.hash) |
| 148 | |
| 149 | assert ".gitignore" in paths |
| 150 | assert "ignored.txt" in paths |
| 151 | assert "ignored-dir/note.txt" in paths |
| 152 | assert ".env" not in paths |
| 153 | |
| 154 | |
| 155 | def test_shadow_repo_empty_head_is_repaired_without_losing_history(workspace): |
| 156 | root, service = workspace |
| 157 | (root / "a.txt").write_text("one\n", encoding="utf-8") |
| 158 | first = service.snapshot(trigger="manual") |
| 159 | (service.workspace.repo_git_path / "HEAD").write_text("", encoding="utf-8") |
| 160 | |
| 161 | (root / "a.txt").write_text("one\ntwo\n", encoding="utf-8") |
| 162 | second = service.snapshot(trigger="manual") |
| 163 | |
| 164 | assert second.created is True |
| 165 | assert service.current_hash() == second.hash |
| 166 | assert [commit["hash"] for commit in service.history_list(limit=10)["commits"][:2]] == [ |
| 167 | second.hash, |
| 168 | first.hash, |
| 169 | ] |
| 170 | |
| 171 | |
| 172 | def test_workspace_identity_canonicalizes_symlink_aliases(): |
| 173 | name = f"tt-{uuid.uuid4().hex}" |
| 174 | root = PROJECT_ROOT / "usr" / "time-travel-tests" / name |
| 175 | target = root / "target" |
| 176 | alias = root / "alias" |
| 177 | target.mkdir(parents=True) |
| 178 | os.symlink(target, alias) |
| 179 | |
| 180 | target_workspace = _workspace_from_display(f"/a0/usr/time-travel-tests/{name}/target") |
| 181 | alias_workspace = _workspace_from_display(f"/a0/usr/time-travel-tests/{name}/alias") |
| 182 | try: |
| 183 | assert alias_workspace.id == target_workspace.id |
| 184 | assert alias_workspace.display_path == target_workspace.display_path |
| 185 | finally: |
| 186 | shutil.rmtree(root, ignore_errors=True) |
| 187 | shutil.rmtree(target_workspace.shadow_path, ignore_errors=True) |
| 188 | |
| 189 | |
| 190 | def test_usr_root_snapshot_skips_plugins_and_nested_git_projects(tmp_path: Path): |
| 191 | root = tmp_path / "usr" |
| 192 | root.mkdir() |
| 193 | (root / "workdir").mkdir() |
| 194 | (root / "workdir" / "note.txt").write_text("note\n", encoding="utf-8") |
| 195 | (root / "plugins" / "demo").mkdir(parents=True) |
| 196 | (root / "plugins" / "demo" / "plugin.yaml").write_text("name: demo\n", encoding="utf-8") |
| 197 | (root / "projects" / "git-project").mkdir(parents=True) |
| 198 | (root / "projects" / "git-project" / ".git").mkdir() |
| 199 | (root / "projects" / "git-project" / "app.py").write_text("print('tracked elsewhere')\n", encoding="utf-8") |
| 200 | (root / "projects" / "plain-project").mkdir(parents=True) |
| 201 | (root / "projects" / "plain-project" / "app.py").write_text("print('plain')\n", encoding="utf-8") |
| 202 | |
| 203 | paths = set(tt.iter_snapshot_paths(root, display_path="/a0/usr")) |
| 204 | |
| 205 | assert "workdir/note.txt" in paths |
| 206 | assert "projects/plain-project/app.py" in paths |
| 207 | assert "plugins/demo/plugin.yaml" not in paths |
| 208 | assert "projects/git-project/app.py" not in paths |
| 209 | |
| 210 | |
| 211 | def test_metadata_policy_tracks_safe_project_files_and_preserves_exclusions(workspace): |
| 212 | root, service = workspace |
| 213 | (root / "src").mkdir() |
| 214 | (root / "src" / "app.py").write_text("print('one')\n", encoding="utf-8") |
| 215 | (root / ".a0proj" / "instructions").mkdir(parents=True) |
| 216 | (root / ".a0proj" / "knowledge").mkdir(parents=True) |
| 217 | (root / ".a0proj" / "skills" / "demo").mkdir(parents=True) |
| 218 | (root / ".a0proj" / "plugins" / "demo").mkdir(parents=True) |
| 219 | (root / ".a0proj" / "memory").mkdir(parents=True) |
| 220 | (root / "node_modules").mkdir() |
| 221 | (root / "dist").mkdir() |
| 222 | (root / "__pycache__").mkdir() |
| 223 | (root / ".a0proj" / "project.json").write_text("{}", encoding="utf-8") |
| 224 | (root / ".a0proj" / "agents.json").write_text("{}", encoding="utf-8") |
| 225 | (root / ".a0proj" / "instructions" / "one.md").write_text("i\n", encoding="utf-8") |
| 226 | (root / ".a0proj" / "knowledge" / "one.md").write_text("k\n", encoding="utf-8") |
| 227 | (root / ".a0proj" / "skills" / "demo" / "SKILL.md").write_text("s\n", encoding="utf-8") |
| 228 | (root / ".a0proj" / "plugins" / "demo" / "config.json").write_text("{}", encoding="utf-8") |
| 229 | (root / ".a0proj" / "plugins" / "demo" / "presets.yaml").write_text("[]\n", encoding="utf-8") |
| 230 | (root / ".a0proj" / "plugins" / "demo" / "state.json").write_text('{"state": true}\n', encoding="utf-8") |
| 231 | (root / ".a0proj" / "secrets.env").write_text("SECRET=one\n", encoding="utf-8") |
| 232 | (root / ".a0proj" / "variables.env").write_text("VAR=one\n", encoding="utf-8") |
| 233 | (root / ".a0proj" / "memory" / "index.faiss").write_bytes(b"memory") |
| 234 | (root / ".env").write_text("TOKEN=one\n", encoding="utf-8") |
| 235 | (root / "node_modules" / "pkg.js").write_text("pkg\n", encoding="utf-8") |
| 236 | (root / "dist" / "bundle.js").write_text("dist\n", encoding="utf-8") |
| 237 | (root / "__pycache__" / "app.pyc").write_bytes(b"pyc") |
| 238 | |
| 239 | first = service.snapshot(trigger="manual") |
| 240 | paths = tracked_paths(service, first.hash) |
| 241 | |
| 242 | assert "src/app.py" in paths |
| 243 | assert ".a0proj/project.json" in paths |
| 244 | assert ".a0proj/agents.json" in paths |
| 245 | assert ".a0proj/instructions/one.md" in paths |
| 246 | assert ".a0proj/knowledge/one.md" in paths |
| 247 | assert ".a0proj/skills/demo/SKILL.md" in paths |
| 248 | assert ".a0proj/plugins/demo/config.json" in paths |
| 249 | assert ".a0proj/plugins/demo/presets.yaml" in paths |
| 250 | assert ".a0proj/plugins/demo/state.json" not in paths |
| 251 | assert ".a0proj/secrets.env" not in paths |
| 252 | assert ".a0proj/variables.env" not in paths |
| 253 | assert ".a0proj/memory/index.faiss" not in paths |
| 254 | assert ".env" not in paths |
| 255 | assert "node_modules/pkg.js" not in paths |
| 256 | assert "dist/bundle.js" not in paths |
| 257 | assert "__pycache__/app.pyc" not in paths |
| 258 | |
| 259 | (root / "src" / "app.py").write_text("print('two')\n", encoding="utf-8") |
| 260 | (root / ".a0proj" / "secrets.env").write_text("SECRET=two\n", encoding="utf-8") |
| 261 | service.snapshot(trigger="manual") |
| 262 | service.travel(commit_hash=first.hash) |
| 263 | |
| 264 | assert (root / "src" / "app.py").read_text(encoding="utf-8") == "print('one')\n" |
| 265 | assert (root / ".a0proj" / "secrets.env").read_text(encoding="utf-8") == "SECRET=two\n" |
| 266 | |
| 267 | |
| 268 | def test_symlink_entries_are_snapshotted_and_deleted_without_following_targets(workspace, tmp_path: Path): |
| 269 | root, service = workspace |
| 270 | outside = tmp_path / "outside.txt" |
| 271 | outside.write_text("outside\n", encoding="utf-8") |
| 272 | os.symlink(outside, root / "outside-link") |
| 273 | |
| 274 | first = service.snapshot(trigger="manual") |
| 275 | assert "outside-link" in tracked_paths(service, first.hash) |
| 276 | assert service._git("ls-tree", "HEAD", "outside-link").stdout.startswith("120000") |
| 277 | |
| 278 | (root / "outside-link").unlink() |
| 279 | second = service.snapshot(trigger="manual") |
| 280 | assert outside.exists() |
| 281 | |
| 282 | service.travel(commit_hash=first.hash) |
| 283 | assert (root / "outside-link").is_symlink() |
| 284 | assert outside.exists() |
| 285 | |
| 286 | service.travel(commit_hash=second.hash) |
| 287 | assert not (root / "outside-link").exists() |
| 288 | assert outside.exists() |
| 289 | |
| 290 | |
| 291 | def test_pagination_large_diff_and_invalid_inputs(workspace, monkeypatch: pytest.MonkeyPatch): |
| 292 | root, service = workspace |
| 293 | (root / "file.txt").write_text("0\n", encoding="utf-8") |
| 294 | hashes = [service.snapshot(trigger="manual").hash] |
| 295 | for index in range(1, 4): |
| 296 | (root / "file.txt").write_text(("x\n" * index), encoding="utf-8") |
| 297 | hashes.append(service.snapshot(trigger="manual").hash) |
| 298 | |
| 299 | page = service.history_list(limit=2) |
| 300 | assert len(page["commits"]) == 2 |
| 301 | assert page["has_more"] is True |
| 302 | page2 = service.history_list(limit=2, offset=2) |
| 303 | assert page2["commits"][0]["hash"] == hashes[1] |
| 304 | |
| 305 | monkeypatch.setattr(tt, "MAX_RENDERED_PATCH_BYTES", 30) |
| 306 | diff = service.history_diff(commit_hash=hashes[-1], path="file.txt", mode="commit") |
| 307 | assert diff["too_large"] is True |
| 308 | assert len(diff["patch"].encode("utf-8")) <= 30 |
| 309 | |
| 310 | with pytest.raises(TimeTravelError): |
| 311 | service.history_diff(commit_hash="not-a-commit", path="file.txt", mode="commit") |
| 312 | with pytest.raises(TimeTravelError): |
| 313 | service.history_diff(commit_hash=hashes[-1], path="../file.txt", mode="commit") |
| 314 | |
| 315 | |
| 316 | def test_debounced_snapshots_coalesce_to_one_commit(workspace): |
| 317 | root, service = workspace |
| 318 | tt.clear_debounced_snapshots() |
| 319 | try: |
| 320 | (root / "file.txt").write_text("one\n", encoding="utf-8") |
| 321 | tt.schedule_debounced_snapshot( |
| 322 | service.workspace, |
| 323 | trigger="watchdog", |
| 324 | metadata={"source": "watchdog", "changed_path_hints": ["/a0/usr/file.txt"]}, |
| 325 | delay=60, |
| 326 | ) |
| 327 | assert service.current_hash() == "" |
| 328 | |
| 329 | (root / "file.txt").write_text("two\n", encoding="utf-8") |
| 330 | tt.schedule_debounced_snapshot( |
| 331 | service.workspace, |
| 332 | trigger="text_editor_write", |
| 333 | metadata={"source": "text_editor", "changed_path_hints": ["/a0/usr/other.txt"]}, |
| 334 | delay=60, |
| 335 | ) |
| 336 | tt.flush_debounced_snapshots() |
| 337 | |
| 338 | current = service.current_hash() |
| 339 | assert current |
| 340 | commits = service.history_list(limit=10)["commits"] |
| 341 | assert len(commits) == 1 |
| 342 | assert commits[0]["hash"] == current |
| 343 | assert commits[0]["metadata"]["trigger"] == "text_editor_write" |
| 344 | assert commits[0]["metadata"]["source"] == "text_editor" |
| 345 | assert commits[0]["metadata"]["changed_path_hints"] == [ |
| 346 | "/a0/usr/file.txt", |
| 347 | "/a0/usr/other.txt", |
| 348 | ] |
| 349 | assert "two" in service.history_diff(commit_hash=current, path="file.txt", mode="commit")["patch"] |
| 350 | finally: |
| 351 | tt.clear_debounced_snapshots() |
| 352 | |
| 353 | |
| 354 | def test_debounced_snapshot_skips_removed_workspace(workspace, monkeypatch): |
| 355 | root, service = workspace |
| 356 | errors = [] |
| 357 | monkeypatch.setattr(tt.PrintStyle, "error", lambda message: errors.append(message)) |
| 358 | tt.clear_debounced_snapshots() |
| 359 | try: |
| 360 | (root / "file.txt").write_text("one\n", encoding="utf-8") |
| 361 | tt.schedule_debounced_snapshot( |
| 362 | service.workspace, |
| 363 | trigger="watchdog", |
| 364 | metadata={"source": "watchdog"}, |
| 365 | delay=60, |
| 366 | ) |
| 367 | shutil.rmtree(root) |
| 368 | |
| 369 | tt.flush_debounced_snapshots() |
| 370 | |
| 371 | assert errors == [] |
| 372 | assert not service.workspace.repo_git_path.exists() |
| 373 | finally: |
| 374 | tt.clear_debounced_snapshots() |
| 375 | |
| 376 | |
| 377 | def test_workspace_resolution_prefers_project_and_rejects_external_paths(monkeypatch: pytest.MonkeyPatch, workspace): |
| 378 | root, _service = workspace |
| 379 | projects_mod = ModuleType("helpers.projects") |
| 380 | projects_mod.get_context_project_name = lambda _context: "demo" |
| 381 | projects_mod.get_project_folder = lambda _name: str(root) |
| 382 | settings_mod = ModuleType("helpers.settings") |
| 383 | settings_mod.get_settings = lambda: {"workdir_path": "/tmp/not-a0"} |
| 384 | |
| 385 | import helpers |
| 386 | |
| 387 | monkeypatch.setitem(sys.modules, "helpers.projects", projects_mod) |
| 388 | monkeypatch.setitem(sys.modules, "helpers.settings", settings_mod) |
| 389 | monkeypatch.setattr(helpers, "projects", projects_mod, raising=False) |
| 390 | monkeypatch.setattr(helpers, "settings", settings_mod, raising=False) |
| 391 | |
| 392 | resolved = resolve_workspace("ctx", context_loader=lambda _ctxid: SimpleNamespace(id="ctx")) |
| 393 | assert resolved.project_name == "demo" |
| 394 | assert resolved.display_path.startswith("/a0/usr/time-travel-tests/") |
| 395 | |
| 396 | projects_mod.get_context_project_name = lambda _context: "" |
| 397 | with pytest.raises(WorkspaceRejectedError): |
| 398 | resolve_workspace("ctx", context_loader=lambda _ctxid: SimpleNamespace(id="ctx")) |
| 399 | |
| 400 | |
| 401 | def test_selectable_workspaces_list_workdir_first_and_default_to_context_project( |
| 402 | monkeypatch: pytest.MonkeyPatch, |
| 403 | workspace, |
| 404 | ): |
| 405 | root, _service = workspace |
| 406 | (root / "workdir").mkdir() |
| 407 | (root / "demo").mkdir() |
| 408 | (root / "other").mkdir() |
| 409 | |
| 410 | projects_mod = ModuleType("helpers.projects") |
| 411 | projects_mod.get_active_projects_list = lambda: [ |
| 412 | {"name": "demo", "title": "Demo Project", "color": "#336699"}, |
| 413 | {"name": "other", "title": "Other Project", "color": ""}, |
| 414 | ] |
| 415 | projects_mod.get_context_project_name = lambda _context: "demo" |
| 416 | projects_mod.get_project_folder = lambda name: str(root / name) |
| 417 | settings_mod = ModuleType("helpers.settings") |
| 418 | settings_mod.get_settings = lambda: {"workdir_path": str(root / "workdir")} |
| 419 | |
| 420 | import helpers |
| 421 | |
| 422 | monkeypatch.setitem(sys.modules, "helpers.projects", projects_mod) |
| 423 | monkeypatch.setitem(sys.modules, "helpers.settings", settings_mod) |
| 424 | monkeypatch.setattr(helpers, "projects", projects_mod, raising=False) |
| 425 | monkeypatch.setattr(helpers, "settings", settings_mod, raising=False) |
| 426 | |
| 427 | data = tt.list_selectable_workspaces( |
| 428 | "ctx", |
| 429 | context_loader=lambda _ctxid: SimpleNamespace(id="ctx"), |
| 430 | ) |
| 431 | workspaces = data["workspaces"] |
| 432 | demo_workspace = next(item for item in workspaces if item["project_name"] == "demo") |
| 433 | |
| 434 | assert workspaces[0]["kind"] == "workdir" |
| 435 | assert workspaces[0]["display_path"].endswith("/workdir") |
| 436 | assert [item["project_name"] for item in workspaces[1:]] == ["demo", "other"] |
| 437 | assert data["default_workspace_id"] == demo_workspace["id"] |
| 438 | |
| 439 | resolved = resolve_workspace( |
| 440 | "ctx", |
| 441 | workspace_id=demo_workspace["id"], |
| 442 | context_loader=lambda _ctxid: SimpleNamespace(id="ctx"), |
| 443 | ) |
| 444 | |
| 445 | assert resolved.project_name == "demo" |
| 446 | assert resolved.display_path == demo_workspace["display_path"] |
| 447 | |
| 448 | with pytest.raises(WorkspaceRejectedError): |
| 449 | resolve_workspace( |
| 450 | "ctx", |
| 451 | workspace_id="missing", |
| 452 | context_loader=lambda _ctxid: SimpleNamespace(id="ctx"), |
| 453 | ) |
| 454 | |
| 455 | |
| 456 | def test_external_workdir_workspace_option_is_locked(monkeypatch: pytest.MonkeyPatch): |
| 457 | projects_mod = ModuleType("helpers.projects") |
| 458 | projects_mod.get_active_projects_list = lambda: [] |
| 459 | projects_mod.get_context_project_name = lambda _context: "" |
| 460 | settings_mod = ModuleType("helpers.settings") |
| 461 | settings_mod.get_settings = lambda: {"workdir_path": "/tmp/not-a0"} |
| 462 | |
| 463 | import helpers |
| 464 | |
| 465 | monkeypatch.setitem(sys.modules, "helpers.projects", projects_mod) |
| 466 | monkeypatch.setitem(sys.modules, "helpers.settings", settings_mod) |
| 467 | monkeypatch.setattr(helpers, "projects", projects_mod, raising=False) |
| 468 | monkeypatch.setattr(helpers, "settings", settings_mod, raising=False) |
| 469 | |
| 470 | data = tt.list_selectable_workspaces("") |
| 471 | workdir = data["workspaces"][0] |
| 472 | |
| 473 | assert workdir["kind"] == "workdir" |
| 474 | assert workdir["locked"] is True |
| 475 | assert data["default_workspace_id"] == workdir["id"] |
| 476 | |
| 477 | with pytest.raises(WorkspaceRejectedError): |
| 478 | resolve_workspace("", workspace_id=workdir["id"]) |