felixm
1b79530214
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>
137 lines
5.1 KiB
Python
137 lines
5.1 KiB
Python
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from hqt.db.models import Base
|
|
from hqt.errors import ServiceError
|
|
from hqt.mcp.service import McpService
|
|
from hqt.projects.service import ProjectService
|
|
|
|
|
|
@pytest.fixture
|
|
def factory():
|
|
# StaticPool + check_same_thread=False: every session created by the
|
|
# factory shares the single in-memory connection, so per-operation
|
|
# sessions all see the same database.
|
|
engine = create_engine(
|
|
"sqlite://",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
Base.metadata.create_all(engine)
|
|
return sessionmaker(bind=engine, expire_on_commit=False)
|
|
|
|
|
|
class TestProjectService:
|
|
def test_create_and_get(self, factory, tmp_path):
|
|
svc = ProjectService(factory)
|
|
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, tmp_path):
|
|
a = tmp_path / "a"
|
|
a.mkdir()
|
|
b = tmp_path / "b"
|
|
b.mkdir()
|
|
svc = ProjectService(factory)
|
|
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, tmp_path):
|
|
old = tmp_path / "old"
|
|
old.mkdir()
|
|
new = tmp_path / "new"
|
|
new.mkdir()
|
|
svc = ProjectService(factory)
|
|
p = svc.create("old", str(old))
|
|
updated = svc.update(p.id, "new", str(new))
|
|
assert updated.name == "new"
|
|
assert updated.path == str(new.resolve())
|
|
assert svc.get(p.id).path == str(new.resolve())
|
|
|
|
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", str(a))
|
|
p2 = svc.create("b", str(b))
|
|
with pytest.raises(ServiceError, match="already uses path"):
|
|
svc.update(p2.id, "b", str(a))
|
|
# DB session must remain usable after rollback, values unchanged
|
|
assert svc.get(p2.id).path == str(b.resolve())
|
|
assert svc.get(p2.id).name == "b"
|
|
|
|
def test_update_unknown_id_raises(self, factory, tmp_path):
|
|
svc = ProjectService(factory)
|
|
with pytest.raises(ServiceError, match="not found"):
|
|
svc.update(9999, "x", str(tmp_path))
|
|
|
|
def test_create_duplicate_path_raises_service_error(self, factory, tmp_path):
|
|
dup = tmp_path / "dup"
|
|
dup.mkdir()
|
|
svc = ProjectService(factory)
|
|
svc.create("a", str(dup))
|
|
with pytest.raises(ServiceError, match="already uses path"):
|
|
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:
|
|
def test_crud(self, factory):
|
|
svc = McpService(factory)
|
|
server = svc.create("test-mcp", "stdio", command="node", args=["server.js"])
|
|
assert server.id is not None
|
|
assert svc.get("test-mcp") is not None
|
|
assert len(svc.list_all()) == 1
|
|
svc.delete("test-mcp")
|
|
assert svc.get("test-mcp") is None
|
|
|
|
def test_bind_unbind(self, factory, tmp_path):
|
|
psvc = ProjectService(factory)
|
|
msvc = McpService(factory)
|
|
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
|
|
msvc.unbind_from_project(project.id, server.id)
|
|
assert len(msvc.get_project_mcps(project.id)) == 0
|