warehack.ing/openocd-python
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Apply ruff format to existing source and test files

Browse Source
This commit is contained in:
Ryan Malloy 2026-02-15 16:36:12 -07:00
parent 988234f4c3
commit 7a893cb328
22 changed files with 89 additions and 133 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/openocd/breakpoints.py
View File

@ -203,6 +203,7 @@ class BreakpointManager:
# Sync wrapper
# ======================================================================
class SyncBreakpointManager:
"""Synchronous wrapper around BreakpointManager."""

8
src/openocd/cli.py
View File

@ -27,12 +27,8 @@ def main() -> None:
prog="openocd-python",
description=f"OpenOCD Python bindings v{pkg_version}",
)
parser.add_argument(
"--version", action="version", version=f"openocd-python {pkg_version}"
)
parser.add_argument(
"--host", default="localhost", help="OpenOCD host (default: localhost)"
)
parser.add_argument("--version", action="version", version=f"openocd-python {pkg_version}")
parser.add_argument("--host", default="localhost", help="OpenOCD host (default: localhost)")
parser.add_argument(
"--port", type=int, default=6666, help="OpenOCD TCL RPC port (default: 6666)"
)

12
src/openocd/connection/tcl_rpc.py
View File

@ -131,9 +131,7 @@ class TclRpcConnection(Connection):
timeout=self._timeout,
)
except TimeoutError as exc:
raise OcdTimeoutError(
f"Timed out waiting for response to: {command}"
) from exc
raise OcdTimeoutError(f"Timed out waiting for response to: {command}") from exc
response = raw.decode("utf-8", errors="replace")
log.debug("RX: %s", response[:200])
@ -209,13 +207,9 @@ class TclRpcConnection(Connection):
# Read and discard the acknowledgement
ack_buf = bytearray()
while True:
chunk = await asyncio.wait_for(
self._notif_reader.read(4096), timeout=self._timeout
)
chunk = await asyncio.wait_for(self._notif_reader.read(4096), timeout=self._timeout)
if not chunk:
raise ConnectionError(
"Notification connection closed during setup"
)
raise ConnectionError("Notification connection closed during setup")
ack_buf.extend(chunk)
if ack_buf.find(SEPARATOR) != -1:
break

4
src/openocd/connection/telnet.py
View File

@ -92,9 +92,7 @@ class TelnetConnection(Connection):
raise ConnectionError("OpenOCD closed the connection")
buf.extend(chunk)
if len(buf) > MAX_RESPONSE_SIZE:
raise ConnectionError(
f"Response exceeded {MAX_RESPONSE_SIZE} bytes without prompt"
)
raise ConnectionError(f"Response exceeded {MAX_RESPONSE_SIZE} bytes without prompt")
if buf.endswith(PROMPT):
return bytes(buf[: -len(PROMPT)])

8
src/openocd/flash.py
View File

@ -168,12 +168,7 @@ class Flash:
# Read the file back through TCL to handle remote OpenOCD instances.
# Use ocd_find + binary read if available, otherwise fall back to
# reading the local file.
tcl_read = (
f"set fp [open {tmp_path} rb]; "
f"set data [read $fp]; "
f"close $fp; "
f"set data"
)
tcl_read = f"set fp [open {tmp_path} rb]; set data [read $fp]; close $fp; set data"
try:
raw = await self._conn.send(tcl_read)
# TCL returns binary as string; try base64 approach if garbled
@ -343,6 +338,7 @@ class Flash:
# Sync wrapper
# ======================================================================
class SyncFlash:
"""Synchronous wrapper around Flash for use outside async contexts."""

2
src/openocd/jtag/chain.py
View File

@ -93,6 +93,7 @@ def _parse_scan_chain(raw: str) -> list[TAPInfo]:
# JTAGController — unified facade
# ======================================================================
class JTAGController:
"""High-level async interface to all JTAG operations.
@ -161,6 +162,7 @@ class JTAGController:
# SyncJTAGController — blocking wrappers
# ======================================================================
class SyncJTAGController:
"""Synchronous wrapper around :class:`JTAGController`.

8
src/openocd/memory.py
View File

