Files
progpib/tests/test_controller.py
S Groesz 97a632435d fix: compute instrument read timeout before sending the read command
The previous fix computed the host-side wait (via a read_tmo_ms query)
after already sending "++read eoi" -- but the controller doesn't answer
any further command while that read is outstanding, so the read_tmo_ms
query itself raced the read it was meant to be timing and could time
out. Reordered so the timeout is resolved first in both query() and
read(); added a regression test pinning the wire ordering.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-24 18:55:26 -05:00

100 lines
3.6 KiB
Python

import time
import pytest
from bits import Bit
from progpib.controller import GpibController
from progpib.exceptions import CommandError
def _wait_until(condition, timeout=2.0, interval=0.02):
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if condition():
return True
time.sleep(interval)
return condition()
def test_mode_property_roundtrip(fake_gpib_server):
server, tcp_server = fake_gpib_server
with GpibController.ethernet(tcp_server.host, tcp_server.port) as ctrl:
ctrl.mode = 1
assert ctrl.mode == 1
ctrl.mode = 0
assert ctrl.mode == 0
def test_addr_property_roundtrip_as_tuple(fake_gpib_server):
server, tcp_server = fake_gpib_server
with GpibController.ethernet(tcp_server.host, tcp_server.port) as ctrl:
ctrl.addr = (5, 96)
assert ctrl.addr == (5, 96)
ctrl.addr = 12
assert ctrl.addr == 12
def test_ver_returns_server_version(fake_gpib_server):
server, tcp_server = fake_gpib_server
with GpibController.ethernet(tcp_server.host, tcp_server.port) as ctrl:
assert ctrl.ver() == server.version
def test_write_escapes_special_bytes_on_the_wire(fake_gpib_server):
server, tcp_server = fake_gpib_server
with GpibController.ethernet(tcp_server.host, tcp_server.port) as ctrl:
ctrl.write("*IDN?\r\n+test")
expected = b"*IDN?\x1b\r\x1b\n\x1b+test\n"
assert _wait_until(lambda: bytes(server.raw_received) == expected)
def test_read_tmo_ms_range_validation_without_network(fake_gpib_server):
server, tcp_server = fake_gpib_server
with GpibController.ethernet(tcp_server.host, tcp_server.port) as ctrl:
with pytest.raises(ValueError):
ctrl.read_tmo_ms = 5000
# unchanged: the invalid value was never sent
assert ctrl.read_tmo_ms == 500
def test_status_and_lon_raise_in_controller_mode(fake_gpib_server):
server, tcp_server = fake_gpib_server
server.state["mode"] = "1" # CONTROLLER
with GpibController.ethernet(tcp_server.host, tcp_server.port) as ctrl:
with pytest.raises(CommandError):
_ = ctrl.status
with pytest.raises(CommandError):
_ = ctrl.lon
def test_query_fetches_read_tmo_ms_before_sending_read_eoi(fake_gpib_server):
# Regression test: found against real hardware that the controller
# doesn't answer any further command while a "++read" it issued is
# still outstanding (waiting for EOI or its own read_tmo_ms timeout).
# So query()/read() must fetch read_tmo_ms (to size the host-side wait)
# *before* sending "++read eoi" -- fetching it after races the read
# it's meant to be timing. Assert that ordering on the wire.
server, tcp_server = fake_gpib_server
with GpibController.ethernet(tcp_server.host, tcp_server.port) as ctrl:
ctrl.query("*IDN?", read_timeout=None)
assert _wait_until(lambda: b"++read eoi" in server.raw_received)
raw = bytes(server.raw_received)
assert raw.index(b"++read_tmo_ms") < raw.index(b"++read eoi")
def test_status_and_spoll_return_bits_with_rqs_at_index_6(fake_gpib_server):
server, tcp_server = fake_gpib_server
server.state["mode"] = "0" # DEVICE
server.state["status"] = "72" # 0x48: RQS (bit 6) and bit 3 set
server.spoll_response = 72
with GpibController.ethernet(tcp_server.host, tcp_server.port) as ctrl:
status = ctrl.status
assert int(status) == 72
assert status[6] == Bit(1)
assert status[5] == Bit(0)
result = ctrl.spoll()
assert int(result) == 72
assert result[6] == Bit(1)