Have Device.write accept a full byte sequence, not just a single byte

pyvisa_sim/devices.py

@@ -232,10 +232,6 @@ def write(self, data: bytes) -> None:
         if not isinstance(data, bytes):
             raise TypeError("data must be an instance of bytes")
 
-        if len(data) != 1:
-            msg = "data must have a length of 1, not %d"
-            raise ValueError(msg % len(data))
-
         self._input_buffer.extend(data)
 
         le = len(self._query_eom)

pyvisa_sim/sessions/serial.py

@@ -84,11 +84,10 @@ def write(self, data: bytes) -> Tuple[int, constants.StatusCode]:
         if asrl_end == constants.SerialTermination.last_bit:
             last_bit, _ = self.get_attribute(constants.ResourceAttribute.asrl_data_bits)
             mask = 1 << (last_bit - 1)
-            for val in common.iter_bytes(data, mask, send_end):
-                self.device.write(val)
+            val = b"".join(common.iter_bytes(data, mask, send_end))
+            self.device.write(val)
         else:
-            for i in range(len(data)):
-                self.device.write(data[i : i + 1])
+            self.device.write(data)
 
             if asrl_end == constants.SerialTermination.termination_char:
                 if send_end:

pyvisa_sim/sessions/session.py

@@ -239,8 +239,7 @@ def read(self, count: int) -> Tuple[bytes, constants.StatusCode]:
     def write(self, data: bytes) -> Tuple[int, constants.StatusCode]:
         send_end = self.get_attribute(constants.ResourceAttribute.send_end_enabled)
 
-        for i in range(len(data)):
-            self.device.write(data[i : i + 1])
+        self.device.write(data)
 
         if send_end:
             # EOM4882

pyvisa_sim/testsuite/test_all.py

@@ -1,4 +1,6 @@
 # -*- coding: utf-8 -*-
+import logging
+
 import pytest
 from pyvisa.errors import VisaIOError
 
@@ -179,3 +181,18 @@ def test_instrument_for_error_state(resource, resource_manager):
 
     inst.write(":VOLT:IMM:AMPL 0")
     assert_instrument_response(inst, ":SYST:ERR?", "1, Command error")
+
+
+def test_device_write_logging(caplog, resource_manager) -> None:
+    instr = resource_manager.open_resource(
+        "USB0::0x1111::0x2222::0x4444::0::INSTR",
+        read_termination="\n",
+        write_termination="\n",
+    )
+
+    with caplog.at_level(logging.DEBUG):
+        instr.write("*IDN?")
+        instr.read()
+
+    assert "input buffer: b'D'" not in caplog.text
+    assert r"input buffer: b'*IDN?\n'" in caplog.text

pyvisa_sim/testsuite/test_common.py

@@ -0,0 +1,40 @@
+from typing import List, Optional
+
+import pytest
+
+from pyvisa_sim import common
+
+
+@pytest.mark.parametrize(
+    "data, mask, send_end, want",
+    [
+        (b"1234", None, False, b"1234"),
+        (b"41234", 3, False, b"40004"),
+        # Can't figure this one out. Getting ValueError: bytes must in in range(0, 256)
+        # If I knew more about the purpose of `iter_bytes` then maybe I could
+        # reconcile it, but I don't have time to investigate right now.
+        pytest.param(
+            b"1234",
+            3,
+            True,
+            b"1234",
+            marks=pytest.mark.xfail(
+                reason=(
+                    "ValueError bytes must be in range(0, 256). TODO: figure"
+                    " out correct 'want' value."
+                )
+            ),
+        ),
+    ],
+)
+def test_iter_bytes(
+    data: bytes, mask: Optional[int], send_end: bool, want: List[bytes]
+) -> None:
+    got = b"".join(common.iter_bytes(data, mask=mask, send_end=send_end))
+    assert got == want
+
+
+def test_iter_bytes_with_send_end_requires_mask() -> None:
+    with pytest.raises(ValueError):
+        # Need to wrap in list otherwise the iterator is never called.
+        list(common.iter_bytes(b"", mask=None, send_end=True))

pyvisa_sim/testsuite/test_serial.py

@@ -0,0 +1,29 @@
+# -*- coding: utf-8 -*-
+import pyvisa
+
+from pyvisa_sim.sessions import serial
+
+serial.SerialInstrumentSession
+
+
+def test_serial_write_with_termination_last_bit(resource_manager):
+    instr = resource_manager.open_resource(
+        "ASRL4::INSTR",
+        read_termination="\n",
+        write_termination="\r\n",
+    )
+
+    # Ensure that we test the `asrl_end` block of serial.SerialInstrumentSession.write
+    instr.set_visa_attribute(
+        pyvisa.constants.ResourceAttribute.asrl_end_out,
+        pyvisa.constants.SerialTermination.last_bit,
+    )
+
+    # There's a bug (maybe?) in common.iter_bytes that we want to avoid for now.
+    instr.set_visa_attribute(
+        pyvisa.constants.ResourceAttribute.send_end_enabled,
+        pyvisa.constants.VI_FALSE,
+    )
+
+    instr.write("*IDN?")
+    assert instr.read() == "SCPI,MOCK,VERSION_1.0"