Files
Jonathan Hui c9d3215b0f [otci] drain pending lines in OtCliCommandRunner before timing out (#13494)
In `OtCliCommandRunner.__expect_line()`, asynchronous commands loop
for `timeout` iterations, calling `self.wait(1)` in each step.
Under virtual-time simulation (`VIRTUAL_TIME=1`), `VirtualTime.go()`
advances simulation time without delay when no timer events or radio
traffic are scheduled on the node.

In each iteration, `self.wait(1)` calls `__pending_lines.get_nowait()`.
Because the background reader thread (`__otcli_reader`) transfers
output asynchronously from the subprocess stdout pipe to the queue,
a fast-forwarding simulation can exhaust all timeout iterations in
sub-millisecond wall-clock time before the reader thread has pushed
the buffered lines (such as `Done`) into `__pending_lines`. This
results in an intermittent and spurious `ExpectLineTimeoutError`.

This commit updates `__expect_line()` to drain `__pending_lines` with
a short timeout (0.1s total deadline) after the timeout loop
concludes, giving the reader thread sufficient time to process any
remaining stdout lines before raising `ExpectLineTimeoutError`.

TAG=agy
CONV=9ffc5984-d5ce-4533-ae04-87de08ed9ba7
2026-08-13 10:51:48 -07:00
..
2024-09-18 08:11:02 -07:00

OpenThread Control Interface

The OpenThread Control Interface (OTCI) is a library which provides uniform Python interfaces to connect and control various kinds of devices running OpenThread.

Supported device types

Example

import otci

# Connect to an OTBR device via SSH
node1 = otci.connect_otbr_ssh("192.168.1.101")

# Connect to an OpenThread CLI device via Serial
node2 = otci.connect_cli_serial("/dev/ttyACM0"))

# Start node1 and wait for it to become the Leader
node1.dataset_init_buffer()
node1.dataset_set_buffer(network_name='test', network_key='00112233445566778899aabbccddeeff', panid=0xface, channel=11)
node1.dataset_commit_buffer('active')

node1.ifconfig_up()
node1.thread_start()
node1.wait(5)
assert node1.get_state() == "leader"

# Start the Commissioner role on node1
node1.commissioner_start()
node1.wait(3)

node1.commissioner_add_joiner("TEST123",eui64='*')

# Start node2
node2.ifconfig_up()
node2.set_router_selection_jitter(1)

# Start the Joiner role on node2 and wait for it to join the network
node2.joiner_start("TEST123")
node2.wait(10, expect_line="Join success")

# Wait for node 2 to become a Router
node2.thread_start()
node2.wait(5)
assert node2.get_state() == "router"