@ -151,9 +151,7 @@ class Memory:
hex_str = hex_str.ljust(49)
# ASCII portion
ascii_str = "".join(
chr(b) if 0x20 <= b < 0x7F else "." for b in chunk
)
ascii_str = "".join(chr(b) if 0x20 <= b < 0x7F else "." for b in chunk)
lines.append(f"{line_addr:08X}: {hex_str} |{ascii_str}|")
@ -179,9 +177,7 @@ class Memory:
try:
return [int(t, 16) for t in tokens]
except ValueError as exc:
raise TargetError(
f"Cannot parse read_memory response: {resp!r}"
) from exc
raise TargetError(f"Cannot parse read_memory response: {resp!r}") from exc
async def _write(self, addr: int, width: int, values: int | list[int]) -> None:
"""Write values of *width* bits using the TCL ``write_memory`` API.

12
src/openocd/process.py
View File

@ -59,9 +59,7 @@ class OpenOCDProcess:
self._tcl_port = tcl_port
binary = openocd_bin or shutil.which("openocd")
if not binary:
raise ProcessError(
"OpenOCD binary not found. Install it or pass openocd_bin="
)
raise ProcessError("OpenOCD binary not found. Install it or pass openocd_bin=")
args = [binary]
@ -77,9 +75,7 @@ class OpenOCDProcess:
part = config_parts[i]
if part in ("-f", "-c"):
if i + 1 >= len(config_parts):
raise ProcessError(
f"Config flag '{part}' requires an argument"
)
raise ProcessError(f"Config flag '{part}' requires an argument")
args.extend([part, config_parts[i + 1]])
i += 2
else:
@ -129,9 +125,7 @@ class OpenOCDProcess:
except (OSError, TimeoutError):
await asyncio.sleep(READY_POLL_INTERVAL)
raise OpenOCDTimeoutError(
f"OpenOCD did not become ready within {timeout}s"
)
raise OpenOCDTimeoutError(f"OpenOCD did not become ready within {timeout}s")
async def stop(self) -> None:
"""Terminate the OpenOCD process."""

8
src/openocd/registers.py
View File

@ -17,9 +17,7 @@ from openocd.types import Register
log = logging.getLogger(__name__)
# Matches "reg <name>" output: "pc (/32): 0x08001234"
_REG_VALUE_RE = re.compile(
r"(\S+)\s+\(/(\d+)\):\s*(0x[0-9a-fA-F]+)"
)
_REG_VALUE_RE = re.compile(r"(\S+)\s+\(/(\d+)\):\s*(0x[0-9a-fA-F]+)")
# Matches a row in "reg" (list all) output.
# Typical formats:
@ -149,9 +147,7 @@ class Registers:
"""
lower = resp.lower()
if "not halted" in lower or "target not halted" in lower:
raise TargetNotHaltedError(
"Target must be halted to access registers"
)
raise TargetNotHaltedError("Target must be halted to access registers")
class SyncRegisters:

10
src/openocd/rtt.py
View File

