ahmed
cc2018aae7
- SwiftUI popover: apps grid w/ live search, sources, glass remote, gradient volume slider, animations - Persistent python protocol bridge (vidaa-control, MQTT/TLS :36669) - Wake-on-LAN power-on, one-time PIN pairing with token auto-refresh - MCP server for AI agents (11 tv_* tools) + live test script - Gitea Actions CI workflow
85 lines
2.8 KiB
Python
85 lines
2.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Live test for mcp_tv_server.py: handshake, tools/list, real TV calls. Exit 0 = pass."""
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
|
|
SERVER = "/Users/ahmedmohammed/Documents/rae/HisenseBar/mcp_tv_server.py"
|
|
PY = "/Users/ahmedmohammed/Documents/rae/.venv/bin/python3"
|
|
|
|
proc = subprocess.Popen([PY, SERVER], stdin=subprocess.PIPE, stdout=subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL, text=True, bufsize=1)
|
|
failures = []
|
|
|
|
|
|
def send(msg):
|
|
proc.stdin.write(json.dumps(msg) + "\n")
|
|
proc.stdin.flush()
|
|
|
|
|
|
def recv(expect_id, timeout=30):
|
|
import time
|
|
start = time.time()
|
|
while time.time() - start < timeout:
|
|
line = proc.stdout.readline()
|
|
if not line:
|
|
break
|
|
try:
|
|
msg = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
if msg.get("id") == expect_id:
|
|
return msg
|
|
raise TimeoutError(f"no response for id {expect_id}")
|
|
|
|
|
|
def check(name, cond, detail=""):
|
|
print(("PASS" if cond else "FAIL"), name, detail)
|
|
if not cond:
|
|
failures.append(name)
|
|
|
|
|
|
# 1. initialize handshake
|
|
send({"jsonrpc": "2.0", "id": 1, "method": "initialize",
|
|
"params": {"protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "test"}}})
|
|
r = recv(1)
|
|
check("initialize", "serverInfo" in r.get("result", {}), r.get("result", {}).get("serverInfo", {}).get("name"))
|
|
|
|
send({"jsonrpc": "2.0", "method": "notifications/initialized"})
|
|
|
|
# 2. tools/list
|
|
send({"jsonrpc": "2.0", "id": 2, "method": "tools/list"})
|
|
r = recv(2)
|
|
tools = [t["name"] for t in r["result"]["tools"]]
|
|
check("tools_list", len(tools) == 11, f"{len(tools)} tools")
|
|
|
|
# 3. live TV calls
|
|
send({"jsonrpc": "2.0", "id": 3, "method": "tools/call",
|
|
"params": {"name": "tv_get_volume", "arguments": {}}})
|
|
r = recv(3)
|
|
vol_text = r["result"]["content"][0]["text"]
|
|
check("tv_get_volume", "Volume:" in vol_text and not r["result"].get("isError", False), vol_text)
|
|
|
|
send({"jsonrpc": "2.0", "id": 4, "method": "tools/call",
|
|
"params": {"name": "tv_get_state", "arguments": {}}})
|
|
r = recv(4)
|
|
check("tv_get_state", not r["result"].get("isError", False), r["result"]["content"][0]["text"][:60])
|
|
|
|
# wake the screen first (fake_sleep blocks sourcelist)
|
|
send({"jsonrpc": "2.0", "id": 41, "method": "tools/call",
|
|
"params": {"name": "tv_send_key", "arguments": {"key": "KEY_HOME"}}})
|
|
recv(41)
|
|
import time as _t; _t.sleep(5)
|
|
|
|
send({"jsonrpc": "2.0", "id": 5, "method": "tools/call",
|
|
"params": {"name": "tv_list_sources", "arguments": {}}})
|
|
r = recv(5)
|
|
srcs = r["result"]["content"][0]["text"]
|
|
check("tv_list_sources", "HDMI1" in srcs, srcs[:60])
|
|
|
|
proc.stdin.close()
|
|
proc.wait(timeout=5)
|
|
print("=" * 30)
|
|
print("RESULT:", "PASS" if not failures else f"FAIL ({failures})")
|
|
sys.exit(0 if not failures else 1)
|