test: end-to-end integration test — full session flow

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

tests/test_integration.py

@@ -0,0 +1,120 @@
+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()