@ -126,10 +126,7 @@ class RTTManager:
"""
# Escape TCL special characters to prevent injection
escaped = (
data.replace("\\", "\\\\")
.replace('"', '\\"')
.replace("[", "\\[")
.replace("$", "\\$")
data.replace("\\", "\\\\").replace('"', '\\"').replace("[", "\\[").replace("$", "\\$")
)
cmd = f'rtt channelwrite {channel} "{escaped}"'
response = await self._conn.send(cmd)
@ -149,9 +146,7 @@ class SyncRTTManager:
size: int,
id_string: str = "SEGGER RTT",
) -> None:
self._loop.run_until_complete(
self._manager.setup(address, size, id_string)
)
self._loop.run_until_complete(self._manager.setup(address, size, id_string))
def start(self) -> None:
self._loop.run_until_complete(self._manager.start())
@ -173,6 +168,7 @@ class SyncRTTManager:
# Helpers
# ---------------------------------------------------------------------------
def _check_rtt_response(response: str, command: str) -> None:
"""Raise on error responses from RTT commands."""
if response and "error" in response.lower():

3
src/openocd/svd/parser.py
View File

@ -75,8 +75,7 @@ class SVDParserWrapper:
periph = self._peripherals.get(name)
if periph is None:
raise SVDError(
f"Peripheral '{name}' not found. "
f"Available: {', '.join(sorted(self._peripherals))}"
f"Peripheral '{name}' not found. Available: {', '.join(sorted(self._peripherals))}"
)
return periph

8
src/openocd/svd/peripheral.py
View File

@ -173,14 +173,10 @@ class SyncSVDManager:
return self._manager.list_registers(peripheral)
def read_register(self, peripheral: str, register: str) -> DecodedRegister:
return self._loop.run_until_complete(
self._manager.read_register(peripheral, register)
)
return self._loop.run_until_complete(self._manager.read_register(peripheral, register))
def read_peripheral(self, peripheral: str) -> dict[str, DecodedRegister]:
return self._loop.run_until_complete(
self._manager.read_peripheral(peripheral)
)
return self._loop.run_until_complete(self._manager.read_peripheral(peripheral))
def decode(self, peripheral: str, register: str, value: int) -> DecodedRegister:
return self._manager.decode(peripheral, register, value)

1
src/openocd/transport.py
View File

@ -125,6 +125,7 @@ class Transport:
# Helpers
# ---------------------------------------------------------------------------
def _parse_speed(response: str) -> int | None:
"""Extract a numeric kHz value from an adapter speed response.

1
tests/conftest.py
View File

@ -1,4 +1,5 @@
"""Shared pytest fixtures for openocd-python tests."""
from __future__ import annotations
import pytest

2
tests/test_connection.py
View File

@ -1,4 +1,5 @@
"""Tests for the TclRpcConnection class."""
from __future__ import annotations
import asyncio
@ -63,6 +64,7 @@ async def test_send_before_connect_raises():
async def test_timeout_on_hung_server():
"""A server that never sends \\x1a should trigger a TimeoutError."""
# Start a server that accepts connections but never responds
async def _hang(reader, writer):
# Read the command but never reply

62
tests/test_error_paths.py
View File

