From a46d3f22f4c079dc213cf1e3abcf4c48977481a2 Mon Sep 17 00:00:00 2001 From: sam0rr Date: Wed, 22 Jul 2026 14:13:38 -0400 Subject: [PATCH 1/2] Fix strict quality gate --- README.md | 41 ++++-- docs/configuration.md | 12 +- docs/usage.md | 52 ++++++-- examples/main.py | 4 +- pyproject.toml | 125 +++++++++++------- src/xkc_kl200_python/__init__.py | 24 ++-- src/xkc_kl200_python/config.py | 2 +- src/xkc_kl200_python/constants.py | 2 +- src/xkc_kl200_python/errors.py | 6 +- src/xkc_kl200_python/sensor.py | 58 +++++---- src/xkc_kl200_python/serial_manager.py | 98 +++++++++------ src/xkc_kl200_python/utils.py | 4 +- tests/test_dataclass_policy.py | 146 +++++++++++++++++++++ tests/test_dead_code.py | 21 ++++ tests/test_documentation_examples.py | 167 +++++++++++++++++++++++++ tests/test_sensor_init_modes.py | 18 ++- tests/test_sensor_misc.py | 46 +++---- tests/test_sensor_reading.py | 46 +++---- tests/test_serial_manager.py | 56 ++++----- 19 files changed, 687 insertions(+), 241 deletions(-) create mode 100644 tests/test_dataclass_policy.py create mode 100644 tests/test_dead_code.py create mode 100644 tests/test_documentation_examples.py diff --git a/README.md b/README.md index 783185e..0477099 100644 --- a/README.md +++ b/README.md @@ -44,11 +44,17 @@ Before running the library on a Pi: - connect `TX -> RX`, `RX -> TX`, and `GND -> GND` ```python -from xkc_kl200_python import XKC_KL200 +"""Read one XKC-KL200 measurement.""" -with XKC_KL200(port="/dev/serial0", baudrate=9600) as sensor: - distance_mm = sensor.read_distance() - print(f"Distance: {distance_mm} mm") +import logging + +from xkc_kl200_python import XkcKl200 + +logging.basicConfig(level=logging.INFO, format="%(message)s") +_LOGGER = logging.getLogger(__name__) + +with XkcKl200(port="/dev/serial0", baudrate=9600) as sensor: + _LOGGER.info("Distance: %s mm", sensor.read_distance()) ``` The library requests exclusive access when opening the serial port, so another @@ -58,16 +64,31 @@ mixed reads later. If you want continuous data, loop in your own application: ```python +"""Continuously read XKC-KL200 measurements.""" + +import logging import time -from xkc_kl200_python import XKC_KL200, XKC_KL200_ReadError +from xkc_kl200_python import XkcKl200, XkcKl200ReadError + +_LOGGER = logging.getLogger(__name__) + + +def log_distance(sensor: XkcKl200) -> None: + """Read and log one measurement or the resulting sensor error.""" + try: + distance_mm = sensor.read_distance() + except XkcKl200ReadError: + _LOGGER.exception("Sensor read failed") + else: + _LOGGER.info("Distance: %s mm", distance_mm) + + +logging.basicConfig(level=logging.INFO, format="%(message)s") -with XKC_KL200(port="/dev/serial0", baudrate=9600) as sensor: +with XkcKl200(port="/dev/serial0", baudrate=9600) as sensor: while True: - try: - print(sensor.read_distance()) - except XKC_KL200_ReadError: - print("Read failed") + log_distance(sensor) time.sleep(0.1) ``` diff --git a/docs/configuration.md b/docs/configuration.md index d41d021..b98559a 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -7,7 +7,14 @@ `SensorConfig` stores the serial connection parameters used by the library: ```python -from xkc_kl200_python import SensorConfig, XKC_KL200 +"""Configure an XKC-KL200 sensor connection.""" + +import logging + +from xkc_kl200_python import SensorConfig, XkcKl200 + +logging.basicConfig(level=logging.INFO, format="%(message)s") +_LOGGER = logging.getLogger(__name__) config = SensorConfig( port="/dev/ttyUSB0", @@ -17,7 +24,8 @@ config = SensorConfig( startup_delay_s=0.1, ) -sensor = XKC_KL200(config=config) +with XkcKl200(config=config) as sensor: + _LOGGER.info("Sensor configured at address %#06x", sensor.address) ``` --- diff --git a/docs/usage.md b/docs/usage.md index bc4046e..fdb7857 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -5,9 +5,17 @@ ## Basic Initialization ```python -from xkc_kl200_python import XKC_KL200 +"""Initialize an XKC-KL200 sensor connection.""" -sensor = XKC_KL200(port="/dev/serial0", baudrate=9600) +import logging + +from xkc_kl200_python import XkcKl200 + +logging.basicConfig(level=logging.INFO, format="%(message)s") +_LOGGER = logging.getLogger(__name__) + +with XkcKl200(port="/dev/serial0", baudrate=9600) as sensor: + _LOGGER.info("Sensor ready at address %#06x", sensor.address) ``` The default connection settings are: @@ -31,10 +39,17 @@ Before using the library on a Pi: ## Manual Distance Read ```python -from xkc_kl200_python import XKC_KL200 +"""Read one XKC-KL200 measurement.""" + +import logging + +from xkc_kl200_python import XkcKl200 + +logging.basicConfig(level=logging.INFO, format="%(message)s") +_LOGGER = logging.getLogger(__name__) -with XKC_KL200(port="/dev/serial0", baudrate=9600) as sensor: - print(sensor.read_distance()) +with XkcKl200(port="/dev/serial0", baudrate=9600) as sensor: + _LOGGER.info("Distance: %s mm", sensor.read_distance()) ``` --- @@ -42,16 +57,31 @@ with XKC_KL200(port="/dev/serial0", baudrate=9600) as sensor: ## Looping In Your App ```python +"""Continuously read XKC-KL200 measurements.""" + +import logging import time -from xkc_kl200_python import XKC_KL200, XKC_KL200_ReadError +from xkc_kl200_python import XkcKl200, XkcKl200ReadError + +_LOGGER = logging.getLogger(__name__) + + +def log_distance(sensor: XkcKl200) -> None: + """Read and log one measurement or the resulting sensor error.""" + try: + distance_mm = sensor.read_distance() + except XkcKl200ReadError: + _LOGGER.exception("Sensor read failed") + else: + _LOGGER.info("Distance: %s mm", distance_mm) + + +logging.basicConfig(level=logging.INFO, format="%(message)s") -with XKC_KL200(port="/dev/serial0", baudrate=9600) as sensor: +with XkcKl200(port="/dev/serial0", baudrate=9600) as sensor: while True: - try: - print(sensor.read_distance()) - except XKC_KL200_ReadError: - print("Read failed") + log_distance(sensor) time.sleep(0.1) ``` diff --git a/examples/main.py b/examples/main.py index 7b271fa..81a3951 100644 --- a/examples/main.py +++ b/examples/main.py @@ -2,7 +2,7 @@ import logging -from xkc_kl200_python import XKC_KL200 +from xkc_kl200_python import XkcKl200 _LOGGER = logging.getLogger(__name__) @@ -10,7 +10,7 @@ def main() -> None: """Read and display one distance measurement.""" logging.basicConfig(level=logging.INFO, format="%(message)s") - with XKC_KL200(port="/dev/ttyUSB0") as sensor: + with XkcKl200(port="/dev/ttyUSB0") as sensor: _LOGGER.info("Distance: %s mm", sensor.read_distance()) diff --git a/pyproject.toml b/pyproject.toml index 089f7b7..04a3727 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,20 +1,20 @@ +[build-system] +requires = ["uv_build>=0.11.19,<0.12.0"] +build-backend = "uv_build" + [project] name = "xkc-kl200-python" version = "6.0.0" description = "A typed Python library for controlling the XKC-KL200 UART laser distance sensor." readme = "README.md" authors = [ - { name = "sam0rr", email = "samor140@hotmail.com" } + { name = "sam0rr", email = "samor140@hotmail.com" }, ] requires-python = ">=3.13" dependencies = [ "pyserial>=3.5", ] -[build-system] -requires = ["uv_build>=0.11.19,<0.12.0"] -build-backend = "uv_build" - [dependency-groups] dev = [ "mypy>=2.1.0", @@ -25,83 +25,122 @@ dev = [ "vulture>=2.16", ] -[tool.pytest.ini_options] -addopts = "-v --maxfail=1 --cov=xkc_kl200_python --cov-fail-under=100" - -[tool.mypy] -strict = true -disallow_any_explicit = true -files = ["src", "tests"] - -[tool.ruff] -target-version = "py313" - [tool.ruff.lint] -# Based on the official Ruff configurations used by FastAPI, Pydantic, -# Pydantic AI, and HTTPX: +# Informed by the official Ruff configurations used by FastAPI, Pydantic, +# Pydantic AI, and HTTPX, then tailored for strict typed projects: # https://github.com/fastapi/fastapi/blob/master/pyproject.toml # https://github.com/pydantic/pydantic/blob/main/pyproject.toml # https://github.com/pydantic/pydantic-ai/blob/main/pyproject.toml # https://github.com/encode/httpx/blob/master/pyproject.toml select = [ - # Core correctness. + # Core correctness and safety. "E", # Pycodestyle errors "W", # Pycodestyle warnings "F", # Pyflakes correctness - "B", # Flake8 Bugbear + "B", # Flake8 Bugbear correctness checks + "A004", # Prevent imports from shadowing built-ins + "A005", # Prevent modules from shadowing the standard library + "ARG", # Reject unused callable arguments "PLE", # Pylint error-level correctness checks - - # FastAPI and asynchronous correctness. - "FAST", # FastAPI correctness and modern parameter declarations - "ASYNC", # Blocking and unsafe async patterns - "RUF006", # Dangling asyncio tasks + "PLW", # Pylint warning-level correctness checks + "S", # Bandit-derived security checks + "TRY004", # Use TypeError for invalid argument types + + # Typing and dataclass correctness. + "PYI034", # Use Self for self-returning special methods + "RUF008", # Reject mutable dataclass defaults + "RUF009", # Reject function calls in dataclass defaults + "RUF012", # Require ClassVar for mutable class attributes + "RUF033", # Reject defaults on __post_init__ parameters + "RUF049", # Reject dataclass-decorated Enum classes + + # Targeted runtime and data correctness. + "PTH124", # Reject legacy py.path usage + "PTH210", # Reject invalid Path.with_suffix() arguments + "RUF024", # Reject mutable values passed to dict.fromkeys() + "RUF026", # Pass defaultdict factories positionally + "RUF032", # Avoid constructing Decimal from float literals + "RUF064", # Require octal literals for file-permission modes + + # Asynchronous and FastAPI correctness. + "ASYNC", # Detect blocking and unsafe async patterns + "FAST", # FastAPI-specific correctness checks + "RUF006", # Retain references to created asyncio tasks # Imports, modernization, and unnecessary code. "I", # Import ordering - "UP", # Python 3.13 syntax modernization - "C4", # Comprehension improvements - "PIE", # Unnecessary code + "RUF022", # Sort __all__ exports + "UP", # Modernize syntax for Python 3.13 + "YTT", # Validate Python-version checks + "FURB", # Stable Refurb modernization checks + "C4", # Improve comprehensions + "RUF005", # Prefer unpacking over collection concatenation + "PIE", # Remove unnecessary code + "RUF059", # Mark intentionally unused unpacked variables explicitly + + # Control-flow and resource safety. + "RUF021", # Parenthesize mixed and/or expressions + "RUF060", # Reject membership tests against empty collections + "SIM105", # Use contextlib.suppress for intentional suppression + "SIM107", # Prevent finally returns from masking exceptions + "SIM114", # Combine branches with identical bodies + "SIM115", # Require context managers for locally opened files + "SIM220", # Reject expressions equivalent to x and not x + "SIM221", # Reject expressions equivalent to x or not x + "TRY300", # Keep success-only statements outside try blocks # Documentation, maintainability, and performance. "D", # Docstring conventions + "N", # PEP 8 naming conventions "C90", # McCabe complexity "PLR0913", # Excessive function arguments "PERF", # Performance anti-patterns + "RUF017", # Prevent quadratic list flattening with sum() # Logging and development hygiene. - "LOG", # Logging correctness - "G", # Logging format correctness + "LOG", # Logging API correctness + "G", # Logging format-string correctness "T10", # Debugger calls "T20", # Print statements # Test and assertion correctness. - "PT", # Pytest correctness and style - "RUF018", # Assignments inside assertions - "RUF043", # Ambiguous pytest.raises patterns - - # Suppression and project metadata hygiene. - "PGH", # Blanket suppressions and invalid mock access - "RUF10", # Noqa and suppression-comment checks - "RUF200", # Invalid pyproject.toml metadata + "PT", # Pytest correctness and style checks + "RUF018", # Reject assignments inside assertions + "RUF030", # Reject print() calls used as assert messages + "RUF040", # Reject non-string literal assert messages + "RUF043", # Require unambiguous pytest.raises() patterns + + # Suppression hygiene. + "PGH", # Detect blanket suppressions and invalid mock access + "RUF028", # Validate Ruff formatter-suppression comments + "RUF10", # Validate noqa and suppression directives + + # Project configuration correctness. + "RUF200", # Validate pyproject.toml ] ignore = [ "E501", # Line length is handled by the Ruff formatter ] -[tool.ruff.lint.mccabe] -# Pydantic AI uses 15: strict enough to expose genuinely complex functions -# without forcing ordinary orchestration code into unnecessary abstractions. -max-complexity = 15 - [tool.ruff.lint.pydocstyle] convention = "google" [tool.ruff.lint.per-file-ignores] +# Pytest assertions are test checks, not runtime security guards. +"tests/**/*.py" = ["S101"] # Vulture whitelist files intentionally use undefined bare-name expressions # to mark dynamically consumed symbols as used. "vulture_whitelist.py" = ["B018", "F821"] +[tool.mypy] +strict = true +disallow_any_explicit = true +files = ["src", "tests"] + [tool.vulture] min_confidence = 60 paths = ["src", "tests", "vulture_whitelist.py"] sort_by_size = true + +[tool.pytest.ini_options] +addopts = "-v --maxfail=1 --cov=xkc_kl200_python --cov-fail-under=100" diff --git a/src/xkc_kl200_python/__init__.py b/src/xkc_kl200_python/__init__.py index 5f60453..1d51b5b 100644 --- a/src/xkc_kl200_python/__init__.py +++ b/src/xkc_kl200_python/__init__.py @@ -5,24 +5,24 @@ CommunicationMode, LedMode, RelayMode, - XKC_KL200_Status, + XkcKl200Status, ) from .errors import ( - XKC_KL200_ReadError, - XKC_KL200_ResponseError, - XKC_KL200_TimeoutError, + XkcKl200ReadError, + XkcKl200ResponseError, + XkcKl200TimeoutError, ) -from .sensor import XKC_KL200 +from .sensor import XkcKl200 # Re-export the supported public API from one stable package entrypoint. __all__ = [ - "XKC_KL200", - "SensorConfig", - "XKC_KL200_Status", - "XKC_KL200_ReadError", - "XKC_KL200_TimeoutError", - "XKC_KL200_ResponseError", + "CommunicationMode", "LedMode", "RelayMode", - "CommunicationMode", + "SensorConfig", + "XkcKl200", + "XkcKl200ReadError", + "XkcKl200ResponseError", + "XkcKl200Status", + "XkcKl200TimeoutError", ] diff --git a/src/xkc_kl200_python/config.py b/src/xkc_kl200_python/config.py index 3f0fcfd..de99aca 100644 --- a/src/xkc_kl200_python/config.py +++ b/src/xkc_kl200_python/config.py @@ -10,7 +10,7 @@ ) -@dataclass(frozen=True, slots=True, kw_only=True) +@dataclass(frozen=True, kw_only=True, slots=True) class SensorConfig: """Serial connection settings for an XKC-KL200 sensor instance.""" diff --git a/src/xkc_kl200_python/constants.py b/src/xkc_kl200_python/constants.py index 4ff1a0b..69e38fb 100644 --- a/src/xkc_kl200_python/constants.py +++ b/src/xkc_kl200_python/constants.py @@ -37,7 +37,7 @@ CODE_TO_BAUD_RATE = {code: baudrate for baudrate, code in BAUD_RATE_TO_CODE.items()} -class XKC_KL200_Status(IntEnum): +class XkcKl200Status(IntEnum): """Protocol-level status codes returned by the library API.""" SUCCESS = 0 diff --git a/src/xkc_kl200_python/errors.py b/src/xkc_kl200_python/errors.py index a5f7bf8..5b33c99 100644 --- a/src/xkc_kl200_python/errors.py +++ b/src/xkc_kl200_python/errors.py @@ -1,13 +1,13 @@ """Public exception types for fresh-read failures in the XKC-KL200 library.""" -class XKC_KL200_ReadError(Exception): +class XkcKl200ReadError(Exception): """Base exception for failures while reading a fresh sensor measurement.""" -class XKC_KL200_TimeoutError(XKC_KL200_ReadError): +class XkcKl200TimeoutError(XkcKl200ReadError): """Raised when the sensor does not return a full measurement frame in time.""" -class XKC_KL200_ResponseError(XKC_KL200_ReadError): +class XkcKl200ResponseError(XkcKl200ReadError): """Raised when the sensor returns a malformed or unexpected measurement frame.""" diff --git a/src/xkc_kl200_python/sensor.py b/src/xkc_kl200_python/sensor.py index 5700b69..4043c86 100644 --- a/src/xkc_kl200_python/sensor.py +++ b/src/xkc_kl200_python/sensor.py @@ -2,7 +2,7 @@ import time from dataclasses import replace -from typing import TypeVar +from typing import Self, TypeVar from .config import SensorConfig from .constants import ( @@ -22,9 +22,9 @@ CommunicationMode, LedMode, RelayMode, - XKC_KL200_Status, + XkcKl200Status, ) -from .errors import XKC_KL200_ResponseError, XKC_KL200_TimeoutError +from .errors import XkcKl200ResponseError, XkcKl200TimeoutError from .serial_manager import SerialFactory, SerialManager from .utils import ( EMPTY_FRAME_PAYLOAD, @@ -37,7 +37,7 @@ EnumValue = TypeVar("EnumValue", bound=int) -class XKC_KL200: +class XkcKl200: """High-level interface for controlling an XKC-KL200 UART sensor.""" def __init__( @@ -65,8 +65,8 @@ def __init__( if config.startup_delay_s > 0: time.sleep(config.startup_delay_s) - def __enter__(self) -> "XKC_KL200": - """Support ``with XKC_KL200(...)`` context management.""" + def __enter__(self) -> Self: + """Support ``with XkcKl200(...)`` context management.""" return self def __exit__(self, *_: object) -> None: @@ -99,24 +99,24 @@ def reset_output_buffer(self) -> None: """Clear pending serial output bytes.""" self._serial_manager.reset_output_buffer() - def hard_reset(self) -> XKC_KL200_Status: + def hard_reset(self) -> XkcKl200Status: """Request a factory reset on the sensor.""" return self._send_ack_command( command=RESET_COMMAND, payload=FramePayload(tail=0xFE), ) - def soft_reset(self) -> XKC_KL200_Status: + def soft_reset(self) -> XkcKl200Status: """Request a user-settings reset on the sensor.""" return self._send_ack_command( command=RESET_COMMAND, payload=FramePayload(tail=0xFD), ) - def change_address(self, address: int) -> XKC_KL200_Status: + def change_address(self, address: int) -> XkcKl200Status: """Change the sensor address if the requested value is valid.""" if not MIN_ADDRESS <= address <= MAX_ADDRESS: - return XKC_KL200_Status.INVALID_PARAMETER + return XkcKl200Status.INVALID_PARAMETER result = self._send_ack_command( command=CHANGE_ADDRESS_COMMAND, @@ -125,52 +125,52 @@ def change_address(self, address: int) -> XKC_KL200_Status: data_low=address & 0xFF, ), ) - if result == XKC_KL200_Status.SUCCESS: + if result == XkcKl200Status.SUCCESS: self.config = replace(self.config, address=address) self._address = address return result - def change_baud_rate(self, baud_rate: int) -> XKC_KL200_Status: + def change_baud_rate(self, baud_rate: int) -> XkcKl200Status: """Change the sensor baud rate using a baud value or protocol code.""" baud_code = self._resolve_baud_rate_code(baud_rate) if baud_code is None: - return XKC_KL200_Status.INVALID_PARAMETER + return XkcKl200Status.INVALID_PARAMETER result = self._send_ack_command( command=CHANGE_BAUD_RATE_COMMAND, payload=FramePayload(data_low=baud_code), ) - if result == XKC_KL200_Status.SUCCESS: + if result == XkcKl200Status.SUCCESS: new_baudrate = CODE_TO_BAUD_RATE[baud_code] self.config = replace(self.config, baudrate=new_baudrate) self._serial_manager.set_baudrate(new_baudrate) return result - def set_led_mode(self, mode: int | LedMode) -> XKC_KL200_Status: + def set_led_mode(self, mode: int | LedMode) -> XkcKl200Status: """Configure the sensor LED behavior.""" value = self._coerce_enum_value(mode, LedMode) if value is None: - return XKC_KL200_Status.INVALID_PARAMETER + return XkcKl200Status.INVALID_PARAMETER return self._send_ack_command( command=SET_LED_MODE_COMMAND, payload=FramePayload(data_low=value), ) - def set_relay_mode(self, mode: int | RelayMode) -> XKC_KL200_Status: + def set_relay_mode(self, mode: int | RelayMode) -> XkcKl200Status: """Configure the relay output behavior.""" value = self._coerce_enum_value(mode, RelayMode) if value is None: - return XKC_KL200_Status.INVALID_PARAMETER + return XkcKl200Status.INVALID_PARAMETER return self._send_ack_command( command=SET_RELAY_MODE_COMMAND, payload=FramePayload(data_low=value), ) - def set_communication_mode(self, mode: int | CommunicationMode) -> XKC_KL200_Status: + def set_communication_mode(self, mode: int | CommunicationMode) -> XkcKl200Status: """Switch the device between relay mode and UART mode.""" value = self._coerce_enum_value(mode, CommunicationMode) if value is None: - return XKC_KL200_Status.INVALID_PARAMETER + return XkcKl200Status.INVALID_PARAMETER return self._send_ack_command( header=SYSTEM_HEADER, command=SET_COMMUNICATION_MODE_COMMAND, @@ -181,8 +181,8 @@ def read_distance(self, timeout: float | None = None) -> int: """Request one fresh distance measurement. Raises: - XKC_KL200_TimeoutError: The sensor did not reply before the timeout. - XKC_KL200_ResponseError: The reply frame was malformed or unexpected. + XkcKl200TimeoutError: The sensor did not reply before the timeout. + XkcKl200ResponseError: The reply frame was malformed or unexpected. """ self._serial_manager.reset_input_buffer() self._serial_manager.write_frame( @@ -199,11 +199,9 @@ def read_distance(self, timeout: float | None = None) -> int: timeout=self.config.timeout if timeout is None else timeout, ) if response is None: - if status == XKC_KL200_Status.TIMEOUT: - raise XKC_KL200_TimeoutError( - "Timed out waiting for a measurement frame" - ) - raise XKC_KL200_ResponseError("Received an invalid measurement frame") + if status == XkcKl200Status.TIMEOUT: + raise XkcKl200TimeoutError("Timed out waiting for a measurement frame") + raise XkcKl200ResponseError("Received an invalid measurement frame") address, distance_mm = parse_measurement_frame(response) @@ -217,7 +215,7 @@ def _send_ack_command( command: int, header: int = COMMAND_HEADER, payload: FramePayload = EMPTY_FRAME_PAYLOAD, - ) -> XKC_KL200_Status: + ) -> XkcKl200Status: """Send a command frame and wait for its acknowledgement.""" frame = build_command_frame( header=header, @@ -234,7 +232,7 @@ def _wait_for_response( *, expected_header: int = COMMAND_HEADER, expected_command: int, - ) -> XKC_KL200_Status: + ) -> XkcKl200Status: """Read and classify the acknowledgement for a configuration command.""" # Command 0x30 is shared by baud-rate and communication-mode ACKs. allow_header_mismatch_skip = expected_command == CHANGE_BAUD_RATE_COMMAND @@ -252,7 +250,7 @@ def _wait_for_response( expected_command=expected_command, ) self._address = parsed.address - return XKC_KL200_Status.SUCCESS + return XkcKl200Status.SUCCESS @staticmethod def _resolve_baud_rate_code(baud_rate: int) -> int | None: diff --git a/src/xkc_kl200_python/serial_manager.py b/src/xkc_kl200_python/serial_manager.py index 14ce2f5..d1b49e0 100644 --- a/src/xkc_kl200_python/serial_manager.py +++ b/src/xkc_kl200_python/serial_manager.py @@ -7,7 +7,7 @@ import serial from .config import SensorConfig -from .constants import COMMAND_HEADER, FRAME_LENGTH, SYSTEM_HEADER, XKC_KL200_Status +from .constants import COMMAND_HEADER, FRAME_LENGTH, SYSTEM_HEADER, XkcKl200Status from .utils import calculate_checksum @@ -113,11 +113,11 @@ def read_frame( expected_header: int | None = None, expected_command: int | None = None, allow_header_mismatch_skip: bool = False, - ) -> tuple[bytes | None, XKC_KL200_Status]: + ) -> tuple[bytes | None, XkcKl200Status]: """Read one valid frame or return the protocol-level failure status.""" deadline = time.monotonic() + timeout buffered_bytes_at_deadline: int | None = None - deferred_error: XKC_KL200_Status | None = None + deferred_error: XkcKl200Status | None = None while True: frame, frame_status = self._scan_buffer( @@ -126,14 +126,11 @@ def read_frame( expected_command=expected_command, ) if frame is not None: - return frame, XKC_KL200_Status.SUCCESS - if frame_status == XKC_KL200_Status.CHECKSUM_ERROR: - deferred_error = XKC_KL200_Status.CHECKSUM_ERROR - elif ( - frame_status == XKC_KL200_Status.RESPONSE_ERROR - and deferred_error != XKC_KL200_Status.CHECKSUM_ERROR - ): - deferred_error = XKC_KL200_Status.RESPONSE_ERROR + return frame, XkcKl200Status.SUCCESS + deferred_error = self._prefer_protocol_error( + current=deferred_error, + candidate=frame_status, + ) if time.monotonic() >= deadline: # Snapshot how many bytes were already queued when the timeout @@ -149,7 +146,7 @@ def read_frame( self._buffer.clear() if deferred_error is not None: return None, deferred_error - return None, XKC_KL200_Status.TIMEOUT + return None, XkcKl200Status.TIMEOUT if self._read_from_serial() > 0: continue @@ -162,9 +159,9 @@ def _scan_buffer( allow_header_mismatch_skip: bool, expected_header: int | None, expected_command: int | None, - ) -> tuple[bytes | None, XKC_KL200_Status | None]: + ) -> tuple[bytes | None, XkcKl200Status | None]: """Scan buffered bytes until a frame is found or parsing stalls.""" - deferred_error: XKC_KL200_Status | None = None + deferred_error: XkcKl200Status | None = None while True: frame, frame_status, consumed_data = self._extract_frame( @@ -174,13 +171,10 @@ def _scan_buffer( ) if frame is not None: return frame, None - if frame_status == XKC_KL200_Status.CHECKSUM_ERROR: - deferred_error = XKC_KL200_Status.CHECKSUM_ERROR - elif ( - frame_status == XKC_KL200_Status.RESPONSE_ERROR - and deferred_error != XKC_KL200_Status.CHECKSUM_ERROR - ): - deferred_error = XKC_KL200_Status.RESPONSE_ERROR + deferred_error = self._prefer_protocol_error( + current=deferred_error, + candidate=frame_status, + ) if not consumed_data: return None, deferred_error @@ -205,24 +199,14 @@ def _extract_frame( allow_header_mismatch_skip: bool, expected_header: int | None, expected_command: int | None, - ) -> tuple[bytes | None, XKC_KL200_Status | None, bool]: + ) -> tuple[bytes | None, XkcKl200Status | None, bool]: """Return the next valid frame, one protocol error, and whether data was consumed.""" if not self._buffer: return None, None, False - header_index = self._find_next_header() - if header_index < 0: - if len(self._buffer) >= FRAME_LENGTH: - del self._buffer[0] - return None, XKC_KL200_Status.RESPONSE_ERROR, True - self._buffer.clear() - return None, None, True - if header_index > 0: - malformed_frame = len(self._buffer) >= FRAME_LENGTH - del self._buffer[:header_index] - if malformed_frame: - return None, XKC_KL200_Status.RESPONSE_ERROR, True - return None, None, True + consumed_data, prefix_status = self._consume_misaligned_prefix() + if consumed_data: + return None, prefix_status, True if len(self._buffer) < 3: return None, None, False @@ -231,7 +215,7 @@ def _extract_frame( if frame_length != FRAME_LENGTH: del self._buffer[0] if len(self._buffer) >= FRAME_LENGTH - 1: - return None, XKC_KL200_Status.RESPONSE_ERROR, True + return None, XkcKl200Status.RESPONSE_ERROR, True return None, None, True if len(self._buffer) < FRAME_LENGTH: @@ -240,7 +224,7 @@ def _extract_frame( candidate = bytes(self._buffer[:FRAME_LENGTH]) if candidate[-1] != calculate_checksum(candidate[:-1]): del self._buffer[0] - return None, XKC_KL200_Status.CHECKSUM_ERROR, True + return None, XkcKl200Status.CHECKSUM_ERROR, True if expected_command is not None and candidate[1] != expected_command: return self._consume_mismatched_candidate(error_status=None) @@ -250,18 +234,52 @@ def _extract_frame( error_status=( None if allow_header_mismatch_skip - else XKC_KL200_Status.RESPONSE_ERROR + else XkcKl200Status.RESPONSE_ERROR ) ) del self._buffer[:FRAME_LENGTH] return candidate, None, True + def _consume_misaligned_prefix( + self, + ) -> tuple[bool, XkcKl200Status | None]: + """Discard bytes before a protocol header and report malformed data.""" + header_index = self._find_next_header() + if header_index == 0: + return False, None + + malformed_frame = len(self._buffer) >= FRAME_LENGTH + if header_index < 0: + discard_count = 1 if malformed_frame else len(self._buffer) + else: + discard_count = header_index + + del self._buffer[:discard_count] + status = XkcKl200Status.RESPONSE_ERROR if malformed_frame else None + return True, status + + @staticmethod + def _prefer_protocol_error( + *, + current: XkcKl200Status | None, + candidate: XkcKl200Status | None, + ) -> XkcKl200Status | None: + """Keep the most specific protocol error observed while scanning.""" + if candidate == XkcKl200Status.CHECKSUM_ERROR: + return XkcKl200Status.CHECKSUM_ERROR + if ( + candidate == XkcKl200Status.RESPONSE_ERROR + and current != XkcKl200Status.CHECKSUM_ERROR + ): + return XkcKl200Status.RESPONSE_ERROR + return current + def _consume_mismatched_candidate( self, *, - error_status: XKC_KL200_Status | None, - ) -> tuple[None, XKC_KL200_Status | None, bool]: + error_status: XkcKl200Status | None, + ) -> tuple[None, XkcKl200Status | None, bool]: """Resynchronize within a checksum-valid mismatched candidate.""" overlap_header_index = self._find_next_header(start=1, stop=FRAME_LENGTH) if overlap_header_index < 0: diff --git a/src/xkc_kl200_python/utils.py b/src/xkc_kl200_python/utils.py index adb43b0..ebef39e 100644 --- a/src/xkc_kl200_python/utils.py +++ b/src/xkc_kl200_python/utils.py @@ -12,7 +12,7 @@ ) -@dataclass(frozen=True, slots=True, kw_only=True) +@dataclass(frozen=True, kw_only=True, slots=True) class FramePayload: """Three data bytes carried by an XKC-KL200 protocol frame.""" @@ -35,7 +35,7 @@ def to_bytes(self) -> bytes: return bytes((self.data_high, self.data_low, self.tail)) -@dataclass(frozen=True, slots=True, kw_only=True) +@dataclass(frozen=True, kw_only=True, slots=True) class ProtocolFrame: """Decoded view of a raw 9-byte XKC-KL200 protocol frame.""" diff --git a/tests/test_dataclass_policy.py b/tests/test_dataclass_policy.py new file mode 100644 index 0000000..4e91314 --- /dev/null +++ b/tests/test_dataclass_policy.py @@ -0,0 +1,146 @@ +"""Repository-wide dataclass options and ordering enforcement tests.""" + +from __future__ import annotations + +import ast +from pathlib import Path + +_DATACLASS_OPTION_ORDER = ( + "init", + "repr", + "eq", + "order", + "unsafe_hash", + "frozen", + "match_args", + "kw_only", + "slots", + "weakref_slot", +) +_DATACLASS_OPTION_POSITION = { + option: position for position, option in enumerate(_DATACLASS_OPTION_ORDER) +} +_REQUIRED_OPTIONS = ("kw_only", "slots") + + +def _is_literal_true(expression: ast.expr | None) -> bool: + """Return whether an AST expression is the literal value true.""" + return isinstance(expression, ast.Constant) and expression.value is True + + +def _dataclass_bindings(tree: ast.Module) -> tuple[set[str], set[str]]: + """Return direct decorator names and imported dataclasses module names.""" + decorator_names: set[str] = set() + module_names: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom) and node.module == "dataclasses": + decorator_names.update( + alias.asname or alias.name + for alias in node.names + if alias.name == "dataclass" + ) + elif isinstance(node, ast.Import): + module_names.update( + alias.asname or alias.name + for alias in node.names + if alias.name == "dataclasses" + ) + return decorator_names, module_names + + +def _is_dataclass_reference( + expression: ast.expr, + decorator_names: set[str], + module_names: set[str], +) -> bool: + """Recognize imported direct and module-qualified dataclass references.""" + if isinstance(expression, ast.Name): + return expression.id in decorator_names + return ( + isinstance(expression, ast.Attribute) + and expression.attr == "dataclass" + and isinstance(expression.value, ast.Name) + and expression.value.id in module_names + ) + + +def _dataclass_keywords( + decorator: ast.expr, + decorator_names: set[str], + module_names: set[str], +) -> list[ast.keyword] | None: + """Return explicit keywords when `decorator` is an imported dataclass.""" + reference = decorator.func if isinstance(decorator, ast.Call) else decorator + if not _is_dataclass_reference(reference, decorator_names, module_names): + return None + if not isinstance(decorator, ast.Call): + return [] + return decorator.keywords + + +def _violations( + path: Path, + repository_root: Path, +) -> list[str]: + """Return invalid dataclass declarations from one Python file.""" + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + decorator_names, module_names = _dataclass_bindings(tree) + relative_path = path.relative_to(repository_root) + violations: list[str] = [] + + for node in ast.walk(tree): + if not isinstance(node, ast.ClassDef): + continue + for decorator in node.decorator_list: + keywords = _dataclass_keywords( + decorator, + decorator_names, + module_names, + ) + if keywords is None: + continue + options = { + keyword.arg: keyword.value + for keyword in keywords + if keyword.arg is not None + } + missing = [ + option + for option in _REQUIRED_OPTIONS + if not _is_literal_true(options.get(option)) + ] + if missing: + violations.append( + f"{relative_path}:{node.lineno}: {node.name} must set " + + ", ".join(f"{option}=True" for option in missing) + ) + option_names = [ + keyword.arg for keyword in keywords if keyword.arg is not None + ] + canonical_option_names = sorted( + option_names, + key=lambda option: _DATACLASS_OPTION_POSITION.get( + option, + len(_DATACLASS_OPTION_POSITION), + ), + ) + if option_names != canonical_option_names: + violations.append( + f"{relative_path}:{node.lineno}: {node.name} must order " + "dataclass options as " + ", ".join(canonical_option_names) + ) + + return violations + + +def test_dataclasses_follow_policy() -> None: + """Require consistent dataclass construction and option ordering.""" + repository_root = Path(__file__).resolve().parents[1] + violations = [ + violation + for directory in ("src", "tests") + for path in sorted((repository_root / directory).rglob("*.py")) + for violation in _violations(path, repository_root) + ] + + assert not violations, "Dataclass policy violations:\n" + "\n".join(violations) diff --git a/tests/test_dead_code.py b/tests/test_dead_code.py new file mode 100644 index 0000000..05e4d74 --- /dev/null +++ b/tests/test_dead_code.py @@ -0,0 +1,21 @@ +"""Repository-wide dead-code enforcement tests.""" + +from __future__ import annotations + +import subprocess +import sys +from pathlib import Path + + +def test_vulture_reports_no_dead_code() -> None: + """Reject unreviewed dead Python code across source and tests.""" + repository_root = Path(__file__).resolve().parents[1] + result = subprocess.run( + [sys.executable, "-m", "vulture"], + cwd=repository_root, + capture_output=True, + text=True, + check=False, + ) + output = "\n".join(part for part in (result.stdout, result.stderr) if part) + assert result.returncode == 0, output diff --git a/tests/test_documentation_examples.py b/tests/test_documentation_examples.py new file mode 100644 index 0000000..91acb52 --- /dev/null +++ b/tests/test_documentation_examples.py @@ -0,0 +1,167 @@ +"""Quality checks for executable Python examples in project documentation.""" + +from __future__ import annotations + +import subprocess +import sys +from pathlib import Path + + +def _python_fence_start(line: str) -> tuple[str, int, int] | None: + """Return marker, width, and indentation for a Python opening fence.""" + content = line.rstrip("\r\n") + stripped = content.lstrip(" ") + indentation = len(content) - len(stripped) + if indentation > 3 or not stripped: + return None + + marker = stripped[0] + if marker not in ("`", "~"): + return None + + fence_width = len(stripped) - len(stripped.lstrip(marker)) + if fence_width < 3: + return None + + info_string = stripped[fence_width:].strip() + if not info_string or info_string.split(maxsplit=1)[0] != "python": + return None + return marker, fence_width, indentation + + +def _is_closing_fence(line: str, *, marker: str, minimum_width: int) -> bool: + """Return whether a line closes the active Markdown code fence.""" + content = line.rstrip("\r\n") + stripped = content.lstrip(" ") + indentation = len(content) - len(stripped) + if indentation > 3 or not stripped.startswith(marker): + return False + + fence_width = len(stripped) - len(stripped.lstrip(marker)) + return fence_width >= minimum_width and not stripped[fence_width:].strip() + + +def _strip_fence_indentation(line: str, indentation: int) -> str: + """Remove the opening fence indentation from one example line.""" + leading_spaces = len(line) - len(line.lstrip(" ")) + return line[min(indentation, leading_spaces) :] + + +def _python_examples(path: Path) -> list[str]: + """Extract fenced Python examples from one Markdown document.""" + examples: list[str] = [] + active_fence: tuple[str, int, int] | None = None + example_lines: list[str] = [] + + for line in path.read_text(encoding="utf-8").splitlines(keepends=True): + if active_fence is None: + active_fence = _python_fence_start(line) + if active_fence is not None: + example_lines = [] + continue + + marker, minimum_width, indentation = active_fence + if _is_closing_fence( + line, + marker=marker, + minimum_width=minimum_width, + ): + examples.append("".join(example_lines)) + active_fence = None + continue + example_lines.append(_strip_fence_indentation(line, indentation)) + + if active_fence is not None: + examples.append("".join(example_lines)) + return examples + + +def _run_ruff_lint( + source: str, repository_root: Path +) -> subprocess.CompletedProcess[str]: + """Run the configured Ruff linter against one documentation example.""" + return subprocess.run( + [ + sys.executable, + "-m", + "ruff", + "check", + "--no-cache", + "--stdin-filename", + "documentation_example.py", + "-", + ], + cwd=repository_root, + input=source, + capture_output=True, + text=True, + check=False, + ) + + +def _run_ruff_format( + source: str, + repository_root: Path, +) -> subprocess.CompletedProcess[str]: + """Run the configured Ruff formatter check against one documentation example.""" + return subprocess.run( + [ + sys.executable, + "-m", + "ruff", + "format", + "--check", + "--no-cache", + "--stdin-filename", + "documentation_example.py", + "-", + ], + cwd=repository_root, + input=source, + capture_output=True, + text=True, + check=False, + ) + + +def test_python_examples_support_markdown_fence_variants(tmp_path: Path) -> None: + """Recognize indentation, attributes, tildes, and longer closing fences.""" + markdown_path = tmp_path / "examples.md" + markdown_path.write_text( + '```python title="basic"\nfirst = 1\n```\n' + "\n" + ' ~~~python linenums="1"\n second = 2\n ~~~~\n' + "\n" + "```bash\necho ignored\n```\n", + encoding="utf-8", + ) + + assert _python_examples(markdown_path) == ["first = 1\n", "second = 2\n"] + + +def test_documentation_python_examples_pass_ruff() -> None: + """Require every documented Python example to satisfy the Ruff policy.""" + repository_root = Path(__file__).resolve().parents[1] + markdown_paths = [ + repository_root / "README.md", + *sorted((repository_root / "docs").rglob("*.md")), + ] + failures: list[str] = [] + + for path in markdown_paths: + for example_number, example in enumerate(_python_examples(path), start=1): + checks = ( + ("lint", _run_ruff_lint(example, repository_root)), + ("format", _run_ruff_format(example, repository_root)), + ) + for check_name, result in checks: + if result.returncode != 0: + output = "\n".join( + part for part in (result.stdout, result.stderr) if part + ) + failures.append( + f"{path.relative_to(repository_root)} example " + f"{example_number} {check_name}:\n{output}" + ) + + assert not failures, "\n\n".join(failures) diff --git a/tests/test_sensor_init_modes.py b/tests/test_sensor_init_modes.py index cb66b7a..6fe6760 100644 --- a/tests/test_sensor_init_modes.py +++ b/tests/test_sensor_init_modes.py @@ -2,8 +2,8 @@ from conftest import FakeSerialFactory -from xkc_kl200_python import XKC_KL200, CommunicationMode, LedMode, RelayMode -from xkc_kl200_python.constants import XKC_KL200_Status +from xkc_kl200_python import CommunicationMode, LedMode, RelayMode, XkcKl200 +from xkc_kl200_python.constants import XkcKl200Status from xkc_kl200_python.utils import build_command_frame @@ -11,14 +11,14 @@ def test_invalid_led_mode_returns_invalid_parameter( serial_factory: FakeSerialFactory, ) -> None: """Verify that invalid LED enum values are rejected before any I/O.""" - sensor = XKC_KL200(port="/dev/ttyUSB0", serial_factory=serial_factory) + sensor = XkcKl200(port="/dev/ttyUSB0", serial_factory=serial_factory) - assert sensor.set_led_mode(9) == XKC_KL200_Status.INVALID_PARAMETER + assert sensor.set_led_mode(9) == XkcKl200Status.INVALID_PARAMETER def test_set_control_modes_acknowledge(serial_factory: FakeSerialFactory) -> None: """Verify that the supported mode-setting commands all receive acknowledgements.""" - sensor = XKC_KL200(port="/dev/ttyUSB0", timeout=0.01, serial_factory=serial_factory) + sensor = XkcKl200(port="/dev/ttyUSB0", timeout=0.01, serial_factory=serial_factory) serial_port = serial_factory.holder["serial"] serial_port.queue_read( build_command_frame(header=0x62, command=0x37, address=0xFFFF) @@ -30,12 +30,10 @@ def test_set_control_modes_acknowledge(serial_factory: FakeSerialFactory) -> Non build_command_frame(header=0x61, command=0x30, address=0xFFFF) ) - assert sensor.set_led_mode(LedMode.ALWAYS_ON) == XKC_KL200_Status.SUCCESS + assert sensor.set_led_mode(LedMode.ALWAYS_ON) == XkcKl200Status.SUCCESS assert ( - sensor.set_relay_mode(RelayMode.ACTIVE_WHEN_DETECTED) - == XKC_KL200_Status.SUCCESS + sensor.set_relay_mode(RelayMode.ACTIVE_WHEN_DETECTED) == XkcKl200Status.SUCCESS ) assert ( - sensor.set_communication_mode(CommunicationMode.UART) - == XKC_KL200_Status.SUCCESS + sensor.set_communication_mode(CommunicationMode.UART) == XkcKl200Status.SUCCESS ) diff --git a/tests/test_sensor_misc.py b/tests/test_sensor_misc.py index 31d6a01..a4691b3 100644 --- a/tests/test_sensor_misc.py +++ b/tests/test_sensor_misc.py @@ -3,14 +3,14 @@ import pytest from conftest import FakeSerialFactory -from xkc_kl200_python import XKC_KL200, CommunicationMode, SensorConfig -from xkc_kl200_python.constants import XKC_KL200_Status +from xkc_kl200_python import CommunicationMode, SensorConfig, XkcKl200 +from xkc_kl200_python.constants import XkcKl200Status from xkc_kl200_python.utils import FramePayload, build_command_frame def make_sensor( serial_factory: FakeSerialFactory, *, timeout: float = 0.01 -) -> XKC_KL200: +) -> XkcKl200: """Create a zero-startup-delay sensor for focused behavior tests.""" config = SensorConfig( port="/dev/ttyUSB0", @@ -18,18 +18,18 @@ def make_sensor( timeout=timeout, startup_delay_s=0.0, ) - return XKC_KL200(config=config, serial_factory=serial_factory) + return XkcKl200(config=config, serial_factory=serial_factory) def test_init_without_port_or_config_raises() -> None: """Verify that callers must provide either a port or a full config object.""" with pytest.raises(ValueError, match="port is required"): - XKC_KL200() + XkcKl200() def test_context_manager_closes_serial(serial_factory: FakeSerialFactory) -> None: """Verify that context-manager use always closes the serial port.""" - sensor: XKC_KL200 + sensor: XkcKl200 with make_sensor(serial_factory) as sensor: assert sensor is not None assert sensor.address == 0xFFFF @@ -85,8 +85,8 @@ def test_hard_and_soft_reset(serial_factory: FakeSerialFactory) -> None: build_command_frame(header=0x62, command=0x39, address=0xFFFF) ) - assert sensor.hard_reset() == XKC_KL200_Status.SUCCESS - assert sensor.soft_reset() == XKC_KL200_Status.SUCCESS + assert sensor.hard_reset() == XkcKl200Status.SUCCESS + assert sensor.soft_reset() == XkcKl200Status.SUCCESS assert serial_port.reset_input_count == 2 @@ -102,7 +102,7 @@ def test_change_address_success_updates_config_and_state( result = sensor.change_address(0x1234) - assert result == XKC_KL200_Status.SUCCESS + assert result == XkcKl200Status.SUCCESS assert sensor.config.address == 0x1234 assert sensor.address == 0x1234 @@ -111,14 +111,14 @@ def test_change_address_invalid_parameter(serial_factory: FakeSerialFactory) -> """Verify that out-of-range addresses are rejected before touching the port.""" sensor = make_sensor(serial_factory) - assert sensor.change_address(0x1_0000) == XKC_KL200_Status.INVALID_PARAMETER + assert sensor.change_address(0x1_0000) == XkcKl200Status.INVALID_PARAMETER def test_change_baud_rate_invalid_parameter(serial_factory: FakeSerialFactory) -> None: """Verify that unsupported baud values are rejected before sending a command.""" sensor = make_sensor(serial_factory) - assert sensor.change_baud_rate(12345) == XKC_KL200_Status.INVALID_PARAMETER + assert sensor.change_baud_rate(12345) == XkcKl200Status.INVALID_PARAMETER def test_invalid_relay_and_communication_modes( @@ -127,8 +127,8 @@ def test_invalid_relay_and_communication_modes( """Verify that invalid enum values for relay and communication mode are rejected.""" sensor = make_sensor(serial_factory) - assert sensor.set_relay_mode(3) == XKC_KL200_Status.INVALID_PARAMETER - assert sensor.set_communication_mode(9) == XKC_KL200_Status.INVALID_PARAMETER + assert sensor.set_relay_mode(3) == XkcKl200Status.INVALID_PARAMETER + assert sensor.set_communication_mode(9) == XkcKl200Status.INVALID_PARAMETER def test_wait_for_response_timeout_and_errors( @@ -138,20 +138,20 @@ def test_wait_for_response_timeout_and_errors( sensor = make_sensor(serial_factory, timeout=0.0) serial_port = serial_factory.holder["serial"] - assert sensor._wait_for_response(expected_command=0x34) == XKC_KL200_Status.TIMEOUT + assert sensor._wait_for_response(expected_command=0x34) == XkcKl200Status.TIMEOUT serial_port.queue_read( bytes([0x62, 0x34, 0x09, 0xFF, 0xFF, 0x00, 0x00, 0x00, 0x00]) ) assert ( sensor._wait_for_response(expected_command=0x34) - == XKC_KL200_Status.CHECKSUM_ERROR + == XkcKl200Status.CHECKSUM_ERROR ) serial_port.queue_read(b"\x62\x34\x08\xff\xff\x00\x00\x00\x00") assert ( sensor._wait_for_response(expected_command=0x34) - == XKC_KL200_Status.RESPONSE_ERROR + == XkcKl200Status.RESPONSE_ERROR ) @@ -173,7 +173,7 @@ def test_wait_for_response_skips_stale_valid_frame( build_command_frame(header=0x62, command=0x39, address=0xFFFF) ) - assert sensor._wait_for_response(expected_command=0x39) == XKC_KL200_Status.SUCCESS + assert sensor._wait_for_response(expected_command=0x39) == XkcKl200Status.SUCCESS def test_wait_for_response_stale_frame_then_timeout( @@ -186,14 +186,14 @@ def test_wait_for_response_stale_frame_then_timeout( build_command_frame(header=0x62, command=0x35, address=0xFFFF) ) - assert sensor._wait_for_response(expected_command=0x34) == XKC_KL200_Status.TIMEOUT + assert sensor._wait_for_response(expected_command=0x34) == XkcKl200Status.TIMEOUT def test_resolve_baud_rate_code_static_helper() -> None: """Verify the small baud helper supports both values and raw codes.""" - assert XKC_KL200._resolve_baud_rate_code(9600) == 2 - assert XKC_KL200._resolve_baud_rate_code(8) == 8 - assert XKC_KL200._resolve_baud_rate_code(99999) is None + assert XkcKl200._resolve_baud_rate_code(9600) == 2 + assert XkcKl200._resolve_baud_rate_code(8) == 8 + assert XkcKl200._resolve_baud_rate_code(99999) is None def test_close_delegates_to_serial_manager( @@ -223,7 +223,7 @@ def test_communication_mode_success_uses_system_header( result = sensor.set_communication_mode(CommunicationMode.UART) - assert result == XKC_KL200_Status.SUCCESS + assert result == XkcKl200Status.SUCCESS assert serial_port.written_frames[-1][0] == 0x61 @@ -239,4 +239,4 @@ def test_set_communication_mode_requires_system_header_ack( result = sensor.set_communication_mode(CommunicationMode.UART) - assert result == XKC_KL200_Status.TIMEOUT + assert result == XkcKl200Status.TIMEOUT diff --git a/tests/test_sensor_reading.py b/tests/test_sensor_reading.py index 3224ade..641b35f 100644 --- a/tests/test_sensor_reading.py +++ b/tests/test_sensor_reading.py @@ -4,17 +4,17 @@ from conftest import FakeSerialFactory from xkc_kl200_python import ( - XKC_KL200, - XKC_KL200_ResponseError, - XKC_KL200_TimeoutError, + XkcKl200, + XkcKl200ResponseError, + XkcKl200TimeoutError, ) -from xkc_kl200_python.constants import XKC_KL200_Status +from xkc_kl200_python.constants import XkcKl200Status from xkc_kl200_python.utils import FramePayload, build_command_frame def test_read_distance_updates_state(serial_factory: FakeSerialFactory) -> None: """Verify that a valid measurement updates the cached value and address.""" - sensor = XKC_KL200(port="/dev/ttyUSB0", timeout=0.01, serial_factory=serial_factory) + sensor = XkcKl200(port="/dev/ttyUSB0", timeout=0.01, serial_factory=serial_factory) serial_port = serial_factory.holder["serial"] serial_port.queue_read( bytes([0x62, 0x33, 0x09, 0xFF, 0xFF, 0x01, 0x2C, 0x00, 0x75]) @@ -35,10 +35,10 @@ def test_read_distance_timeout_returns_last_distance( serial_factory: FakeSerialFactory, ) -> None: """Verify that timeouts fail explicitly instead of returning cached data.""" - sensor = XKC_KL200(port="/dev/ttyUSB0", timeout=0.0, serial_factory=serial_factory) + sensor = XkcKl200(port="/dev/ttyUSB0", timeout=0.0, serial_factory=serial_factory) sensor._last_received_distance_mm = 123 - with pytest.raises(XKC_KL200_TimeoutError): + with pytest.raises(XkcKl200TimeoutError): sensor.read_distance(timeout=0.0) assert sensor.last_received_distance == 123 @@ -48,14 +48,14 @@ def test_read_distance_invalid_frame_returns_last_distance( serial_factory: FakeSerialFactory, ) -> None: """Verify that corrupted expected frames fail explicitly and preserve the last good value.""" - sensor = XKC_KL200(port="/dev/ttyUSB0", timeout=0.01, serial_factory=serial_factory) + sensor = XkcKl200(port="/dev/ttyUSB0", timeout=0.01, serial_factory=serial_factory) serial_port = serial_factory.holder["serial"] sensor._last_received_distance_mm = 55 serial_port.queue_read( bytes([0x62, 0x33, 0x09, 0xFF, 0xFF, 0x00, 0x64, 0x00, 0x00]) ) - with pytest.raises(XKC_KL200_ResponseError): + with pytest.raises(XkcKl200ResponseError): sensor.read_distance() assert sensor.last_received_distance == 55 @@ -65,7 +65,7 @@ def test_read_distance_recovers_from_checksum_error_with_buffered_reply( serial_factory: FakeSerialFactory, ) -> None: """Verify that a valid measurement buffered behind a bad frame is still returned.""" - sensor = XKC_KL200(port="/dev/ttyUSB0", timeout=0.0, serial_factory=serial_factory) + sensor = XkcKl200(port="/dev/ttyUSB0", timeout=0.0, serial_factory=serial_factory) serial_port = serial_factory.holder["serial"] serial_port.queue_read( bytes([0x62, 0x33, 0x09, 0xFF, 0xFF, 0x00, 0x64, 0x00, 0x00]) @@ -87,12 +87,12 @@ def test_read_distance_malformed_complete_frame_raises_response_error( serial_factory: FakeSerialFactory, ) -> None: """Verify that a complete malformed measurement is not misreported as a timeout.""" - sensor = XKC_KL200(port="/dev/ttyUSB0", timeout=0.0, serial_factory=serial_factory) + sensor = XkcKl200(port="/dev/ttyUSB0", timeout=0.0, serial_factory=serial_factory) serial_port = serial_factory.holder["serial"] sensor._last_received_distance_mm = 55 serial_port.queue_read(b"\x62\x33\x08\xff\xff\x00\x64\x00\x00") - with pytest.raises(XKC_KL200_ResponseError): + with pytest.raises(XkcKl200ResponseError): sensor.read_distance(timeout=0.0) assert sensor.last_received_distance == 55 @@ -102,7 +102,7 @@ def test_read_distance_wrong_header_reply_raises_response_error( serial_factory: FakeSerialFactory, ) -> None: """Verify that a complete wrong-header reply is surfaced as a response error.""" - sensor = XKC_KL200(port="/dev/ttyUSB0", timeout=0.0, serial_factory=serial_factory) + sensor = XkcKl200(port="/dev/ttyUSB0", timeout=0.0, serial_factory=serial_factory) serial_port = serial_factory.holder["serial"] serial_port.queue_read( build_command_frame( @@ -113,7 +113,7 @@ def test_read_distance_wrong_header_reply_raises_response_error( ) ) - with pytest.raises(XKC_KL200_ResponseError): + with pytest.raises(XkcKl200ResponseError): sensor.read_distance(timeout=0.0) @@ -121,11 +121,11 @@ def test_read_distance_junk_before_timeout( serial_factory: FakeSerialFactory, ) -> None: """Verify that stray junk still ends as a timeout when no measurement follows.""" - sensor = XKC_KL200(port="/dev/ttyUSB0", timeout=0.0, serial_factory=serial_factory) + sensor = XkcKl200(port="/dev/ttyUSB0", timeout=0.0, serial_factory=serial_factory) serial_port = serial_factory.holder["serial"] serial_port.queue_read(b"\x60\x33\x09") - with pytest.raises(XKC_KL200_TimeoutError): + with pytest.raises(XkcKl200TimeoutError): sensor.read_distance() @@ -133,7 +133,7 @@ def test_read_distance_skips_stale_valid_frame( serial_factory: FakeSerialFactory, ) -> None: """Verify that stale valid frames are skipped until the measurement arrives.""" - sensor = XKC_KL200(port="/dev/ttyUSB0", timeout=0.01, serial_factory=serial_factory) + sensor = XkcKl200(port="/dev/ttyUSB0", timeout=0.01, serial_factory=serial_factory) serial_port = serial_factory.holder["serial"] serial_port.queue_read( build_command_frame(header=0x62, command=0x39, address=0xFFFF) @@ -157,7 +157,7 @@ def test_read_distance_skips_stale_valid_frame_in_same_chunk( serial_factory: FakeSerialFactory, ) -> None: """Verify that a coalesced stale frame and measurement still succeed at zero timeout.""" - sensor = XKC_KL200(port="/dev/ttyUSB0", timeout=0.0, serial_factory=serial_factory) + sensor = XkcKl200(port="/dev/ttyUSB0", timeout=0.0, serial_factory=serial_factory) serial_port = serial_factory.holder["serial"] serial_port.queue_read( build_command_frame(header=0x62, command=0x39, address=0xFFFF) @@ -179,7 +179,7 @@ def test_read_distance_resynchronizes_within_checksum_valid_overlap( serial_factory: FakeSerialFactory, ) -> None: """Verify that an overlapped checksum-valid window still recovers the buffered measurement.""" - sensor = XKC_KL200(port="/dev/ttyUSB0", timeout=0.0, serial_factory=serial_factory) + sensor = XkcKl200(port="/dev/ttyUSB0", timeout=0.0, serial_factory=serial_factory) serial_port = serial_factory.holder["serial"] measurement_frame = build_command_frame( header=0x62, @@ -200,7 +200,7 @@ def test_change_baud_rate_accepts_real_baudrate( serial_factory: FakeSerialFactory, ) -> None: """Verify that change_baud_rate accepts human-readable baud values.""" - sensor = XKC_KL200(port="/dev/ttyUSB0", timeout=0.01, serial_factory=serial_factory) + sensor = XkcKl200(port="/dev/ttyUSB0", timeout=0.01, serial_factory=serial_factory) serial_port = serial_factory.holder["serial"] serial_port.queue_read( build_command_frame( @@ -213,7 +213,7 @@ def test_change_baud_rate_accepts_real_baudrate( result = sensor.change_baud_rate(115200) - assert result == XKC_KL200_Status.SUCCESS + assert result == XkcKl200Status.SUCCESS assert sensor.config.baudrate == 115200 assert serial_port.baudrate == 115200 @@ -222,7 +222,7 @@ def test_change_baud_rate_requires_command_header_ack( serial_factory: FakeSerialFactory, ) -> None: """Verify that change_baud_rate ignores a stale system-header ACK sharing command 0x30.""" - sensor = XKC_KL200(port="/dev/ttyUSB0", timeout=0.0, serial_factory=serial_factory) + sensor = XkcKl200(port="/dev/ttyUSB0", timeout=0.0, serial_factory=serial_factory) serial_port = serial_factory.holder["serial"] serial_port.queue_read( build_command_frame( @@ -235,6 +235,6 @@ def test_change_baud_rate_requires_command_header_ack( result = sensor.change_baud_rate(115200) - assert result == XKC_KL200_Status.TIMEOUT + assert result == XkcKl200Status.TIMEOUT assert sensor.config.baudrate == 9600 assert serial_port.baudrate == 9600 diff --git a/tests/test_serial_manager.py b/tests/test_serial_manager.py index a8a2bf2..b9b3e73 100644 --- a/tests/test_serial_manager.py +++ b/tests/test_serial_manager.py @@ -4,7 +4,7 @@ from conftest import FakeSerial, FakeSerialFactory from xkc_kl200_python.config import SensorConfig -from xkc_kl200_python.constants import XKC_KL200_Status +from xkc_kl200_python.constants import XkcKl200Status from xkc_kl200_python.serial_manager import SerialManager, default_serial_factory from xkc_kl200_python.utils import FramePayload, build_command_frame @@ -74,13 +74,13 @@ def test_read_frame_handles_empty_read_with_waiting( ) serial_port = serial_factory.holder["serial"] - monkeypatch.setattr(type(serial_port), "in_waiting", property(lambda self: 1)) - monkeypatch.setattr(serial_port, "read", lambda size=1: b"") + monkeypatch.setattr(type(serial_port), "in_waiting", property(lambda _self: 1)) + monkeypatch.setattr(serial_port, "read", lambda _size=1: b"") result, status = manager.read_frame(expected_command=0x33, timeout=0.0) assert result is None - assert status == XKC_KL200_Status.TIMEOUT + assert status == XkcKl200Status.TIMEOUT def test_read_frame_sleeps_before_timeout( @@ -100,13 +100,13 @@ def test_read_frame_sleeps_before_timeout( ) monkeypatch.setattr( "xkc_kl200_python.serial_manager.time.sleep", - lambda duration: sleep_calls.append(duration), + sleep_calls.append, ) result, status = manager.read_frame(expected_command=0x33, timeout=0.001) assert result is None - assert status == XKC_KL200_Status.TIMEOUT + assert status == XkcKl200Status.TIMEOUT assert sleep_calls == [0.001] @@ -135,7 +135,7 @@ def test_read_frame_skips_leading_junk(serial_factory: FakeSerialFactory) -> Non address=0xFFFF, payload=FramePayload(data_low=17), ) - assert status == XKC_KL200_Status.SUCCESS + assert status == XkcKl200Status.SUCCESS def test_extract_frame_keeps_partial_header_after_leading_junk( @@ -174,7 +174,7 @@ def test_read_frame_waits_for_frame_prefix(serial_factory: FakeSerialFactory) -> result, status = manager.read_frame(expected_command=0x33, timeout=0.01) assert result == frame - assert status == XKC_KL200_Status.SUCCESS + assert status == XkcKl200Status.SUCCESS def test_read_frame_zero_timeout_drains_buffered_chunks( @@ -193,7 +193,7 @@ def test_read_frame_zero_timeout_drains_buffered_chunks( result, status = manager.read_frame(expected_command=0x33, timeout=0.0) assert result == frame - assert status == XKC_KL200_Status.SUCCESS + assert status == XkcKl200Status.SUCCESS def test_read_frame_waits_for_partial_frame(serial_factory: FakeSerialFactory) -> None: @@ -210,7 +210,7 @@ def test_read_frame_waits_for_partial_frame(serial_factory: FakeSerialFactory) - result, status = manager.read_frame(expected_command=0x33, timeout=0.01) assert result == frame - assert status == XKC_KL200_Status.SUCCESS + assert status == XkcKl200Status.SUCCESS def test_read_frame_skips_invalid_length_then_recovers( @@ -240,7 +240,7 @@ def test_read_frame_skips_invalid_length_then_recovers( address=0xFFFF, payload=FramePayload(data_low=18), ) - assert status == XKC_KL200_Status.SUCCESS + assert status == XkcKl200Status.SUCCESS def test_read_frame_invalid_length_reports_response_error( @@ -257,7 +257,7 @@ def test_read_frame_invalid_length_reports_response_error( result, status = manager.read_frame(expected_command=0x33, timeout=0.0) assert result is None - assert status == XKC_KL200_Status.RESPONSE_ERROR + assert status == XkcKl200Status.RESPONSE_ERROR def test_read_frame_checksum_error(serial_factory: FakeSerialFactory) -> None: @@ -274,7 +274,7 @@ def test_read_frame_checksum_error(serial_factory: FakeSerialFactory) -> None: result, status = manager.read_frame(expected_command=0x33, timeout=0.0) assert result is None - assert status == XKC_KL200_Status.CHECKSUM_ERROR + assert status == XkcKl200Status.CHECKSUM_ERROR def test_read_frame_recovers_from_checksum_error_with_buffered_valid_reply( @@ -299,7 +299,7 @@ def test_read_frame_recovers_from_checksum_error_with_buffered_valid_reply( result, status = manager.read_frame(expected_command=0x33, timeout=0.0) assert result == valid_frame - assert status == XKC_KL200_Status.SUCCESS + assert status == XkcKl200Status.SUCCESS def test_read_frame_waits_through_interframe_gap_after_checksum_error( @@ -333,7 +333,7 @@ def fake_sleep(duration: float) -> None: result, status = manager.read_frame(expected_command=0x33, timeout=0.01) assert result == valid_frame - assert status == XKC_KL200_Status.SUCCESS + assert status == XkcKl200Status.SUCCESS assert sleep_calls == [0.001] @@ -360,7 +360,7 @@ def test_read_frame_recovers_from_checksum_error_with_next_ready_chunk( result, status = manager.read_frame(expected_command=0x33, timeout=1.0) assert result == valid_frame - assert status == XKC_KL200_Status.SUCCESS + assert status == XkcKl200Status.SUCCESS def test_read_frame_returns_deferred_checksum_error_at_timeout( @@ -383,13 +383,13 @@ def test_read_frame_returns_deferred_checksum_error_at_timeout( ) monkeypatch.setattr( "xkc_kl200_python.serial_manager.time.sleep", - lambda duration: sleep_calls.append(duration), + sleep_calls.append, ) result, status = manager.read_frame(expected_command=0x33, timeout=0.001) assert result is None - assert status == XKC_KL200_Status.CHECKSUM_ERROR + assert status == XkcKl200Status.CHECKSUM_ERROR assert sleep_calls == [] @@ -409,7 +409,7 @@ def test_read_frame_invalid_header_reports_response_error( result, status = manager.read_frame(expected_command=0x33, timeout=0.0) assert result is None - assert status == XKC_KL200_Status.RESPONSE_ERROR + assert status == XkcKl200Status.RESPONSE_ERROR def test_read_frame_wrong_header_for_unique_command_reports_response_error( @@ -432,7 +432,7 @@ def test_read_frame_wrong_header_for_unique_command_reports_response_error( ) assert result is None - assert status == XKC_KL200_Status.RESPONSE_ERROR + assert status == XkcKl200Status.RESPONSE_ERROR def test_read_frame_matches_expected_header_and_command_for_ambiguous_opcode( @@ -458,7 +458,7 @@ def test_read_frame_matches_expected_header_and_command_for_ambiguous_opcode( ) assert result == expected_frame - assert status == XKC_KL200_Status.SUCCESS + assert status == XkcKl200Status.SUCCESS def test_read_frame_resynchronizes_within_checksum_valid_mismatched_candidate( @@ -481,7 +481,7 @@ def test_read_frame_resynchronizes_within_checksum_valid_mismatched_candidate( result, status = manager.read_frame(expected_command=0x33, timeout=0.0) assert result == expected_frame - assert status == XKC_KL200_Status.SUCCESS + assert status == XkcKl200Status.SUCCESS def test_read_frame_zero_timeout_rescans_buffer_after_skipping_stale_frame( @@ -501,7 +501,7 @@ def test_read_frame_zero_timeout_rescans_buffer_after_skipping_stale_frame( result, status = manager.read_frame(expected_command=0x33, timeout=0.0) assert result == expected_frame - assert status == XKC_KL200_Status.SUCCESS + assert status == XkcKl200Status.SUCCESS def test_read_frame_reports_checksum_error_for_corrupted_command( @@ -520,7 +520,7 @@ def test_read_frame_reports_checksum_error_for_corrupted_command( result, status = manager.read_frame(expected_command=0x33, timeout=0.0) assert result is None - assert status == XKC_KL200_Status.CHECKSUM_ERROR + assert status == XkcKl200Status.CHECKSUM_ERROR def test_read_frame_unexpected_command_error( @@ -539,7 +539,7 @@ def test_read_frame_unexpected_command_error( result, status = manager.read_frame(expected_command=0x33, timeout=0.0) assert result is None - assert status == XKC_KL200_Status.TIMEOUT + assert status == XkcKl200Status.TIMEOUT def test_read_frame_honors_deadline_while_draining_stale_frames( @@ -564,7 +564,7 @@ def test_read_frame_honors_deadline_while_draining_stale_frames( result, status = manager.read_frame(expected_command=0x34, timeout=0.0) assert result is None - assert status == XKC_KL200_Status.TIMEOUT + assert status == XkcKl200Status.TIMEOUT def test_read_frame_clears_partial_buffer_on_timeout( @@ -596,6 +596,6 @@ def test_read_frame_clears_partial_buffer_on_timeout( ) assert timed_out_result is None - assert timed_out_status == XKC_KL200_Status.TIMEOUT + assert timed_out_status == XkcKl200Status.TIMEOUT assert next_result is None - assert next_status == XKC_KL200_Status.TIMEOUT + assert next_status == XkcKl200Status.TIMEOUT From 951904f9c917df0ef117aaa18c81d2f04b854a44 Mon Sep 17 00:00:00 2001 From: sam0rr Date: Wed, 22 Jul 2026 14:16:55 -0400 Subject: [PATCH 2/2] Harden documentation example parsing --- tests/test_documentation_examples.py | 29 +++++++++++++++------------- 1 file changed, 16 insertions(+), 13 deletions(-) diff --git a/tests/test_documentation_examples.py b/tests/test_documentation_examples.py index 91acb52..c0a621d 100644 --- a/tests/test_documentation_examples.py +++ b/tests/test_documentation_examples.py @@ -7,8 +7,8 @@ from pathlib import Path -def _python_fence_start(line: str) -> tuple[str, int, int] | None: - """Return marker, width, and indentation for a Python opening fence.""" +def _fence_start(line: str) -> tuple[str, int, int, bool] | None: + """Return the structure and Python status of a Markdown opening fence.""" content = line.rstrip("\r\n") stripped = content.lstrip(" ") indentation = len(content) - len(stripped) @@ -24,9 +24,8 @@ def _python_fence_start(line: str) -> tuple[str, int, int] | None: return None info_string = stripped[fence_width:].strip() - if not info_string or info_string.split(maxsplit=1)[0] != "python": - return None - return marker, fence_width, indentation + language = info_string.split(maxsplit=1)[0] if info_string else "" + return marker, fence_width, indentation, language == "python" def _is_closing_fence(line: str, *, marker: str, minimum_width: int) -> bool: @@ -50,28 +49,30 @@ def _strip_fence_indentation(line: str, indentation: int) -> str: def _python_examples(path: Path) -> list[str]: """Extract fenced Python examples from one Markdown document.""" examples: list[str] = [] - active_fence: tuple[str, int, int] | None = None + active_fence: tuple[str, int, int, bool] | None = None example_lines: list[str] = [] for line in path.read_text(encoding="utf-8").splitlines(keepends=True): if active_fence is None: - active_fence = _python_fence_start(line) - if active_fence is not None: + active_fence = _fence_start(line) + if active_fence is not None and active_fence[3]: example_lines = [] continue - marker, minimum_width, indentation = active_fence + marker, minimum_width, indentation, is_python = active_fence if _is_closing_fence( line, marker=marker, minimum_width=minimum_width, ): - examples.append("".join(example_lines)) + if is_python: + examples.append("".join(example_lines)) active_fence = None continue - example_lines.append(_strip_fence_indentation(line, indentation)) + if is_python: + example_lines.append(_strip_fence_indentation(line, indentation)) - if active_fence is not None: + if active_fence is not None and active_fence[3]: examples.append("".join(example_lines)) return examples @@ -132,7 +133,9 @@ def test_python_examples_support_markdown_fence_variants(tmp_path: Path) -> None "\n" ' ~~~python linenums="1"\n second = 2\n ~~~~\n' "\n" - "```bash\necho ignored\n```\n", + "```bash\necho ignored\n```\n" + "\n" + "````markdown\n```python\nnot_python_here = 1\n```\n````\n", encoding="utf-8", )