refactor: per-operation DB sessions in Project/Mcp services

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

tests/test_services.py

@@ -1,62 +1,76 @@
 import pytest
 from sqlalchemy import create_engine
-from sqlalchemy.orm import Session
+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 db():
-    engine = create_engine("sqlite:///:memory:")
+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)
-    with Session(engine) as session:
-        yield session
+    return sessionmaker(bind=engine, expire_on_commit=False)
 
 
 class TestProjectService:
-    def test_create_and_get(self, db):
-        svc = ProjectService(db)
+    def test_create_and_get(self, factory):
+        svc = ProjectService(factory)
         p = svc.create("test", "/tmp/test")
         assert p.id is not None
         assert svc.get(p.id).name == "test"
 
-    def test_list_excludes_archived(self, db):
-        svc = ProjectService(db)
+    def test_list_excludes_archived(self, factory):
+        svc = ProjectService(factory)
         svc.create("a", "/a")
         p2 = svc.create("b", "/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, db):
-        svc = ProjectService(db)
+    def test_update_name_and_path(self, factory):
+        svc = ProjectService(factory)
         p = svc.create("old", "/old/path")
         updated = svc.update(p.id, "new", "/new/path")
         assert updated.name == "new"
         assert updated.path == "/new/path"
         assert svc.get(p.id).path == "/new/path"
 
-    def test_update_duplicate_path_raises(self, db):
-        svc = ProjectService(db)
+    def test_update_duplicate_path_raises(self, factory):
+        svc = ProjectService(factory)
         svc.create("a", "/a")
         p2 = svc.create("b", "/b")
-        with pytest.raises(ValueError, match="already uses path"):
+        with pytest.raises(ServiceError, match="already uses path"):
             svc.update(p2.id, "b", "/a")
         # DB session must remain usable after rollback, values unchanged
         assert svc.get(p2.id).path == "/b"
         assert svc.get(p2.id).name == "b"
 
-    def test_update_unknown_id_raises(self, db):
-        svc = ProjectService(db)
-        with pytest.raises(ValueError, match="not found"):
+    def test_update_unknown_id_raises(self, factory):
+        svc = ProjectService(factory)
+        with pytest.raises(ServiceError, match="not found"):
             svc.update(9999, "x", "/x")
 
+    def test_create_duplicate_path_raises_service_error(self, factory):
+        svc = ProjectService(factory)
+        svc.create("a", "/dup")
+        with pytest.raises(ServiceError, match="already uses path"):
+            svc.create("b", "/dup")
+
 
 class TestMcpService:
-    def test_crud(self, db):
-        svc = McpService(db)
+    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
@@ -64,9 +78,9 @@ class TestMcpService:
         svc.delete("test-mcp")
         assert svc.get("test-mcp") is None
 
-    def test_bind_unbind(self, db):
-        psvc = ProjectService(db)
-        msvc = McpService(db)
+    def test_bind_unbind(self, factory):
+        psvc = ProjectService(factory)
+        msvc = McpService(factory)
         project = psvc.create("proj", "/proj")
         server = msvc.create("s1", "stdio", command="cmd")
         msvc.bind_to_project(project.id, server.id)