@ -8,6 +8,7 @@ Each test configures a mock server to return error responses (or
misbehave at the protocol level) and asserts that the correct
exception type is raised with a meaningful message.
"""
from __future__ import annotations
import asyncio
@ -289,9 +290,7 @@ class TestMemoryErrors:
async def test_read_u32_target_not_halted(self, error_server, error_conn):
"""read_u32 with 'target not halted' in response raises TargetError."""
error_server.add_response(
r"^read_memory\s+", "error: target not halted"
)
error_server.add_response(r"^read_memory\s+", "error: target not halted")
mem = Memory(error_conn)
with pytest.raises(TargetError, match="read_memory failed"):
@ -299,9 +298,7 @@ class TestMemoryErrors:
async def test_write_u32_error_response(self, error_server, error_conn):
"""write_u32 with an error response raises TargetError."""
error_server.add_response(
r"^write_memory\s+", "error: target not halted"
)
error_server.add_response(r"^write_memory\s+", "error: target not halted")
mem = Memory(error_conn)
with pytest.raises(TargetError, match="write_memory failed"):
@ -309,9 +306,7 @@ class TestMemoryErrors:
async def test_read_u32_non_hex_tokens(self, error_server, error_conn):
"""read_memory returning non-hex garbage raises TargetError."""
error_server.add_response(
r"^read_memory\s+", "not_a_hex_value xyz !!!"
)
error_server.add_response(r"^read_memory\s+", "not_a_hex_value xyz !!!")
mem = Memory(error_conn)
with pytest.raises(TargetError, match="Cannot parse read_memory"):
@ -319,9 +314,7 @@ class TestMemoryErrors:
async def test_read_u8_error_response(self, error_server, error_conn):
"""read_u8 with an error response raises TargetError."""
error_server.add_response(
r"^read_memory\s+", "error: bus fault during memory read"
)
error_server.add_response(r"^read_memory\s+", "error: bus fault during memory read")
mem = Memory(error_conn)
with pytest.raises(TargetError, match="read_memory failed"):
@ -329,9 +322,7 @@ class TestMemoryErrors:
async def test_write_bytes_error_response(self, error_server, error_conn):
"""write_bytes with an error response raises TargetError."""
error_server.add_response(
r"^write_memory\s+", "error: write access violation"
)
error_server.add_response(r"^write_memory\s+", "error: write access violation")
mem = Memory(error_conn)
with pytest.raises(TargetError, match="write_memory failed"):
@ -339,9 +330,7 @@ class TestMemoryErrors:
async def test_read_u16_error_response(self, error_server, error_conn):
"""read_u16 with an error response raises TargetError."""
error_server.add_response(
r"^read_memory\s+", "error: alignment fault"
)
error_server.add_response(r"^read_memory\s+", "error: alignment fault")
mem = Memory(error_conn)
with pytest.raises(TargetError, match="read_memory failed"):
@ -358,9 +347,7 @@ class TestRegisterErrors:
async def test_read_not_halted(self, error_server, error_conn):
"""read('pc') when target is not halted raises TargetNotHaltedError."""
error_server.add_response(
r"^reg\s+pc$", "target not halted"
)
error_server.add_response(r"^reg\s+pc$", "target not halted")
regs = Registers(error_conn)
with pytest.raises(TargetNotHaltedError, match="halted"):
@ -368,9 +355,7 @@ class TestRegisterErrors:
async def test_read_nonexistent_register(self, error_server, error_conn):
"""read('nonexistent') with unparseable response raises TargetError."""
error_server.add_response(
r"^reg\s+nonexistent$", "invalid command name \"nonexistent\""
)
error_server.add_response(r"^reg\s+nonexistent$", 'invalid command name "nonexistent"')
regs = Registers(error_conn)
with pytest.raises(TargetError, match="Cannot parse register"):
@ -378,9 +363,7 @@ class TestRegisterErrors:
async def test_write_not_halted(self, error_server, error_conn):
"""write('pc', val) when target is not halted raises TargetNotHaltedError."""
error_server.add_response(
r"^reg\s+pc\s+0x", "target not halted"
)
error_server.add_response(r"^reg\s+pc\s+0x", "target not halted")
regs = Registers(error_conn)
with pytest.raises(TargetNotHaltedError, match="halted"):
@ -388,9 +371,7 @@ class TestRegisterErrors:
async def test_write_generic_error(self, error_server, error_conn):
"""write() with a non-halted-related error raises TargetError."""
error_server.add_response(
r"^reg\s+r0\s+0x", "error: register write failed"
)
error_server.add_response(r"^reg\s+r0\s+0x", "error: register write failed")
regs = Registers(error_conn)
with pytest.raises(TargetError, match="reg write failed"):
@ -455,20 +436,17 @@ class TestFlashErrors:
async def test_write_image_error(self, error_server, error_conn):
"""write_image with error from server raises FlashError."""
error_server.add_response(
r"^flash write_image\s+", "error: flash write failed"
)
error_server.add_response(r"^flash write_image\s+", "error: flash write failed")
flash = Flash(error_conn)
with pytest.raises(FlashError, match="flash write_image"):
from pathlib import Path
await flash.write_image(Path("/tmp/fake_firmware.bin"), verify=False)
async def test_protect_error(self, error_server, error_conn):
"""flash.protect() with error response raises FlashError."""
error_server.add_response(
r"^flash protect\s+", "error: protection change not supported"
)
error_server.add_response(r"^flash protect\s+", "error: protection change not supported")
flash = Flash(error_conn)
with pytest.raises(FlashError, match="flash protect"):
@ -495,9 +473,7 @@ class TestBreakpointErrors:
async def test_remove_breakpoint_error(self, error_server, error_conn):
"""remove() with error response raises BreakpointError."""
error_server.add_response(
r"^rbp\s+", "error: no breakpoint at address"
)
error_server.add_response(r"^rbp\s+", "error: no breakpoint at address")
bp = BreakpointManager(error_conn)
with pytest.raises(BreakpointError, match="rbp 0x"):
@ -505,9 +481,7 @@ class TestBreakpointErrors:
async def test_add_watchpoint_error(self, error_server, error_conn):
"""add_watchpoint() with error response raises BreakpointError."""
error_server.add_response(
r"^wp\s+0x", "error: no free watchpoint comparator"
)
error_server.add_response(r"^wp\s+0x", "error: no free watchpoint comparator")
bp = BreakpointManager(error_conn)
with pytest.raises(BreakpointError, match="wp 0x"):
@ -515,9 +489,7 @@ class TestBreakpointErrors:
async def test_remove_watchpoint_error(self, error_server, error_conn):
"""remove_watchpoint() with error response raises BreakpointError."""
error_server.add_response(
r"^rwp\s+", "error: no watchpoint at address"
)
error_server.add_response(r"^rwp\s+", "error: no watchpoint at address")
bp = BreakpointManager(error_conn)
with pytest.raises(BreakpointError, match="rwp 0x"):

1
tests/test_jtag.py
View File

@ -1,4 +1,5 @@
"""Tests for the JTAG subsystem."""
from __future__ import annotations
import pytest

1
tests/test_memory.py
View File

@ -1,4 +1,5 @@
"""Tests for the Memory subsystem."""
from __future__ import annotations

1
tests/test_registers.py
View File

@ -1,4 +1,5 @@
"""Tests for the Registers subsystem."""
from __future__ import annotations
from openocd.types import Register

38
tests/test_svd.py
View File

@ -3,6 +3,7 @@
These tests exercise the bitfield decoder and DecodedRegister formatting
using synthetic data, without needing an SVD file or a mock server.
"""
from __future__ import annotations
from dataclasses import dataclass
@ -14,6 +15,7 @@ from openocd.types import BitField, DecodedRegister
# -- Fake SVD objects to avoid needing a real .svd file -----------------------
@dataclass
class FakeSVDField:
name: str
@ -41,22 +43,26 @@ def gpioa_odr():
"""A fake GPIOA.ODR register with two bitfields."""
fields = [
FakeSVDField(
name="ODR0", bit_offset=0, bit_width=1,
name="ODR0",
bit_offset=0,
bit_width=1,
description="Port output data bit 0",
),
FakeSVDField(
name="ODR1", bit_offset=1, bit_width=1,
name="ODR1",
bit_offset=1,
bit_width=1,
description="Port output data bit 1",
),
FakeSVDField(
name="ODR15_2", bit_offset=2, bit_width=14,
name="ODR15_2",
bit_offset=2,
bit_width=14,
description="Port output data bits 15:2",
),
]
register = FakeSVDRegister(name="ODR", address_offset=0x14, fields=fields)
peripheral = FakeSVDPeripheral(
name="GPIOA", base_address=0x40010800, registers=[register]
)
peripheral = FakeSVDPeripheral(name="GPIOA", base_address=0x40010800, registers=[register])
return peripheral, register
@ -68,27 +74,33 @@ def usart_cr1():
FakeSVDField(name="RE", bit_offset=2, bit_width=1, description="Receiver enable"),
FakeSVDField(name="TE", bit_offset=3, bit_width=1, description="Transmitter enable"),
FakeSVDField(
name="RXNEIE", bit_offset=5, bit_width=1,
name="RXNEIE",
bit_offset=5,
bit_width=1,
description="RXNE interrupt enable",
),
FakeSVDField(
name="TCIE", bit_offset=6, bit_width=1,
name="TCIE",
bit_offset=6,
bit_width=1,
description="Transmission complete IE",
),
FakeSVDField(
name="TXEIE", bit_offset=7, bit_width=1,
name="TXEIE",
bit_offset=7,
bit_width=1,
description="TXE interrupt enable",
),
FakeSVDField(name="M", bit_offset=12, bit_width=1, description="Word length"),
FakeSVDField(
name="OVER8", bit_offset=15, bit_width=1,
name="OVER8",
bit_offset=15,
bit_width=1,
description="Oversampling mode",
),
]
register = FakeSVDRegister(name="CR1", address_offset=0x0C, fields=fields)
peripheral = FakeSVDPeripheral(
name="USART1", base_address=0x40013800, registers=[register]
)
peripheral = FakeSVDPeripheral(name="USART1", base_address=0x40013800, registers=[register])
return peripheral, register

1
tests/test_target.py
View File

@ -1,4 +1,5 @@
"""Tests for the Target subsystem."""
from __future__ import annotations
import pytest