diff --git a/src/hqt/projects/service.py b/src/hqt/projects/service.py index cfd2286..40496ee 100644 --- a/src/hqt/projects/service.py +++ b/src/hqt/projects/service.py @@ -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: diff --git a/tests/test_services.py b/tests/test_services.py index 989c7da..0710a0a 100644 --- a/tests/test_services.py +++ b/tests/test_services.py @@ -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