90 lines
2.8 KiB
Python
90 lines
2.8 KiB
Python
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from hqt.db.models import Base, Harness
|
|
from hqt.projects.service import ProjectService
|
|
from hqt.sessions.service import SessionService
|
|
from hqt.tmux.manager import TmuxManager, SpawnResult
|
|
from hqt.harnesses.configurators.claude import ClaudeConfigurator
|
|
|
|
|
|
@pytest.fixture
|
|
def setup():
|
|
engine = create_engine(
|
|
"sqlite://",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
Base.metadata.create_all(engine)
|
|
factory = sessionmaker(bind=engine, expire_on_commit=False)
|
|
db = factory()
|
|
db.add(Harness(name="claude-code", display_name="Claude Code"))
|
|
db.commit()
|
|
|
|
tmux = MagicMock(spec=TmuxManager)
|
|
tmux.spawn = AsyncMock(return_value=SpawnResult(ok=True, window_id="@1", error=""))
|
|
tmux.kill = AsyncMock()
|
|
tmux.is_alive = AsyncMock(return_value=True)
|
|
tmux.window_exists = AsyncMock(return_value=True)
|
|
tmux.poll_info = AsyncMock(return_value={})
|
|
tmux.capture_pane = AsyncMock(return_value="")
|
|
tmux.attach = AsyncMock(return_value=True)
|
|
|
|
configurator = ClaudeConfigurator()
|
|
harnesses = {"claude-code": configurator}
|
|
|
|
proj_svc = ProjectService(factory)
|
|
sess_svc = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
|
|
return proj_svc, sess_svc, tmux
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_full_lifecycle(setup, tmp_path):
|
|
proj_svc, sess_svc, tmux = setup
|
|
|
|
# Create project
|
|
project = proj_svc.create(name="test-proj", path=str(tmp_path))
|
|
assert project.id == 1
|
|
|
|
# Create session
|
|
create_result = await sess_svc.create_session(
|
|
project_id=project.id, harness_name="claude-code", nickname="e2e"
|
|
)
|
|
session = create_result.session
|
|
assert session.id is not None
|
|
tmux.spawn.assert_called_once()
|
|
|
|
# List sessions - alive
|
|
from hqt.tmux.runner import WindowInfo
|
|
|
|
tmux.poll_info.return_value = {
|
|
session.tmux_session_name: WindowInfo(alive=True, last_activity=0)
|
|
}
|
|
infos = await sess_svc.list_sessions(project_id=project.id)
|
|
assert len(infos) == 1
|
|
assert infos[0].alive is True
|
|
|
|
# Stop session
|
|
await sess_svc.stop_session(session_id=session.id)
|
|
tmux.kill.assert_called_once_with(session.tmux_session_name)
|
|
|
|
# List sessions - dead
|
|
tmux.poll_info.return_value = {
|
|
session.tmux_session_name: WindowInfo(alive=False, last_activity=0)
|
|
}
|
|
infos = await sess_svc.list_sessions(project_id=project.id)
|
|
assert len(infos) == 1
|
|
assert infos[0].alive is False
|
|
|
|
# Delete session
|
|
tmux.window_exists.return_value = False
|
|
await sess_svc.delete_session(session_id=session.id)
|
|
|
|
# List sessions - empty
|
|
infos = await sess_svc.list_sessions(project_id=project.id)
|
|
assert len(infos) == 0
|