335 lines
12 KiB
Python
335 lines
12 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Run a no-hardware cmvr_es process against the local MsQuic test gateway."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import json
|
||
|
|
import signal
|
||
|
|
import subprocess
|
||
|
|
import tempfile
|
||
|
|
import time
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
|
||
|
|
def parse_args() -> argparse.Namespace:
|
||
|
|
parser = argparse.ArgumentParser()
|
||
|
|
parser.add_argument("--gateway", type=Path, required=True)
|
||
|
|
parser.add_argument("--cmvr-es", type=Path, required=True)
|
||
|
|
parser.add_argument("--cert", type=Path, required=True)
|
||
|
|
parser.add_argument("--key", type=Path, required=True)
|
||
|
|
parser.add_argument("--timeout-seconds", type=float, default=15.0)
|
||
|
|
return parser.parse_args()
|
||
|
|
|
||
|
|
|
||
|
|
def wait_for_json(
|
||
|
|
path: Path, process: subprocess.Popen[str], deadline: float
|
||
|
|
) -> dict[str, object]:
|
||
|
|
last_error: Exception | None = None
|
||
|
|
while time.monotonic() < deadline:
|
||
|
|
if process.poll() is not None:
|
||
|
|
raise RuntimeError(
|
||
|
|
f"gateway exited before publishing {path.name}: "
|
||
|
|
f"returncode={process.returncode}"
|
||
|
|
)
|
||
|
|
if path.is_file():
|
||
|
|
try:
|
||
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
||
|
|
except (OSError, json.JSONDecodeError) as error:
|
||
|
|
last_error = error
|
||
|
|
time.sleep(0.02)
|
||
|
|
raise TimeoutError(
|
||
|
|
f"timed out waiting for {path}"
|
||
|
|
+ (f": {last_error}" if last_error else "")
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def stop_process(process: subprocess.Popen[str] | None) -> int | None:
|
||
|
|
if process is None:
|
||
|
|
return None
|
||
|
|
if process.poll() is not None:
|
||
|
|
return process.returncode
|
||
|
|
process.send_signal(signal.SIGTERM)
|
||
|
|
try:
|
||
|
|
process.wait(timeout=3.0)
|
||
|
|
except subprocess.TimeoutExpired:
|
||
|
|
process.kill()
|
||
|
|
process.wait(timeout=3.0)
|
||
|
|
return process.returncode
|
||
|
|
|
||
|
|
|
||
|
|
def write_runtime_config(config_dir: Path, port: int) -> Path:
|
||
|
|
config_dir.mkdir(parents=True, exist_ok=True)
|
||
|
|
(config_dir / "cmvr_es.pb.txt").write_text(
|
||
|
|
"""
|
||
|
|
cmvr_es {
|
||
|
|
logger_config_file: "logger.pb.txt"
|
||
|
|
device_manager_config_file: "device_manager.pb.txt"
|
||
|
|
task_manager_config_file: "task_manager.pb.txt"
|
||
|
|
}
|
||
|
|
""".lstrip(),
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
(config_dir / "logger.pb.txt").write_text(
|
||
|
|
"""
|
||
|
|
logger {
|
||
|
|
minimum_level: LOG_LEVEL_INFO
|
||
|
|
routes { level: LOG_LEVEL_INFO terminal: true }
|
||
|
|
routes { level: LOG_LEVEL_WARNING terminal: true }
|
||
|
|
routes { level: LOG_LEVEL_ERROR terminal: true }
|
||
|
|
routes { level: LOG_LEVEL_FATAL terminal: true }
|
||
|
|
max_file_size_mb: 1
|
||
|
|
flush_interval_seconds: 1
|
||
|
|
format {
|
||
|
|
show_time: false
|
||
|
|
show_level: true
|
||
|
|
show_thread_id: false
|
||
|
|
show_source_location: true
|
||
|
|
}
|
||
|
|
}
|
||
|
|
""".lstrip(),
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
(config_dir / "device_manager.pb.txt").write_text(
|
||
|
|
"""
|
||
|
|
device_manager {
|
||
|
|
name: "cmvr-quic-process-smoke"
|
||
|
|
version: "test"
|
||
|
|
description: "no-hardware QUIC process smoke test"
|
||
|
|
devices {
|
||
|
|
id: "disabled-smoke-camera"
|
||
|
|
type: DEVICE_TYPE_CAMERA
|
||
|
|
enable: false
|
||
|
|
}
|
||
|
|
devices {
|
||
|
|
id: "disabled-smoke-microphone"
|
||
|
|
type: DEVICE_TYPE_MICROPHONE
|
||
|
|
enable: false
|
||
|
|
}
|
||
|
|
devices {
|
||
|
|
id: "enabled-missing-camera"
|
||
|
|
type: DEVICE_TYPE_CAMERA
|
||
|
|
enable: true
|
||
|
|
}
|
||
|
|
}
|
||
|
|
""".lstrip(),
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
(config_dir / "task_manager.pb.txt").write_text(
|
||
|
|
"""
|
||
|
|
task_manager {
|
||
|
|
tasks {
|
||
|
|
id: "quic_edge"
|
||
|
|
type: TASK_TYPE_QUIC_EDGE
|
||
|
|
run_mode: TASK_RUN_MODE_BLOCKING_SERVICE
|
||
|
|
config_file: "quic_edge.pb.txt"
|
||
|
|
enable: true
|
||
|
|
}
|
||
|
|
}
|
||
|
|
""".lstrip(),
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
(config_dir / "quic_edge.pb.txt").write_text(
|
||
|
|
f"""
|
||
|
|
quic_edge {{
|
||
|
|
id: "quic_edge"
|
||
|
|
server_host: "127.0.0.1"
|
||
|
|
server_port: {port}
|
||
|
|
alpn: "cmvr-quic-edge/1"
|
||
|
|
node_id: "cmvr-process-smoke"
|
||
|
|
software_version: "test"
|
||
|
|
grpc_endpoint_host: "127.0.0.1"
|
||
|
|
grpc_endpoint_port: 50052
|
||
|
|
grpc_endpoint_tls: false
|
||
|
|
include_loopback_interfaces: true
|
||
|
|
heartbeat_interval_ms: 250
|
||
|
|
control_response_timeout_ms: 1000
|
||
|
|
tls {{ allow_insecure: true }}
|
||
|
|
reconnect {{
|
||
|
|
initial_delay_ms: 50
|
||
|
|
maximum_delay_ms: 250
|
||
|
|
multiplier: 2.0
|
||
|
|
jitter_percent: 0
|
||
|
|
connect_timeout_ms: 2000
|
||
|
|
}}
|
||
|
|
maximum_datagram_bytes: 1200
|
||
|
|
maximum_control_frame_bytes: 1048576
|
||
|
|
maximum_frame_bytes: 16384
|
||
|
|
datagram_send_queue_depth: 64
|
||
|
|
media_poll_interval_ms: 2
|
||
|
|
}}
|
||
|
|
""".lstrip(),
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
return config_dir / "cmvr_es.pb.txt"
|
||
|
|
|
||
|
|
|
||
|
|
def require_count(summary: dict[str, object], key: str, minimum: int) -> None:
|
||
|
|
value = summary.get(key)
|
||
|
|
if not isinstance(value, int) or value < minimum:
|
||
|
|
raise RuntimeError(
|
||
|
|
f"gateway summary {key}={value!r}, expected at least {minimum}"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def run() -> int:
|
||
|
|
args = parse_args()
|
||
|
|
for path in (args.gateway, args.cmvr_es, args.cert, args.key):
|
||
|
|
if not path.is_file():
|
||
|
|
raise FileNotFoundError(path)
|
||
|
|
|
||
|
|
deadline = time.monotonic() + args.timeout_seconds
|
||
|
|
gateway_process: subprocess.Popen[str] | None = None
|
||
|
|
edge_process: subprocess.Popen[str] | None = None
|
||
|
|
|
||
|
|
with tempfile.TemporaryDirectory(prefix="cmvr-es-quic-smoke-") as temp:
|
||
|
|
temp_dir = Path(temp)
|
||
|
|
ready_file = temp_dir / "ready.json"
|
||
|
|
summary_file = temp_dir / "summary.json"
|
||
|
|
gateway_log = temp_dir / "gateway.log"
|
||
|
|
edge_log = temp_dir / "cmvr_es.log"
|
||
|
|
|
||
|
|
try:
|
||
|
|
with gateway_log.open("w", encoding="utf-8") as gateway_output:
|
||
|
|
gateway_process = subprocess.Popen(
|
||
|
|
[
|
||
|
|
str(args.gateway),
|
||
|
|
"--bind",
|
||
|
|
"127.0.0.1",
|
||
|
|
"--port",
|
||
|
|
"0",
|
||
|
|
"--cert",
|
||
|
|
str(args.cert),
|
||
|
|
"--key",
|
||
|
|
str(args.key),
|
||
|
|
"--heartbeat-interval-ms",
|
||
|
|
# A zero response keeps the edge-side value below,
|
||
|
|
# proving heartbeat_interval_ms is configurable.
|
||
|
|
"0",
|
||
|
|
"--ready-file",
|
||
|
|
str(ready_file),
|
||
|
|
"--summary-file",
|
||
|
|
str(summary_file),
|
||
|
|
"--exit-after-heartbeats",
|
||
|
|
# The production client keeps only one heartbeat
|
||
|
|
# outstanding. Receiving heartbeat 3 therefore proves
|
||
|
|
# that ACKs 1 and 2 were processed by cmvr_es.
|
||
|
|
"3",
|
||
|
|
],
|
||
|
|
stdout=gateway_output,
|
||
|
|
stderr=subprocess.STDOUT,
|
||
|
|
text=True,
|
||
|
|
)
|
||
|
|
ready = wait_for_json(ready_file, gateway_process, deadline)
|
||
|
|
port = ready.get("port")
|
||
|
|
if not isinstance(port, int) or not 0 < port <= 65535:
|
||
|
|
raise RuntimeError(f"invalid gateway ready payload: {ready}")
|
||
|
|
|
||
|
|
root_config = write_runtime_config(temp_dir / "config", port)
|
||
|
|
with edge_log.open("w", encoding="utf-8") as edge_output:
|
||
|
|
edge_process = subprocess.Popen(
|
||
|
|
[str(args.cmvr_es), str(root_config)],
|
||
|
|
stdout=edge_output,
|
||
|
|
stderr=subprocess.STDOUT,
|
||
|
|
text=True,
|
||
|
|
)
|
||
|
|
remaining = max(0.1, deadline - time.monotonic())
|
||
|
|
gateway_returncode = gateway_process.wait(timeout=remaining)
|
||
|
|
if gateway_returncode != 0:
|
||
|
|
raise RuntimeError(
|
||
|
|
f"gateway exited with {gateway_returncode}"
|
||
|
|
)
|
||
|
|
|
||
|
|
edge_returncode = stop_process(edge_process)
|
||
|
|
edge_process = None
|
||
|
|
if edge_returncode != 0:
|
||
|
|
raise RuntimeError(
|
||
|
|
f"cmvr_es exited with {edge_returncode}"
|
||
|
|
)
|
||
|
|
|
||
|
|
summary = json.loads(summary_file.read_text(encoding="utf-8"))
|
||
|
|
if summary.get("runtime_failed") is not False:
|
||
|
|
raise RuntimeError(f"gateway runtime failure: {summary}")
|
||
|
|
require_count(summary, "registrations_accepted", 1)
|
||
|
|
require_count(summary, "heartbeats_received", 3)
|
||
|
|
require_count(summary, "heartbeat_acks_sent", 3)
|
||
|
|
require_count(summary, "registration_interface_count", 1)
|
||
|
|
require_count(summary, "heartbeat_interface_count", 1)
|
||
|
|
if summary.get("heartbeat_has_device_manager") is not True:
|
||
|
|
raise RuntimeError(
|
||
|
|
f"DeviceManager snapshot was not received: {summary}"
|
||
|
|
)
|
||
|
|
if (
|
||
|
|
summary.get("heartbeat_device_manager_name")
|
||
|
|
!= "cmvr-quic-process-smoke"
|
||
|
|
or summary.get("heartbeat_device_manager_version") != "test"
|
||
|
|
or summary.get("heartbeat_device_count") != 1
|
||
|
|
or summary.get("heartbeat_enabled_device_count") != 1
|
||
|
|
or summary.get("heartbeat_disabled_device_count") != 0
|
||
|
|
or summary.get("heartbeat_error_device_count") != 1
|
||
|
|
or summary.get("heartbeat_unknown_health_device_count") != 1
|
||
|
|
):
|
||
|
|
raise RuntimeError(
|
||
|
|
f"unexpected DeviceManager heartbeat snapshot: {summary}"
|
||
|
|
)
|
||
|
|
devices = summary.get("heartbeat_devices")
|
||
|
|
if not isinstance(devices, list) or len(devices) != 1:
|
||
|
|
raise RuntimeError(
|
||
|
|
f"heartbeat device rows are missing: {summary}"
|
||
|
|
)
|
||
|
|
row = devices[0]
|
||
|
|
if (
|
||
|
|
not isinstance(row, dict)
|
||
|
|
or row.get("device_id") != "enabled-missing-camera"
|
||
|
|
or row.get("kind") != 5
|
||
|
|
or row.get("type_name") != "Camera"
|
||
|
|
or row.get("enabled") is not True
|
||
|
|
or row.get("manager_state") != 7
|
||
|
|
or row.get("health") != 0
|
||
|
|
or row.get("has_error") is not True
|
||
|
|
or row.get("error_message") != "device creation failed"
|
||
|
|
or not isinstance(
|
||
|
|
row.get("status_updated_at_unix_ms"), int
|
||
|
|
)
|
||
|
|
or row["status_updated_at_unix_ms"] <= 0
|
||
|
|
):
|
||
|
|
raise RuntimeError(
|
||
|
|
f"unexpected heartbeat device row: {row!r}"
|
||
|
|
)
|
||
|
|
if summary.get("protocol_violations") != 0:
|
||
|
|
raise RuntimeError(f"protocol violation: {summary}")
|
||
|
|
if summary.get("last_node_id") != "cmvr-process-smoke":
|
||
|
|
raise RuntimeError(f"unexpected registered node: {summary}")
|
||
|
|
if (
|
||
|
|
summary.get("last_grpc_endpoint_host") != "127.0.0.1"
|
||
|
|
or summary.get("last_grpc_endpoint_port") != 50052
|
||
|
|
or summary.get("last_observed_source_ip") != "127.0.0.1"
|
||
|
|
):
|
||
|
|
raise RuntimeError(
|
||
|
|
f"IP/gRPC endpoint report was not observed: {summary}"
|
||
|
|
)
|
||
|
|
|
||
|
|
edge_output_text = edge_log.read_text(encoding="utf-8")
|
||
|
|
if "[QuicEdgeTask] Started" not in edge_output_text:
|
||
|
|
raise RuntimeError(
|
||
|
|
"cmvr_es did not start the QUIC task:\n" + edge_output_text
|
||
|
|
)
|
||
|
|
print(
|
||
|
|
"cmvr_es_quic_process_smoke_test: PASS "
|
||
|
|
+ json.dumps(summary, sort_keys=True)
|
||
|
|
)
|
||
|
|
return 0
|
||
|
|
except Exception:
|
||
|
|
if gateway_log.is_file():
|
||
|
|
print("gateway log:\n" + gateway_log.read_text(encoding="utf-8"))
|
||
|
|
if edge_log.is_file():
|
||
|
|
print("cmvr_es log:\n" + edge_log.read_text(encoding="utf-8"))
|
||
|
|
raise
|
||
|
|
finally:
|
||
|
|
stop_process(edge_process)
|
||
|
|
stop_process(gateway_process)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
raise SystemExit(run())
|