Adolfo Reyna
116 lines
3.2 KiB
Python
116 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Host smoke test for the Tactility MCP Screen Phase 1 API."""
|
|
|
|
import argparse
|
|
import json
|
|
import socket
|
|
import urllib.request
|
|
|
|
|
|
def discover(timeout: float = 2.0) -> str | None:
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
|
|
sock.settimeout(timeout)
|
|
try:
|
|
sock.sendto(b"DISCOVER_SCREEN", ("255.255.255.255", 5000))
|
|
data, address = sock.recvfrom(128)
|
|
if data == b"SCREEN_IP_80":
|
|
return address[0]
|
|
except TimeoutError:
|
|
return None
|
|
finally:
|
|
sock.close()
|
|
return None
|
|
|
|
|
|
def rpc(ip: str, request_id: int, method: str, params: dict | None = None) -> dict:
|
|
payload = {
|
|
"jsonrpc": "2.0",
|
|
"id": request_id,
|
|
"method": method,
|
|
}
|
|
if params is not None:
|
|
payload["params"] = params
|
|
|
|
request = urllib.request.Request(
|
|
f"http://{ip}/api/mcp",
|
|
data=json.dumps(payload).encode(),
|
|
headers={"Content-Type": "application/json"},
|
|
method="POST",
|
|
)
|
|
with urllib.request.urlopen(request, timeout=5) as response:
|
|
return json.loads(response.read())
|
|
|
|
|
|
def require_result(response: dict, label: str) -> None:
|
|
if "error" in response:
|
|
raise RuntimeError(f"{label} failed: {response['error']}")
|
|
if "result" not in response:
|
|
raise RuntimeError(f"{label} returned no result: {response}")
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--ip", help="Device IP; omit to use UDP discovery")
|
|
parser.add_argument(
|
|
"--draw",
|
|
action="store_true",
|
|
help="Also clear the app canvas and draw visible test text",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
ip = args.ip or discover()
|
|
if not ip:
|
|
raise SystemExit("MCP Screen was not discovered. Pass --ip DEVICE_IP to test directly.")
|
|
|
|
print(f"Testing MCP Screen at {ip}")
|
|
|
|
tools = rpc(ip, 1, "tools/list")
|
|
require_result(tools, "tools/list")
|
|
names = [tool["name"] for tool in tools["result"]["tools"]]
|
|
expected = {"get_capabilities", "clear_screen", "draw_text"}
|
|
if not expected.issubset(names):
|
|
raise RuntimeError(f"Missing tools: {sorted(expected - set(names))}")
|
|
print(f"tools/list: {', '.join(names)}")
|
|
|
|
capabilities = rpc(
|
|
ip,
|
|
2,
|
|
"tools/call",
|
|
{"name": "get_capabilities", "arguments": {}},
|
|
)
|
|
require_result(capabilities, "get_capabilities")
|
|
print("get_capabilities:", capabilities["result"]["content"][0]["text"])
|
|
|
|
if args.draw:
|
|
cleared = rpc(
|
|
ip,
|
|
3,
|
|
"tools/call",
|
|
{"name": "clear_screen", "arguments": {"color": 1}},
|
|
)
|
|
require_result(cleared, "clear_screen")
|
|
|
|
drawn = rpc(
|
|
ip,
|
|
4,
|
|
"tools/call",
|
|
{
|
|
"name": "draw_text",
|
|
"arguments": {
|
|
"text": "Tactility MCP is alive",
|
|
"x": 20,
|
|
"y": 30,
|
|
"size": 2,
|
|
},
|
|
},
|
|
)
|
|
require_result(drawn, "draw_text")
|
|
print("Visible draw test sent.")
|
|
|
|
print("Phase 1 smoke test passed.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|