270 lines
12 KiB
Python
270 lines
12 KiB
Python
"""Exercise the RRCS client against a mock gateway that speaks spec-shaped XML."""
|
|
|
|
import asyncio
|
|
import sys
|
|
import xmlrpc.client
|
|
|
|
import aiohttp
|
|
from aiohttp import web
|
|
|
|
import pathlib
|
|
|
|
ROOT = pathlib.Path(__file__).resolve().parents[1] / "custom_components"
|
|
DIR = str(ROOT / "riedel_rrcs")
|
|
|
|
|
|
def _load_module():
|
|
"""Load rrcs.py without dragging in the Home Assistant package __init__."""
|
|
import importlib.util
|
|
import types
|
|
|
|
pkg = types.ModuleType("rrcspkg")
|
|
pkg.__path__ = [DIR]
|
|
sys.modules["rrcspkg"] = pkg
|
|
for name in ("const", "rrcs"):
|
|
spec = importlib.util.spec_from_file_location(f"rrcspkg.{name}", f"{DIR}/{name}.py")
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[f"rrcspkg.{name}"] = module
|
|
spec.loader.exec_module(module)
|
|
return sys.modules["rrcspkg.rrcs"]
|
|
|
|
|
|
R = _load_module()
|
|
|
|
RECEIVED = []
|
|
|
|
|
|
def resp(body: str) -> web.Response:
|
|
return web.Response(
|
|
body=('<?xml version="1.0"?><methodResponse><params><param>'
|
|
f"{body}</param></params></methodResponse>").encode(),
|
|
content_type="text/xml",
|
|
)
|
|
|
|
|
|
async def handler(request: web.Request) -> web.Response:
|
|
raw = await request.read()
|
|
params, method = xmlrpc.client.loads(raw)
|
|
RECEIVED.append((method, params))
|
|
tk = params[0]
|
|
|
|
if method == "GetVersion":
|
|
return resp(f"<value><array><data><value><string>{tk}</string></value>"
|
|
"<value><int>0</int></value>"
|
|
"<value><string>8.9.1.0</string></value></data></array></value>")
|
|
if method == "GetState":
|
|
return resp(f"<value><array><data><value><string>{tk}</string></value>"
|
|
"<value><int>0</int></value>"
|
|
"<value><string>Working</string></value></data></array></value>")
|
|
if method == "IsConnectedToArtist":
|
|
return resp("<value><struct>"
|
|
"<member><name>IsConnected</name><value><boolean>1</boolean></value></member>"
|
|
f"<member><name>TransKey</name><value><string>{tk}</string></value></member>"
|
|
"</struct></value>")
|
|
if method == "GetAllActiveXpsCount":
|
|
return resp(f"<value><array><data><value><string>{tk}</string></value>"
|
|
"<value><int>0</int></value>"
|
|
"<value><i4>42</i4></value></data></array></value>")
|
|
if method == "GetAllLogicSources_v2":
|
|
return resp("<value><struct>"
|
|
"<member><name>ErrorCode</name><value><int>0</int></value></member>"
|
|
f"<member><name>TransKey</name><value><string>{tk}</string></value></member>"
|
|
"<member><name>LogicSourceCount</name><value><int>2</int></value></member>"
|
|
"<member><name>LogicSource#1</name><value><array><data>"
|
|
"<value><string>Ben Busy</string></value>"
|
|
"<value><string>BenBusy</string></value>"
|
|
"<value><i4>1454317500</i4></value>"
|
|
"<value><boolean>1</boolean></value>"
|
|
"</data></array></value></member>"
|
|
"<member><name>LogicSource#2</name><value><array><data>"
|
|
"<value><string>Studio Red Light</string></value>"
|
|
"<value><string>RedLight</string></value>"
|
|
"<value><i4>1454317501</i4></value>"
|
|
"<value><boolean>0</boolean></value>"
|
|
"</data></array></value></member>"
|
|
"</struct></value>")
|
|
if method in ("SetLogicSourceState", "SetGpOutput", "SetXp", "SetKeyLabel"):
|
|
return resp(f"<value><array><data><value><string>{tk}</string></value>"
|
|
"<value><int>0</int></value></data></array></value>")
|
|
if method == "SetInputGain":
|
|
return resp(f"<value><array><data><value><string>{tk}</string></value>"
|
|
"<value><int>6</int></value></data></array></value>")
|
|
if method in ("GetGpInputState", "GetGpOutputState"):
|
|
return resp(f"<value><array><data><value><string>{tk}</string></value>"
|
|
"<value><int>0</int></value>"
|
|
"<value><boolean>1</boolean></value></data></array></value>")
|
|
if method == "GetLevelMeterValues":
|
|
# Deliberately puts a payload int where ErrorCode usually sits.
|
|
return resp(f"<value><array><data><value><string>{tk}</string></value>"
|
|
"<value><int>17</int></value><value><int>4</int></value>"
|
|
"<value><boolean>0</boolean></value>"
|
|
"<value><int>21</int></value></data></array></value>")
|
|
if method == "GetAllGpIns":
|
|
return resp(f"<value><array><data><value><string>{tk}</string></value>"
|
|
"<value><array><data>"
|
|
"<value><struct>"
|
|
"<member><name>Index</name><value><i4>5</i4></value></member>"
|
|
"<member><name>IsInput</name><value><boolean>1</boolean></value></member>"
|
|
"<member><name>Node</name><value><i4>2</i4></value></member>"
|
|
"<member><name>Bay</name><value><i4>6</i4></value></member>"
|
|
"</struct></value>"
|
|
"<value><struct>"
|
|
"<member><name>PortAddress</name><value><struct>"
|
|
"<member><name>IsInput</name><value><boolean>1</boolean></value></member>"
|
|
"<member><name>Node</name><value><i4>4</i4></value></member>"
|
|
"<member><name>Port</name><value><i4>9</i4></value></member>"
|
|
"</struct></value></member>"
|
|
"<member><name>Index</name><value><i4>0</i4></value></member>"
|
|
"<member><name>IsInput</name><value><boolean>1</boolean></value></member>"
|
|
"</struct></value>"
|
|
"</data></array></value></data></array></value>")
|
|
if method == "GetAllGpOuts":
|
|
raise web.HTTPInternalServerError()
|
|
if method == "RegisterForAllEvents":
|
|
return resp(f"<value><string>{tk}</string></value>")
|
|
if method == "IsRegisteredForAllEvents":
|
|
return resp("<value><struct>"
|
|
"<member><name>IsRegistered</name><value><boolean>1</boolean></value></member>"
|
|
f"<member><name>TransKey</name><value><string>{tk}</string></value></member>"
|
|
"</struct></value>")
|
|
if method == "BadLabel":
|
|
# UTF-8 declared, Latin-1 emitted: the tolerant parser must recover.
|
|
body = ('<?xml version="1.0" encoding="UTF-8"?><methodResponse><params><param>'
|
|
f'<value><array><data><value><string>{tk}</string></value>'
|
|
'<value><int>0</int></value>'
|
|
'<value><string>Caf\xe9 Mic</string></value>'
|
|
'</data></array></value></param></params></methodResponse>')
|
|
return web.Response(body=body.encode("latin-1"), content_type="text/xml")
|
|
if method == "Faulty":
|
|
return web.Response(
|
|
body=xmlrpc.client.dumps(
|
|
xmlrpc.client.Fault(14, "XML-RPC parameters wrong")
|
|
).encode(),
|
|
content_type="text/xml",
|
|
)
|
|
return resp(f"<value><string>{tk}</string></value>")
|
|
|
|
|
|
async def main() -> int:
|
|
app = web.Application()
|
|
app.router.add_post("/", handler)
|
|
runner = web.AppRunner(app)
|
|
await runner.setup()
|
|
site = web.TCPSite(runner, "127.0.0.1", 18193)
|
|
await site.start()
|
|
|
|
failures = []
|
|
|
|
def check(label, got, want):
|
|
ok = got == want
|
|
print(f"{'PASS' if ok else 'FAIL'} {label}: {got!r}")
|
|
if not ok:
|
|
failures.append((label, got, want))
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
c = R.RRCSClient(session, "127.0.0.1", 18193)
|
|
|
|
check("url", c.url, "http://127.0.0.1:18193/")
|
|
check("transkey format", len(c.next_transkey()), 11)
|
|
|
|
check("get_version", await c.get_version(), "8.9.1.0")
|
|
check("get_state", await c.get_state(), "Working")
|
|
check("is_connected_to_artist", await c.is_connected_to_artist(), True)
|
|
check("get_active_xp_count", await c.get_active_xp_count(), 42)
|
|
|
|
sources = await c.get_logic_sources()
|
|
check("logic source count", len(sources), 2)
|
|
check("logic source name", sources[1454317500].long_name, "Ben Busy")
|
|
check("logic source label", sources[1454317500].label, "BenBusy")
|
|
check("logic source state on", sources[1454317500].state, True)
|
|
check("logic source state off", sources[1454317501].state, False)
|
|
|
|
await c.set_logic_source(1454317500, True)
|
|
method, params = RECEIVED[-1]
|
|
check("SetLogicSourceState params", params[1:], (1454317500, True))
|
|
|
|
addr = R.GpioAddress(1, 2, 128, 0, 5, True)
|
|
check("gp input read", await c.get_gp_input_state(addr), True)
|
|
await c.set_gp_output(R.GpioAddress(1, 2, 128, 0, 5, False), True)
|
|
check("SetGpOutput params", RECEIVED[-1][1][1:], (1, 2, 128, 0, 5, 1))
|
|
|
|
# A payload integer in slot 1 must not be mistaken for an error code.
|
|
level = await c.call("GetLevelMeterValues", 2, 0, 0)
|
|
check("level meter untouched", level[1:], [17, 4, False, 21])
|
|
|
|
# ...but a real error code on a checked call must raise.
|
|
try:
|
|
await c.call("SetInputGain", 1, 2, 3, 99, check=True)
|
|
check("error raised", False, True)
|
|
except R.RRCSError as err:
|
|
check("error code", err.code, 6)
|
|
check("error text", str(err), "SetInputGain: Input gain invalid")
|
|
|
|
gpios = await c.discover_gpios()
|
|
keys = sorted(g.key for g in gpios)
|
|
check("discovered client-card gpio", "1_2_128_5_5_in" in keys, True)
|
|
check("discovered port gpio", "1_4_9_0_0_in" in keys, True)
|
|
check("failed GetAllGpOuts tolerated", len(gpios), 2)
|
|
|
|
check("mojibake tolerated", (await c.call("BadLabel"))[2], "Caf\ufffd Mic")
|
|
|
|
try:
|
|
await c.call("Faulty")
|
|
check("fault raised", False, True)
|
|
except R.RRCSFaultError as err:
|
|
check("fault code", err.code, 14)
|
|
|
|
await c.register_for_all_events(8123, "/api/riedel_rrcs/abc123")
|
|
check("register params", RECEIVED[-1][1][1:],
|
|
(8123, "/api/riedel_rrcs/abc123", False, False))
|
|
check("still registered",
|
|
await c.is_registered_for_all_events(8123, "/api/riedel_rrcs/abc123"), True)
|
|
|
|
# Unreachable host must surface as a connection error, not a crash.
|
|
dead = R.RRCSClient(session, "127.0.0.1", 18194, timeout=2)
|
|
try:
|
|
await dead.get_version()
|
|
check("dead host raises", False, True)
|
|
except R.RRCSConnectionError:
|
|
check("dead host raises", True, True)
|
|
|
|
# --- Manual GPIO config parsing ---
|
|
parsed = R.parse_gpio_config(
|
|
'[{"name":"Tally A","direction":"in","node":2,"index":3},'
|
|
' {"name":"Red Light","direction":"out","net":1,"node":2,"port":128,'
|
|
' "slot":1,"index":5}]'
|
|
)
|
|
check("parsed count", len(parsed), 2)
|
|
check("parsed defaults", parsed[0][0], R.GpioAddress(1, 2, 128, 0, 3, True))
|
|
check("parsed name", parsed[1][1], "Red Light")
|
|
check("parsed direction", parsed[1][0].is_input, False)
|
|
check("empty config", R.parse_gpio_config(""), [])
|
|
for bad, label in (("not json", "bad json"), ('[{"node":1}]', "missing index"),
|
|
('[{"node":1,"index":2,"direction":"sideways"}]', "bad direction")):
|
|
try:
|
|
R.parse_gpio_config(bad)
|
|
check(label, False, True)
|
|
except ValueError:
|
|
check(label, True, True)
|
|
|
|
# --- Inbound notification decoding ---
|
|
call = xmlrpc.client.dumps(("R0000000001", 1454317500, True), "LogicSourceChange")
|
|
params, method = R.parse_xmlrpc(call.encode())
|
|
check("notification method", method, "LogicSourceChange")
|
|
check("notification params", params, ("R0000000001", 1454317500, True))
|
|
|
|
ack = xmlrpc.client.dumps((["R0000000001", 0],), methodresponse=True)
|
|
check("ack parses", xmlrpc.client.loads(ack.encode())[0][0], ["R0000000001", 0])
|
|
|
|
await runner.cleanup()
|
|
|
|
print()
|
|
if failures:
|
|
print(f"{len(failures)} FAILURE(S)")
|
|
return 1
|
|
print("all checks passed")
|
|
return 0
|
|
|
|
|
|
sys.exit(asyncio.run(main()))
|