honor line_buffered on QEMU devices

Documentation/config-yaml.rst

@@ -126,7 +126,8 @@ for serial targets without flow control and remains the default.
 
 Set ``line_buffered: true`` for a core to write a complete command in one
 transport operation instead. This reduces host-side overhead for ``sim`` and
-can also be enabled for serial targets with reliable flow control.
+``qemu`` and can also be enabled for serial targets with reliable flow
+control.
 
 .. code-block:: yaml
 
@@ -138,9 +139,12 @@ can also be enabled for serial targets with reliable flow control.
          device: "sim"
          line_buffered: true
 
-For the simulator, line-buffered mode also disables the pexpect per-send
-delay. Do not enable this option for serial targets that cannot
-reliably accept a full command at once.
+For host-based devices (``sim`` and ``qemu``), line-buffered mode also
+disables the pexpect per-send delay. Do not enable this option for serial
+targets that cannot reliably accept a full command at once.
+
+The target serial RX buffer must hold a whole command line, otherwise the
+tail of long commands is dropped and they time out.
 
 **SMP (Symmetric Multi-Processing)**
 

Documentation/config.yaml

@@ -91,6 +91,7 @@ product:                          # many products can be supported in tests (pro
       line_buffered: false        # (optional) send each command in one transport write.
                                   # Keep false for the default byte-wise behavior.
                                   # Enable for sim or a serial transport with flow control.
+                                  # Target serial RX buffer must hold a whole command line.
       dcmake:                     # (optional) Defines passed to CMake build
         DEFINE1: "VALUE1"
         DEFINE2: "VALUE2"

src/ntfc/device/host.py

@@ -43,6 +43,9 @@
 class DeviceHost(DeviceCommon):
     """This class implements common interface for host emulated devices."""
 
+    # appended when a command lacks a trailing newline
+    NEWLINE_PAD = b"\n"
+
     def __init__(self, conf: "CoreConfig"):
         """Initialize host based device.
 
@@ -77,6 +80,24 @@ def _dev_reopen(self) -> pexpect.spawn:
 
         return self.host_open(self._cmd)
 
+    def _write(self, data: bytes) -> None:
+        """Write to the host device."""
+        if not self.dev_is_health():
+            return
+
+        assert self._child
+
+        if data[-1] != ord("\n"):
+            data += self.NEWLINE_PAD
+
+        if self._conf.line_buffered:
+            self._child.send(data)
+            return
+
+        # send char by char to avoid line length full
+        for c in data:
+            self._child.send(bytes([c]))
+
     def _write_ctrl(self, c: str) -> None:
         """Write a control character to the host device."""
         if not self.dev_is_health():

src/ntfc/device/qemu.py

@@ -80,18 +80,3 @@ def _start_impl(self) -> None:
     def name(self) -> str:
         """Get device name."""
         return "qemu"
-
-    def _write(self, data: bytes) -> None:  # pragma: no cover
-        """Write to the host device."""
-        if not self.dev_is_health():
-            return
-
-        assert self._child
-
-        # send char by char to avoid line length full
-        for c in data:
-            self._child.send(bytes([c]))
-
-        # add new line if missing
-        if data[-1] != ord("\n"):
-            self._child.send(b"\n")

src/ntfc/device/sim.py

@@ -35,6 +35,9 @@
 class DeviceSim(DeviceHost):
     """This class implements host-based sim emulator."""
 
+    # sometimes sim misses a single trailing newline, so send two
+    NEWLINE_PAD = b"\n\n"
+
     def __init__(self, conf: "CoreConfig"):
         """Initialize sim emulator device."""
         DeviceHost.__init__(self, conf)
@@ -55,29 +58,3 @@ def _start_impl(self) -> None:
     def name(self) -> str:
         """Get device name."""
         return "sim"
-
-    def _write(self, data: bytes) -> None:
-        """Write to the host device."""
-        if not self.dev_is_health():
-            return
-
-        assert self._child
-
-        if self._conf.line_buffered:
-            if data[-1] != ord("\n"):
-                # Sometimes sim misses a single trailing newline.
-                data += b"\n\n"
-
-            self._child.send(data)
-            return
-
-        # send char by char to avoid line length full
-        for c in data:
-            self._child.send(bytes([c]))
-
-        # add new line if missing
-        if data[-1] != ord("\n"):
-            # sometimes new line send to sim is missing
-            # so we have to send more than one new line
-            self._child.send(b"\n")
-            self._child.send(b"\n")

tests/device/test_qemu.py

@@ -25,6 +25,60 @@
 from ntfc.device.qemu import DeviceQemu
 
 
+def test_device_qemu_write_adds_newline():
+    with patch("ntfc.coreconfig.CoreConfig") as mockdevice:
+        config = mockdevice.return_value
+        config.os = "nuttx"
+        config.read_poll_interval = 0.1
+        config.line_buffered = False
+        qemu = DeviceQemu(config)
+
+        sent = []
+
+        class FakeChild:
+            def isalive(self):
+                return True
+
+            def send(self, data):
+                sent.append(data)
+
+        qemu._child = FakeChild()
+
+        qemu._write(b"abc")
+        assert sent == [b"a", b"b", b"c", b"\n"]
+
+        sent.clear()
+        qemu._write(b"abc\n")
+        assert sent == [b"a", b"b", b"c", b"\n"]
+
+
+def test_device_qemu_line_buffered_write():
+    with patch("ntfc.coreconfig.CoreConfig") as mockdevice:
+        config = mockdevice.return_value
+        config.os = "nuttx"
+        config.read_poll_interval = 0.1
+        config.line_buffered = True
+        qemu = DeviceQemu(config)
+
+        sent = []
+
+        class FakeChild:
+            def isalive(self):
+                return True
+
+            def send(self, data):
+                sent.append(data)
+
+        qemu._child = FakeChild()
+
+        qemu._write(b"abc")
+        assert sent == [b"abc\n"]
+
+        sent.clear()
+        qemu._write(b"abc\n")
+        assert sent == [b"abc\n"]
+
+
 def test_device_qemu_open():
 
     with patch("ntfc.coreconfig.CoreConfig") as mockdevice: