139 lines
4.1 KiB
Python
139 lines
4.1 KiB
Python
from pathlib import Path
|
|
|
|
import pytest
|
|
from sqlalchemy import inspect
|
|
|
|
from hqt.config import Settings
|
|
from hqt.db import migrations
|
|
from hqt.db.engine import ensure_db, get_engine, get_session_factory
|
|
from hqt.db.migrations import migrate
|
|
from hqt.db.models import Base, Harness, Project, Session
|
|
|
|
|
|
def _tmp_settings(tmp_path: Path) -> Settings:
|
|
return Settings(db_path=tmp_path / "test.db")
|
|
|
|
|
|
def test_ensure_db_creates_tables(tmp_path):
|
|
settings = _tmp_settings(tmp_path)
|
|
ensure_db(settings)
|
|
engine = get_engine(settings)
|
|
tables = inspect(engine).get_table_names()
|
|
assert "projects" in tables
|
|
assert "sessions" in tables
|
|
assert "harnesses" in tables
|
|
assert "mcp_servers" in tables
|
|
assert "project_mcp_servers" in tables
|
|
assert "project_skills" in tables
|
|
|
|
|
|
def test_project_crud(tmp_path):
|
|
settings = _tmp_settings(tmp_path)
|
|
ensure_db(settings)
|
|
engine = get_engine(settings)
|
|
factory = get_session_factory(engine)
|
|
|
|
with factory() as session:
|
|
p = Project(name="test", path="/tmp/test")
|
|
session.add(p)
|
|
session.commit()
|
|
assert p.id is not None
|
|
|
|
p.name = "updated"
|
|
session.commit()
|
|
|
|
fetched = session.get(Project, p.id)
|
|
assert fetched.name == "updated"
|
|
|
|
session.delete(fetched)
|
|
session.commit()
|
|
assert session.get(Project, p.id) is None
|
|
|
|
|
|
def test_session_with_fk(tmp_path):
|
|
settings = _tmp_settings(tmp_path)
|
|
ensure_db(settings)
|
|
engine = get_engine(settings)
|
|
factory = get_session_factory(engine)
|
|
|
|
with factory() as session:
|
|
p = Project(name="proj", path="/tmp/proj")
|
|
h = Harness(name="claude-code")
|
|
session.add_all([p, h])
|
|
session.commit()
|
|
|
|
s = Session(
|
|
project_id=p.id,
|
|
harness_id=h.id,
|
|
tmux_session_name="hqt-test-1",
|
|
)
|
|
session.add(s)
|
|
session.commit()
|
|
|
|
assert s.project.name == "proj"
|
|
assert s.harness.name == "claude-code"
|
|
assert s in p.sessions
|
|
|
|
|
|
def _user_version(engine) -> int:
|
|
with engine.connect() as conn:
|
|
return conn.exec_driver_sql("PRAGMA user_version").scalar() or 0
|
|
|
|
|
|
def test_fresh_db_stamped_latest_version(tmp_path):
|
|
settings = _tmp_settings(tmp_path)
|
|
ensure_db(settings)
|
|
assert _user_version(get_engine(settings)) == migrations.LATEST_VERSION
|
|
|
|
|
|
def test_unversioned_existing_db_stamped_baseline(tmp_path):
|
|
# Simulates a user DB created before versioning existed: tables present,
|
|
# user_version still 0.
|
|
settings = _tmp_settings(tmp_path)
|
|
settings.db_path.parent.mkdir(parents=True, exist_ok=True)
|
|
engine = get_engine(settings)
|
|
Base.metadata.create_all(engine)
|
|
migrate(engine)
|
|
assert _user_version(engine) == migrations.BASELINE_VERSION
|
|
|
|
|
|
def test_db_from_newer_hqt_raises(tmp_path):
|
|
settings = _tmp_settings(tmp_path)
|
|
ensure_db(settings)
|
|
engine = get_engine(settings)
|
|
with engine.begin() as conn:
|
|
conn.exec_driver_sql("PRAGMA user_version = 9999")
|
|
with pytest.raises(RuntimeError, match="newer"):
|
|
migrate(engine)
|
|
|
|
|
|
def test_pending_migrations_applied_in_order_once(tmp_path, monkeypatch):
|
|
settings = _tmp_settings(tmp_path)
|
|
ensure_db(settings) # stamps BASELINE_VERSION
|
|
applied: list[int] = []
|
|
monkeypatch.setattr(
|
|
migrations,
|
|
"MIGRATIONS",
|
|
[(2, lambda conn: applied.append(2)), (3, lambda conn: applied.append(3))],
|
|
)
|
|
monkeypatch.setattr(migrations, "LATEST_VERSION", 3)
|
|
engine = get_engine(settings)
|
|
migrate(engine)
|
|
assert applied == [2, 3]
|
|
assert _user_version(engine) == 3
|
|
migrate(engine) # re-run is a no-op
|
|
assert applied == [2, 3]
|
|
|
|
|
|
def test_rows_stay_readable_after_session_closes(tmp_path):
|
|
# Services use per-operation sessions; rows they return must remain
|
|
# readable after the originating session closes (expire_on_commit=False).
|
|
settings = _tmp_settings(tmp_path)
|
|
ensure_db(settings)
|
|
factory = get_session_factory(get_engine(settings))
|
|
with factory() as session:
|
|
p = Project(name="x", path="/x")
|
|
session.add(p)
|
|
session.commit()
|
|
assert p.name == "x" # would raise DetachedInstanceError without the flag
|