Mortdecai
121 lines
4.0 KiB
Python
121 lines
4.0 KiB
Python
import asyncio
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from kitty_workbench.server import KittWorkbenchServer
|
|
from kitty_workbench.protocol import (
|
|
encode_message, decode_message, ReadyEvent, ChecklistToggleEvent,
|
|
InitCmd, DisplayCmd, LogCmd,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_full_session_flow(tmp_workbench):
|
|
"""End-to-end: open → display → log → events → close."""
|
|
srv = KittWorkbenchServer(
|
|
workbench_dir=tmp_workbench,
|
|
socket_dir=str(tmp_workbench),
|
|
)
|
|
|
|
# Mock the backend so it doesn't actually launch a pane
|
|
from unittest.mock import MagicMock
|
|
mock_backend = MagicMock()
|
|
mock_backend.name = "test"
|
|
mock_backend.image_protocol.return_value = "none"
|
|
mock_backend.launch_pane.return_value = 0
|
|
srv.backend = mock_backend
|
|
|
|
# Override _launch_tui to start socket server and connect as fake TUI
|
|
async def fake_launch(name, title):
|
|
import os
|
|
sock_path = srv._socket_path(name)
|
|
if os.path.exists(sock_path):
|
|
os.unlink(sock_path)
|
|
|
|
ready_event = asyncio.Event()
|
|
|
|
async def handle(reader, writer):
|
|
srv._connections[name] = writer
|
|
while True:
|
|
line = await reader.readline()
|
|
if not line:
|
|
break
|
|
msg = decode_message(line.decode().strip())
|
|
if msg is None:
|
|
continue
|
|
if isinstance(msg, ReadyEvent):
|
|
init_cmd = InitCmd(project=name, title=title, image_protocol="none")
|
|
writer.write((encode_message(init_cmd) + "\n").encode())
|
|
await writer.drain()
|
|
ready_event.set()
|
|
else:
|
|
from dataclasses import asdict
|
|
if name not in srv._event_queues:
|
|
srv._event_queues[name] = []
|
|
srv._event_queues[name].append(asdict(msg))
|
|
|
|
server = await asyncio.start_unix_server(handle, path=sock_path)
|
|
srv._socket_servers[name] = server
|
|
srv._pane_ids[name] = 0
|
|
|
|
# Connect as fake TUI
|
|
reader, writer = await asyncio.open_unix_connection(sock_path)
|
|
writer.write((encode_message(ReadyEvent()) + "\n").encode())
|
|
await writer.drain()
|
|
await asyncio.wait_for(ready_event.wait(), timeout=5.0)
|
|
|
|
# Read init command
|
|
line = await reader.readline()
|
|
init = decode_message(line.decode().strip())
|
|
assert isinstance(init, InitCmd)
|
|
assert init.project == name
|
|
|
|
srv._test_reader = reader
|
|
srv._test_writer = writer
|
|
|
|
srv._launch_tui = fake_launch
|
|
|
|
# --- Open ---
|
|
result = json.loads(await srv.kitt_open("test-proj", "Integration Test"))
|
|
assert result["status"] == "ready"
|
|
assert result["project"] == "test-proj"
|
|
|
|
# --- Display ---
|
|
result = json.loads(await srv.kitt_display("test-proj", "markdown", content="# Hello"))
|
|
assert result["ok"] is True
|
|
|
|
line = await srv._test_reader.readline()
|
|
cmd = decode_message(line.decode().strip())
|
|
assert isinstance(cmd, DisplayCmd)
|
|
assert cmd.content == "# Hello"
|
|
|
|
# --- Log ---
|
|
result = json.loads(await srv.kitt_log("test-proj", "Test entry", level="info"))
|
|
assert result["ok"] is True
|
|
|
|
entries = json.loads(await srv.kitt_read_log("test-proj"))["entries"]
|
|
assert len(entries) == 1
|
|
assert entries[0]["entry"] == "Test entry"
|
|
|
|
# --- Events ---
|
|
evt = ChecklistToggleEvent(pane="sidebar", index=0, label="Test", checked=True)
|
|
srv._test_writer.write((encode_message(evt) + "\n").encode())
|
|
await srv._test_writer.drain()
|
|
await asyncio.sleep(0.1)
|
|
|
|
events = json.loads(await srv.kitt_events("test-proj"))["events"]
|
|
assert len(events) == 1
|
|
assert events[0]["checked"] is True
|
|
|
|
# --- List ---
|
|
projects = json.loads(await srv.kitt_list())["projects"]
|
|
assert any(p["name"] == "test-proj" and p["active"] for p in projects)
|
|
|
|
# --- Close ---
|
|
result = json.loads(await srv.kitt_close("test-proj"))
|
|
assert result["ok"] is True
|
|
|
|
# Cleanup
|
|
srv._test_writer.close()
|