88 lines
4.1 KiB
Python
88 lines
4.1 KiB
Python
"""Opt-in title-batch protocol/isolation adapter for multi_match_harness.py.
|
|
|
|
Run only after the complete Python suite is green, using the original CLI:
|
|
python tests/multi_match_title_harness.py --run --ticks 2 --fields 64 \
|
|
--phase-timeout 30 --output /absolute/path/title-result.json
|
|
|
|
This reuses every original phase, socket, assertion and cleanup operation. The
|
|
socket is an ACK double with simulated gather(), not the Go dispatcher's bounded
|
|
HTTP workers. Real Agent HTTP scheduling is tested separately in Go.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import multi_match_harness as original
|
|
|
|
|
|
MODE = "mapping_title_batch_v1_protocol_isolation"
|
|
LIMITATIONS = [
|
|
"Protocol/isolation simulation only; no Windows Agent executable, physical vMix or real HTTP worker scheduling",
|
|
"The unchanged HarnessSocket uses simulated concurrent gather(), not the Go five-bulk-plus-one-interactive worker pool",
|
|
"Go dispatcher HTTP concurrency, target ordering, cancellation and Countdown protection require the separate Go tests",
|
|
]
|
|
|
|
|
|
class TitleScenario(original.Scenario):
|
|
def __init__(self, project, directory, args):
|
|
# This adapter intentionally exercises exactly the newly capable Agent.
|
|
# The original CLI/parser and the legacy scenario remain untouched.
|
|
args.agent_version = "1.5.5"
|
|
super().__init__(project, directory, args)
|
|
|
|
def hello(self, name):
|
|
payload = super().hello(name)
|
|
payload["agent_version"] = "1.5.5"
|
|
payload["capabilities"] = {"mapping_title_batch_v1": True}
|
|
return payload
|
|
|
|
def title_checks(self):
|
|
live_flags = {device: getattr(live, "mapping_title_batch_v1", False) is True
|
|
for device, live in self.hub._live.items()}
|
|
batches = [packet for packet in self.frames if packet.get("type") == "vmix.batch"]
|
|
bulk = [packet for packet in batches if packet.get("mapping_bulk") is True]
|
|
return {
|
|
"six_live_agents_with_strict_capability": len(live_flags) == 6 and all(live_flags.values()),
|
|
"at_least_one_bulk_packet_larger_than_six": any(len(packet.get("commands", [])) > 6 for packet in bulk),
|
|
"every_mapping_batch_is_explicit_bulk": bool(batches) and len(bulk) == len(batches),
|
|
"all_packets_within_500_commands_and_48kib": bool(bulk) and all(
|
|
1 <= len(packet.get("commands", [])) <= 500 and
|
|
len(json.dumps(packet, ensure_ascii=False, separators=(",", ":")).encode("utf-8")) <= 48 * 1024
|
|
for packet in bulk),
|
|
"all_bulk_deadlines_are_bounded_integers": bool(bulk) and all(
|
|
type(packet.get("mapping_deadline_ms")) is int and 0 < packet["mapping_deadline_ms"] <= 59000
|
|
for packet in bulk),
|
|
"one_mapping_request_pending_per_agent": max((socket.max_pending for socket in self.sockets.values()), default=0) == 1,
|
|
"all_six_original_phases_passed": tuple(phase["phase"] for phase in self.phase_results) == (
|
|
"bootstrap", "steady_load", "negative_ack", "slow_source", "reconnect", "handoff") and all(
|
|
phase["ok"] is True for phase in self.phase_results),
|
|
"no_source_or_transport_isolation_errors": not self.source_errors and not self.transport_contract_errors
|
|
and not self.background_mapping_failures and not self.polling_errors,
|
|
}
|
|
|
|
def assert_isolation(self):
|
|
super().assert_isolation()
|
|
checks = self.title_checks()
|
|
assert all(checks.values()), checks
|
|
|
|
def report(self, ok):
|
|
report = super().report(ok)
|
|
checks = self.title_checks()
|
|
report.update(mode=MODE, title_transport_checks=checks,
|
|
live_capabilities={device: getattr(live, "mapping_title_batch_v1", False) is True
|
|
for device, live in self.hub._live.items()})
|
|
report["limits"].extend(LIMITATIONS)
|
|
if ok and not all(checks.values()):
|
|
report["ok"] = False
|
|
report["unexpected_failures"] += 1
|
|
return report
|
|
|
|
|
|
def main():
|
|
original.Scenario = TitleScenario
|
|
original.main()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|