wehub-resource-sync
287 lines
11 KiB
Python
287 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import dataclasses
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from opensquilla.cli.tui.opentui.messages import (
|
|
HOST_TO_PYTHON_TYPES,
|
|
PYTHON_TO_HOST_TYPES,
|
|
ApprovalDismiss,
|
|
CompletionCandidate,
|
|
CompletionContext,
|
|
HostApprovalResponse,
|
|
HostInputCancel,
|
|
HostInputEof,
|
|
HostInputSubmit,
|
|
HostProtocolUnknown,
|
|
HostReady,
|
|
HostResize,
|
|
HostToPythonMessageError,
|
|
RouterPluginState,
|
|
host_message_from_json,
|
|
python_message_to_json,
|
|
)
|
|
|
|
_PACKAGE_SRC = (
|
|
Path(__file__).resolve().parents[4] / "src/opensquilla/cli/tui/opentui/package/src"
|
|
)
|
|
|
|
|
|
def _production_host_sources() -> list[Path]:
|
|
return [
|
|
path
|
|
for path in sorted(_PACKAGE_SRC.glob("*.mjs"))
|
|
if not path.name.endswith(".test.mjs")
|
|
]
|
|
|
|
|
|
def test_python_message_to_json_serializes_router_update() -> None:
|
|
payload = python_message_to_json(
|
|
"router.update",
|
|
RouterPluginState(
|
|
model="gpt-5.5",
|
|
route="T3 | 91%",
|
|
saving="42% | -$0.021",
|
|
context="128k | 37%",
|
|
style="normal",
|
|
),
|
|
)
|
|
|
|
assert payload.endswith("\n")
|
|
assert '"type":"router.update"' in payload
|
|
assert '"model":"gpt-5.5"' in payload
|
|
assert '"route":"T3 | 91%"' in payload
|
|
|
|
|
|
def test_python_message_to_json_serializes_completion_context() -> None:
|
|
payload = python_message_to_json(
|
|
"completion.context",
|
|
CompletionContext(
|
|
catalog=(
|
|
CompletionCandidate(
|
|
label="/compact",
|
|
description="Compact chat context.",
|
|
insert_text="/compact",
|
|
category="command",
|
|
),
|
|
CompletionCandidate(
|
|
label="/code-review",
|
|
description="Run a comprehensive code review",
|
|
insert_text="use the code-review skill: ",
|
|
category="skill",
|
|
),
|
|
),
|
|
files=("src/main.py",),
|
|
filters_sensitive_paths=True,
|
|
),
|
|
)
|
|
|
|
assert payload.endswith("\n")
|
|
assert '"type":"completion.context"' in payload
|
|
assert '"label":"/compact"' in payload
|
|
assert '"category":"command"' in payload
|
|
assert '"insert_text":"use the code-review skill: "' in payload
|
|
assert '"files":["src/main.py"]' in payload
|
|
assert '"filters_sensitive_paths":true' in payload
|
|
|
|
|
|
def test_host_message_from_json_parses_ready_and_submit() -> None:
|
|
assert host_message_from_json('{"type":"ready"}') == HostReady()
|
|
assert host_message_from_json(
|
|
'{"type":"input.submit","text":"中文 prompt"}'
|
|
) == HostInputSubmit(text="中文 prompt")
|
|
|
|
|
|
def test_host_message_from_json_parses_control_messages() -> None:
|
|
assert host_message_from_json('{"type":"input.cancel"}') == HostInputCancel()
|
|
assert host_message_from_json('{"type":"input.eof"}') == HostInputEof()
|
|
assert host_message_from_json('{"type":"resize","width":120,"height":36}') == (
|
|
HostResize(width=120, height=36)
|
|
)
|
|
|
|
|
|
def test_host_message_rejects_malformed_control_payloads() -> None:
|
|
with pytest.raises(HostToPythonMessageError, match="input.submit.text"):
|
|
host_message_from_json('{"type":"input.submit"}')
|
|
|
|
with pytest.raises(HostToPythonMessageError, match="resize.width"):
|
|
host_message_from_json('{"type":"resize","height":36}')
|
|
|
|
with pytest.raises(HostToPythonMessageError, match="Unknown OpenTUI host"):
|
|
host_message_from_json('{"type":"surprise"}')
|
|
|
|
|
|
def test_host_message_from_json_parses_protocol_unknown() -> None:
|
|
parsed = host_message_from_json('{"type":"protocol.unknown","messageType":"tool.call"}')
|
|
assert parsed == HostProtocolUnknown(message_type="tool.call")
|
|
|
|
with pytest.raises(HostToPythonMessageError, match="protocol.unknown.messageType"):
|
|
host_message_from_json('{"type":"protocol.unknown"}')
|
|
|
|
|
|
def test_host_message_from_json_parses_approval_response() -> None:
|
|
parsed = host_message_from_json(
|
|
'{"type":"approval.response","id":"appr-1","approved":true,"choice":"allow_once"}'
|
|
)
|
|
assert parsed == HostApprovalResponse(id="appr-1", approved=True, choice="allow_once")
|
|
|
|
denied = host_message_from_json(
|
|
'{"type":"approval.response","id":"appr-2","approved":false,"choice":null}'
|
|
)
|
|
assert denied == HostApprovalResponse(id="appr-2", approved=False, choice=None)
|
|
|
|
|
|
def test_host_message_rejects_malformed_approval_response() -> None:
|
|
with pytest.raises(HostToPythonMessageError, match="approval.response.id"):
|
|
host_message_from_json('{"type":"approval.response","approved":true}')
|
|
|
|
with pytest.raises(HostToPythonMessageError, match="approval.response.approved"):
|
|
host_message_from_json('{"type":"approval.response","id":"appr-1"}')
|
|
|
|
with pytest.raises(HostToPythonMessageError, match="approval.response.approved"):
|
|
host_message_from_json('{"type":"approval.response","id":"appr-1","approved":"yes"}')
|
|
|
|
|
|
def test_python_message_to_json_serializes_approval_request() -> None:
|
|
payload = python_message_to_json(
|
|
"approval.request",
|
|
{
|
|
"id": "appr-1",
|
|
"tool": "shell",
|
|
"summary": "touch demo.txt",
|
|
"choices": ["allow_once", "deny"],
|
|
},
|
|
)
|
|
assert payload.endswith("\n")
|
|
assert '"type":"approval.request"' in payload
|
|
assert '"id":"appr-1"' in payload
|
|
assert '"choices":["allow_once","deny"]' in payload
|
|
|
|
|
|
def test_python_message_to_json_serializes_approval_dismiss() -> None:
|
|
payload = python_message_to_json("approval.dismiss", ApprovalDismiss(id="appr-1"))
|
|
assert payload.endswith("\n")
|
|
assert '"type":"approval.dismiss"' in payload
|
|
assert '"id":"appr-1"' in payload
|
|
|
|
|
|
def test_python_message_to_json_serializes_structured_blocks() -> None:
|
|
from opensquilla.cli.tui.opentui.messages import (
|
|
ModelText,
|
|
PromptEcho,
|
|
TurnBegin,
|
|
TurnEnd,
|
|
TurnStatusState,
|
|
)
|
|
|
|
assert '"type":"turn.begin"' in python_message_to_json("turn.begin", TurnBegin(id="t1"))
|
|
assert '"type":"prompt.echo"' in python_message_to_json(
|
|
"prompt.echo", PromptEcho(text="帮我分析架构")
|
|
)
|
|
assert '"id":"t1"' in python_message_to_json("turn.end", TurnEnd(id="t1", cancelled=False))
|
|
model = python_message_to_json("model.text", ModelText(text="先扫描结构"))
|
|
assert '"type":"model.text"' in model and '"text":"先扫描结构"' in model
|
|
status = python_message_to_json(
|
|
"turn.status", TurnStatusState(phase="tool", label="read_file", active=True)
|
|
)
|
|
assert '"phase":"tool"' in status and '"active":true' in status
|
|
|
|
|
|
def test_block_messages_serialize_with_kind_and_fields() -> None:
|
|
from opensquilla.cli.tui.opentui.messages import (
|
|
BlockAppend,
|
|
BlockBegin,
|
|
BlockEnd,
|
|
BlockUpdate,
|
|
python_message_to_json,
|
|
)
|
|
begin = python_message_to_json(
|
|
"block.begin",
|
|
BlockBegin(id="b1", kind="tool", meta={"name": "ls", "args": "src"}),
|
|
)
|
|
assert '"type":"block.begin"' in begin
|
|
assert '"kind":"tool"' in begin
|
|
assert '"name":"ls"' in begin
|
|
append = python_message_to_json("block.append", BlockAppend(id="b1", delta="line"))
|
|
assert '"delta":"line"' in append
|
|
update = python_message_to_json("block.update", BlockUpdate(id="b1", patch={"status": "ok"}))
|
|
assert '"status":"ok"' in update
|
|
end = python_message_to_json("block.end", BlockEnd(id="b1"))
|
|
assert '"type":"block.end"' in end
|
|
|
|
|
|
def test_python_to_host_registry_matches_js_dispatcher_cases() -> None:
|
|
"""Every outbound type must have a dispatcher case in the host, and vice
|
|
versa — a type without a handler is one release skew away from a live
|
|
protocol error, and a handler without a sender is dead contract."""
|
|
text = (_PACKAGE_SRC / "ipc.mjs").read_text(encoding="utf-8")
|
|
dispatcher = text.split("createDispatcher", 1)[1]
|
|
cases = set(re.findall(r'case "([a-z][a-z0-9._-]*)":', dispatcher))
|
|
assert cases, "could not parse dispatcher cases from ipc.mjs"
|
|
assert set(PYTHON_TO_HOST_TYPES) == cases
|
|
|
|
|
|
def test_host_emitted_types_are_all_parseable_by_python() -> None:
|
|
"""Every type literal the host can emit must have a Python parse branch
|
|
(extra Python branches are harmless forward tolerance)."""
|
|
emitted: set[str] = set()
|
|
for path in _production_host_sources():
|
|
emitted.update(
|
|
re.findall(
|
|
r'\{\s*type:\s*"([a-z][a-z0-9._-]*)"',
|
|
path.read_text(encoding="utf-8"),
|
|
)
|
|
)
|
|
assert emitted, "could not parse emitter literals from the host sources"
|
|
unparseable = emitted - set(HOST_TO_PYTHON_TYPES)
|
|
assert not unparseable, f"host emits types Python cannot parse: {sorted(unparseable)}"
|
|
|
|
|
|
def test_host_to_python_registry_round_trips_canonical_frames() -> None:
|
|
samples = {
|
|
"ready": '{"type":"ready"}',
|
|
"input.submit": '{"type":"input.submit","text":"hi"}',
|
|
"input.cancel": '{"type":"input.cancel"}',
|
|
"input.eof": '{"type":"input.eof"}',
|
|
"resize": '{"type":"resize","width":80,"height":24}',
|
|
"completion.request": (
|
|
'{"type":"completion.request","kind":"file","query":"a","request_id":1}'
|
|
),
|
|
"error": '{"type":"error","message":"boom"}',
|
|
"protocol.unknown": '{"type":"protocol.unknown","messageType":"mystery.type"}',
|
|
"approval.response": (
|
|
'{"type":"approval.response","id":"appr-1","approved":true,"choice":"allow_once"}'
|
|
),
|
|
}
|
|
assert set(samples) == set(HOST_TO_PYTHON_TYPES)
|
|
for wire_type, raw in samples.items():
|
|
assert isinstance(host_message_from_json(raw), HOST_TO_PYTHON_TYPES[wire_type])
|
|
|
|
|
|
def test_js_snake_case_payload_reads_match_python_dataclass_fields() -> None:
|
|
"""Payloads travel via dataclasses.asdict and JS reads fall back with `??`,
|
|
so a renamed Python field silently blanks the host UI. Pin every snake_case
|
|
payload read in the host against the registered dataclass fields."""
|
|
reads: set[str] = set()
|
|
for path in _production_host_sources():
|
|
reads.update(
|
|
re.findall(
|
|
r"\b[a-zA-Z_$][a-zA-Z0-9_$]*\.([a-z][a-z0-9]*_[a-z0-9_]+)\b",
|
|
path.read_text(encoding="utf-8"),
|
|
)
|
|
)
|
|
assert reads, "could not find any snake_case payload reads in the host sources"
|
|
allowed: set[str] = set()
|
|
for payload_cls in PYTHON_TO_HOST_TYPES.values():
|
|
if payload_cls is not None:
|
|
allowed.update(field.name for field in dataclasses.fields(payload_cls))
|
|
# Nested completion candidates plus the ad-hoc mapping payloads
|
|
# (completion.response and theme.set carry dicts, not dataclasses).
|
|
allowed.update(field.name for field in dataclasses.fields(CompletionCandidate))
|
|
allowed.update({"request_id", "kind", "items", "name"})
|
|
stale = reads - allowed
|
|
assert not stale, f"host reads payload fields Python does not send: {sorted(stale)}"
|