This is an automated email from the ASF dual-hosted git repository.
acassis pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nuttx-ntfc.git
The following commit(s) were added to refs/heads/main by this push:
new f68efe4 ntfc: honor line_buffered on QEMU devices
f68efe4 is described below
commit f68efe4e6080c0fa7018664cd284789d3314cf55
Author: raiden00pl <[email protected]>
AuthorDate: Mon Aug 31 18:29:39 2026 +0200
ntfc: honor line_buffered on QEMU devices
honor line_buffered on QEMU devices
Signed-off-by: raiden00pl <[email protected]>
Assisted-by: Claude Code
---
Documentation/config-yaml.rst | 12 ++++++----
Documentation/config.yaml | 1 +
src/ntfc/device/host.py | 21 +++++++++++++++++
src/ntfc/device/qemu.py | 15 ------------
src/ntfc/device/sim.py | 29 +++--------------------
tests/device/test_qemu.py | 54 +++++++++++++++++++++++++++++++++++++++++++
6 files changed, 87 insertions(+), 45 deletions(-)
diff --git a/Documentation/config-yaml.rst b/Documentation/config-yaml.rst
index e23ad76..8383404 100644
--- a/Documentation/config-yaml.rst
+++ b/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)**
diff --git a/Documentation/config.yaml b/Documentation/config.yaml
index 6d17628..b091634 100644
--- a/Documentation/config.yaml
+++ b/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"
diff --git a/src/ntfc/device/host.py b/src/ntfc/device/host.py
index 6ccba99..d632387 100644
--- a/src/ntfc/device/host.py
+++ b/src/ntfc/device/host.py
@@ -43,6 +43,9 @@ if TYPE_CHECKING:
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 @@ class DeviceHost(DeviceCommon):
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():
diff --git a/src/ntfc/device/qemu.py b/src/ntfc/device/qemu.py
index 81521b8..a655d26 100644
--- a/src/ntfc/device/qemu.py
+++ b/src/ntfc/device/qemu.py
@@ -80,18 +80,3 @@ class DeviceQemu(DeviceHost):
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")
diff --git a/src/ntfc/device/sim.py b/src/ntfc/device/sim.py
index 57cb226..d4ebeb5 100644
--- a/src/ntfc/device/sim.py
+++ b/src/ntfc/device/sim.py
@@ -35,6 +35,9 @@ if TYPE_CHECKING:
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 @@ class DeviceSim(DeviceHost):
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")
diff --git a/tests/device/test_qemu.py b/tests/device/test_qemu.py
index ac062fb..c99741e 100644
--- a/tests/device/test_qemu.py
+++ b/tests/device/test_qemu.py
@@ -25,6 +25,60 @@ import pytest
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:
