ruizanthonyandAnthony Ruiz
eee72f942e
* fix(workspace): new chat inherits the current conversation's workspace
`newSession()` resolved the workspace of a new chat as:
switchWs || S._profileDefaultWorkspace || S.session.workspace
`S._profileDefaultWorkspace` is global to the profile, hydrated from
`/api/profile/active`. Preferring it over the current conversation meant a new
chat opened from a MES conversation landed on the profile-global value instead
of MES. With several conversations running in parallel on different workspaces,
the context the user was actually in was systematically lost.
The precedence is worse than it looks, because that "profile default" is not a
stable configured value: `get_profile_default_workspace()` reads
`last_workspace.txt` BEFORE the configured key (api/workspace.py ~419-427), and
for the `default` profile that file is the GLOBAL one. `set_last_workspace()`
rewrites it on every `/api/chat/start` and `/api/session/update` of ANY
conversation. So "profile default wins over the current session" degrades in
practice to "whatever workspace another conversation last used wins over the
conversation I am in" -- observed live: the pointer flipped mid-operation
because a concurrent turn wrote to it.
New precedence:
switchWs || S.session.workspace || S._profileDefaultWorkspace
Preserved invariants:
- the one-shot profile-switch flag still wins first and is still consumed (#823);
- a blank page (no loaded session) still falls back to the profile default,
which remains required for the empty-composer workspace chip (#804/#5169);
- `_profileDefaultWorkspace` is never nulled by `newSession()` (#823).
This deliberately diverges from upstream #4755, which made the profile default
win. That decision assumed a stable configured default; on a shared/global
`last_workspace.txt` it inverts its own intent. The upstream test is rewritten
with the rationale inline so a future rebase does not silently restore the bug,
plus a new test that evaluates the real expression under node and proves the
blank-page fallback still holds.
Pre-existing failures in tests/test_workspace_stale_recovery.py (3) were
verified identical with and without this change.
* fix(workspace): recover stale inherited workspace
---------
Co-authored-by: Anthony Ruiz <ruizanthony@users.noreply.github.com>
626 lines
21 KiB
Python
626 lines
21 KiB
Python
from collections import OrderedDict
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from urllib.parse import urlparse
|
|
|
|
import pytest
|
|
|
|
from api import config as api_config
|
|
from api import models, routes, workspace
|
|
|
|
|
|
def _write_session_sidecar(monkeypatch, tmp_path, session):
|
|
session_dir = tmp_path / "sessions"
|
|
session_dir.mkdir(exist_ok=True)
|
|
monkeypatch.setattr(models, "SESSION_DIR", session_dir)
|
|
monkeypatch.setattr(models, "_write_session_index", lambda **_kwargs: None)
|
|
sidecar = session_dir / f"{session.session_id}.json"
|
|
sidecar.write_text(
|
|
json.dumps(
|
|
{
|
|
"session_id": session.session_id,
|
|
"workspace": session.workspace,
|
|
"messages": [{"role": "user", "content": "preserve me"}],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
return sidecar
|
|
|
|
|
|
def test_profile_default_workspace_uses_live_config_default(monkeypatch, tmp_path):
|
|
live_default = tmp_path / "live-default"
|
|
live_default.mkdir()
|
|
|
|
monkeypatch.setattr(api_config, "DEFAULT_WORKSPACE", live_default)
|
|
monkeypatch.setattr(api_config, "get_config", lambda: {})
|
|
|
|
assert workspace._profile_default_workspace() == str(live_default.resolve())
|
|
|
|
|
|
def test_implicit_workspace_recovery_keeps_fallback_lazy(monkeypatch, tmp_path):
|
|
valid = tmp_path / "valid"
|
|
valid.mkdir()
|
|
monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
|
|
|
|
resolved, recovered = workspace.resolve_implicit_workspace_with_recovery(
|
|
valid,
|
|
lambda: (_ for _ in ()).throw(AssertionError("fallback should stay lazy")),
|
|
)
|
|
|
|
assert resolved == valid.resolve()
|
|
assert recovered is False
|
|
|
|
|
|
def test_resolve_chat_workspace_with_recovery_repairs_missing_implicit_workspace(monkeypatch, tmp_path):
|
|
fallback = tmp_path / "fallback"
|
|
fallback.mkdir()
|
|
stale = tmp_path / "deleted-workspace"
|
|
|
|
session = SimpleNamespace(session_id="sess-1", workspace=str(stale))
|
|
sidecar = _write_session_sidecar(monkeypatch, tmp_path, session)
|
|
|
|
monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
|
|
monkeypatch.setattr(workspace, "load_workspaces", lambda: [])
|
|
monkeypatch.setattr(routes, "get_last_workspace", lambda: str(fallback))
|
|
|
|
resolved = routes._resolve_chat_workspace_with_recovery(session, None)
|
|
|
|
assert resolved == str(fallback.resolve())
|
|
assert session.workspace == str(fallback.resolve())
|
|
assert str(fallback.resolve()) in sidecar.read_text(encoding="utf-8")
|
|
|
|
|
|
def test_chat_recovery_persistence_failure_fails_closed(monkeypatch, tmp_path):
|
|
fallback = tmp_path / "fallback"
|
|
fallback.mkdir()
|
|
stale = tmp_path / "deleted-workspace"
|
|
session = SimpleNamespace(
|
|
session_id="sess-chat-save-fails",
|
|
workspace=str(stale),
|
|
save=lambda **_kwargs: (_ for _ in ()).throw(OSError("disk full")),
|
|
)
|
|
|
|
monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
|
|
monkeypatch.setattr(workspace, "load_workspaces", lambda: [])
|
|
monkeypatch.setattr(routes, "get_last_workspace", lambda: str(fallback))
|
|
|
|
with pytest.raises(routes.WorkspaceBindingPersistenceError):
|
|
routes._resolve_chat_workspace_with_recovery(session, None)
|
|
|
|
assert session.workspace == str(stale)
|
|
|
|
|
|
def test_resolve_chat_workspace_with_recovery_preserves_explicit_errors(monkeypatch, tmp_path):
|
|
fallback = tmp_path / "fallback"
|
|
fallback.mkdir()
|
|
stale = tmp_path / "deleted-workspace"
|
|
|
|
def fake_resolve(value):
|
|
if value == str(stale):
|
|
raise ValueError(f"Path does not exist: {stale}")
|
|
return Path(value).resolve()
|
|
|
|
saved = {"count": 0}
|
|
|
|
def fake_save():
|
|
saved["count"] += 1
|
|
|
|
session = SimpleNamespace(session_id="sess-2", workspace=str(fallback), save=fake_save)
|
|
|
|
monkeypatch.setattr(routes, "resolve_trusted_workspace", fake_resolve)
|
|
monkeypatch.setattr(routes, "get_last_workspace", lambda: str(fallback))
|
|
|
|
with pytest.raises(ValueError, match="Path does not exist"):
|
|
routes._resolve_chat_workspace_with_recovery(session, str(stale))
|
|
|
|
assert session.workspace == str(fallback)
|
|
assert saved["count"] == 0
|
|
|
|
|
|
def _post_new_session_with_workspace(
|
|
monkeypatch,
|
|
body,
|
|
previous_session,
|
|
fallback,
|
|
*,
|
|
previous_visible=True,
|
|
):
|
|
import api.session_lifecycle as session_lifecycle
|
|
|
|
captured = {"new": [], "fallback": 0}
|
|
|
|
class _NewSession:
|
|
session_id = "new-session"
|
|
profile = "default"
|
|
messages = []
|
|
|
|
def __init__(self, workspace):
|
|
self.workspace = workspace
|
|
|
|
def compact(self):
|
|
return {
|
|
"session_id": self.session_id,
|
|
"profile": self.profile,
|
|
"workspace": self.workspace,
|
|
}
|
|
|
|
def get_fallback():
|
|
captured["fallback"] += 1
|
|
return str(fallback)
|
|
|
|
def create_session(**kwargs):
|
|
captured["new"].append(kwargs)
|
|
return _NewSession(kwargs["workspace"])
|
|
|
|
monkeypatch.setattr(routes, "read_body", lambda _handler: body)
|
|
monkeypatch.setattr(routes, "_check_csrf", lambda _handler: True)
|
|
monkeypatch.setattr(routes, "_csrf_exempt_path", lambda _path: False)
|
|
monkeypatch.setattr(routes, "_handle_extension_sidecar_proxy", lambda *_a, **_k: False)
|
|
monkeypatch.setattr(routes, "_guard_request_session_visibility", lambda *_a, **_k: True)
|
|
monkeypatch.setattr(
|
|
routes,
|
|
"_session_id_visible_to_request_profile",
|
|
lambda *_a, **_k: previous_visible,
|
|
)
|
|
monkeypatch.setattr(
|
|
routes,
|
|
"get_session",
|
|
lambda sid, metadata_only=False: previous_session
|
|
if sid == body.get("prev_session_id")
|
|
else (_ for _ in ()).throw(KeyError(sid)),
|
|
)
|
|
monkeypatch.setattr(routes, "get_last_workspace", get_fallback)
|
|
monkeypatch.setattr(routes, "_session_model_state_from_request", lambda *_a: (None, None))
|
|
monkeypatch.setattr(routes, "new_session", create_session)
|
|
monkeypatch.setattr(
|
|
session_lifecycle,
|
|
"_register_background_commit_thread",
|
|
lambda _thread: False,
|
|
)
|
|
monkeypatch.setattr(
|
|
routes,
|
|
"j",
|
|
lambda _handler, payload, status=200, **_kwargs: captured.setdefault(
|
|
"response", (status, payload)
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
routes,
|
|
"bad",
|
|
lambda _handler, message, status=400, **_kwargs: captured.setdefault(
|
|
"error", (status, message)
|
|
),
|
|
)
|
|
|
|
handler = SimpleNamespace(command="POST", headers={})
|
|
routes.handle_post(handler, urlparse("/api/session/new"))
|
|
return captured
|
|
|
|
|
|
def test_session_new_recovers_deleted_workspace_inherited_from_visible_previous_session(
|
|
monkeypatch, tmp_path
|
|
):
|
|
fallback = tmp_path / "fallback"
|
|
fallback.mkdir()
|
|
stale = tmp_path / "deleted-workspace"
|
|
previous = SimpleNamespace(
|
|
session_id="previous-session",
|
|
profile="default",
|
|
workspace=str(stale),
|
|
)
|
|
body = {
|
|
"workspace": str(stale),
|
|
"workspace_inherited_from_prev_session": True,
|
|
"prev_session_id": previous.session_id,
|
|
"profile": "default",
|
|
"worktree": False,
|
|
}
|
|
|
|
monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
|
|
monkeypatch.setattr(workspace, "load_workspaces", lambda: [])
|
|
captured = _post_new_session_with_workspace(
|
|
monkeypatch,
|
|
body,
|
|
previous,
|
|
fallback,
|
|
)
|
|
|
|
assert "error" not in captured
|
|
assert captured["response"][0] == 200
|
|
assert captured["response"][1]["session"]["workspace"] == str(fallback.resolve())
|
|
assert captured["new"][0]["workspace"] == str(fallback.resolve())
|
|
assert captured["fallback"] == 1
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("mark_inherited", "previous_visible", "stored_workspace"),
|
|
[
|
|
pytest.param(False, True, "requested", id="explicit-request"),
|
|
pytest.param(True, False, "requested", id="foreign-previous-session"),
|
|
pytest.param(True, True, "different", id="stored-workspace-mismatch"),
|
|
],
|
|
)
|
|
def test_session_new_keeps_deleted_unverified_workspace_strict(
|
|
monkeypatch,
|
|
tmp_path,
|
|
mark_inherited,
|
|
previous_visible,
|
|
stored_workspace,
|
|
):
|
|
fallback = tmp_path / "fallback"
|
|
fallback.mkdir()
|
|
stale = tmp_path / "deleted-workspace"
|
|
previous = SimpleNamespace(
|
|
session_id="previous-session",
|
|
profile="default",
|
|
workspace=(str(stale) if stored_workspace == "requested" else str(tmp_path / "other")),
|
|
)
|
|
body = {
|
|
"workspace": str(stale),
|
|
"prev_session_id": previous.session_id,
|
|
"profile": "default",
|
|
"worktree": False,
|
|
}
|
|
if mark_inherited:
|
|
body["workspace_inherited_from_prev_session"] = True
|
|
|
|
monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
|
|
monkeypatch.setattr(workspace, "load_workspaces", lambda: [])
|
|
captured = _post_new_session_with_workspace(
|
|
monkeypatch,
|
|
body,
|
|
previous,
|
|
fallback,
|
|
previous_visible=previous_visible,
|
|
)
|
|
|
|
assert captured["error"][0] == 400
|
|
assert "does not exist" in captured["error"][1].lower()
|
|
assert captured["new"] == []
|
|
assert captured["fallback"] == 0
|
|
|
|
|
|
def test_chat_recovery_preserves_existing_implicit_trust_error(monkeypatch, tmp_path):
|
|
home = tmp_path / "home"
|
|
fallback = home / "fallback"
|
|
fallback.mkdir(parents=True)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
saved = {"count": 0}
|
|
session = SimpleNamespace(
|
|
session_id="sess-untrusted",
|
|
workspace=str(outside),
|
|
save=lambda: saved.__setitem__("count", saved["count"] + 1),
|
|
)
|
|
|
|
monkeypatch.setattr(workspace, "_home_path", lambda: home)
|
|
monkeypatch.setattr(workspace, "load_workspaces", lambda: [])
|
|
monkeypatch.setattr(workspace, "_BOOT_DEFAULT_WORKSPACE", fallback)
|
|
monkeypatch.setattr(routes, "get_last_workspace", lambda: str(fallback))
|
|
|
|
with pytest.raises(ValueError, match="outside the user home directory"):
|
|
routes._resolve_chat_workspace_with_recovery(session, None)
|
|
|
|
assert session.workspace == str(outside)
|
|
assert saved["count"] == 0
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"terminal_cfg",
|
|
[
|
|
pytest.param(
|
|
{"backend": "ssh", "cwd": "/Users/joeyshiue"},
|
|
id="cwd-absolute",
|
|
),
|
|
pytest.param({"backend": "ssh"}, id="cwd-omitted"),
|
|
pytest.param({"backend": "ssh", "cwd": ""}, id="cwd-empty"),
|
|
pytest.param({"backend": "ssh", "cwd": "."}, id="cwd-dot"),
|
|
],
|
|
)
|
|
def test_chat_recovery_preserves_remote_workspace_rejection(
|
|
monkeypatch, tmp_path, terminal_cfg
|
|
):
|
|
candidate = "/Users/other/projects/demo"
|
|
fallback_path = tmp_path / "fallback"
|
|
fallback_path.mkdir()
|
|
fallback_calls = {"count": 0}
|
|
saved = {"count": 0}
|
|
session = SimpleNamespace(
|
|
session_id="sess-remote-untrusted",
|
|
workspace=candidate,
|
|
save=lambda: saved.__setitem__("count", saved["count"] + 1),
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
api_config,
|
|
"get_config",
|
|
lambda: {"terminal": terminal_cfg},
|
|
)
|
|
monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
|
|
|
|
def fallback():
|
|
fallback_calls["count"] += 1
|
|
return fallback_path
|
|
|
|
monkeypatch.setattr(routes, "get_last_workspace", fallback)
|
|
|
|
with pytest.raises(ValueError, match="Path does not exist"):
|
|
routes._resolve_chat_workspace_with_recovery(session, None)
|
|
|
|
assert fallback_calls["count"] == 0
|
|
assert session.workspace == candidate
|
|
assert saved["count"] == 0
|
|
|
|
|
|
def test_list_dir_recovers_missing_implicit_session_workspace(monkeypatch, tmp_path):
|
|
fallback = tmp_path / "fallback"
|
|
fallback.mkdir()
|
|
stale = tmp_path / "deleted-workspace"
|
|
session = SimpleNamespace(
|
|
session_id="sess-list",
|
|
workspace=str(stale),
|
|
)
|
|
sidecar = _write_session_sidecar(monkeypatch, tmp_path, session)
|
|
captured = {}
|
|
|
|
monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
|
|
monkeypatch.setattr(workspace, "load_workspaces", lambda: [])
|
|
monkeypatch.setattr(routes, "get_session", lambda _sid: session)
|
|
monkeypatch.setattr(routes, "get_last_workspace", lambda: str(fallback))
|
|
|
|
def fake_list_dir(workspace_path, rel_path):
|
|
captured["workspace"] = workspace_path
|
|
captured["rel_path"] = rel_path
|
|
return []
|
|
|
|
monkeypatch.setattr(routes, "list_dir", fake_list_dir)
|
|
monkeypatch.setattr(routes, "dir_signature", lambda *_args: "sig")
|
|
monkeypatch.setattr(routes, "j", lambda _handler, payload, **_kwargs: payload)
|
|
|
|
payload = routes._handle_list_dir(
|
|
object(), urlparse("/api/list?session_id=sess-list&path=.")
|
|
)
|
|
|
|
assert captured == {"workspace": fallback.resolve(), "rel_path": "."}
|
|
assert session.workspace == str(fallback.resolve())
|
|
assert str(fallback.resolve()) in sidecar.read_text(encoding="utf-8")
|
|
assert payload == {
|
|
"entries": [],
|
|
"signature": "sig",
|
|
"path": ".",
|
|
"workspace": str(fallback.resolve()),
|
|
"workspace_recovered": True,
|
|
}
|
|
|
|
|
|
def test_list_recovery_stays_bound_when_global_fallback_changes(
|
|
monkeypatch, tmp_path
|
|
):
|
|
"""A later mutation must use the root that the Files pane displayed."""
|
|
fallback_a = tmp_path / "fallback-a"
|
|
fallback_b = tmp_path / "fallback-b"
|
|
fallback_a.mkdir()
|
|
fallback_b.mkdir()
|
|
stale = tmp_path / "deleted-workspace"
|
|
selected = {"workspace": str(fallback_a)}
|
|
session = SimpleNamespace(
|
|
session_id="sess-authority",
|
|
workspace=str(stale),
|
|
profile=None,
|
|
)
|
|
sidecar = _write_session_sidecar(monkeypatch, tmp_path, session)
|
|
captured = {}
|
|
|
|
monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
|
|
monkeypatch.setattr(workspace, "load_workspaces", lambda: [])
|
|
monkeypatch.setattr(routes, "get_session", lambda _sid, **_kwargs: session)
|
|
monkeypatch.setattr(
|
|
routes, "get_last_workspace", lambda: selected["workspace"]
|
|
)
|
|
def capture_list(workspace_path, _rel):
|
|
captured["listed"] = workspace_path
|
|
return []
|
|
|
|
monkeypatch.setattr(routes, "list_dir", capture_list)
|
|
monkeypatch.setattr(routes, "dir_signature", lambda *_args: "sig")
|
|
monkeypatch.setattr(routes, "j", lambda _handler, payload, **_kwargs: payload)
|
|
|
|
payload = routes._handle_list_dir(
|
|
object(), urlparse("/api/list?session_id=sess-authority&path=.")
|
|
)
|
|
selected["workspace"] = str(fallback_b)
|
|
|
|
assert payload["workspace"] == str(fallback_a.resolve())
|
|
assert session.workspace == str(fallback_a.resolve())
|
|
assert captured["listed"] == fallback_a.resolve()
|
|
assert str(fallback_a.resolve()) in sidecar.read_text(encoding="utf-8")
|
|
|
|
|
|
def test_persisted_list_recovery_anchors_later_create_dir_to_fallback_a(
|
|
monkeypatch, tmp_path
|
|
):
|
|
"""Reproduce the Maintainer's A→B sequence across fresh sidecar loads."""
|
|
session_dir = tmp_path / "sessions"
|
|
session_dir.mkdir()
|
|
fallback_a = tmp_path / "fallback-a"
|
|
fallback_b = tmp_path / "fallback-b"
|
|
fallback_a.mkdir()
|
|
fallback_b.mkdir()
|
|
stale = tmp_path / "deleted-workspace"
|
|
selected = {"workspace": str(fallback_a)}
|
|
sid = "sess-http-authority"
|
|
|
|
monkeypatch.setattr(models, "SESSION_DIR", session_dir)
|
|
monkeypatch.setattr(models, "SESSIONS", OrderedDict())
|
|
monkeypatch.setattr(models, "_write_session_index", lambda **_kwargs: None)
|
|
monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
|
|
monkeypatch.setattr(workspace, "load_workspaces", lambda: [])
|
|
monkeypatch.setattr(routes, "get_last_workspace", lambda: selected["workspace"])
|
|
monkeypatch.setattr(models, "get_last_workspace", lambda: selected["workspace"])
|
|
monkeypatch.setattr(routes, "j", lambda _handler, payload, **_kwargs: payload)
|
|
monkeypatch.setattr(
|
|
routes,
|
|
"bad",
|
|
lambda _handler, message, status=400: {"error": message, "status": status},
|
|
)
|
|
|
|
session = models.Session(
|
|
session_id=sid,
|
|
workspace=str(stale),
|
|
messages=[{"role": "user", "content": "preserve transcript"}],
|
|
)
|
|
session.save(skip_index=True)
|
|
models.SESSIONS.clear()
|
|
|
|
listed = routes._handle_list_dir(
|
|
object(), urlparse(f"/api/list?session_id={sid}&path=.")
|
|
)
|
|
persisted = models.Session.load(sid)
|
|
assert listed["workspace"] == str(fallback_a.resolve())
|
|
assert listed["workspace_recovered"] is True
|
|
assert persisted is not None
|
|
assert persisted.workspace == str(fallback_a.resolve())
|
|
assert persisted.messages == [{"role": "user", "content": "preserve transcript"}]
|
|
|
|
models.SESSIONS.clear()
|
|
selected["workspace"] = str(fallback_b)
|
|
created = routes._handle_create_dir(
|
|
object(), {"session_id": sid, "path": "gate-wrong-root"}
|
|
)
|
|
|
|
assert created == {"ok": True, "path": "gate-wrong-root"}
|
|
assert (fallback_a / "gate-wrong-root").is_dir()
|
|
assert not (fallback_b / "gate-wrong-root").exists()
|
|
|
|
|
|
def test_list_recovery_persistence_failure_fails_closed(monkeypatch, tmp_path):
|
|
fallback = tmp_path / "fallback"
|
|
fallback.mkdir()
|
|
stale = tmp_path / "deleted-workspace"
|
|
calls = {"list_dir": 0}
|
|
session = SimpleNamespace(
|
|
session_id="sess-save-fails",
|
|
workspace=str(stale),
|
|
save=lambda **_kwargs: (_ for _ in ()).throw(OSError("disk full")),
|
|
)
|
|
|
|
monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
|
|
monkeypatch.setattr(workspace, "load_workspaces", lambda: [])
|
|
monkeypatch.setattr(routes, "get_session", lambda _sid, **_kwargs: session)
|
|
monkeypatch.setattr(routes, "get_last_workspace", lambda: str(fallback))
|
|
monkeypatch.setattr(
|
|
routes,
|
|
"list_dir",
|
|
lambda *_args: calls.__setitem__("list_dir", calls["list_dir"] + 1),
|
|
)
|
|
monkeypatch.setattr(
|
|
routes,
|
|
"bad",
|
|
lambda _handler, message, status=400: {"error": message, "status": status},
|
|
)
|
|
|
|
payload = routes._handle_list_dir(
|
|
object(), urlparse("/api/list?session_id=sess-save-fails&path=.")
|
|
)
|
|
|
|
assert payload["status"] >= 400
|
|
assert "persist" in payload["error"].lower()
|
|
assert calls["list_dir"] == 0
|
|
assert session.workspace == str(stale)
|
|
|
|
|
|
def test_list_dir_does_not_recover_unpersistable_cli_workspace(
|
|
monkeypatch, tmp_path
|
|
):
|
|
stale = tmp_path / "deleted-cli-workspace"
|
|
fallback = tmp_path / "fallback"
|
|
fallback.mkdir()
|
|
calls = {"fallback": 0, "list_dir": 0}
|
|
|
|
monkeypatch.setattr(routes, "get_session", lambda _sid: (_ for _ in ()).throw(KeyError(_sid)))
|
|
monkeypatch.setattr(
|
|
routes,
|
|
"get_cli_sessions",
|
|
lambda: [{"session_id": "cli-stale", "workspace": str(stale)}],
|
|
)
|
|
|
|
def get_fallback():
|
|
calls["fallback"] += 1
|
|
return str(fallback)
|
|
|
|
monkeypatch.setattr(routes, "get_last_workspace", get_fallback)
|
|
monkeypatch.setattr(
|
|
routes,
|
|
"list_dir",
|
|
lambda *_args: calls.__setitem__("list_dir", calls["list_dir"] + 1),
|
|
)
|
|
monkeypatch.setattr(
|
|
routes,
|
|
"bad",
|
|
lambda _handler, message, status=400: {"error": message, "status": status},
|
|
)
|
|
|
|
payload = routes._handle_list_dir(
|
|
object(), urlparse("/api/list?session_id=cli-stale&path=.")
|
|
)
|
|
|
|
assert payload["status"] == 404
|
|
assert calls == {"fallback": 0, "list_dir": 0}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"terminal_cfg",
|
|
[
|
|
pytest.param(
|
|
{"backend": "ssh", "cwd": "/Users/joeyshiue"},
|
|
id="cwd-absolute",
|
|
),
|
|
pytest.param({"backend": "ssh"}, id="cwd-omitted"),
|
|
pytest.param({"backend": "ssh", "cwd": ""}, id="cwd-empty"),
|
|
pytest.param({"backend": "ssh", "cwd": "."}, id="cwd-dot"),
|
|
],
|
|
)
|
|
def test_list_dir_preserves_remote_workspace_rejection(
|
|
monkeypatch, tmp_path, terminal_cfg
|
|
):
|
|
candidate = "/Users/other/projects/demo"
|
|
fallback_path = tmp_path / "fallback"
|
|
fallback_path.mkdir()
|
|
session = SimpleNamespace(session_id="sess-list-remote", workspace=candidate)
|
|
calls = {"fallback": 0, "list_dir": 0}
|
|
|
|
monkeypatch.setattr(
|
|
api_config,
|
|
"get_config",
|
|
lambda: {"terminal": terminal_cfg},
|
|
)
|
|
monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
|
|
monkeypatch.setattr(routes, "get_session", lambda _sid: session)
|
|
|
|
def fallback():
|
|
calls["fallback"] += 1
|
|
return fallback_path
|
|
|
|
def fake_list_dir(*_args):
|
|
calls["list_dir"] += 1
|
|
return []
|
|
|
|
monkeypatch.setattr(routes, "get_last_workspace", fallback)
|
|
monkeypatch.setattr(routes, "list_dir", fake_list_dir)
|
|
monkeypatch.setattr(
|
|
routes,
|
|
"bad",
|
|
lambda _handler, message, status=400: {"error": message, "status": status},
|
|
)
|
|
|
|
payload = routes._handle_list_dir(
|
|
object(), urlparse("/api/list?session_id=sess-list-remote&path=.")
|
|
)
|
|
|
|
assert isinstance(payload, dict)
|
|
assert payload["status"] == 404
|
|
assert "Path does not exist" in payload["error"]
|
|
assert calls == {"fallback": 0, "list_dir": 0}
|