| #!/usr/bin/env python3 |
| # SPDX-License-Identifier: MIT |
| # One MTP test session for one board, in a disposable process. Every libmtp call is |
| # synchronous ctypes in our own address space and blocks in a usbfs ioctl in D state on |
| # a wedged device, where not even SIGKILL is delivered — so the session must be |
| # something the harness can abandon: hil_test.test_device_mtp runs it under |
| # hil_util.run_cmd (killpg + bounded reap, rc 124 on timeout). Imports stay stdlib + |
| # pymtp: nothing here may pull in the harness. |
| # |
| # Exit 0 on a fully passing session; 1 with the failure on stdout/stderr otherwise. |
| import argparse |
| import ctypes |
| import glob |
| import hashlib |
| import os |
| import signal |
| import subprocess |
| import sys |
| import threading |
| import time |
| |
| sys.path.append(os.path.dirname(os.path.abspath(__file__))) # PYTHONSAFEPATH drops it |
| # -- APPEND so PYTHONPATH still wins (the tests steer a fake pymtp that way) |
| |
| from pathlib import Path |
| from pymtp import LIBMTP_DeviceEntry, LIBMTP_RawDevice, MTP |
| |
| FILE1_EXPECT = b'TinyUSB MTP Filesystem example' |
| FILE2_MD5_EXPECT = '40ef23fc2891018d41a05d4a0d5f822f' # md5sum of logo.png |
| |
| |
| # Real paths by default; the offline tests point these at a fixture tree, the same way |
| # they steer the pymtp fake through FAKE_PYMTP_*. |
| # The one test seam: '' in production, a tmpdir in the offline tests, which mirror the |
| # real layout beneath it. This runs as a SUBPROCESS (a libmtp call blocked in a usbfs |
| # ioctl hangs its thread forever, so the session must be somewhere killable), and neither |
| # monkeypatching nor import shadowing crosses that boundary -- unlike the fake pymtp, |
| # which the tests inject through PYTHONPATH alone. |
| _ROOT = os.environ.get('HIL_MTP_FAKE_ROOT', '') |
| _MARKER_GLOB = f'{_ROOT}/dev/libmtp-*' |
| _SYS_USB = Path(f'{_ROOT}/sys/bus/usb/devices') |
| _USB_DEV = Path(f'{_ROOT}/dev/bus/usb') |
| |
| |
| def _bounded_read(path, grace: float = 2.0): |
| """Read a sysfs attribute with a wall-clock bound, or return None. |
| |
| `serial` is served under the device lock a wedged usbfs ioctl holds, and EVERY MTP DUT |
| is cafe:4017 -- so the vid/pid filter below cannot rule out a wedged NEIGHBOUR, and an |
| unbounded read of its serial would burn this session's whole budget and report a |
| healthy board as wedged. Stdlib only by design (this file never imports the harness), |
| so this is a small local twin of hil_util.read_sysfs. |
| """ |
| out = {} |
| |
| def _read(): |
| try: |
| out['v'] = path.read_text().strip() |
| except OSError: |
| pass |
| |
| t = threading.Thread(target=_read, daemon=True) |
| t.start() |
| t.join(grace) |
| return out.get('v') |
| |
| |
| def _ready_marker(uid: str): |
| """(busnum, devnum) of the udev-ready MTP device with this serial, or None. |
| |
| /dev/libmtp-<sysname> is published by libmtp-runtime AFTER its synchronous mtp-probe |
| accepts the device, so this set is both small and ready -- unlike a sysfs-wide scan, |
| which races re-enumerations from other boards' jobs. Requires the libmtp-runtime |
| package. |
| """ |
| for marker_name in glob.glob(_MARKER_GLOB): |
| marker = Path(marker_name) |
| try: |
| dev = _SYS_USB / marker.name[len('libmtp-'):] |
| # vid/pid first: lock-free descriptor fields, so they rule out every other |
| # device before the `serial` read, which the kernel serves under the device |
| # lock a wedged usbfs ioctl would hold |
| if ((dev / 'idVendor').read_text().strip() != 'cafe' |
| or (dev / 'idProduct').read_text().strip() != '4017'): |
| continue |
| # bounded: this one CAN block, and a wedged neighbour shares the vid/pid above |
| serial = _bounded_read(dev / 'serial') |
| if serial is None or serial.lower() != uid.lower(): |
| continue |
| busnum = int((dev / 'busnum').read_text()) |
| devnum = int((dev / 'devnum').read_text()) |
| node = _USB_DEV / f'{busnum:03d}' / f'{devnum:03d}' |
| if marker.resolve(strict=True) != node or not os.access(node, os.R_OK | os.W_OK): |
| continue |
| return busnum, devnum |
| except (OSError, ValueError): |
| # a marker can vanish while another board flashes: not our device's problem |
| continue |
| return None |
| |
| |
| def _gvfs_unmount(uid: str, deadline: float) -> None: |
| """Drop any gvfs claim on this device, immediately before opening it. |
| |
| Called only once the udev marker exists. gvfs claims an MTP device AFTER udev |
| probing, so before the marker there is nothing to unmount: an earlier call is a |
| guaranteed no-op that still forks a process, and it leaves the gap between the |
| unmount and the open unprotected -- the hang this exists to prevent. Per-iteration |
| calls also forked one gio per second of the enumeration budget. |
| """ |
| # Popen, not run(timeout=): run's post-timeout reap is an unbounded wait(), and a gio |
| # blocked in D state on a wedged usbfs node does not die on SIGKILL, so run(timeout=2) |
| # can hang for good. Bounded by at most HALF of what is LEFT of our own budget, never |
| # a fixed sub-bound: the parent gives us --timeout 8 (4 on a retry), so anything larger |
| # collapsed the poll loop to one attempt and made a slow gio look like a wedged session. |
| gio_bound = max(0.5, min(3.0, (deadline - time.monotonic()) / 2)) |
| try: |
| # argv, not shell=True: uid comes from a hand-edited roster and is board firmware |
| # output, so a space or $(...) would unmount the wrong URI (leaving the gvfs mount |
| # held) or run as us. |
| gio = subprocess.Popen(['gio', 'mount', '-u', |
| f'mtp://TinyUsb_TinyUsb_Device_{uid}/'], |
| stdout=subprocess.DEVNULL, |
| stderr=subprocess.DEVNULL, start_new_session=True) |
| except OSError: |
| # glib2.0-bin absent (ci.lan has no gio at all): nothing holds a gvfs mount |
| # either, so go straight on to the open. |
| return |
| try: |
| gio.wait(timeout=gio_bound) |
| except subprocess.TimeoutExpired: |
| try: |
| os.killpg(gio.pid, signal.SIGKILL) |
| except OSError: |
| gio.kill() |
| try: |
| gio.wait(timeout=2) # reap it: an abandoned gio leaves a zombie |
| except subprocess.TimeoutExpired: |
| pass |
| print('gio unmount timed out; continuing', file=sys.stderr) |
| |
| |
| def open_mtp_dev(uid: str, timeout: float): |
| mtp = MTP() |
| deadline = time.monotonic() + timeout |
| while True: |
| try: |
| # pymtp raises USB_LAYER/PTP_LAYER/GENERAL/AlreadyConnected on a board still |
| # settling right after a flash; an unguarded raise would skip the rest of the |
| # enumeration budget (and the disconnect) instead of retrying. |
| # |
| # Never detect_devices(): that PROBES every MTP device on the rig, so a board |
| # still initialising in a parallel job answers our scan (the race #3790 fixed). |
| # libmtp-runtime publishes /dev/libmtp-<sysname> only after its own mtp-probe |
| # has accepted a device, so start from that small, ready-only set and open OUR |
| # device directly by bus/dev address. |
| target = _ready_marker(uid) |
| if target: |
| # ready first, THEN unmount, then open -- see _gvfs_unmount |
| _gvfs_unmount(uid, deadline) |
| busnum, devnum = target |
| # TinyUSB needs no libmtp quirks, so the raw entry can be built here |
| entry = LIBMTP_DeviceEntry(None, 0xcafe, None, 0x4017, 0) |
| raw = LIBMTP_RawDevice(entry, busnum, devnum) |
| mtp.device = mtp.mtp.LIBMTP_Open_Raw_Device(ctypes.byref(raw)) |
| if mtp.device: |
| serial = mtp.get_serialnumber() |
| if (serial.decode('utf-8') if serial else '').lower() == uid.lower(): |
| return mtp |
| mtp.disconnect() |
| except Exception as e: |
| print(f'mtp poll: {type(e).__name__}: {e}', file=sys.stderr) |
| # only when a device was actually opened: pymtp's `self.device == None` |
| # guard does NOT catch a ctypes NULL pointer (falsy, but != None), so |
| # disconnecting blindly calls LIBMTP_Release_Device(NULL) |
| if getattr(mtp, 'device', None): |
| try: |
| mtp.disconnect() |
| except Exception: |
| pass |
| mtp.device = None |
| if time.monotonic() >= deadline: |
| return None |
| time.sleep(1) |
| |
| |
| def run_session(uid: str, timeout: float) -> int: |
| mtp = open_mtp_dev(uid, timeout) |
| if mtp is None or mtp.device is None: |
| print('MTP device not found') |
| return 1 |
| |
| try: |
| assert b"TinyUSB" == mtp.get_manufacturer(), 'MTP wrong manufacturer' |
| assert b"MTP Example" == mtp.get_modelname(), 'MTP wrong model' |
| assert b'1.0' == mtp.get_deviceversion(), 'MTP wrong version' |
| assert b'TinyUSB MTP' == mtp.get_devicename(), 'MTP wrong device name' |
| |
| f1 = uid.encode("utf-8") + b'_file1' |
| f2 = uid.encode("utf-8") + b'_file2' |
| f3 = uid.encode("utf-8") + b'_file3' |
| mtp.get_file_to_file(1, f1) |
| with open(f1, 'rb') as file: |
| f1_data = file.read() |
| os.remove(f1) |
| assert f1_data == FILE1_EXPECT, 'MTP file1 wrong data' |
| mtp.get_file_to_file(2, f2) |
| with open(f2, 'rb') as file: |
| f2_data = file.read() |
| os.remove(f2) |
| assert FILE2_MD5_EXPECT == hashlib.md5(f2_data).hexdigest(), 'MTP file2 wrong data' |
| with open(f3, "wb") as file: |
| # 1524-byte payload + 12-byte MTP header = 3 full 512-byte buffers, so this |
| # exercises delivery of the final OUT payload before its ZLP. Deliberate and |
| # FIXED: a random size hits that boundary in ~0.2% of runs, which is not a test |
| # of it. Deterministic content so a mismatch is reproducible. |
| f3_data = bytes((i % 251) + 1 for i in range(1524)) |
| file.write(f3_data) |
| file.close() |
| fid = mtp.send_file_from_file(f3, b'file3') |
| f3_readback = f3 + b'_readback' |
| mtp.get_file_to_file(fid, f3_readback) |
| with open(f3_readback, 'rb') as f: |
| f3_rb_data = f.read() |
| os.remove(f3_readback) |
| assert f3_rb_data == f3_data, 'MTP file3 wrong data' |
| os.remove(f3) |
| mtp.delete_object(fid) |
| except AssertionError as e: |
| print(e) |
| return 1 |
| finally: |
| mtp.disconnect() |
| return 0 |
| |
| |
| def main() -> int: |
| parser = argparse.ArgumentParser() |
| parser.add_argument('--uid', required=True, help='board_get_unique_id serial to match') |
| parser.add_argument('--timeout', type=float, default=30, help='enumeration wait budget (s)') |
| args = parser.parse_args() |
| return run_session(args.uid, args.timeout) |
| |
| |
| if __name__ == '__main__': |
| sys.exit(main()) |
