357 lines
12 KiB
Python
357 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import struct
|
|
import sys
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
import usb.core
|
|
|
|
|
|
def trace_reply(
|
|
*,
|
|
magic=b"NHTR",
|
|
version=1,
|
|
status=0,
|
|
slot=4,
|
|
reserved=0,
|
|
time_us=123456,
|
|
generation=17,
|
|
):
|
|
return struct.pack(
|
|
"<4sBBBBII", magic, version, status, slot, reserved, time_us, generation
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def trace_rig(monkeypatch, tmp_path):
|
|
monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[1] / "tools"))
|
|
import native_joycon_hub_check as check
|
|
|
|
rig = SimpleNamespace(
|
|
check=check,
|
|
clock=0.0,
|
|
child_duration=0.25,
|
|
events=[],
|
|
child_error=usb.core.USBError("child descriptor timed out"),
|
|
capture=io.StringIO(),
|
|
)
|
|
monkeypatch.setattr(check, "time", SimpleNamespace(monotonic=lambda: rig.clock))
|
|
args = SimpleNamespace(
|
|
pairs=2,
|
|
timeout=10.0,
|
|
duration=2.0,
|
|
usb_timeout_ms=500,
|
|
rumble_sample=None,
|
|
input_only=True,
|
|
neutral=False,
|
|
output=tmp_path / "trace.json",
|
|
build_dir=tmp_path,
|
|
)
|
|
# Intentionally omit the new option: existing Namespace callers remain valid.
|
|
rig.scenario = check.Check(args, rig.capture)
|
|
|
|
class ControlPipe:
|
|
def __init__(self, owner, response):
|
|
self.owner = owner
|
|
self.response = response
|
|
self.transfers = []
|
|
|
|
def ctrl_transfer(
|
|
self, request_type, request, value, index, length, *, timeout
|
|
):
|
|
assert request_type & 0x80, "trace tests must never issue USB OUT transfers"
|
|
self.transfers.append(
|
|
(request_type, request, value, index, length, timeout)
|
|
)
|
|
rig.events.append(self.owner)
|
|
if self.owner != "root":
|
|
rig.clock += rig.child_duration
|
|
if isinstance(self.response, Exception):
|
|
raise self.response
|
|
return self.response
|
|
|
|
def attach_kernel_driver(self, interface):
|
|
rig.events.append("reattach")
|
|
|
|
rig.root = ControlPipe("root", trace_reply())
|
|
rig.child = ControlPipe("B_L", rig.child_error)
|
|
rig.scenario.devices = {"root": rig.root, "B_L": rig.child}
|
|
rig.scenario.util = SimpleNamespace(
|
|
release_interface=lambda device, interface: rig.events.append("release"),
|
|
dispose_resources=lambda device: rig.events.append("dispose"),
|
|
)
|
|
rig.scenario.claimed = [("B_L", 0)]
|
|
rig.scenario.detached = [("B_L", 0)]
|
|
|
|
def control(owner="B_L", name="device_descriptor"):
|
|
return rig.scenario.control(owner, name, 0x80, 6, 0x0100, 0, 18)
|
|
|
|
rig.control = control
|
|
return rig
|
|
|
|
|
|
@pytest.mark.parametrize("explicit_default", [False, True])
|
|
def test_child_error_does_not_mark_without_opt_in(trace_rig, explicit_default):
|
|
rig = trace_rig
|
|
if explicit_default:
|
|
rig.scenario.args.capture_trace_on_error = False
|
|
|
|
with pytest.raises(usb.core.USBError) as caught:
|
|
rig.control()
|
|
|
|
assert caught.value is rig.child_error
|
|
assert rig.root.transfers == []
|
|
assert len(rig.child.transfers) == 1
|
|
assert rig.scenario.result["failure_trace"] is None
|
|
|
|
|
|
def test_opt_in_captures_before_error_propagation_and_cleanup(trace_rig):
|
|
rig = trace_rig
|
|
rig.scenario.args.capture_trace_on_error = True
|
|
|
|
with pytest.raises(usb.core.USBError) as caught:
|
|
try:
|
|
rig.control()
|
|
except usb.core.USBError:
|
|
rig.events.append("propagated")
|
|
# The receipt is durable before the caller begins resource cleanup.
|
|
audit = json.loads(rig.capture.getvalue())
|
|
raise
|
|
finally:
|
|
rig.scenario.cleanup()
|
|
|
|
assert caught.value is rig.child_error
|
|
assert rig.events == [
|
|
"B_L",
|
|
"root",
|
|
"propagated",
|
|
"release",
|
|
"reattach",
|
|
"dispose",
|
|
"dispose",
|
|
]
|
|
assert len(rig.child.transfers) == 1
|
|
assert rig.root.transfers == [(0xC0, 0x5E, 0x5452, 4, 16, 500)]
|
|
trace = audit["failure_trace"]
|
|
assert trace["status"] == "captured" and trace["captured"]
|
|
assert trace["request_attempted"]
|
|
assert trace["response_hex"] == trace_reply().hex()
|
|
assert trace["receipt"] == {
|
|
"version": 1,
|
|
"status": 0,
|
|
"slot": 4,
|
|
"time_us": 123456,
|
|
"control_generation": 17,
|
|
}
|
|
assert trace["failed_control"]["setup"] == [0x80, 6, 0x0100, 0, 18]
|
|
assert trace["failed_control"]["side"] == "B_L"
|
|
assert trace["failed_control"]["error"] == str(rig.child_error)
|
|
|
|
|
|
def test_root_error_does_not_consume_the_single_child_marker(trace_rig):
|
|
rig = trace_rig
|
|
rig.scenario.args.capture_trace_on_error = True
|
|
root_error = usb.core.USBError("root descriptor failed")
|
|
rig.root.response = root_error
|
|
with pytest.raises(usb.core.USBError) as caught:
|
|
rig.control("root")
|
|
assert caught.value is root_error
|
|
assert len(rig.root.transfers) == 1
|
|
assert rig.scenario.result["failure_trace"] is None
|
|
|
|
rig.root.response = trace_reply()
|
|
for name in ("first_child_error", "later_child_error"):
|
|
with pytest.raises(usb.core.USBError) as caught:
|
|
rig.control(name=name)
|
|
assert caught.value is rig.child_error
|
|
|
|
assert [transfer[1] for transfer in rig.root.transfers] == [6, 0x5E]
|
|
audit = json.loads(rig.capture.getvalue())
|
|
assert audit["failure_trace"]["failed_control"]["name"] == "first_child_error"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"marker_error",
|
|
[usb.core.USBError("trace request stalled"), OSError("root disconnected")],
|
|
)
|
|
def test_failed_marker_preserves_original_error_and_is_not_retried(
|
|
trace_rig, marker_error
|
|
):
|
|
rig = trace_rig
|
|
rig.scenario.args.capture_trace_on_error = True
|
|
rig.root.response = marker_error
|
|
|
|
for name in ("first_child_error", "later_child_error"):
|
|
with pytest.raises(usb.core.USBError) as caught:
|
|
rig.control(name=name)
|
|
assert caught.value is rig.child_error
|
|
|
|
assert len(rig.root.transfers) == 1
|
|
trace = json.loads(rig.capture.getvalue())["failure_trace"]
|
|
assert trace["status"] == "error" and not trace["captured"]
|
|
assert trace["response_hex"] is None and trace["receipt"] is None
|
|
assert str(marker_error) in trace["error"]
|
|
assert trace["failed_control"]["name"] == "first_child_error"
|
|
|
|
|
|
def test_busy_marker_is_an_explicit_refusal_not_capture(trace_rig):
|
|
rig = trace_rig
|
|
rig.scenario.args.capture_trace_on_error = True
|
|
rig.root.response = trace_reply(status=1, time_us=0, generation=0)
|
|
|
|
with pytest.raises(usb.core.USBError) as caught:
|
|
rig.control()
|
|
|
|
assert caught.value is rig.child_error
|
|
trace = json.loads(rig.capture.getvalue())["failure_trace"]
|
|
assert trace["status"] == "busy" and not trace["captured"]
|
|
assert trace["response_hex"] == rig.root.response.hex()
|
|
assert trace["receipt"]["status"] == 1
|
|
assert trace["receipt"]["time_us"] == trace["receipt"]["control_generation"] == 0
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"response",
|
|
[
|
|
trace_reply()[:-1],
|
|
trace_reply() + b"\x00",
|
|
trace_reply(magic=b"NOPE"),
|
|
trace_reply(version=2),
|
|
trace_reply(slot=3),
|
|
trace_reply(reserved=1),
|
|
trace_reply(status=2),
|
|
trace_reply(status=1, time_us=1, generation=0),
|
|
trace_reply(status=1, time_us=0, generation=1),
|
|
],
|
|
ids=[
|
|
"short",
|
|
"long",
|
|
"magic",
|
|
"version",
|
|
"slot",
|
|
"reserved",
|
|
"status",
|
|
"busy-time",
|
|
"busy-generation",
|
|
],
|
|
)
|
|
def test_malformed_marker_never_qualifies_as_a_capture(trace_rig, response):
|
|
rig = trace_rig
|
|
rig.scenario.args.capture_trace_on_error = True
|
|
rig.root.response = response
|
|
|
|
with pytest.raises(usb.core.USBError) as caught:
|
|
rig.control()
|
|
|
|
assert caught.value is rig.child_error
|
|
trace = json.loads(rig.capture.getvalue())["failure_trace"]
|
|
assert trace["status"] == "malformed" and not trace["captured"]
|
|
assert trace["response_hex"] == response.hex()
|
|
assert trace["receipt"] is None
|
|
assert "error" in trace
|
|
|
|
|
|
def test_marker_timeout_uses_only_remaining_deadline(trace_rig):
|
|
rig = trace_rig
|
|
rig.scenario.args.capture_trace_on_error = True
|
|
rig.child_duration = rig.scenario.args.timeout - 0.125
|
|
|
|
with pytest.raises(usb.core.USBError) as caught:
|
|
rig.control()
|
|
|
|
assert caught.value is rig.child_error
|
|
assert rig.root.transfers == [(0xC0, 0x5E, 0x5452, 4, 16, 125)]
|
|
|
|
|
|
def test_expired_deadline_after_child_error_records_without_request(trace_rig):
|
|
rig = trace_rig
|
|
rig.scenario.args.capture_trace_on_error = True
|
|
rig.child_duration = rig.scenario.args.timeout
|
|
|
|
with pytest.raises(usb.core.USBError) as caught:
|
|
rig.control()
|
|
|
|
assert caught.value is rig.child_error
|
|
assert rig.root.transfers == []
|
|
trace = json.loads(rig.capture.getvalue())["failure_trace"]
|
|
assert trace["status"] == "deadline_expired" and not trace["captured"]
|
|
assert not trace["request_attempted"]
|
|
assert trace["failed_control"]["error"] == str(rig.child_error)
|
|
|
|
|
|
def test_pretransfer_deadline_does_not_count_as_a_child_usb_failure(trace_rig):
|
|
rig = trace_rig
|
|
rig.scenario.args.capture_trace_on_error = True
|
|
rig.clock = rig.scenario.deadline
|
|
|
|
with pytest.raises(TimeoutError):
|
|
rig.control()
|
|
|
|
assert rig.child.transfers == rig.root.transfers == []
|
|
assert rig.scenario.result["failure_trace"] is None
|
|
|
|
|
|
def test_capture_file_error_cannot_replace_child_transfer_error(trace_rig):
|
|
rig = trace_rig
|
|
rig.scenario.args.capture_trace_on_error = True
|
|
rig.capture.close()
|
|
|
|
with pytest.raises(usb.core.USBError) as caught:
|
|
rig.control()
|
|
|
|
assert caught.value is rig.child_error
|
|
trace = rig.scenario.result["failure_trace"]
|
|
assert trace["status"] == "captured"
|
|
assert "ValueError" in trace["checkpoint_error"]
|
|
|
|
|
|
def test_capture_option_marks_cli_failure_without_recovering(
|
|
trace_rig, monkeypatch, tmp_path
|
|
):
|
|
rig = trace_rig
|
|
capture = tmp_path / "option.json"
|
|
rig.child.owner = "R"
|
|
rig.root.response = trace_reply(slot=1)
|
|
|
|
def discover(scenario):
|
|
scenario.devices = {"root": rig.root, "R": rig.child, "L": rig.child}
|
|
scenario.util = rig.scenario.util
|
|
|
|
def claim(scenario):
|
|
scenario.claimed = [("R", 0)]
|
|
scenario.detached = [("R", 0)]
|
|
|
|
# Replace only private build references and the physical USB boundary. The
|
|
# CLI, descriptor control, failure path, audit and cleanup execute normally.
|
|
monkeypatch.setattr(
|
|
rig.check, "model_references", lambda *args, **kwargs: rig.check.child_models(1)
|
|
)
|
|
monkeypatch.setattr(rig.check.Check, "discover", discover)
|
|
monkeypatch.setattr(rig.check.Check, "permissions", lambda scenario: None)
|
|
monkeypatch.setattr(rig.check.Check, "claim", claim)
|
|
monkeypatch.setattr(
|
|
sys,
|
|
"argv",
|
|
[
|
|
"native_joycon_hub_check.py",
|
|
"--capture-trace-on-error",
|
|
"--output",
|
|
str(capture),
|
|
],
|
|
)
|
|
|
|
assert rig.check.main() == 2
|
|
audit = json.loads(capture.read_text())
|
|
assert not audit["success"]
|
|
assert audit["failure"] == str(rig.child_error)
|
|
assert audit["failure_trace"]["status"] == "captured"
|
|
assert audit["failure_trace"]["failed_control"]["side"] == "R"
|
|
assert audit["parameters"]["capture_trace_on_error"]
|
|
assert audit["safety"]["trace_marker_requested"]
|
|
assert len(rig.child.transfers) == 1
|
|
assert rig.root.transfers == [(0xC0, 0x5E, 0x5452, 1, 16, 500)]
|
|
assert rig.events[:4] == ["R", "root", "release", "reattach"]
|