diff --git a/CHANGES.md b/CHANGES.md index 56f7b7b..db752b5 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -4,6 +4,7 @@ - Add Renode device support. - Refuse commands longer than the target CONFIG_LINE_MAX. +- Reopen serial ports that disappear on target reset (USB re-enumeration). ## 0.0.3 diff --git a/src/ntfc/device/serial.py b/src/ntfc/device/serial.py index 808bb98..5e7c107 100644 --- a/src/ntfc/device/serial.py +++ b/src/ntfc/device/serial.py @@ -43,6 +43,7 @@ def __init__(self, conf: "CoreConfig"): """Initialize sim emulator device.""" DeviceCommon.__init__(self, conf, echo=False) self._ser = None + self._lost = False def _decode_exec_args(self, args: str) -> Dict[str, Any]: """Decode a serial port configuration string.""" @@ -87,8 +88,41 @@ def _decode_exec_args(self, args: str) -> Dict[str, Any]: except Exception as e: raise ValueError(f"Invalid format '{args}': {e}") from e + def _open(self) -> None: + """Open the serial port.""" + path = self._conf.exec_path + args = self._conf.exec_args + + if args: # pragma: no cover + self._ser = serial.Serial( + path, timeout=0, **self._decode_exec_args(args) + ) + else: + self._ser = serial.Serial(path, timeout=0) + + def _port_lost(self) -> None: + """Drop a port that failed, e.g. a USB device re-enumerating.""" + logger.info("serial port lost, reopening") + + assert self._ser + self._ser.close() + self._ser = None + self._lost = True + self._reopen() + + def _reopen(self) -> None: + """Try to reopen a lost port.""" + try: + self._open() + self._lost = False + except serial.SerialException: + self._ser = None + def _dev_is_health_priv(self) -> bool: """Check if the serial device is OK.""" + if not self._ser and self._lost: + self._reopen() + if not self._ser: return False @@ -101,19 +135,23 @@ def _write(self, data: bytes) -> None: assert self._ser - if self._conf.line_buffered: - if data[-1] != ord("\n"): - data += b"\n" - - self._ser.write(data) - else: - # send char by char to avoid line length full - for c in data: - self._ser.write(bytes([c])) - - # add new line if missing - if data[-1] != ord("\n"): - self._ser.write(b"\n") # pragma: no cover + try: + if self._conf.line_buffered: + if data[-1] != ord("\n"): + data += b"\n" + + self._ser.write(data) + else: + # send char by char to avoid line length full + for c in data: + self._ser.write(bytes([c])) + + # add new line if missing + if data[-1] != ord("\n"): + self._ser.write(b"\n") # pragma: no cover + except (serial.SerialException, OSError): + self._port_lost() + return # read all garbage left by character echo _ = self._read_all(timeout=0) @@ -136,22 +174,20 @@ def _read(self) -> bytes: assert self._ser - return self._ser.read(size=1024) + try: + return self._ser.read(size=1024) + except (serial.SerialException, OSError): + self._port_lost() + return b"" def _start_impl(self) -> None: """Start serial communication implementation.""" - timeout = 0 - path = self._conf.exec_path - args = self._conf.exec_args + logger.info(f"serial path: {self._conf.exec_path}") + logger.info(f"serial args: {self._conf.exec_args}") - logger.info(f"serial path: {path}") - logger.info(f"serial args: {args}") - - if args: # pragma: no cover - args = self._decode_exec_args(args) - self._ser = serial.Serial(path, timeout=timeout, **args) - else: - self._ser = serial.Serial(path, timeout=timeout) + # A USB port may show up only after the device boots + self._lost = True + self._reopen() # reboot device if possible self.reboot() @@ -177,6 +213,7 @@ def _stop_impl(self) -> None: if self._ser: # pragma: no cover self._ser.close() self._ser = None + self._lost = False logger.info("serial device closed") def _poweroff_impl(self) -> bool: diff --git a/tests/device/test_serial.py b/tests/device/test_serial.py index b08c56d..11f577f 100644 --- a/tests/device/test_serial.py +++ b/tests/device/test_serial.py @@ -123,8 +123,8 @@ def test_device_sim_init(serial_config, serial_pair): ser = DeviceSerial(serial_config) - # no path and args - with pytest.raises(serial.serialutil.SerialException): + # no port yet (USB device not enumerated): wait for boot, then time out + with pytest.raises(TimeoutError): serial_config._config["exec_path"] = "" serial_config._config["exec_args"] = "" ser.start() @@ -191,3 +191,38 @@ def fake_wait(timeout=5): # boot wait uses the configured boot_timeout assert boot_timeouts == [9] + + +def test_device_serial_reconnect(serial_config, serial_pair): + + serial_config._config["exec_path"] = serial_pair[1] + ser = DeviceSerial(serial_config) + ser._open() + first = ser._ser + + # a failed read (USB device gone) reopens the port + def fail(*args, **kwargs): + raise serial.SerialException("device disconnected") + + ser._ser.read = fail + assert ser._read() == b"" + assert ser._ser is not None + assert ser._ser is not first + assert ser._lost is False + + # a failed write while the port is still missing leaves it lost + serial_config._config["exec_path"] = "/dev/ntfc-no-such-port" + ser._ser.write = fail + assert ser._write(b"a") is None + assert ser._ser is None + assert ser._lost is True + assert ser._dev_is_health_priv() is False + + # the health check reconnects once the port is back + serial_config._config["exec_path"] = serial_pair[1] + assert ser._dev_is_health_priv() is True + assert ser._lost is False + + ser.stop() + assert ser._lost is False + assert ser.notalive is True