feat: validate project paths at create/update
Reject non-existent or non-directory paths with ServiceError (already surfaced as a TUI notification), and store the resolved absolute path. Closes the P2 finding where bad paths became dead sessions later. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -1,3 +1,5 @@
|
||||
from pathlib import Path
|
||||
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
@@ -11,7 +13,20 @@ class ProjectService:
|
||||
# from these methods stay readable after their session closes.
|
||||
self.factory = factory
|
||||
|
||||
def _normalize_path(self, raw: str) -> str:
|
||||
"""Expand ~, require an existing directory, return the resolved absolute path.
|
||||
|
||||
Raises ServiceError if the path does not exist or is not a directory, so
|
||||
the TUI surfaces it as a notification instead of letting the bad path
|
||||
become a dead session later.
|
||||
"""
|
||||
path = Path(raw).expanduser()
|
||||
if not path.is_dir():
|
||||
raise ServiceError(f"Path does not exist or is not a directory: {raw}")
|
||||
return str(path.resolve())
|
||||
|
||||
def create(self, name: str, path: str) -> Project:
|
||||
path = self._normalize_path(path)
|
||||
with self.factory() as db:
|
||||
project = Project(name=name, path=path)
|
||||
db.add(project)
|
||||
@@ -34,6 +49,7 @@ class ProjectService:
|
||||
return db.get(Project, project_id)
|
||||
|
||||
def update(self, project_id: int, name: str, path: str) -> Project:
|
||||
path = self._normalize_path(path)
|
||||
with self.factory() as db:
|
||||
project = db.get(Project, project_id)
|
||||
if project is None:
|
||||
|
||||
+69
-22
@@ -24,48 +24,95 @@ def factory():
|
||||
|
||||
|
||||
class TestProjectService:
|
||||
def test_create_and_get(self, factory):
|
||||
def test_create_and_get(self, factory, tmp_path):
|
||||
svc = ProjectService(factory)
|
||||
p = svc.create("test", "/tmp/test")
|
||||
p = svc.create("test", str(tmp_path))
|
||||
assert p.id is not None
|
||||
assert svc.get(p.id).name == "test"
|
||||
|
||||
def test_list_excludes_archived(self, factory):
|
||||
def test_list_excludes_archived(self, factory, tmp_path):
|
||||
a = tmp_path / "a"
|
||||
a.mkdir()
|
||||
b = tmp_path / "b"
|
||||
b.mkdir()
|
||||
svc = ProjectService(factory)
|
||||
svc.create("a", "/a")
|
||||
p2 = svc.create("b", "/b")
|
||||
svc.create("a", str(a))
|
||||
p2 = svc.create("b", str(b))
|
||||
svc.archive(p2.id)
|
||||
assert len(svc.list_all()) == 1
|
||||
assert len(svc.list_all(include_archived=True)) == 2
|
||||
|
||||
def test_update_name_and_path(self, factory):
|
||||
def test_update_name_and_path(self, factory, tmp_path):
|
||||
old = tmp_path / "old"
|
||||
old.mkdir()
|
||||
new = tmp_path / "new"
|
||||
new.mkdir()
|
||||
svc = ProjectService(factory)
|
||||
p = svc.create("old", "/old/path")
|
||||
updated = svc.update(p.id, "new", "/new/path")
|
||||
p = svc.create("old", str(old))
|
||||
updated = svc.update(p.id, "new", str(new))
|
||||
assert updated.name == "new"
|
||||
assert updated.path == "/new/path"
|
||||
assert svc.get(p.id).path == "/new/path"
|
||||
assert updated.path == str(new.resolve())
|
||||
assert svc.get(p.id).path == str(new.resolve())
|
||||
|
||||
def test_update_duplicate_path_raises(self, factory):
|
||||
def test_update_duplicate_path_raises(self, factory, tmp_path):
|
||||
a = tmp_path / "a"
|
||||
a.mkdir()
|
||||
b = tmp_path / "b"
|
||||
b.mkdir()
|
||||
svc = ProjectService(factory)
|
||||
svc.create("a", "/a")
|
||||
p2 = svc.create("b", "/b")
|
||||
svc.create("a", str(a))
|
||||
p2 = svc.create("b", str(b))
|
||||
with pytest.raises(ServiceError, match="already uses path"):
|
||||
svc.update(p2.id, "b", "/a")
|
||||
svc.update(p2.id, "b", str(a))
|
||||
# DB session must remain usable after rollback, values unchanged
|
||||
assert svc.get(p2.id).path == "/b"
|
||||
assert svc.get(p2.id).path == str(b.resolve())
|
||||
assert svc.get(p2.id).name == "b"
|
||||
|
||||
def test_update_unknown_id_raises(self, factory):
|
||||
def test_update_unknown_id_raises(self, factory, tmp_path):
|
||||
svc = ProjectService(factory)
|
||||
with pytest.raises(ServiceError, match="not found"):
|
||||
svc.update(9999, "x", "/x")
|
||||
svc.update(9999, "x", str(tmp_path))
|
||||
|
||||
def test_create_duplicate_path_raises_service_error(self, factory):
|
||||
def test_create_duplicate_path_raises_service_error(self, factory, tmp_path):
|
||||
dup = tmp_path / "dup"
|
||||
dup.mkdir()
|
||||
svc = ProjectService(factory)
|
||||
svc.create("a", "/dup")
|
||||
svc.create("a", str(dup))
|
||||
with pytest.raises(ServiceError, match="already uses path"):
|
||||
svc.create("b", "/dup")
|
||||
svc.create("b", str(dup))
|
||||
|
||||
def test_create_rejects_nonexistent_path(self, factory, tmp_path):
|
||||
svc = ProjectService(factory)
|
||||
missing = tmp_path / "does-not-exist"
|
||||
with pytest.raises(ServiceError, match="does not exist or is not a directory"):
|
||||
svc.create("p", str(missing))
|
||||
|
||||
def test_create_rejects_file_path(self, factory, tmp_path):
|
||||
svc = ProjectService(factory)
|
||||
a_file = tmp_path / "afile"
|
||||
a_file.write_text("x")
|
||||
with pytest.raises(ServiceError, match="does not exist or is not a directory"):
|
||||
svc.create("p", str(a_file))
|
||||
|
||||
def test_create_stores_resolved_absolute_path(self, factory, tmp_path):
|
||||
svc = ProjectService(factory)
|
||||
p = svc.create("p", str(tmp_path))
|
||||
assert p.path == str(tmp_path.resolve())
|
||||
|
||||
def test_create_expands_user_home(self, factory, tmp_path, monkeypatch):
|
||||
# Point ~ at tmp_path; "~" must expand to the existing tmp_path dir.
|
||||
monkeypatch.setenv("HOME", str(tmp_path))
|
||||
svc = ProjectService(factory)
|
||||
p = svc.create("p", "~")
|
||||
assert p.path == str(tmp_path.resolve())
|
||||
|
||||
def test_update_rejects_nonexistent_path_and_leaves_row(self, factory, tmp_path):
|
||||
svc = ProjectService(factory)
|
||||
p = svc.create("p", str(tmp_path))
|
||||
missing = tmp_path / "nope"
|
||||
with pytest.raises(ServiceError, match="does not exist or is not a directory"):
|
||||
svc.update(p.id, "p", str(missing))
|
||||
assert svc.get(p.id).path == str(tmp_path.resolve())
|
||||
|
||||
|
||||
class TestMcpService:
|
||||
@@ -78,10 +125,10 @@ class TestMcpService:
|
||||
svc.delete("test-mcp")
|
||||
assert svc.get("test-mcp") is None
|
||||
|
||||
def test_bind_unbind(self, factory):
|
||||
def test_bind_unbind(self, factory, tmp_path):
|
||||
psvc = ProjectService(factory)
|
||||
msvc = McpService(factory)
|
||||
project = psvc.create("proj", "/proj")
|
||||
project = psvc.create("proj", str(tmp_path))
|
||||
server = msvc.create("s1", "stdio", command="cmd")
|
||||
msvc.bind_to_project(project.id, server.id)
|
||||
assert len(msvc.get_project_mcps(project.id)) == 1
|
||||
|
||||
Reference in New Issue
Block a user