diff --git a/docs/superpowers/contracts/brother-printer-status-accuracy-2026-07-17.json b/docs/superpowers/contracts/brother-printer-status-accuracy-2026-07-17.json new file mode 100644 index 00000000..cb95d1d0 --- /dev/null +++ b/docs/superpowers/contracts/brother-printer-status-accuracy-2026-07-17.json @@ -0,0 +1,23 @@ +{ + "title": "brother_printer: report the truth about printer state", + "objective": "./run.sh must run standalone from any cwd, and every status it reports must be traceable to something the printer or the HP PJL reference actually says. The tool previously failed to start at all, and once patched to start, reported a sleeping printer as a critical paper jam from a status table whose meanings were invented. Reading status must never hang the tool or destroy an in-flight print job, and the report must surface pages the printer silently discards - a failure mode that CUPS, the printer's error bit, and the LEDs all report as success. Success: the sleeping printer reads healthy, an open cover reads critical, consumable estimates derive from the printer's own counter, and a dropped-page run names both the gap and the workaround.", + "acceptance_criteria": [ + "./run.sh works from the package directory and from an unrelated cwd, with no PYTHONPATH or venv setup by the user", + "Every status code in the table is transcribed from the HP PJL Technical Reference Manual, Appendix D; codes absent from it report 'Unknown status - check printer display' instead of an invented meaning", + "Codes verified on the real HL-1110 decode correctly: 10001 READY, 10003 WARMING UP, 10023 PRINTING, 40000 SLEEP (severity ok, not a jam), 40021 TOP COVER OPEN (critical), 41213 NO PAPER (critical)", + "CUPS-derived and self-inferred states never borrow a PJL code to describe themselves", + "Consumable estimates use @PJL INFO PAGECOUNT; stored baselines migrate onto that scale exactly once, preserving the reported percentages across the switch", + "query_usb_pjl never blocks indefinitely: an unavailable printer produces an actionable message within seconds", + "Reading status never stops CUPS, never resets the USB device, and never runs the pyusb probe while a job is printing", + "A run after pages were dropped reports the gap between CUPS's claim and the printer's counter, and names the lp -o Resolution=300dpi workaround", + "402 tests pass with 100% statement and branch coverage; pre-commit clean with no suppressions" + ], + "out_of_scope": [ + "Changing the CUPS queue default resolution (user chose to keep 600dpi and pass -o Resolution=300dpi per job)", + "Any lpadmin, CUPS, or system configuration change", + "Identifying the original transient yellow '!' warning, which cleared before it could be read and was not reproduced", + "Verifying PJL codes that could not be induced on the hardware; those are spec-derived and marked as such", + "The network/SNMP path, which is untouched" + ], + "verifier": "./run.sh against the real Brother HL-1110 (USB) for each induced printer state; python3 -m pytest python_pkg/brother_printer/tests/ --cov=python_pkg/brother_printer (402 tests, 100% branch coverage); pre-commit run --files (ruff, ruff format, mypy, pylint, bandit, shellcheck, codespell, coverage enforcement)" +} diff --git a/docs/superpowers/evidence/brother-printer-status-accuracy-2026-07-17.json b/docs/superpowers/evidence/brother-printer-status-accuracy-2026-07-17.json new file mode 100644 index 00000000..7c705379 --- /dev/null +++ b/docs/superpowers/evidence/brother-printer-status-accuracy-2026-07-17.json @@ -0,0 +1,73 @@ +{ + "intent": "Make brother_printer report the truth: run ./run.sh standalone, stop reporting a sleeping printer as a critical paper jam, base consumable estimates on the printer's own page counter, never hang or destroy a print job while reading status, and surface pages the printer silently drops.", + "scope": [ + "python_pkg/brother_printer/run.sh, __main__.py: standalone entry point", + "python_pkg/brother_printer/constants.py: PJL status table rebuilt from the HP PJL Technical Reference Manual, Appendix D", + "python_pkg/brother_printer/usb_query.py: non-blocking device access, @PJL INFO PAGECOUNT", + "python_pkg/brother_printer/cups_service.py: non-destructive port status, state migration, dropped-page detection", + "python_pkg/brother_printer/display.py, data_classes.py: reporting", + "Non-goal: queue default stays 600dpi (user's choice); the 300dpi workaround is documented, not enforced", + "Non-goal: no lpadmin/CUPS configuration changes" + ], + "changes": [ + "run.sh: resolve REPO_ROOT and pass PYTHONPATH as a sudo argument (sudo env_reset strips an exported one), running the package via -m; fixes ModuleNotFoundError: No module named 'python_pkg'. __main__.py: correct the import prefix and guard main() behind __name__.", + "constants.py: rebuild BROTHER_STATUS_CODES from the HP PJL reference. 40000 is SLEEP MODE (the reference explicitly flags it 'not an error'), previously reported as a critical Paper Jam; the real jam code 40022 was absent entirely. 10006 is TONER LOW, previously rendered as a benign 'Processing'. Nine codes present in no reference (40310, 40300, 40302, 40309, 42000, 41000, 41001, 30038, 40201, 30201) are dropped and now fall back to 'Unknown status - check printer display' rather than asserting an invented meaning. Only 2 of the original 25 entries were correct.", + "constants.py: add a range rule for the 41xyy 'foreground paper loading' family, because the HL-1110 reports 41213 for an empty tray and its x/yy digits do not decode per HP's tray/media tables.", + "constants.py: CUPS-derived and self-inferred states no longer borrow PJL numbers. media-jam mapped to 40000, which after the rebuild would have reported a genuine jam as 'Sleep'; it now maps to 40022. Page-count estimates and unidentified errors use explicit non-PJL tags (estimate:toner-end, cups:error).", + "usb_query.py: send @PJL INFO PAGECOUNT and record it on USBResult. cups_service.py/display.py: derive consumable life from the printer's lifetime counter instead of the CUPS page log, which only sees jobs CUPS logged and counts pages the printer never printed.", + "cups_service.py: migrate replacement baselines from the CUPS scale onto the printer's scale once (schema v2), shifting by the measured gap so percentages do not jump. Zero baselines mean 'never replaced' and are left alone. reset_consumable records the printer's counter; check_brother_printer refuses to reset without it rather than writing a mis-scaled baseline.", + "usb_query.py: open the device O_NONBLOCK and deadline-bound every write. A blocking open() hung forever whenever the printer was not ready - the exact moment a status report is wanted. EBUSY/EACCES now produce actionable messages.", + "cups_service.py: _query_usb_port_status_raw no longer stops CUPS or calls dev.reset() to read a status byte; that destroyed the in-flight job and then misreported the damage as a printer error. It reads the same byte via the free usblp LPGETSTATUS ioctl, skips the probe entirely while a job is printing, and re-attaches the kernel driver so /dev/usb/lp* stops disappearing. Removed the now-orphaned _stop_cups helper.", + "cups_service.py/display.py: add check_page_delivery, comparing pages CUPS logged against the printer's counter between jobs. The HL-1110 silently discards pages whose 600dpi raster exceeds its ~1MB memory while staying READY, and CUPS reports the job successful, so this comparison is the only available signal. Reports the gap and the lp -o Resolution=300dpi workaround.", + "cups_service.py: express a CUPS-log figure on the printer's scale in fallback mode. Comparing the raw CUPS total against a printer-scale baseline went negative and clamped to zero, reporting a part-used cartridge as ~100% full." + ], + "verification": [ + { + "command": "./run.sh (real Brother HL-1110, USB)", + "result": "pass", + "evidence": "Sleeping printer now reports '. Sleep (standby) / Printer is healthy' with CODE=40000, previously '. Paper Jam (critical)'. Page count 2087 from the printer vs 1778 from the CUPS log; toner 68%, drum 80%." + }, + { + "command": "PJL status codes induced on real hardware and compared against the HP reference", + "result": "pass", + "evidence": "Six states verified, all matching the reference and none matching the old table: 10001 READY, 10003 WARMING UP, 10023 PRINTING, 40000 SLEEP, 40021 TOP COVER OPEN, 41213 NO PAPER." + }, + { + "command": "query_usb_pjl() with /dev/usb/lp0 held by another process", + "result": "pass", + "evidence": "Returned in 0.0s with '/dev/usb/lp0 is busy: another process (usually the CUPS backend mid-job) holds the printer.' Previously hung indefinitely." + }, + { + "command": "check_page_delivery replaying the observed failure", + "result": "pass", + "evidence": "Printer counter static while CUPS logged 63 pages -> cups_pages=63, printer_pages=0, dropped=63, suspected=True. Silent when queue_idle=False." + }, + { + "command": "estimate_consumable_life(0) after the scale fix", + "result": "pass", + "evidence": "Reports 68% matching the authoritative reading, instead of the ~100% the raw CUPS total produced." + }, + { + "command": "python3 -m pytest python_pkg/brother_printer/tests/ --cov=python_pkg/brother_printer", + "result": "pass", + "evidence": "402 passed; 100% statement and branch coverage across all 11 modules." + }, + { + "command": "pre-commit run --files <17 changed files>", + "result": "pass", + "evidence": "42 hooks passed/skipped, 0 failed (ruff, ruff format, mypy, pylint, bandit, shellcheck, codespell, pytest coverage enforcement)." + } + ], + "risks": [ + "The status table is transcribed from the HP reference and verified for six states on one HL-1110. Codes we could not induce are spec-derived but unverified on this device; wrong entries now degrade to an honest 'Unknown status' rather than a confident wrong answer.", + "The baseline migration assumes the pages the CUPS log never saw were printed before the last toner replacement. Unknowable in principle; the user chose this over discarding their replacement history. It runs once (schema v2) and preserved the reported toner percentage across the switch.", + "check_page_delivery could false-positive if a job finishes exactly as the check runs. Mitigated by skipping while the queue is non-empty and by a 5-page threshold; it reports a suspicion with a suggested action, never a certainty.", + "The pyusb port-status path still claims the USB interface, but only when CUPS is idle, and it re-attaches the kernel driver afterwards.", + "Fallback-mode percentages remain approximate (shifted by the last measured gap) and are labelled as such in the report." + ], + "rollback": [ + "git revert ; the package is self-contained and no CUPS or system configuration was modified.", + "state.json gains schema/last_printer_count/last_cups_total keys. Older code ignores unknown keys but would read the migrated printer-scale baselines as CUPS-scale ones, understating pages since replacement; to fully revert, restore toner_replaced_at to 1408 and delete the added keys.", + "After rollback, validate with ./run.sh: expect the total to fall back to the CUPS page-log figure and the sleeping printer to be misreported as a paper jam again." + ] +} diff --git a/python_pkg/brother_printer/__main__.py b/python_pkg/brother_printer/__main__.py index ec1bd506..b794a7af 100644 --- a/python_pkg/brother_printer/__main__.py +++ b/python_pkg/brother_printer/__main__.py @@ -1,10 +1,13 @@ """Brother laser printer status checker. Usage: - sudo python3 -m brother_printer # auto-detect - sudo python3 -m brother_printer # network/SNMP mode + sudo python3 -m python_pkg.brother_printer # auto-detect + sudo python3 -m python_pkg.brother_printer # network/SNMP mode """ -from brother_printer.check_brother_printer import main +from __future__ import annotations -main() +from python_pkg.brother_printer.check_brother_printer import main + +if __name__ == "__main__": + main() diff --git a/python_pkg/brother_printer/check_brother_printer.py b/python_pkg/brother_printer/check_brother_printer.py index 332ea7ef..ae8ad516 100644 --- a/python_pkg/brother_printer/check_brother_printer.py +++ b/python_pkg/brother_printer/check_brother_printer.py @@ -63,6 +63,31 @@ def _discover_network_printer() -> str: return "" +def _reset_consumable_from_printer(name: str) -> None: + """Reset a consumable baseline against the printer's own page counter. + + Baselines are stored on the printer's lifetime-counter scale, so recording + one from any other source (such as the CUPS page log, which undercounts) + would silently skew every later estimate. Refuse rather than guess. + + Args: + name: Consumable to reset, "toner" or "drum". + """ + result = query_usb_pjl() + printer_total = int(result.page_count) if result.page_count.isdigit() else 0 + if printer_total <= 0: + _out( + f"{RED}Could not read the printer's page counter" + f" (@PJL INFO PAGECOUNT).{RESET}", + ) + _out("The printer must be powered on and connected via USB to reset a") + _out("consumable, so the new baseline is recorded on the right scale.") + if result.error: + _out(f" {result.error}") + sys.exit(1) + reset_consumable(name, printer_total) + + def _run_network_mode(printer_ip: str) -> None: """Handle explicit network/SNMP mode.""" if not shutil.which("snmpwalk"): diff --git a/python_pkg/brother_printer/constants.py b/python_pkg/brother_printer/constants.py index bac7ba2f..3f2530a9 100644 --- a/python_pkg/brother_printer/constants.py +++ b/python_pkg/brother_printer/constants.py @@ -29,6 +29,11 @@ CUPS_PAGE_LOG_PATH = "/var/log/cups/page_log" CONSUMABLE_STATE_DIR = ".config/brother_printer" MIN_LPSTAT_JOB_PARTS = 4 +# How many pages CUPS may claim over the printer's own counter before we call +# it dropping pages. A small gap is just timing (a job finishing as we look); +# a large one means the printer is discarding work and saying nothing. +PAGE_DROP_WARN_THRESHOLD = 5 + BROTHER_USB_VENDOR_ID = 0x04F9 @@ -44,127 +49,183 @@ def _prompt(text: str) -> str: return sys.stdin.readline().strip() -# ── Brother PJL status codes ──────────────────────────────────────── -# Documented in Brother PJL Technical Reference. +# ── PJL status codes ───────────────────────────────────────────────── +# Transcribed from the HP PJL Technical Reference Manual, Appendix D +# ("PJL Status Codes"). Brother publishes no PJL status table of its own, +# but an HL-1110 was probed on real hardware (2026-07-17) and matched the HP +# codes for every state that could be induced: +# +# 10001 READY 10003 WARMING UP 10023 PRINTING +# 40000 SLEEP 40021 TOP COVER OPEN 41213 NO PAPER +# +# Only codes documented in that reference appear below. Anything else is +# reported as unknown by get_status_info() rather than guessed at: a previous +# revision of this table invented meanings (40000 "Paper Jam", 30010 "Toner +# Low", 41000 "Cover Open" ...) and consequently reported a sleeping printer +# as a critical paper jam, while a genuine toner-low warning (10006) was +# rendered as a benign "Processing". +# # Format: code -> (severity, short_text, action) # Severities: ok, info, warn, critical +_ORDER_TONER = "Order replacement toner cartridge (TN-1050/TN-1030 compatible)." + BROTHER_STATUS_CODES: dict[int, tuple[str, str, str]] = { + # Informational (10xxx) 10001: ("ok", "Ready", ""), - 10002: ("ok", "Sleep", ""), - 10003: ("info", "Self-test / Calibrating", ""), - 10004: ("ok", "Warming up", ""), - 10005: ("ok", "Cooling down", ""), - 10006: ("info", "Processing", ""), - 10007: ("info", "Printing", ""), - 10014: ("ok", "Cancelling", ""), - 10023: ("info", "Waiting", ""), - # Toner - 30010: ( - "warn", - "Toner Low", - "Order replacement toner cartridge (TN-1050/TN-1030 compatible).", - ), - 30038: ( - "warn", - "Toner Low", - "Order replacement toner cartridge (TN-1050/TN-1030 compatible).", - ), - 40038: ( - "warn", - "Toner Low", - "Order replacement toner cartridge (TN-1050/TN-1030 compatible).", - ), - 40309: ( + 10002: ("info", "Ready (offline)", "Press the power button to bring it online."), + 10003: ("ok", "Warming up", ""), + 10004: ("info", "Self test", ""), + 10005: ("info", "Resetting / clearing memory", ""), + 10006: ("warn", "Toner Low", _ORDER_TONER), + 10007: ("info", "Canceling job", ""), + 10014: ("info", "Printing", ""), + 10023: ("info", "Printing / processing job", ""), + # Auto-continuable conditions (30xxx) + 30010: ("info", "Status buffer overflow", ""), + # Potential operator intervention (35xxx) + 35078: ("ok", "Power save", ""), + # Operator intervention required (40xxx). 40000 is flagged in the + # reference as "not an error - the printer is waiting for data". + 40000: ("ok", "Sleep (standby)", ""), + 40010: ( "critical", - "Replace Toner", - "The toner cartridge needs immediate replacement (TN-1050/TN-1030 compatible).", - ), - 40310: ( - "critical", - "Toner End", - "The toner cartridge is empty. Replace now (TN-1050/TN-1030 compatible).", - ), - # Drum - 30201: ( - "warn", - "Drum End Soon", - "The drum unit is nearing end of life. Order replacement (DR-1050 compatible).", - ), - 40201: ( - "warn", - "Drum End Soon", - "The drum unit is nearing end of life. Order replacement (DR-1050 compatible).", + "No Toner Cartridge", + "Install a toner cartridge (TN-1050/TN-1030 compatible).", ), 40019: ( - "critical", - "Replace Drum", - "The drum unit must be replaced (DR-1050 compatible).", + "warn", + "Remove Paper From Output Bin", + "Take the printed pages out of the output bin.", ), - 40020: ( + 40021: ( "critical", - "Drum Stop", - "The drum unit must be replaced immediately (DR-1050 compatible).", + "Cover Open", + "Close the printer cover and check the toner cartridge is seated.", ), - # Paper / feed - 40000: ("critical", "Paper Jam", "Clear the paper jam and close all covers."), - 40300: ( + 40022: ("critical", "Paper Jam", "Clear the paper jam and close all covers."), + 40026: ("critical", "Install Paper Tray", "Insert the paper tray."), + 40038: ("warn", "Toner Low", _ORDER_TONER), + 40050: ( "critical", - "No Paper / Tray Open", - "Load paper or close the paper tray.", + "Fuser Error (50 SERVICE)", + "Power-cycle the printer. If the error persists, contact service.", ), - 40302: ("critical", "No Paper", "Load paper into the paper tray."), - 40016: ("warn", "Paper Feed Error", "Check paper tray and re-seat paper."), - # Cover - 41000: ("critical", "Cover Open", "Close the top cover of the printer."), - 41001: ("critical", "Cover Open", "Close the front cover of the printer."), - # Others - 35078: ("info", "Manual Feed", "Load paper in the manual feed slot."), - 42000: ( - "critical", - "Machine Error", - "Power-cycle the printer. If error persists, contact service.", + 40079: ( + "warn", + "Offline", + "The printer is offline. Press the power button to bring it online.", ), } +# Code families that encode a sub-status in their trailing digits. The +# HL-1110 reports 41213 for an empty tray, which places it in the reference's +# "Foreground Paper Loading (41xyy)" family - but its x/yy digits do not +# decode per the HP tray/media tables (x=2 "upper cassette", yy=13 "Japan B5" +# on a single-tray A4 printer). So match the family and ignore the digits. +# Each entry is an inclusive low/high bound plus the usual status triple. + +_STATUS_CODE_RANGES: tuple[tuple[int, int, tuple[str, str, str]], ...] = ( + ( + 41000, + 41999, + ("critical", "Out of Paper", "Load paper into the paper tray."), + ), + ( + 50000, + 50999, + ( + "critical", + "Hardware Error", + "Power-cycle the printer. If the error persists, contact service.", + ), + ), +) + # ── CUPS status code mappings ──────────────────────────────────────── +# These translate what CUPS reports into the PJL codes above, for the +# fallback path used when the usblp device is unavailable and the printer +# cannot be asked directly. Every value must name a real code from +# BROTHER_STATUS_CODES (or a range family), otherwise the fallback silently +# degrades to "unknown status". + _CUPS_REASONS_TO_STATUS: dict[str, int] = { - "paused": 10023, - "moving-to-paused": 10023, - "toner-low": 30010, - "toner-empty": 40310, - "marker-supply-low": 30010, - "marker-supply-empty": 40310, - "media-empty": 40302, - "media-needed": 40302, - "media-jam": 40000, - "cover-open": 41000, - "door-open": 41000, - "input-tray-missing": 40300, + "paused": 10002, + "moving-to-paused": 10002, + "toner-low": 40038, + "toner-empty": 40010, + "marker-supply-low": 40038, + "marker-supply-empty": 40010, + "media-empty": 41000, + "media-needed": 41000, + "media-jam": 40022, + "cover-open": 40021, + "door-open": 40021, + "input-tray-missing": 40026, } _CUPS_STATE_TO_STATUS: dict[str, int] = { "idle": 10001, - "processing": 10007, - "stopped": 10023, + "processing": 10023, + "stopped": 40079, } _ERROR_REASON_MAP: tuple[tuple[tuple[str, ...], str, str], ...] = ( - (("media-jam",), "40000", "Paper Jam"), - (("cover-open", "door-open"), "41000", "Cover Open"), - (("toner-empty",), "40310", "Toner End"), - (("toner-low",), "30010", "Toner Low"), + (("media-jam",), "40022", "Paper Jam"), + (("cover-open", "door-open"), "40021", "Cover Open"), + (("toner-empty",), "40010", "No Toner Cartridge"), + (("toner-low",), "40038", "Toner Low"), ) +# States this tool infers itself, rather than reads off the printer: from the +# CUPS page-count estimate, or from an error bit whose cause we cannot name. +# They deliberately are not PJL numbers - borrowing a PJL code for a state the +# printer never reported is how the old table came to claim a sleeping printer +# had jammed. + +DERIVED_TONER_END = "estimate:toner-end" +DERIVED_TONER_LOW = "estimate:toner-low" +DERIVED_CUPS_ERROR = "cups:error" + +_DERIVED_STATUS_CODES: dict[str, tuple[str, str, str]] = { + DERIVED_TONER_END: ( + "critical", + "Toner End (estimated from page count)", + _ORDER_TONER, + ), + DERIVED_TONER_LOW: ("warn", "Toner Low (estimated from page count)", _ORDER_TONER), + DERIVED_CUPS_ERROR: ( + "warn", + "Printer reports an error", + "Check the printer display for details.", + ), +} + + def get_status_info(code: str) -> tuple[str, str, str]: """Look up a PJL status code. Returns (severity, text, action).""" + derived = _DERIVED_STATUS_CODES.get(code) + if derived is not None: + return derived try: - return BROTHER_STATUS_CODES[int(code)] - except (KeyError, ValueError): - return ( - "info", - f"Unknown status (code {code})", - "Check printer display for details.", - ) + numeric = int(code) + except (TypeError, ValueError): + return _unknown_status(code) + exact = BROTHER_STATUS_CODES.get(numeric) + if exact is not None: + return exact + for low, high, info in _STATUS_CODE_RANGES: + if low <= numeric <= high: + return info + return _unknown_status(code) + + +def _unknown_status(code: str) -> tuple[str, str, str]: + """Describe a code that is not in the PJL reference. Never guesses.""" + return ( + "info", + f"Unknown status (code {code})", + "Check printer display for details.", + ) diff --git a/python_pkg/brother_printer/cups_service.py b/python_pkg/brother_printer/cups_service.py index 13ca26e1..68ae5efd 100644 --- a/python_pkg/brother_printer/cups_service.py +++ b/python_pkg/brother_printer/cups_service.py @@ -2,9 +2,12 @@ from __future__ import annotations +import contextlib +import fcntl import importlib import json import logging +import os from pathlib import Path import re import shutil @@ -23,14 +26,19 @@ from python_pkg.brother_printer.constants import ( BROTHER_USB_VENDOR_ID, CONSUMABLE_STATE_DIR, CUPS_PAGE_LOG_PATH, + DERIVED_CUPS_ERROR, + DERIVED_TONER_END, + DERIVED_TONER_LOW, DRUM_RATED_PAGES, GREEN, + PAGE_DROP_WARN_THRESHOLD, RESET, TONER_RATED_PAGES, _out, ) from python_pkg.brother_printer.data_classes import ( PageCountEstimate, + PageDeliveryCheck, USBPortStatus, USBResult, ) @@ -43,6 +51,17 @@ logger = logging.getLogger(__name__) CUPS_PAGE_LOG = Path(CUPS_PAGE_LOG_PATH) CONSUMABLE_STATE_FILE = Path.home() / CONSUMABLE_STATE_DIR / "state.json" +# state.json schema versions. Version 1 (implicit, no "schema" key) recorded +# replacement baselines as CUPS page-log counts; version 2 records them on the +# printer's own lifetime counter. See _migrate_state_to_printer_scale. +STATE_SCHEMA_CUPS_SCALE = 1 +STATE_SCHEMA_PRINTER_SCALE = 2 + +# Directory holding the usblp device nodes, and the ioctl that reads the +# printer's IEEE 1284 status byte from one without disturbing the device. +USB_PRINTER_DEV_GLOB_DIR = "/dev/usb" +LPGETSTATUS = 0x060B + def _import_or_raise(name: str) -> types.ModuleType: """Import a module or raise ImportError with a helpful message.""" @@ -75,23 +94,6 @@ def _get_pyusb_device_info() -> dict[str, str]: # ── CUPS service control ──────────────────────────────────────────── -def _stop_cups() -> bool: - """Stop CUPS service and sockets. Returns True on success.""" - systemctl = shutil.which("systemctl") - if not systemctl: - return False - try: - subprocess.run( - [systemctl, "stop", "cups.service", "cups.socket", "cups.path"], - timeout=15, - check=True, - ) - time.sleep(2) - except (subprocess.TimeoutExpired, subprocess.CalledProcessError, OSError): - return False - return True - - def is_cups_scheduler_running() -> bool: """Check if the CUPS scheduler is currently running.""" lpstat = shutil.which("lpstat") @@ -142,12 +144,75 @@ def _ensure_cups_running() -> bool: # ── USB port status via pyusb ──────────────────────────────────────── -def _query_usb_port_status_raw() -> USBPortStatus | None: - """Query USB printer port status via pyusb control transfer. +def _port_status_from_byte(port_byte: int) -> USBPortStatus: + """Decode a USB printer-class port status byte.""" + return USBPortStatus( + paper_empty=bool(port_byte & 0x20), + online=bool(port_byte & 0x10), + # nFault is active low: the bit is set when there is NO fault. + error=not bool(port_byte & 0x08), + raw_byte=port_byte, + ) - Requires root and temporarily stops CUPS to access the USB device. - Returns None if the query fails. + +def _port_status_via_usblp() -> USBPortStatus | None: + """Read port status through the usblp device node. No side effects. + + The LPGETSTATUS ioctl returns the same byte as the USB control transfer + but costs nothing: no stopping CUPS, no device reset, no driver detach. + Returns None when the node is absent or busy. """ + dev_path = Path(USB_PRINTER_DEV_GLOB_DIR) + devices = sorted(dev_path.glob("lp*")) if dev_path.is_dir() else [] + if not devices: + return None + try: + # O_NONBLOCK: never hang here, the printer may be mid-job. + fd = os.open(str(devices[0]), os.O_RDONLY | os.O_NONBLOCK) + except OSError: + logger.debug("opening %s failed", devices[0], exc_info=True) + return None + try: + buf = bytearray(4) + fcntl.ioctl(fd, LPGETSTATUS, buf) + except OSError: + logger.debug("usblp LPGETSTATUS failed", exc_info=True) + return None + finally: + os.close(fd) + return _port_status_from_byte(int.from_bytes(buf, "little") & 0xFF) + + +def _cups_is_busy(cups_state: str) -> bool: + """Report whether CUPS is currently driving the printer.""" + return "processing" in cups_state.lower() or "printing" in cups_state.lower() + + +def _query_usb_port_status_raw(cups_state: str = "") -> USBPortStatus | None: + """Query the printer's port status without disturbing it. + + Prefers the usblp ioctl, which is free. Falls back to a pyusb control + transfer only when CUPS is idle: claiming the USB interface mid-job kills + the print, and an earlier version of this function did exactly that - it + stopped CUPS and reset the device just to read a status byte, destroying + the job it was trying to report on. + + Args: + cups_state: CUPS printer-state text, used to detect an active job. + + Returns: + The port status, or None when it cannot be read harmlessly. + """ + status = _port_status_via_usblp() + if status is not None: + return status + + if _cups_is_busy(cups_state): + # A job is running and owns the device. Reading the status is not worth + # killing the job for; the caller falls back to what CUPS reports. + logger.debug("skipping USB probe: CUPS is printing") + return None + try: usb_core = _import_or_raise("usb.core") usb_util = _import_or_raise("usb.util") @@ -158,19 +223,12 @@ def _query_usb_port_status_raw() -> USBPortStatus | None: if dev is None: return None - if not _stop_cups(): - return None - + detached = False try: - dev.reset() - time.sleep(2) - dev = usb_core.find(idVendor=BROTHER_USB_VENDOR_ID) - if dev is None: - return None - try: if dev.is_kernel_driver_active(0): dev.detach_kernel_driver(0) + detached = True except (usb_core.USBError, NotImplementedError): pass @@ -178,21 +236,18 @@ def _query_usb_port_status_raw() -> USBPortStatus | None: try: # USB Printer Class GET_PORT_STATUS (bRequest=0x01) raw = dev.ctrl_transfer(0xA1, 0x01, 0, 0, 1, timeout=5000) - port_byte = raw[0] - return USBPortStatus( - paper_empty=bool(port_byte & 0x20), - online=bool(port_byte & 0x10), - error=not bool(port_byte & 0x08), - raw_byte=port_byte, - ) + return _port_status_from_byte(raw[0]) finally: usb_util.release_interface(dev, 0) + if detached: + # Give usblp its device back, or /dev/usb/lp* stays missing and + # every later run is stuck in this fallback path. + with contextlib.suppress(Exception): + dev.attach_kernel_driver(0) usb_util.dispose_resources(dev) except (OSError, ValueError): logger.debug("USB port status query failed", exc_info=True) return None - finally: - start_cups() # ── Consumable state management ────────────────────────────────────── @@ -218,7 +273,16 @@ def _get_cups_total_pages() -> int: def _load_consumable_state() -> dict[str, int]: """Load consumable replacement state from disk.""" - defaults: dict[str, int] = {"toner_replaced_at": 0, "drum_replaced_at": 0} + defaults: dict[str, int] = { + "toner_replaced_at": 0, + "drum_replaced_at": 0, + "schema": STATE_SCHEMA_CUPS_SCALE, + # Snapshot of both counters at the last successful printer read, used to + # spot dropped pages and to place a CUPS-log figure on the printer's + # scale when the printer cannot be reached. + "last_printer_count": 0, + "last_cups_total": 0, + } if not CONSUMABLE_STATE_FILE.exists(): return defaults try: @@ -228,11 +292,50 @@ def _load_consumable_state() -> dict[str, int]: return { "toner_replaced_at": int(data.get("toner_replaced_at", 0)), "drum_replaced_at": int(data.get("drum_replaced_at", 0)), + "schema": int(data.get("schema", STATE_SCHEMA_CUPS_SCALE)), + "last_printer_count": int(data.get("last_printer_count", 0)), + "last_cups_total": int(data.get("last_cups_total", 0)), } except (OSError, json.JSONDecodeError, ValueError, TypeError): return defaults +def _migrate_state_to_printer_scale( + state: dict[str, int], + printer_total: int, +) -> dict[str, int]: + """Rebase replacement baselines from the CUPS page log onto the printer's counter. + + Baselines written before this migration counted pages the CUPS log had seen, + which undercounts the printer's own lifetime counter by however many pages + were printed without CUPS logging them. Shift each baseline by the gap + measured now, so "pages since replacement" - and therefore the reported + percentages - stay put across the switch. + + A zero baseline means "never replaced" rather than "replaced at page zero", + so it is left alone: on the printer's scale it already means "as old as the + printer", which is the truthful reading. + + Args: + state: Loaded state, possibly still on the CUPS page-log scale. + printer_total: Lifetime page count from the printer's own counter. + + Returns: + State on the printer's scale, migrated at most once. + """ + if state.get("schema", STATE_SCHEMA_CUPS_SCALE) >= STATE_SCHEMA_PRINTER_SCALE: + return state + offset = printer_total - _get_cups_total_pages() + migrated = dict(state) + migrated["schema"] = STATE_SCHEMA_PRINTER_SCALE + if offset > 0: + for key in ("toner_replaced_at", "drum_replaced_at"): + if state[key] > 0: + migrated[key] = state[key] + offset + _save_consumable_state(migrated) + return migrated + + def _save_consumable_state(state: dict[str, int]) -> None: """Persist consumable replacement state to disk.""" CONSUMABLE_STATE_FILE.parent.mkdir(parents=True, exist_ok=True) @@ -242,23 +345,131 @@ def _save_consumable_state(state: dict[str, int]) -> None: ) -def reset_consumable(name: str) -> None: - """Record current page count as replacement point for a consumable.""" - total = _get_cups_total_pages() +def reset_consumable(name: str, printer_total: int = 0) -> None: + """Record current page count as replacement point for a consumable. + + Args: + name: Consumable to reset, "toner" or "drum". + printer_total: Lifetime count from the printer's own counter. Falls back + to the CUPS page log when zero, i.e. the printer was unreachable. + """ + total = printer_total if printer_total > 0 else _get_cups_total_pages() state = _load_consumable_state() key = f"{name}_replaced_at" state[key] = total + # Baselines written here are already on the printer's scale, so mark the + # state migrated to stop the rebase shifting them a second time. + if printer_total > 0: + state["schema"] = STATE_SCHEMA_PRINTER_SCALE _save_consumable_state(state) _out(f"{GREEN}✓ {name.capitalize()} counter reset at page count {total}.{RESET}") _out(f" State saved to {CONSUMABLE_STATE_FILE}") -def estimate_consumable_life() -> PageCountEstimate: - """Estimate toner/drum life from CUPS page count since last replacement.""" - total = _get_cups_total_pages() +def _cups_total_on_printer_scale(state: dict[str, int]) -> int: + """Express the CUPS page-log count on the printer's counter scale. + + Replacement baselines are stored against the printer's lifetime counter, so + comparing a raw CUPS figure against them subtracts two different scales. The + CUPS log runs behind the printer (it only sees jobs it printed itself), so + the raw number reads too low - low enough to go negative and clamp to zero, + which reported a spent cartridge as 100% full. + + Shift by the gap measured at the last successful printer read. It is an + approximation, which is why callers flag the estimate as approximate. + + Args: + state: Loaded consumable state holding the last counter snapshot. + + Returns: + The CUPS total shifted onto the printer's scale, or the raw total when + no snapshot exists to shift it by. + """ + cups_total = _get_cups_total_pages() + if cups_total <= 0: + return 0 + offset = state["last_printer_count"] - state["last_cups_total"] + if state["last_printer_count"] <= 0 or offset <= 0: + return cups_total + return cups_total + offset + + +def check_page_delivery(printer_total: int, *, queue_idle: bool) -> PageDeliveryCheck: + """Compare pages CUPS logged against pages the printer actually counted. + + Only meaningful between jobs: mid-job, CUPS has logged pages the printer has + not yet pulled off the wire, which would look identical to dropping them. + Records a fresh snapshot of both counters whenever it runs cleanly. + + Args: + printer_total: Lifetime count from the printer's own counter. + queue_idle: False when a job is queued or printing, which makes any + comparison meaningless. + + Returns: + The comparison. suspected is True only when CUPS claims materially more + pages than the printer recorded. + """ + check = PageDeliveryCheck() + if printer_total <= 0 or not queue_idle: + return check + state = _load_consumable_state() + cups_total = _get_cups_total_pages() + last_printer = state["last_printer_count"] + last_cups = state["last_cups_total"] + + _snapshot_counters(state, printer_total, cups_total) + + if last_printer <= 0 or last_cups <= 0: + # No baseline yet: this run establishes one. + return check + printer_delta = printer_total - last_printer + cups_delta = cups_total - last_cups + if printer_delta < 0 or cups_delta < 0: + # Counter reset or the page log rotated; nothing to conclude. + return check + check.cups_pages = cups_delta + check.printer_pages = printer_delta + check.dropped = cups_delta - printer_delta + check.suspected = check.dropped >= PAGE_DROP_WARN_THRESHOLD + return check + + +def _snapshot_counters( + state: dict[str, int], + printer_total: int, + cups_total: int, +) -> None: + """Persist where both counters stood, for the next run to compare against.""" + if ( + state["last_printer_count"] == printer_total + and state["last_cups_total"] == cups_total + ): + return + updated = dict(state) + updated["last_printer_count"] = printer_total + updated["last_cups_total"] = cups_total + _save_consumable_state(updated) + + +def estimate_consumable_life(printer_total: int = 0) -> PageCountEstimate: + """Estimate toner/drum life from pages printed since the last replacement. + + Args: + printer_total: Lifetime count from @PJL INFO PAGECOUNT. When zero the + printer could not be asked, so the CUPS page log stands in and the + estimate is flagged approximate. + + Returns: + The estimate; total_pages is zero when no counter could be read at all. + """ + approximate = printer_total <= 0 + state = _load_consumable_state() + total = _cups_total_on_printer_scale(state) if approximate else printer_total if total <= 0: return PageCountEstimate() - state = _load_consumable_state() + if not approximate: + state = _migrate_state_to_printer_scale(state, total) toner_pages = max(0, total - state["toner_replaced_at"]) drum_pages = max(0, total - state["drum_replaced_at"]) toner_pct = max(0, 100 - (toner_pages * 100 // TONER_RATED_PAGES)) @@ -269,6 +480,7 @@ def estimate_consumable_life() -> PageCountEstimate: drum_pages=drum_pages, toner_pct_remaining=toner_pct, drum_pct_remaining=drum_pct, + approximate=approximate, toner_exhausted=toner_pages >= TONER_RATED_PAGES, toner_low=toner_pages >= TONER_RATED_PAGES * 80 // 100, drum_near_end=drum_pages >= DRUM_RATED_PAGES * 90 // 100, @@ -339,7 +551,12 @@ def _cups_reasons_to_error(cups_reasons: str) -> tuple[str, str]: for keywords, code, display in _ERROR_REASON_MAP: if any(kw in reasons_lower for kw in keywords): return code, display - return "42000", "Printer Error" + # Nothing recognised. Show what CUPS actually said rather than a bare + # "Printer Error", which tells the reader nothing they can act on. + detail = cups_reasons.strip() + if detail and detail.lower() not in {"none", ""}: + return DERIVED_CUPS_ERROR, f"Printer Error (CUPS reports: {detail})" + return DERIVED_CUPS_ERROR, "Printer Error (CUPS gave no reason)" def _port_status_to_status_code( @@ -347,14 +564,13 @@ def _port_status_to_status_code( cups_reasons: str, ) -> tuple[str, str]: """Map USB port status + CUPS reasons to (status_code, display).""" - if ps.error and ps.paper_empty: - return "40302", "No Paper" - if ps.error and not ps.online: - return "41000", "Cover Open" + # The port status exposes only paper_empty/error/online bits, so anything + # more specific than "out of paper" has to come from the CUPS reasons - + # an error bit alone does not tell us the cover is open. + if ps.paper_empty: + return "41000", "No Paper" if ps.error: return _cups_reasons_to_error(cups_reasons) - if ps.paper_empty: - return "40302", "No Paper" if not ps.online: return "10002", "Offline / Sleep" return "", "" @@ -408,7 +624,7 @@ def query_usb_via_cups() -> USBResult: reasons = ipp.get("printer-state-reasons", "none") result.economode = _get_cups_economode(printer_name) - port_status = _query_usb_port_status_raw() + port_status = _query_usb_port_status_raw(state) if port_status is not None: result.port_status = port_status hw_code, hw_display = _port_status_to_status_code( @@ -422,12 +638,12 @@ def query_usb_via_cups() -> USBResult: return result estimate = estimate_consumable_life() if estimate.toner_exhausted: - result.status_code = "40310" + result.status_code = DERIVED_TONER_END result.display = "Toner End (estimated from page count)" result.online = "TRUE" return result if estimate.toner_low: - result.status_code = "30010" + result.status_code = DERIVED_TONER_LOW result.display = "Toner Low (estimated from page count)" result.online = "TRUE" return result diff --git a/python_pkg/brother_printer/data_classes.py b/python_pkg/brother_printer/data_classes.py index 0ed25a8a..516d4388 100644 --- a/python_pkg/brother_printer/data_classes.py +++ b/python_pkg/brother_printer/data_classes.py @@ -29,7 +29,7 @@ class CUPSQueueStatus: @dataclass class PageCountEstimate: - """Estimated consumable life based on CUPS page count.""" + """Estimated consumable life based on the printer's lifetime page count.""" total_pages: int = 0 toner_pages: int = 0 @@ -39,6 +39,27 @@ class PageCountEstimate: toner_exhausted: bool = False toner_low: bool = False drum_near_end: bool = False + # True when total_pages came from the CUPS page log instead of the + # printer's own counter. The log only sees jobs CUPS itself printed, so + # it undercounts and the percentages read optimistically high. + approximate: bool = False + + +@dataclass +class PageDeliveryCheck: + """Pages CUPS claims it sent versus pages the printer's counter recorded. + + CUPS reports a job as successful once the data leaves the machine, so a page + the printer silently drops still shows up as "printed". The HL-1110 does + exactly that when a page's 600 dpi raster overflows its ~1 MB of memory: it + discards the page, stays READY, and reports no error anywhere. Comparing the + two counters is the only way to notice. + """ + + cups_pages: int = 0 + printer_pages: int = 0 + dropped: int = 0 + suspected: bool = False @dataclass @@ -64,6 +85,10 @@ class USBResult: online: str = "" economode: str = "" error: str = "" + # Lifetime page count from @PJL INFO PAGECOUNT: the printer's own counter, + # and the authoritative one. Empty when the printer could not be asked + # (CUPS fallback mode), which forces the estimate onto the CUPS page log. + page_count: str = "" port_status: USBPortStatus | None = None diff --git a/python_pkg/brother_printer/display.py b/python_pkg/brother_printer/display.py index 2ee3ec14..20dd2edd 100644 --- a/python_pkg/brother_printer/display.py +++ b/python_pkg/brother_printer/display.py @@ -26,7 +26,10 @@ from python_pkg.brother_printer.cups_queue import ( display_cups_queue_status, get_cups_queue_status, ) -from python_pkg.brother_printer.cups_service import estimate_consumable_life +from python_pkg.brother_printer.cups_service import ( + check_page_delivery, + estimate_consumable_life, +) from python_pkg.brother_printer.data_classes import ( NetworkResult, SupplyStatus, @@ -45,9 +48,43 @@ def _display_report_header() -> None: _out() -def _display_page_count_estimate() -> None: - """Show estimated consumable life based on CUPS page count.""" - estimate = estimate_consumable_life() +def _display_page_delivery_warning(printer_total: int, *, queue_idle: bool) -> None: + """Warn when CUPS claims more pages than the printer actually counted. + + Args: + printer_total: Lifetime count from the printer's own counter. + queue_idle: False while a job is queued or printing, when the two + counters legitimately disagree. + """ + check = check_page_delivery(printer_total, queue_idle=queue_idle) + if not check.suspected: + return + _out(f"{BOLD}── Dropped Pages ──{RESET}") + _out() + _out( + f" {RED}{BOLD}⚠ {check.dropped} pages did not print.{RESET}" + f" {RED}CUPS sent {check.cups_pages} pages since the last check;" + f" the printer's own counter advanced by" + f" {check.printer_pages}.{RESET}" + ) + _out() + _out( + f" {DIM}The printer discards pages whose 600 dpi raster does not fit" + f" its memory, stays READY, and reports no error - and CUPS still calls" + f" the job successful. Reprint at a lower resolution:{RESET}" + ) + _out(f" {DIM} lp -o Resolution=300dpi {RESET}") + _out() + + +def _display_page_count_estimate(printer_total: int = 0) -> None: + """Show estimated consumable life based on the printer's page count. + + Args: + printer_total: Lifetime count from the printer's own counter, or zero + when it could not be read and the CUPS page log has to stand in. + """ + estimate = estimate_consumable_life(printer_total) if estimate.total_pages <= 0: return _out(f"{BOLD}── Page Count Estimate ──{RESET}") @@ -57,6 +94,12 @@ def _display_page_count_estimate() -> None: f" (toner: {estimate.toner_pages} since replacement," f" drum: {estimate.drum_pages} since replacement)" ) + if estimate.approximate: + _out( + f" {YELLOW}Approximate: counted from the CUPS log, not the" + f" printer's own counter, so it misses pages CUPS never saw" + f" and reads high.{RESET}" + ) _out() # Toner bar toner_pct = estimate.toner_pct_remaining @@ -236,10 +279,12 @@ def display_usb_results(result: USBResult) -> None: _display_cups_fallback_note(result) _out() - _display_page_count_estimate() + printer_total = int(result.page_count) if result.page_count.isdigit() else 0 + queue = get_cups_queue_status() + _display_page_delivery_warning(printer_total, queue_idle=not queue.jobs) + _display_page_count_estimate(printer_total) _display_consumables_reference() - queue = get_cups_queue_status() display_cups_queue_status(queue) diff --git a/python_pkg/brother_printer/run.sh b/python_pkg/brother_printer/run.sh index af452476..38edc57b 100755 --- a/python_pkg/brother_printer/run.sh +++ b/python_pkg/brother_printer/run.sh @@ -1,13 +1,19 @@ #!/usr/bin/env bash -set -e +set -euo pipefail SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +REPO_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)" # No pip dependencies — script only uses stdlib (+pyusb for fallback info) # Requires root for USB hardware queries and CUPS management +# +# Imports are absolute (python_pkg.brother_printer.*), so the monorepo root must +# be importable. sudo's env_reset drops an exported PYTHONPATH, hence passing it +# as a sudo VAR=value argument instead. +# # Usage: ./run.sh # auto-detect # ./run.sh # network/SNMP mode if [[ $EUID -ne 0 ]]; then - exec sudo python3 "$SCRIPT_DIR/check_brother_printer.py" "$@" + exec sudo PYTHONPATH="$REPO_ROOT" python3 -m python_pkg.brother_printer "$@" fi -exec python3 "$SCRIPT_DIR/check_brother_printer.py" "$@" +exec env PYTHONPATH="$REPO_ROOT" python3 -m python_pkg.brother_printer "$@" diff --git a/python_pkg/brother_printer/tests/test_check_brother_printer.py b/python_pkg/brother_printer/tests/test_check_brother_printer.py index b00daacc..284d56d7 100644 --- a/python_pkg/brother_printer/tests/test_check_brother_printer.py +++ b/python_pkg/brother_printer/tests/test_check_brother_printer.py @@ -11,6 +11,7 @@ import pytest from python_pkg.brother_printer.check_brother_printer import ( _discover_network_printer, _no_printer_found, + _reset_consumable_from_printer, _run_network_mode, _run_usb_mode, main, @@ -209,3 +210,66 @@ class TestMain: pytest.raises(SystemExit), ): main(["1.2.3.4"]) + + +class TestResetConsumableFromPrinter: + """Resetting a baseline must use the printer's own counter, or refuse. + + Baselines live on the printer's lifetime-counter scale. Recording one from + any other source skews every later estimate, so a printer that cannot be + read is a hard error rather than a silent fallback. + """ + + @patch(f"{MOD}.reset_consumable") + @patch(f"{MOD}.query_usb_pjl") + def test_uses_printer_page_count( + self, + mock_query: MagicMock, + mock_reset: MagicMock, + ) -> None: + mock_query.return_value = USBResult(page_count="2017") + _reset_consumable_from_printer("toner") + mock_reset.assert_called_once_with("toner", 2017) + + @patch(f"{MOD}._out") + @patch(f"{MOD}.reset_consumable") + @patch(f"{MOD}.query_usb_pjl") + def test_refuses_without_page_count( + self, + mock_query: MagicMock, + mock_reset: MagicMock, + mock_out: MagicMock, + ) -> None: + mock_query.return_value = USBResult(page_count="") + with pytest.raises(SystemExit): + _reset_consumable_from_printer("toner") + mock_reset.assert_not_called() + + @patch(f"{MOD}._out") + @patch(f"{MOD}.reset_consumable") + @patch(f"{MOD}.query_usb_pjl") + def test_reports_underlying_error( + self, + mock_query: MagicMock, + mock_reset: MagicMock, + mock_out: MagicMock, + ) -> None: + mock_query.return_value = USBResult(page_count="", error="device busy") + with pytest.raises(SystemExit): + _reset_consumable_from_printer("drum") + printed = " ".join(str(c) for c in mock_out.call_args_list) + assert "device busy" in printed + + @patch(f"{MOD}._out") + @patch(f"{MOD}.reset_consumable") + @patch(f"{MOD}.query_usb_pjl") + def test_non_numeric_page_count_refused( + self, + mock_query: MagicMock, + mock_reset: MagicMock, + mock_out: MagicMock, + ) -> None: + mock_query.return_value = USBResult(page_count="???") + with pytest.raises(SystemExit): + _reset_consumable_from_printer("toner") + mock_reset.assert_not_called() diff --git a/python_pkg/brother_printer/tests/test_constants.py b/python_pkg/brother_printer/tests/test_constants.py index f997e2b3..bc6df812 100644 --- a/python_pkg/brother_printer/tests/test_constants.py +++ b/python_pkg/brother_printer/tests/test_constants.py @@ -10,6 +10,7 @@ from python_pkg.brother_printer.constants import ( BROTHER_STATUS_CODES, BROTHER_USB_VENDOR_ID, CYAN, + DERIVED_TONER_END, DIM, DRUM_RATED_PAGES, GREEN, @@ -96,10 +97,62 @@ class TestGetStatusInfo: assert action == "" def test_toner_low(self) -> None: - severity, text, _ = get_status_info("30010") + severity, text, _ = get_status_info("40038") assert severity == "warn" assert "Toner Low" in text + def test_toner_low_informational_variant(self) -> None: + """10006 is TONER LOW in the PJL reference, not a benign 'Processing'.""" + severity, text, _ = get_status_info("10006") + assert severity == "warn" + assert "Toner Low" in text + + def test_sleep_is_not_an_error(self) -> None: + """40000 is SLEEP MODE; the reference flags it as explicitly not an error. + + Verified on a real HL-1110, which reports CODE=40000 DISPLAY="SLEEP". + An earlier table called this a critical Paper Jam. + """ + severity, text, action = get_status_info("40000") + assert severity == "ok" + assert "Sleep" in text + assert action == "" + + def test_paper_jam_is_40022(self) -> None: + severity, text, action = get_status_info("40022") + assert severity == "critical" + assert text == "Paper Jam" + assert action != "" + + def test_cover_open_is_40021(self) -> None: + """Verified on a real HL-1110: CODE=40021 DISPLAY="TOP COVER OPEN".""" + severity, text, _ = get_status_info("40021") + assert severity == "critical" + assert text == "Cover Open" + + def test_out_of_paper_matches_41_family(self) -> None: + """Verified on a real HL-1110: CODE=41213 DISPLAY="NO PAPER". + + The 41xyy sub-digits do not decode per the HP tray/media tables on this + printer, so the whole family maps to one meaning. + """ + severity, text, _ = get_status_info("41213") + assert severity == "critical" + assert text == "Out of Paper" + + def test_derived_codes_are_not_pjl(self) -> None: + """States we infer ourselves must not masquerade as printer reports.""" + severity, text, _ = get_status_info(DERIVED_TONER_END) + assert severity == "critical" + assert "estimated" in text.lower() + + def test_invented_codes_are_unknown(self) -> None: + """Codes absent from the PJL reference must not be given invented meanings.""" + for code in ("40310", "40300", "42000", "30038", "40201"): + severity, text, _ = get_status_info(code) + assert severity == "info" + assert "Unknown" in text + def test_unknown_code(self) -> None: severity, text, action = get_status_info("99999") assert severity == "info" diff --git a/python_pkg/brother_printer/tests/test_cups_service.py b/python_pkg/brother_printer/tests/test_cups_service.py index f868b902..1d165c3e 100644 --- a/python_pkg/brother_printer/tests/test_cups_service.py +++ b/python_pkg/brother_printer/tests/test_cups_service.py @@ -3,21 +3,25 @@ from __future__ import annotations import json +import os import subprocess from unittest.mock import MagicMock, patch from python_pkg.brother_printer.cups_service import ( + STATE_SCHEMA_CUPS_SCALE, + _cups_is_busy, _ensure_cups_running, _get_cups_total_pages, _get_pyusb_device_info, _load_consumable_state, + _port_status_via_usblp, _query_usb_port_status_raw, _save_consumable_state, - _stop_cups, is_cups_scheduler_running, reset_consumable, start_cups, ) +from python_pkg.brother_printer.data_classes import USBPortStatus MOD = "python_pkg.brother_printer.cups_service" @@ -86,37 +90,6 @@ class TestGetPyusbDeviceInfo: assert result == {} -class TestStopCups: - @patch(f"{MOD}.shutil.which", return_value=None) - def test_no_systemctl(self, m: MagicMock) -> None: - assert _stop_cups() is False - - @patch(f"{MOD}.time.sleep") - @patch(f"{MOD}.subprocess.run") - @patch(f"{MOD}.shutil.which", return_value="/usr/bin/systemctl") - def test_success(self, w: MagicMock, mock_run: MagicMock, s: MagicMock) -> None: - mock_run.return_value = MagicMock() - assert _stop_cups() is True - - @patch(f"{MOD}.subprocess.run") - @patch(f"{MOD}.shutil.which", return_value="/usr/bin/systemctl") - def test_timeout(self, w: MagicMock, mock_run: MagicMock) -> None: - mock_run.side_effect = subprocess.TimeoutExpired("systemctl", 15) - assert _stop_cups() is False - - @patch(f"{MOD}.subprocess.run") - @patch(f"{MOD}.shutil.which", return_value="/usr/bin/systemctl") - def test_called_process_error(self, w: MagicMock, mock_run: MagicMock) -> None: - mock_run.side_effect = subprocess.CalledProcessError(1, "systemctl") - assert _stop_cups() is False - - @patch(f"{MOD}.subprocess.run") - @patch(f"{MOD}.shutil.which", return_value="/usr/bin/systemctl") - def test_oserror(self, w: MagicMock, mock_run: MagicMock) -> None: - mock_run.side_effect = OSError("fail") - assert _stop_cups() is False - - class TestIsCupsSchedulerRunning: @patch(f"{MOD}.shutil.which", return_value=None) def test_no_lpstat(self, m: MagicMock) -> None: @@ -210,53 +183,98 @@ class TestEnsureCupsRunning: assert _ensure_cups_running() is False -class TestQueryUsbPortStatusRaw: - def test_import_error(self) -> None: - with ( - patch(f"{MOD}._stop_cups"), - # Simulate ImportError for usb.core - patch.dict( - "sys.modules", {"usb": None, "usb.core": None, "usb.util": None} - ), - ): - result = _query_usb_port_status_raw() - assert result is None +class TestPortStatusViaUsblp: + """The zero-side-effect path: read the status byte from /dev/usb/lp*.""" - @patch(f"{MOD}.start_cups") - @patch(f"{MOD}._stop_cups", return_value=False) - def test_stop_cups_fails(self, st: MagicMock, s: MagicMock) -> None: + @patch(f"{MOD}.Path") + def test_no_dev_dir(self, mock_path: MagicMock) -> None: + mock_path.return_value.is_dir.return_value = False + assert _port_status_via_usblp() is None + + @patch(f"{MOD}.Path") + def test_no_devices(self, mock_path: MagicMock) -> None: + mock_path.return_value.is_dir.return_value = True + mock_path.return_value.glob.return_value = [] + assert _port_status_via_usblp() is None + + @patch(f"{MOD}.os.close") + @patch(f"{MOD}.fcntl.ioctl") + @patch(f"{MOD}.os.open", return_value=7) + @patch(f"{MOD}.Path") + def test_success( + self, + mock_path: MagicMock, + mock_open: MagicMock, + mock_ioctl: MagicMock, + mock_close: MagicMock, + ) -> None: + mock_path.return_value.is_dir.return_value = True + mock_path.return_value.glob.return_value = ["/dev/usb/lp0"] + + def fill(fd: int, req: int, buf: bytearray) -> None: + buf[0] = 0x18 + + mock_ioctl.side_effect = fill + result = _port_status_via_usblp() + assert result is not None + assert result.online is True + assert result.error is False + assert result.paper_empty is False + # Must never block: the printer may be mid-job. + assert mock_open.call_args[0][1] & os.O_NONBLOCK + mock_close.assert_called_once_with(7) + + @patch(f"{MOD}.os.open", side_effect=OSError("busy")) + @patch(f"{MOD}.Path") + def test_open_fails(self, mock_path: MagicMock, mock_open: MagicMock) -> None: + mock_path.return_value.is_dir.return_value = True + mock_path.return_value.glob.return_value = ["/dev/usb/lp0"] + assert _port_status_via_usblp() is None + + +class TestQueryUsbPortStatusRaw: + @patch(f"{MOD}._port_status_via_usblp", return_value=USBPortStatus(raw_byte=0x18)) + def test_prefers_usblp(self, m: MagicMock) -> None: + """The free path wins; pyusb is never touched.""" + result = _query_usb_port_status_raw("idle") + assert result is not None + assert result.raw_byte == 0x18 + + @patch(f"{MOD}._port_status_via_usblp", return_value=None) + def test_never_probes_while_printing(self, m: MagicMock) -> None: + """Regression: probing mid-job used to stop CUPS and kill the print.""" import sys as _sys mock_usb = MagicMock() - mock_usb.core.find.return_value = MagicMock() with patch.dict( _sys.modules, {"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util}, ): - result = _query_usb_port_status_raw() - assert result is None + assert _query_usb_port_status_raw("processing") is None + mock_usb.core.find.assert_not_called() - @patch(f"{MOD}.start_cups") - @patch(f"{MOD}._stop_cups", return_value=True) - def test_dev_none_after_reset(self, st: MagicMock, s: MagicMock) -> None: + @patch(f"{MOD}._port_status_via_usblp", return_value=None) + def test_import_error(self, m: MagicMock) -> None: + with patch.dict( + "sys.modules", {"usb": None, "usb.core": None, "usb.util": None} + ): + assert _query_usb_port_status_raw("idle") is None + + @patch(f"{MOD}._port_status_via_usblp", return_value=None) + def test_dev_none(self, m: MagicMock) -> None: import sys as _sys mock_usb = MagicMock() - mock_dev = MagicMock() - mock_usb.core.find.side_effect = [mock_dev, None] - with ( - patch.dict( - _sys.modules, - {"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util}, - ), - patch(f"{MOD}.time.sleep"), + mock_usb.core.find.return_value = None + with patch.dict( + _sys.modules, + {"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util}, ): - result = _query_usb_port_status_raw() - assert result is None + assert _query_usb_port_status_raw("idle") is None - @patch(f"{MOD}.start_cups") - @patch(f"{MOD}._stop_cups", return_value=True) - def test_success(self, stop: MagicMock, start: MagicMock) -> None: + @patch(f"{MOD}._port_status_via_usblp", return_value=None) + def test_success_reattaches_driver(self, m: MagicMock) -> None: + """usblp must get its device back, or every later run hits the fallback.""" import sys as _sys mock_usb = MagicMock() @@ -265,20 +283,18 @@ class TestQueryUsbPortStatusRaw: mock_dev.ctrl_transfer.return_value = [0x18] mock_usb.core.find.return_value = mock_dev mock_usb.core.USBError = type("USBError", (Exception,), {}) - with ( - patch.dict( - _sys.modules, - {"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util}, - ), - patch(f"{MOD}.time.sleep"), + with patch.dict( + _sys.modules, + {"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util}, ): - result = _query_usb_port_status_raw() - assert result is not None - assert result.online is True + result = _query_usb_port_status_raw("idle") + assert result is not None + assert result.online is True + mock_dev.reset.assert_not_called() + mock_dev.attach_kernel_driver.assert_called_once_with(0) - @patch(f"{MOD}.start_cups") - @patch(f"{MOD}._stop_cups", return_value=True) - def test_kernel_driver_not_active(self, stop: MagicMock, start: MagicMock) -> None: + @patch(f"{MOD}._port_status_via_usblp", return_value=None) + def test_kernel_driver_not_active(self, m: MagicMock) -> None: import sys as _sys mock_usb = MagicMock() @@ -287,19 +303,17 @@ class TestQueryUsbPortStatusRaw: mock_dev.ctrl_transfer.return_value = [0x18] mock_usb.core.find.return_value = mock_dev mock_usb.core.USBError = type("USBError", (Exception,), {}) - with ( - patch.dict( - _sys.modules, - {"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util}, - ), - patch(f"{MOD}.time.sleep"), + with patch.dict( + _sys.modules, + {"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util}, ): - result = _query_usb_port_status_raw() - assert result is not None + result = _query_usb_port_status_raw("idle") + assert result is not None + # Nothing was detached, so nothing should be re-attached. + mock_dev.attach_kernel_driver.assert_not_called() - @patch(f"{MOD}.start_cups") - @patch(f"{MOD}._stop_cups", return_value=True) - def test_kernel_driver_usberror(self, stop: MagicMock, start: MagicMock) -> None: + @patch(f"{MOD}._port_status_via_usblp", return_value=None) + def test_kernel_driver_usberror(self, m: MagicMock) -> None: import sys as _sys mock_usb = MagicMock() @@ -309,19 +323,14 @@ class TestQueryUsbPortStatusRaw: mock_dev.ctrl_transfer.return_value = [0x18] mock_usb.core.find.return_value = mock_dev mock_usb.core.USBError = usb_error_cls - with ( - patch.dict( - _sys.modules, - {"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util}, - ), - patch(f"{MOD}.time.sleep"), + with patch.dict( + _sys.modules, + {"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util}, ): - result = _query_usb_port_status_raw() - assert result is not None + assert _query_usb_port_status_raw("idle") is not None - @patch(f"{MOD}.start_cups") - @patch(f"{MOD}._stop_cups", return_value=True) - def test_oserror_during_transfer(self, stop: MagicMock, start: MagicMock) -> None: + @patch(f"{MOD}._port_status_via_usblp", return_value=None) + def test_oserror_during_transfer(self, m: MagicMock) -> None: import sys as _sys mock_usb = MagicMock() @@ -330,29 +339,42 @@ class TestQueryUsbPortStatusRaw: mock_usb.core.find.return_value = mock_dev mock_usb.core.USBError = type("USBError", (Exception,), {}) mock_usb.util.claim_interface.side_effect = OSError("usb fail") - with ( - patch.dict( - _sys.modules, - {"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util}, - ), - patch(f"{MOD}.time.sleep"), - ): - result = _query_usb_port_status_raw() - assert result is None - - @patch(f"{MOD}.start_cups") - @patch(f"{MOD}._stop_cups", return_value=True) - def test_dev_none_initial(self, stop: MagicMock, start: MagicMock) -> None: - import sys as _sys - - mock_usb = MagicMock() - mock_usb.core.find.return_value = None with patch.dict( _sys.modules, {"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util}, ): - result = _query_usb_port_status_raw() - assert result is None + assert _query_usb_port_status_raw("idle") is None + + @patch(f"{MOD}._port_status_via_usblp", return_value=None) + def test_attach_failure_is_survivable(self, m: MagicMock) -> None: + import sys as _sys + + mock_usb = MagicMock() + mock_dev = MagicMock() + mock_dev.is_kernel_driver_active.return_value = True + mock_dev.ctrl_transfer.return_value = [0x18] + mock_dev.attach_kernel_driver.side_effect = OSError("cannot reattach") + mock_usb.core.find.return_value = mock_dev + mock_usb.core.USBError = type("USBError", (Exception,), {}) + with patch.dict( + _sys.modules, + {"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util}, + ): + assert _query_usb_port_status_raw("idle") is not None + + +class TestCupsIsBusy: + def test_processing(self) -> None: + assert _cups_is_busy("processing") is True + + def test_printing_text(self) -> None: + assert _cups_is_busy("now printing Brother-98") is True + + def test_idle(self) -> None: + assert _cups_is_busy("idle") is False + + def test_empty(self) -> None: + assert _cups_is_busy("") is False class TestGetCupsTotalPages: @@ -389,7 +411,13 @@ class TestLoadConsumableState: def test_no_file(self, mock_file: MagicMock) -> None: mock_file.exists.return_value = False result = _load_consumable_state() - assert result == {"toner_replaced_at": 0, "drum_replaced_at": 0} + assert result == { + "toner_replaced_at": 0, + "drum_replaced_at": 0, + "schema": STATE_SCHEMA_CUPS_SCALE, + "last_printer_count": 0, + "last_cups_total": 0, + } @patch(f"{MOD}.CONSUMABLE_STATE_FILE") def test_valid_file(self, mock_file: MagicMock) -> None: @@ -452,3 +480,22 @@ class TestResetConsumable: reset_consumable("toner") saved_state = mock_save.call_args[0][0] assert saved_state["toner_replaced_at"] == 500 + + +class TestPortStatusViaUsblpIoctlFailure: + @patch(f"{MOD}.os.close") + @patch(f"{MOD}.fcntl.ioctl", side_effect=OSError("not supported")) + @patch(f"{MOD}.os.open", return_value=7) + @patch(f"{MOD}.Path") + def test_ioctl_failure_still_closes_fd( + self, + mock_path: MagicMock, + mock_open: MagicMock, + mock_ioctl: MagicMock, + mock_close: MagicMock, + ) -> None: + """A failed ioctl must not leak the fd, or usblp stays busy.""" + mock_path.return_value.is_dir.return_value = True + mock_path.return_value.glob.return_value = ["/dev/usb/lp0"] + assert _port_status_via_usblp() is None + mock_close.assert_called_once_with(7) diff --git a/python_pkg/brother_printer/tests/test_cups_service_part2.py b/python_pkg/brother_printer/tests/test_cups_service_part2.py index 9daae1e8..734d6bc2 100644 --- a/python_pkg/brother_printer/tests/test_cups_service_part2.py +++ b/python_pkg/brother_printer/tests/test_cups_service_part2.py @@ -5,6 +5,11 @@ from __future__ import annotations import subprocess from unittest.mock import MagicMock, patch +from python_pkg.brother_printer.constants import ( + _CUPS_REASONS_TO_STATUS, + DERIVED_CUPS_ERROR, + get_status_info, +) from python_pkg.brother_printer.cups_service import ( _cups_reasons_to_error, _get_cups_economode, @@ -78,31 +83,40 @@ class TestGetCupsEconomode: class TestMapCupsToStatusCode: - """Tests for _map_cups_to_status_code.""" + """Tests for _map_cups_to_status_code. + + Every code here must exist in BROTHER_STATUS_CODES; mapping CUPS onto a + code the table does not know silently degrades to "unknown status". + """ def test_reason_match(self) -> None: - result = _map_cups_to_status_code("idle", "toner-low-report") - assert result == "30010" + assert _map_cups_to_status_code("idle", "toner-low-report") == "40038" def test_state_match(self) -> None: - result = _map_cups_to_status_code("idle", "none") - assert result == "10001" + assert _map_cups_to_status_code("idle", "none") == "10001" def test_processing_state(self) -> None: - result = _map_cups_to_status_code("processing", "none") - assert result == "10007" + assert _map_cups_to_status_code("processing", "none") == "10023" def test_stopped_state(self) -> None: - result = _map_cups_to_status_code("stopped", "none") - assert result == "10023" + assert _map_cups_to_status_code("stopped", "none") == "40079" def test_unknown_state(self) -> None: - result = _map_cups_to_status_code("mystery", "none") - assert result == "10001" + assert _map_cups_to_status_code("mystery", "none") == "10001" def test_state_with_parenthetical(self) -> None: - result = _map_cups_to_status_code("idle (on fire)", "none") - assert result == "10001" + assert _map_cups_to_status_code("idle (on fire)", "none") == "10001" + + def test_every_mapped_code_is_resolvable(self) -> None: + """Guard the class of bug where a mapping points at a deleted code.""" + for state in ("idle", "processing", "stopped"): + code = _map_cups_to_status_code(state, "none") + _, text, _ = get_status_info(code) + assert "Unknown" not in text + for reason in _CUPS_REASONS_TO_STATUS: + code = _map_cups_to_status_code("idle", reason) + _, text, _ = get_status_info(code) + assert "Unknown" not in text # ── _cups_reasons_to_error ─────────────────────────────────────────── @@ -112,59 +126,82 @@ class TestCupsReasonsToError: """Tests for _cups_reasons_to_error.""" def test_media_jam(self) -> None: + """A real jam is 40022. It used to map to 40000, which is Sleep.""" code, display = _cups_reasons_to_error("media-jam-report") - assert code == "40000" + assert code == "40022" assert display == "Paper Jam" def test_cover_open(self) -> None: code, _ = _cups_reasons_to_error("cover-open") - assert code == "41000" + assert code == "40021" def test_door_open(self) -> None: code, _ = _cups_reasons_to_error("door-open") - assert code == "41000" + assert code == "40021" def test_toner_empty(self) -> None: code, _ = _cups_reasons_to_error("toner-empty") - assert code == "40310" + assert code == "40010" def test_toner_low(self) -> None: code, _ = _cups_reasons_to_error("toner-low") - assert code == "30010" + assert code == "40038" - def test_unknown_reason(self) -> None: + def test_unknown_reason_shows_what_cups_said(self) -> None: + """'Printer Error' alone is useless; quote the reason we were given.""" code, display = _cups_reasons_to_error("something-weird") - assert code == "42000" - assert display == "Printer Error" + assert code == DERIVED_CUPS_ERROR + assert "something-weird" in display + + def test_no_reason_at_all(self) -> None: + code, display = _cups_reasons_to_error("none") + assert code == DERIVED_CUPS_ERROR + assert "no reason" in display.lower() + + def test_every_error_code_is_resolvable(self) -> None: + for reason in ("media-jam", "cover-open", "toner-empty", "toner-low"): + code, _ = _cups_reasons_to_error(reason) + _, text, _ = get_status_info(code) + assert "Unknown" not in text # ── _port_status_to_status_code ────────────────────────────────────── class TestPortStatusToStatusCode: - """Tests for _port_status_to_status_code.""" + """Tests for _port_status_to_status_code. + + The port status only exposes paper/error/online bits, so it must not + pretend to know more than that - it previously reported "Cover Open" for + any error on an offline printer, which was a guess. + """ def test_error_and_paper_empty(self) -> None: ps = USBPortStatus(error=True, paper_empty=True, online=True) code, display = _port_status_to_status_code(ps, "none") - assert code == "40302" + assert code == "41000" assert display == "No Paper" - def test_error_and_not_online(self) -> None: + def test_error_and_not_online_defers_to_cups(self) -> None: ps = USBPortStatus(error=True, paper_empty=False, online=False) - code, display = _port_status_to_status_code(ps, "none") - assert code == "41000" + code, display = _port_status_to_status_code(ps, "cover-open") + assert code == "40021" assert display == "Cover Open" + def test_error_with_no_clue_does_not_guess(self) -> None: + ps = USBPortStatus(error=True, paper_empty=False, online=False) + code, _ = _port_status_to_status_code(ps, "none") + assert code == DERIVED_CUPS_ERROR + def test_error_only(self) -> None: ps = USBPortStatus(error=True, paper_empty=False, online=True) code, _ = _port_status_to_status_code(ps, "media-jam") - assert code == "40000" + assert code == "40022" def test_paper_empty_no_error(self) -> None: ps = USBPortStatus(error=False, paper_empty=True, online=True) code, _ = _port_status_to_status_code(ps, "none") - assert code == "40302" + assert code == "41000" def test_not_online_no_error(self) -> None: ps = USBPortStatus(error=False, paper_empty=False, online=False) diff --git a/python_pkg/brother_printer/tests/test_cups_service_part3.py b/python_pkg/brother_printer/tests/test_cups_service_part3.py index fca92640..a53fe034 100644 --- a/python_pkg/brother_printer/tests/test_cups_service_part3.py +++ b/python_pkg/brother_printer/tests/test_cups_service_part3.py @@ -4,6 +4,10 @@ from __future__ import annotations from unittest.mock import MagicMock, patch +from python_pkg.brother_printer.constants import ( + DERIVED_TONER_END, + DERIVED_TONER_LOW, +) from python_pkg.brother_printer.cups_service import ( query_usb_via_cups, ) @@ -103,7 +107,7 @@ class TestQueryUsbViaCups: ), ): result = query_usb_via_cups() - assert result.status_code == "40302" + assert result.status_code == "41000" assert result.online == "FALSE" def test_port_ok_toner_exhausted(self) -> None: @@ -142,7 +146,7 @@ class TestQueryUsbViaCups: ), ): result = query_usb_via_cups() - assert result.status_code == "40310" + assert result.status_code == DERIVED_TONER_END assert "Toner End" in result.display def test_port_ok_toner_low(self) -> None: @@ -181,7 +185,7 @@ class TestQueryUsbViaCups: ), ): result = query_usb_via_cups() - assert result.status_code == "30010" + assert result.status_code == DERIVED_TONER_LOW assert "Toner Low" in result.display def test_port_ok_normal(self) -> None: @@ -251,6 +255,6 @@ class TestQueryUsbViaCups: ), ): result = query_usb_via_cups() - assert result.status_code == "40000" + assert result.status_code == "40022" assert result.product == "HL-1110" assert result.online == "TRUE" diff --git a/python_pkg/brother_printer/tests/test_cups_service_part4.py b/python_pkg/brother_printer/tests/test_cups_service_part4.py index ae50ceeb..ad3f4f89 100644 --- a/python_pkg/brother_printer/tests/test_cups_service_part4.py +++ b/python_pkg/brother_printer/tests/test_cups_service_part4.py @@ -6,9 +6,16 @@ import subprocess from unittest.mock import MagicMock, patch from python_pkg.brother_printer.cups_service import ( + STATE_SCHEMA_CUPS_SCALE, + STATE_SCHEMA_PRINTER_SCALE, + _cups_total_on_printer_scale, _get_cups_ipp_status, + _migrate_state_to_printer_scale, _parse_ipp_attributes, + _snapshot_counters, + check_page_delivery, estimate_consumable_life, + reset_consumable, ) MOD = "python_pkg.brother_printer.cups_service" @@ -24,7 +31,12 @@ class TestEstimateConsumableLife: @patch(f"{MOD}._load_consumable_state") @patch(f"{MOD}._get_cups_total_pages", return_value=500) def test_mid_life(self, p: MagicMock, mock_load: MagicMock) -> None: - mock_load.return_value = {"toner_replaced_at": 0, "drum_replaced_at": 0} + mock_load.return_value = { + "toner_replaced_at": 0, + "drum_replaced_at": 0, + "last_printer_count": 0, + "last_cups_total": 0, + } result = estimate_consumable_life() assert result.total_pages == 500 assert result.toner_pct_remaining == 50 @@ -34,21 +46,36 @@ class TestEstimateConsumableLife: @patch(f"{MOD}._load_consumable_state") @patch(f"{MOD}._get_cups_total_pages", return_value=1000) def test_toner_exhausted(self, p: MagicMock, mock_load: MagicMock) -> None: - mock_load.return_value = {"toner_replaced_at": 0, "drum_replaced_at": 0} + mock_load.return_value = { + "toner_replaced_at": 0, + "drum_replaced_at": 0, + "last_printer_count": 0, + "last_cups_total": 0, + } result = estimate_consumable_life() assert result.toner_exhausted is True @patch(f"{MOD}._load_consumable_state") @patch(f"{MOD}._get_cups_total_pages", return_value=800) def test_toner_low(self, p: MagicMock, mock_load: MagicMock) -> None: - mock_load.return_value = {"toner_replaced_at": 0, "drum_replaced_at": 0} + mock_load.return_value = { + "toner_replaced_at": 0, + "drum_replaced_at": 0, + "last_printer_count": 0, + "last_cups_total": 0, + } result = estimate_consumable_life() assert result.toner_low is True @patch(f"{MOD}._load_consumable_state") @patch(f"{MOD}._get_cups_total_pages", return_value=9000) def test_drum_near_end(self, p: MagicMock, mock_load: MagicMock) -> None: - mock_load.return_value = {"toner_replaced_at": 8500, "drum_replaced_at": 0} + mock_load.return_value = { + "toner_replaced_at": 8500, + "drum_replaced_at": 0, + "last_printer_count": 0, + "last_cups_total": 0, + } result = estimate_consumable_life() assert result.drum_near_end is True @@ -84,3 +111,340 @@ class TestGetCupsIppStatus: def test_timeout(self, w: MagicMock, mock_run: MagicMock) -> None: mock_run.side_effect = subprocess.TimeoutExpired("ipptool", 10) assert _get_cups_ipp_status("Brother") == {} + + +class TestMigrateStateToPrinterScale: + """Rebasing replacement baselines from the CUPS page log onto the printer. + + The baselines are recorded against whatever counter was in use when they + were written. Switching counters without rebasing them silently changes + every reported percentage, so this is the migration that keeps the numbers + meaning the same thing before and after. + """ + + @patch(f"{MOD}._save_consumable_state") + @patch(f"{MOD}._get_cups_total_pages", return_value=1658) + def test_shifts_baseline_by_offset( + self, + mock_total: MagicMock, + mock_save: MagicMock, + ) -> None: + state = { + "toner_replaced_at": 1408, + "drum_replaced_at": 0, + "schema": STATE_SCHEMA_CUPS_SCALE, + } + result = _migrate_state_to_printer_scale(state, 2017) + # Offset is 2017 - 1658 = 359. + assert result["toner_replaced_at"] == 1767 + assert result["schema"] == STATE_SCHEMA_PRINTER_SCALE + mock_save.assert_called_once() + + @patch(f"{MOD}._save_consumable_state") + @patch(f"{MOD}._get_cups_total_pages", return_value=1658) + def test_zero_baseline_left_alone( + self, + mock_total: MagicMock, + mock_save: MagicMock, + ) -> None: + """Zero means 'never replaced', not 'replaced at page zero'. + + On the printer's scale zero already means "as old as the printer", + which is the truthful reading; shifting it would re-hide the pages the + CUPS log never saw. + """ + state = { + "toner_replaced_at": 0, + "drum_replaced_at": 0, + "schema": STATE_SCHEMA_CUPS_SCALE, + } + result = _migrate_state_to_printer_scale(state, 2017) + assert result["drum_replaced_at"] == 0 + assert result["toner_replaced_at"] == 0 + + @patch(f"{MOD}._save_consumable_state") + @patch(f"{MOD}._get_cups_total_pages", return_value=1658) + def test_runs_only_once( + self, + mock_total: MagicMock, + mock_save: MagicMock, + ) -> None: + """A second run must not shift the baselines again.""" + state = { + "toner_replaced_at": 1767, + "drum_replaced_at": 0, + "schema": STATE_SCHEMA_PRINTER_SCALE, + } + result = _migrate_state_to_printer_scale(state, 2017) + assert result["toner_replaced_at"] == 1767 + mock_save.assert_not_called() + + @patch(f"{MOD}._save_consumable_state") + @patch(f"{MOD}._get_cups_total_pages", return_value=2100) + def test_negative_offset_leaves_baselines( + self, + mock_total: MagicMock, + mock_save: MagicMock, + ) -> None: + """If the CUPS log somehow exceeds the printer, do not corrupt state.""" + state = { + "toner_replaced_at": 1408, + "drum_replaced_at": 0, + "schema": STATE_SCHEMA_CUPS_SCALE, + } + result = _migrate_state_to_printer_scale(state, 2017) + assert result["toner_replaced_at"] == 1408 + assert result["schema"] == STATE_SCHEMA_PRINTER_SCALE + + +class TestEstimateConsumableLifeCounterSource: + """Which counter the estimate came from, and whether it says so.""" + + @patch(f"{MOD}._migrate_state_to_printer_scale") + @patch(f"{MOD}._load_consumable_state") + @patch(f"{MOD}._get_cups_total_pages", return_value=1658) + def test_printer_counter_is_authoritative( + self, + mock_cups: MagicMock, + mock_load: MagicMock, + mock_migrate: MagicMock, + ) -> None: + mock_load.return_value = { + "toner_replaced_at": 1767, + "drum_replaced_at": 0, + "last_printer_count": 0, + "last_cups_total": 0, + } + mock_migrate.side_effect = lambda state, total: state + estimate = estimate_consumable_life(2017) + assert estimate.total_pages == 2017 + assert estimate.approximate is False + assert estimate.toner_pages == 250 + + @patch(f"{MOD}._load_consumable_state") + @patch(f"{MOD}._get_cups_total_pages", return_value=1658) + def test_falls_back_to_cups_log_and_flags_it( + self, + mock_cups: MagicMock, + mock_load: MagicMock, + ) -> None: + """Without the printer's counter, say the number is approximate.""" + mock_load.return_value = { + "toner_replaced_at": 0, + "drum_replaced_at": 0, + "last_printer_count": 0, + "last_cups_total": 0, + } + estimate = estimate_consumable_life(0) + assert estimate.total_pages == 1658 + assert estimate.approximate is True + + @patch(f"{MOD}._get_cups_total_pages", return_value=0) + def test_no_counter_at_all(self, mock_cups: MagicMock) -> None: + assert estimate_consumable_life(0).total_pages == 0 + + +class TestResetConsumableScale: + @patch(f"{MOD}._out") + @patch(f"{MOD}._save_consumable_state") + @patch(f"{MOD}._load_consumable_state") + def test_printer_total_marks_state_migrated( + self, + mock_load: MagicMock, + mock_save: MagicMock, + mock_out: MagicMock, + ) -> None: + """A baseline written from the printer is already on the new scale.""" + mock_load.return_value = { + "toner_replaced_at": 0, + "drum_replaced_at": 0, + "schema": STATE_SCHEMA_CUPS_SCALE, + } + reset_consumable("toner", 2017) + saved = mock_save.call_args[0][0] + assert saved["toner_replaced_at"] == 2017 + assert saved["schema"] == STATE_SCHEMA_PRINTER_SCALE + + @patch(f"{MOD}._out") + @patch(f"{MOD}._save_consumable_state") + @patch(f"{MOD}._load_consumable_state") + @patch(f"{MOD}._get_cups_total_pages", return_value=1658) + def test_without_printer_total_uses_cups_log( + self, + mock_cups: MagicMock, + mock_load: MagicMock, + mock_save: MagicMock, + mock_out: MagicMock, + ) -> None: + mock_load.return_value = { + "toner_replaced_at": 0, + "drum_replaced_at": 0, + "schema": STATE_SCHEMA_CUPS_SCALE, + } + reset_consumable("drum") + saved = mock_save.call_args[0][0] + assert saved["drum_replaced_at"] == 1658 + assert saved["schema"] == STATE_SCHEMA_CUPS_SCALE + + +class TestCheckPageDelivery: + """Spotting pages the printer silently dropped. + + CUPS calls a job successful once the data leaves the machine, so this + comparison against the printer's own counter is the only signal that a page + never actually came out. + """ + + @patch(f"{MOD}._snapshot_counters") + @patch(f"{MOD}._get_cups_total_pages", return_value=1841) + @patch(f"{MOD}._load_consumable_state") + def test_detects_dropped_pages( + self, + mock_load: MagicMock, + mock_cups: MagicMock, + mock_snap: MagicMock, + ) -> None: + """Replay of the real failure: CUPS sent 63, printer printed none.""" + mock_load.return_value = { + "last_printer_count": 2087, + "last_cups_total": 1778, + } + check = check_page_delivery(2087, queue_idle=True) + assert check.cups_pages == 63 + assert check.printer_pages == 0 + assert check.dropped == 63 + assert check.suspected is True + + @patch(f"{MOD}._snapshot_counters") + @patch(f"{MOD}._get_cups_total_pages", return_value=1841) + @patch(f"{MOD}._load_consumable_state") + def test_healthy_job_does_not_warn( + self, + mock_load: MagicMock, + mock_cups: MagicMock, + mock_snap: MagicMock, + ) -> None: + mock_load.return_value = { + "last_printer_count": 2024, + "last_cups_total": 1778, + } + check = check_page_delivery(2087, queue_idle=True) + assert check.dropped == 0 + assert check.suspected is False + + @patch(f"{MOD}._snapshot_counters") + @patch(f"{MOD}._get_cups_total_pages", return_value=1780) + @patch(f"{MOD}._load_consumable_state") + def test_small_gap_is_not_a_warning( + self, + mock_load: MagicMock, + mock_cups: MagicMock, + mock_snap: MagicMock, + ) -> None: + """A page or two of drift is timing, not a fault.""" + mock_load.return_value = { + "last_printer_count": 2087, + "last_cups_total": 1778, + } + check = check_page_delivery(2087, queue_idle=True) + assert check.dropped == 2 + assert check.suspected is False + + def test_skipped_while_printing(self) -> None: + """Mid-job the counters legitimately disagree, so never compare.""" + assert check_page_delivery(2087, queue_idle=False).suspected is False + + def test_skipped_without_printer_count(self) -> None: + assert check_page_delivery(0, queue_idle=True).suspected is False + + @patch(f"{MOD}._snapshot_counters") + @patch(f"{MOD}._get_cups_total_pages", return_value=1841) + @patch(f"{MOD}._load_consumable_state") + def test_first_run_only_establishes_baseline( + self, + mock_load: MagicMock, + mock_cups: MagicMock, + mock_snap: MagicMock, + ) -> None: + mock_load.return_value = {"last_printer_count": 0, "last_cups_total": 0} + check = check_page_delivery(2087, queue_idle=True) + assert check.suspected is False + mock_snap.assert_called_once() + + @patch(f"{MOD}._snapshot_counters") + @patch(f"{MOD}._get_cups_total_pages", return_value=10) + @patch(f"{MOD}._load_consumable_state") + def test_log_rotation_draws_no_conclusion( + self, + mock_load: MagicMock, + mock_cups: MagicMock, + mock_snap: MagicMock, + ) -> None: + """A rotated page log makes the delta negative; do not cry wolf.""" + mock_load.return_value = { + "last_printer_count": 2087, + "last_cups_total": 1778, + } + check = check_page_delivery(2087, queue_idle=True) + assert check.suspected is False + + @patch(f"{MOD}._snapshot_counters") + @patch(f"{MOD}._get_cups_total_pages", return_value=1841) + @patch(f"{MOD}._load_consumable_state") + def test_printer_counter_reset_draws_no_conclusion( + self, + mock_load: MagicMock, + mock_cups: MagicMock, + mock_snap: MagicMock, + ) -> None: + mock_load.return_value = { + "last_printer_count": 5000, + "last_cups_total": 1778, + } + check = check_page_delivery(2087, queue_idle=True) + assert check.suspected is False + + +class TestSnapshotCounters: + @patch(f"{MOD}._save_consumable_state") + def test_writes_new_snapshot(self, mock_save: MagicMock) -> None: + state = {"last_printer_count": 2000, "last_cups_total": 1700} + _snapshot_counters(state, 2087, 1778) + saved = mock_save.call_args[0][0] + assert saved["last_printer_count"] == 2087 + assert saved["last_cups_total"] == 1778 + + @patch(f"{MOD}._save_consumable_state") + def test_unchanged_counters_skip_the_write(self, mock_save: MagicMock) -> None: + """Do not rewrite the file on every run when nothing moved.""" + state = {"last_printer_count": 2087, "last_cups_total": 1778} + _snapshot_counters(state, 2087, 1778) + mock_save.assert_not_called() + + +class TestCupsTotalOnPrinterScale: + """The CUPS figure must be shifted onto the scale the baselines use. + + Without the shift the raw CUPS total sits below a printer-scale baseline, + the subtraction clamps at zero, and a part-used cartridge reads 100%. + """ + + @patch(f"{MOD}._get_cups_total_pages", return_value=1778) + def test_applies_offset(self, mock_cups: MagicMock) -> None: + state = {"last_printer_count": 2087, "last_cups_total": 1778} + assert _cups_total_on_printer_scale(state) == 2087 + + @patch(f"{MOD}._get_cups_total_pages", return_value=1778) + def test_no_snapshot_returns_raw(self, mock_cups: MagicMock) -> None: + state = {"last_printer_count": 0, "last_cups_total": 0} + assert _cups_total_on_printer_scale(state) == 1778 + + @patch(f"{MOD}._get_cups_total_pages", return_value=1778) + def test_negative_offset_returns_raw(self, mock_cups: MagicMock) -> None: + state = {"last_printer_count": 1700, "last_cups_total": 1778} + assert _cups_total_on_printer_scale(state) == 1778 + + @patch(f"{MOD}._get_cups_total_pages", return_value=0) + def test_no_cups_pages(self, mock_cups: MagicMock) -> None: + state = {"last_printer_count": 2087, "last_cups_total": 1778} + assert _cups_total_on_printer_scale(state) == 0 diff --git a/python_pkg/brother_printer/tests/test_display.py b/python_pkg/brother_printer/tests/test_display.py index 1ff2a594..02f040e3 100644 --- a/python_pkg/brother_printer/tests/test_display.py +++ b/python_pkg/brother_printer/tests/test_display.py @@ -10,6 +10,7 @@ import pytest from python_pkg.brother_printer.data_classes import ( NetworkResult, PageCountEstimate, + PageDeliveryCheck, SupplyReadings, USBPortStatus, USBResult, @@ -21,6 +22,7 @@ from python_pkg.brother_printer.display import ( _display_consumables_reference, _display_cups_fallback_note, _display_page_count_estimate, + _display_page_delivery_warning, _display_pjl_status, _display_report_header, _display_supply_levels, @@ -51,6 +53,41 @@ class TestDisplayPageCountEstimate: _display_page_count_estimate() assert out.getvalue() == "" + @patch(f"{MOD}.estimate_consumable_life") + def test_printer_count_passed_through(self, mock_est: MagicMock) -> None: + mock_est.return_value = PageCountEstimate(total_pages=2017) + with patch("sys.stdout", new_callable=StringIO): + _display_page_count_estimate(2017) + mock_est.assert_called_once_with(2017) + + @patch(f"{MOD}.estimate_consumable_life") + def test_approximate_is_labelled(self, mock_est: MagicMock) -> None: + """A CUPS-log figure must not be presented as the printer's own count.""" + mock_est.return_value = PageCountEstimate( + total_pages=1658, + toner_pages=250, + drum_pages=1658, + approximate=True, + ) + with patch("sys.stdout", new_callable=StringIO) as out: + _display_page_count_estimate(0) + assert "Approximate" in out.getvalue() + + @patch(f"{MOD}.estimate_consumable_life") + def test_printer_count_is_not_labelled_approximate( + self, + mock_est: MagicMock, + ) -> None: + mock_est.return_value = PageCountEstimate( + total_pages=2017, + toner_pages=250, + drum_pages=2017, + approximate=False, + ) + with patch("sys.stdout", new_callable=StringIO) as out: + _display_page_count_estimate(2017) + assert "Approximate" not in out.getvalue() + @patch(f"{MOD}.estimate_consumable_life") def test_healthy(self, mock_est: MagicMock) -> None: mock_est.return_value = PageCountEstimate( @@ -436,3 +473,37 @@ class TestDisplaySupplyLevels: _display_supply_levels(result) text = out.getvalue() assert "ACTION NEEDED" in text + + +class TestDisplayPageDeliveryWarning: + """Surfacing pages the printer dropped without telling anyone.""" + + @patch(f"{MOD}.check_page_delivery") + def test_warns_with_actionable_fix(self, mock_check: MagicMock) -> None: + mock_check.return_value = PageDeliveryCheck( + cups_pages=63, + printer_pages=3, + dropped=60, + suspected=True, + ) + with patch("sys.stdout", new_callable=StringIO) as out: + _display_page_delivery_warning(2087, queue_idle=True) + printed = out.getvalue() + assert "60 pages did not print" in printed + assert "63" in printed + # Useless without telling the reader what to actually do about it. + assert "Resolution=300dpi" in printed + + @patch(f"{MOD}.check_page_delivery") + def test_silent_when_healthy(self, mock_check: MagicMock) -> None: + mock_check.return_value = PageDeliveryCheck(suspected=False) + with patch("sys.stdout", new_callable=StringIO) as out: + _display_page_delivery_warning(2087, queue_idle=True) + assert out.getvalue() == "" + + @patch(f"{MOD}.check_page_delivery") + def test_passes_queue_state_through(self, mock_check: MagicMock) -> None: + mock_check.return_value = PageDeliveryCheck(suspected=False) + with patch("sys.stdout", new_callable=StringIO): + _display_page_delivery_warning(2087, queue_idle=False) + mock_check.assert_called_once_with(2087, queue_idle=False) diff --git a/python_pkg/brother_printer/tests/test_main_entry.py b/python_pkg/brother_printer/tests/test_main_entry.py index f57dc987..cb0f6aaa 100644 --- a/python_pkg/brother_printer/tests/test_main_entry.py +++ b/python_pkg/brother_printer/tests/test_main_entry.py @@ -3,27 +3,29 @@ from __future__ import annotations import importlib -import types +import runpy +import sys from unittest.mock import MagicMock, patch class TestMain: - def test_main_called(self) -> None: - """Test that __main__ calls main().""" + def test_main_called_when_run_as_module(self) -> None: + """python3 -m python_pkg.brother_printer runs main().""" mock_main = MagicMock() - # Create a fake brother_printer.check_brother_printer module - fake_module = types.ModuleType("brother_printer.check_brother_printer") - vars(fake_module)["main"] = mock_main - with patch.dict( - "sys.modules", - { - "brother_printer": types.ModuleType("brother_printer"), - "brother_printer.check_brother_printer": fake_module, - }, + with patch( + "python_pkg.brother_printer.check_brother_printer.main", + mock_main, ): - # Remove cached __main__ module so it gets re-imported - import sys + runpy.run_module("python_pkg.brother_printer", run_name="__main__") + mock_main.assert_called_once() + def test_main_not_called_on_plain_import(self) -> None: + """Importing __main__ must not fire off a status report as a side effect.""" + mock_main = MagicMock() + with patch( + "python_pkg.brother_printer.check_brother_printer.main", + mock_main, + ): sys.modules.pop("python_pkg.brother_printer.__main__", None) importlib.import_module("python_pkg.brother_printer.__main__") - mock_main.assert_called_once() + mock_main.assert_not_called() diff --git a/python_pkg/brother_printer/tests/test_usb_query.py b/python_pkg/brother_printer/tests/test_usb_query.py index 8ca42676..c00e40af 100644 --- a/python_pkg/brother_printer/tests/test_usb_query.py +++ b/python_pkg/brother_printer/tests/test_usb_query.py @@ -2,18 +2,22 @@ from __future__ import annotations +import errno +import os from unittest.mock import MagicMock, patch from python_pkg.brother_printer.data_classes import USBResult from python_pkg.brother_printer.usb_query import ( _drain_buffer, _init_usb_result, + _parse_pagecount, _parse_status, _parse_variables, _read_nonblocking, _retry_pjl_query, _run_pjl_queries, _wait_for_pjl_response, + _write_with_deadline, find_brother_usb, find_usb_printer_dev, pjl_query, @@ -99,22 +103,14 @@ class TestFindUsbPrinterDev: class TestDrainBuffer: @patch(f"{MOD}.os.read") - @patch(f"{MOD}.fcntl.fcntl") - def test_drain(self, mock_fcntl: MagicMock, mock_read: MagicMock) -> None: - mock_fcntl.return_value = 0 + def test_drain(self, mock_read: MagicMock) -> None: mock_read.side_effect = [b"data", OSError("done")] _drain_buffer(42) assert mock_read.called @patch(f"{MOD}.os.read") - @patch(f"{MOD}.fcntl.fcntl") - def test_drain_empty_buffer( - self, - mock_fcntl: MagicMock, - mock_read: MagicMock, - ) -> None: - """Buffer is already empty — os.read returns b'' immediately.""" - mock_fcntl.return_value = 0 + def test_drain_empty_buffer(self, mock_read: MagicMock) -> None: + """Buffer is already empty - os.read returns b'' immediately.""" mock_read.return_value = b"" _drain_buffer(42) mock_read.assert_called_once() @@ -122,24 +118,65 @@ class TestDrainBuffer: class TestReadNonblocking: @patch(f"{MOD}.os.read") - @patch(f"{MOD}.fcntl.fcntl") - def test_reads_chunks(self, mock_fcntl: MagicMock, mock_read: MagicMock) -> None: - mock_fcntl.return_value = 0 + def test_reads_chunks(self, mock_read: MagicMock) -> None: mock_read.side_effect = [b"hello", b"", OSError] - result = _read_nonblocking(42, 0) - assert result == b"hello" + assert _read_nonblocking(42) == b"hello" @patch(f"{MOD}.os.read") - @patch(f"{MOD}.fcntl.fcntl") - def test_oserror_suppressed( - self, - mock_fcntl: MagicMock, - mock_read: MagicMock, - ) -> None: - mock_fcntl.return_value = 0 + def test_oserror_suppressed(self, mock_read: MagicMock) -> None: mock_read.side_effect = OSError("would block") - result = _read_nonblocking(42, 0) - assert result == b"" + assert _read_nonblocking(42) == b"" + + +class TestWriteWithDeadline: + """Writes must never block forever on a printer that is not listening.""" + + @patch(f"{MOD}.os.write", return_value=5) + @patch(f"{MOD}.time.time", return_value=0.0) + def test_writes_all(self, t: MagicMock, mock_write: MagicMock) -> None: + assert _write_with_deadline(42, b"hello", 10.0) is True + + @patch(f"{MOD}.os.write") + @patch(f"{MOD}.time.time", return_value=0.0) + def test_partial_writes_resume(self, t: MagicMock, mock_write: MagicMock) -> None: + mock_write.side_effect = [2, 3] + assert _write_with_deadline(42, b"hello", 10.0) is True + assert mock_write.call_count == 2 + + @patch(f"{MOD}.time.time") + def test_deadline_already_passed(self, mock_time: MagicMock) -> None: + mock_time.return_value = 20.0 + assert _write_with_deadline(42, b"hello", 10.0) is False + + @patch(f"{MOD}.select.select", return_value=([], [], [])) + @patch(f"{MOD}.os.write", side_effect=BlockingIOError()) + @patch(f"{MOD}.time.time") + def test_blocked_until_deadline( + self, + mock_time: MagicMock, + mock_write: MagicMock, + mock_select: MagicMock, + ) -> None: + """A printer that never accepts data times out instead of hanging.""" + mock_time.side_effect = [0.0, 1.0, 20.0] + assert _write_with_deadline(42, b"hello", 10.0) is False + + @patch(f"{MOD}.select.select", return_value=([], [42], [])) + @patch(f"{MOD}.os.write") + @patch(f"{MOD}.time.time", return_value=0.0) + def test_blocked_then_writable( + self, + t: MagicMock, + mock_write: MagicMock, + mock_select: MagicMock, + ) -> None: + mock_write.side_effect = [BlockingIOError(), 5] + assert _write_with_deadline(42, b"hello", 10.0) is True + + @patch(f"{MOD}.os.write", side_effect=OSError("gone")) + @patch(f"{MOD}.time.time", return_value=0.0) + def test_oserror_gives_up(self, t: MagicMock, mock_write: MagicMock) -> None: + assert _write_with_deadline(42, b"hello", 10.0) is False class TestWaitForPjlResponse: @@ -155,8 +192,7 @@ class TestWaitForPjlResponse: mock_time.side_effect = [0.0, 0.5, 1.0] mock_select.return_value = ([42], [], []) mock_read.return_value = b"CODE=10001" - result = _wait_for_pjl_response(42, 0, 5.0) - assert b"CODE=10001" in result + assert b"CODE=10001" in _wait_for_pjl_response(42, 5.0) @patch(f"{MOD}._read_nonblocking") @patch(f"{MOD}.select.select") @@ -170,8 +206,7 @@ class TestWaitForPjlResponse: mock_time.side_effect = [0.0, 0.5, 1.0] mock_select.return_value = ([42], [], []) mock_read.return_value = b"@PJL INFO" - result = _wait_for_pjl_response(42, 0, 5.0) - assert b"@PJL" in result + assert b"@PJL" in _wait_for_pjl_response(42, 5.0) @patch(f"{MOD}.select.select") @patch(f"{MOD}.time.time") @@ -181,8 +216,7 @@ class TestWaitForPjlResponse: mock_select: MagicMock, ) -> None: mock_time.side_effect = [10.0, 11.0] - result = _wait_for_pjl_response(42, 0, 5.0) - assert result == b"" + assert _wait_for_pjl_response(42, 5.0) == b"" @patch(f"{MOD}._read_nonblocking") @patch(f"{MOD}.select.select") @@ -195,8 +229,7 @@ class TestWaitForPjlResponse: ) -> None: mock_time.side_effect = [0.0, 0.5, 6.0] mock_select.return_value = ([], [], []) - result = _wait_for_pjl_response(42, 0, 5.0) - assert result == b"" + assert _wait_for_pjl_response(42, 5.0) == b"" @patch(f"{MOD}._read_nonblocking") @patch(f"{MOD}.select.select") @@ -209,8 +242,7 @@ class TestWaitForPjlResponse: ) -> None: """Inner remaining check triggers break.""" mock_time.side_effect = [0.0, 6.0, 6.0] - result = _wait_for_pjl_response(42, 0, 5.0) - assert result == b"" + assert _wait_for_pjl_response(42, 5.0) == b"" mock_select.assert_not_called() @patch(f"{MOD}._read_nonblocking") @@ -222,30 +254,38 @@ class TestWaitForPjlResponse: mock_select: MagicMock, mock_read: MagicMock, ) -> None: - """Data read but no '=' or '@PJL' → continues loop then times out.""" + """Data read but no '=' or '@PJL' -> continues loop then times out.""" mock_time.side_effect = [0.0, 0.5, 1.0, 6.0] mock_select.return_value = ([42], [], []) mock_read.return_value = b"garbage" - result = _wait_for_pjl_response(42, 0, 5.0) - assert result == b"garbage" + assert _wait_for_pjl_response(42, 5.0) == b"garbage" class TestPjlQuery: @patch(f"{MOD}._wait_for_pjl_response") - @patch(f"{MOD}.os.write") - @patch(f"{MOD}.fcntl.fcntl") + @patch(f"{MOD}._write_with_deadline", return_value=True) @patch(f"{MOD}.time.time", return_value=100.0) def test_query( self, t: MagicMock, - mock_fcntl: MagicMock, mock_write: MagicMock, mock_wait: MagicMock, ) -> None: - mock_fcntl.return_value = 0 mock_wait.return_value = b"CODE=10001" - result = pjl_query(42, "@PJL INFO STATUS") - assert "CODE=10001" in result + assert "CODE=10001" in pjl_query(42, "@PJL INFO STATUS") + + @patch(f"{MOD}._wait_for_pjl_response") + @patch(f"{MOD}._write_with_deadline", return_value=False) + @patch(f"{MOD}.time.time", return_value=100.0) + def test_write_timeout_returns_empty( + self, + t: MagicMock, + mock_write: MagicMock, + mock_wait: MagicMock, + ) -> None: + """If we cannot even send the command, do not wait for a reply.""" + assert pjl_query(42, "@PJL INFO STATUS") == "" + mock_wait.assert_not_called() class TestParseStatus: @@ -331,8 +371,8 @@ class TestRunPjlQueries: @patch(f"{MOD}._retry_pjl_query") @patch(f"{MOD}.time.sleep") @patch(f"{MOD}._drain_buffer") - @patch(f"{MOD}.os.write") - def test_runs_both_queries( + @patch(f"{MOD}._write_with_deadline") + def test_runs_status_pagecount_and_variables( self, mock_write: MagicMock, d: MagicMock, @@ -341,7 +381,27 @@ class TestRunPjlQueries: ) -> None: result = USBResult() _run_pjl_queries(42, result, 2) - assert mock_retry.call_count == 2 + assert mock_retry.call_count == 3 + queried = [call.args[1] for call in mock_retry.call_args_list] + assert "@PJL INFO PAGECOUNT" in queried + + +class TestParsePagecount: + def test_found(self) -> None: + result = USBResult() + assert _parse_pagecount("PAGECOUNT=2014\n", result) is True + assert result.page_count == "2014" + + def test_not_found(self) -> None: + result = USBResult() + assert _parse_pagecount("nothing\n", result) is False + assert result.page_count == "" + + def test_non_numeric_rejected(self) -> None: + """A garbled reply must not become a page count.""" + result = USBResult() + assert _parse_pagecount("PAGECOUNT=???\n", result) is False + assert result.page_count == "" class TestInitUsbResult: @@ -366,14 +426,19 @@ class TestQueryUsbPjl: patch(f"{MOD}.find_usb_printer_dev", return_value="/dev/usb/lp0"), patch(f"{MOD}._init_usb_result") as mock_init, patch(f"{MOD}.os.access", return_value=True), - patch(f"{MOD}.os.open", return_value=10), - patch(f"{MOD}.fcntl.fcntl", return_value=0), + patch(f"{MOD}.os.open", return_value=10) as mock_open, patch(f"{MOD}._run_pjl_queries"), patch(f"{MOD}.os.close"), ): - mock_init.return_value = USBResult(device="/dev/usb/lp0") + mock_init.return_value = USBResult( + device="/dev/usb/lp0", + status_code="10001", + ) result = query_usb_pjl() assert result.device == "/dev/usb/lp0" + assert result.error == "" + # The whole point: a blocking open() hangs on an unwell printer. + assert mock_open.call_args[0][1] & os.O_NONBLOCK @patch(f"{MOD}.find_usb_printer_dev", return_value=None) def test_no_dev_falls_back_to_cups(self, f: MagicMock) -> None: @@ -397,18 +462,46 @@ class TestQueryUsbPjl: result = query_usb_pjl() assert "Permission denied" in result.error - def test_oserror_on_open(self) -> None: + def test_silent_printer_is_reported_not_hidden(self) -> None: + """A printer that answers nothing must say so, not look healthy.""" with ( patch(f"{MOD}.find_usb_printer_dev", return_value="/dev/usb/lp0"), patch(f"{MOD}._init_usb_result") as mock_init, patch(f"{MOD}.os.access", return_value=True), patch(f"{MOD}.os.open", return_value=10), - patch(f"{MOD}.fcntl.fcntl", side_effect=OSError("bad fd")), + patch(f"{MOD}._run_pjl_queries"), patch(f"{MOD}.os.close"), ): mock_init.return_value = USBResult(device="/dev/usb/lp0") result = query_usb_pjl() - assert result.error != "" + assert "did not answer" in result.error + + def test_busy_device_explains_why(self) -> None: + """EBUSY means CUPS holds the printer mid-job; say that.""" + exc = OSError("busy") + exc.errno = errno.EBUSY + with ( + patch(f"{MOD}.find_usb_printer_dev", return_value="/dev/usb/lp0"), + patch(f"{MOD}._init_usb_result") as mock_init, + patch(f"{MOD}.os.access", return_value=True), + patch(f"{MOD}.os.open", side_effect=exc), + ): + mock_init.return_value = USBResult(device="/dev/usb/lp0") + result = query_usb_pjl() + assert "busy" in result.error.lower() + + def test_eacces_suggests_sudo(self) -> None: + exc = OSError("denied") + exc.errno = errno.EACCES + with ( + patch(f"{MOD}.find_usb_printer_dev", return_value="/dev/usb/lp0"), + patch(f"{MOD}._init_usb_result") as mock_init, + patch(f"{MOD}.os.access", return_value=True), + patch(f"{MOD}.os.open", side_effect=exc), + ): + mock_init.return_value = USBResult(device="/dev/usb/lp0") + result = query_usb_pjl() + assert "sudo" in result.error @patch(f"{MOD}.os.open", side_effect=OSError("no device")) @patch(f"{MOD}.os.access", return_value=True) @@ -421,7 +514,7 @@ class TestQueryUsbPjl: a: MagicMock, o: MagicMock, ) -> None: - """os.open raises OSError before fd is set → fd stays None.""" + """os.open raises OSError before fd is set -> fd stays None.""" mock_init.return_value = USBResult(device="/dev/usb/lp0") result = query_usb_pjl() - assert result.error == "no device" + assert "no device" in result.error diff --git a/python_pkg/brother_printer/usb_query.py b/python_pkg/brother_printer/usb_query.py index 23983a26..238dbed4 100644 --- a/python_pkg/brother_printer/usb_query.py +++ b/python_pkg/brother_printer/usb_query.py @@ -3,7 +3,7 @@ from __future__ import annotations import contextlib -import fcntl +import errno import importlib import logging import os @@ -48,31 +48,62 @@ def find_usb_printer_dev() -> str | None: # ── PJL over USB ───────────────────────────────────────────────────── +# The printer's fd is opened and kept non-blocking. usblp blocks indefinitely +# on a plain open() and on writes whenever the printer is not ready to talk - +# stuck mid-job, wedged, or simply asleep - so a blocking fd turns "the printer +# is unwell" into "this tool hangs forever", which is the worst possible answer +# when a status report is exactly what you wanted. + + def _drain_buffer(fd: int) -> None: """Read and discard any stale data from the USB buffer.""" - flags = fcntl.fcntl(fd, fcntl.F_GETFL) - fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) with contextlib.suppress(OSError): while os.read(fd, 4096): pass - fcntl.fcntl(fd, fcntl.F_SETFL, flags & ~os.O_NONBLOCK) -def _read_nonblocking(fd: int, flags: int) -> bytes: - """Read all available data from fd in non-blocking mode.""" +def _write_with_deadline(fd: int, data: bytes, deadline: float) -> bool: + """Write all of data to a non-blocking fd. Returns False if the deadline passes. + + Args: + fd: Non-blocking file descriptor for the printer. + data: Bytes to send. + deadline: Absolute monotonic-ish time (time.time()) to give up at. + + Returns: + True when everything was written, False on timeout or write error. + """ + sent = 0 + while sent < len(data): + remaining = deadline - time.time() + if remaining <= 0: + return False + try: + sent += os.write(fd, data[sent:]) + except BlockingIOError: + # Printer's buffer is full or it is not accepting data: wait for it + # to become writable rather than spinning or blocking forever. + _, writable, _ = select.select([], [fd], [], min(remaining, 0.5)) + if not writable: + continue + except OSError: + return False + return True + + +def _read_nonblocking(fd: int) -> bytes: + """Read all currently available data from a non-blocking fd.""" data = b"" - fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) with contextlib.suppress(OSError): while True: chunk = os.read(fd, 4096) if not chunk: break data += chunk - fcntl.fcntl(fd, fcntl.F_SETFL, flags & ~os.O_NONBLOCK) return data -def _wait_for_pjl_response(fd: int, flags: int, deadline: float) -> bytes: +def _wait_for_pjl_response(fd: int, deadline: float) -> bytes: """Poll fd until PJL data arrives or deadline expires.""" response = b"" while time.time() < deadline: @@ -81,24 +112,30 @@ def _wait_for_pjl_response(fd: int, flags: int, deadline: float) -> bytes: break readable, _, _ = select.select([fd], [], [], min(remaining, 1.0)) if readable: - response += _read_nonblocking(fd, flags) + response += _read_nonblocking(fd) if response and (b"=" in response or b"@PJL" in response): break return response def pjl_query(fd: int, cmd: str, timeout_sec: float = 5.0) -> str: - """Send a PJL command via raw fd and read the response.""" - flags = fcntl.fcntl(fd, fcntl.F_GETFL) - fcntl.fcntl(fd, fcntl.F_SETFL, flags & ~os.O_NONBLOCK) + """Send a PJL command via raw fd and read the response. - pjl_cmd = f"\x1b%-12345X@PJL\r\n{cmd}\r\n\x1b%-12345X" - os.write(fd, pjl_cmd.encode()) + Args: + fd: Non-blocking file descriptor for the printer. + cmd: PJL command, e.g. "@PJL INFO STATUS". + timeout_sec: Total budget for sending the command and reading the reply. + Returns: + The printer's reply, or "" if it did not answer in time. A silent + printer is a normal outcome here, not an error: it happens whenever the + printer is busy digesting a job. + """ deadline = time.time() + timeout_sec - response = _wait_for_pjl_response(fd, flags, deadline) - - fcntl.fcntl(fd, fcntl.F_SETFL, flags & ~os.O_NONBLOCK) + pjl_cmd = f"\x1b%-12345X@PJL\r\n{cmd}\r\n\x1b%-12345X" + if not _write_with_deadline(fd, pjl_cmd.encode(), deadline): + return "" + response = _wait_for_pjl_response(fd, deadline) return response.decode("ascii", errors="replace") @@ -117,6 +154,18 @@ def _parse_status(resp: str, result: USBResult) -> bool: return found +def _parse_pagecount(resp: str, result: USBResult) -> bool: + """Parse PAGECOUNT response into result. Returns True if a count was found.""" + for raw_line in resp.splitlines(): + stripped = raw_line.strip() + if stripped.startswith("PAGECOUNT="): + value = stripped.split("=", 1)[1].strip() + if value.isdigit(): + result.page_count = value + return True + return False + + def _parse_variables(resp: str, result: USBResult) -> bool: """Parse VARIABLES response into result. Returns True if data found.""" found = False @@ -149,13 +198,18 @@ def _run_pjl_queries(fd: int, result: USBResult, max_retries: int) -> None: """Execute PJL query sequence on an open file descriptor.""" _drain_buffer(fd) - os.write(fd, b"\x1b%-12345X@PJL\r\n\x1b%-12345X") + # Universal Exit Language, to drop the printer out of any language mode it + # is in and have it listen for PJL. Deadline-bounded like every other write. + _write_with_deadline(fd, b"\x1b%-12345X@PJL\r\n\x1b%-12345X", time.time() + 5.0) time.sleep(0.5) _drain_buffer(fd) _retry_pjl_query(fd, "@PJL INFO STATUS", _parse_status, result, max_retries) _drain_buffer(fd) time.sleep(0.5) + _retry_pjl_query(fd, "@PJL INFO PAGECOUNT", _parse_pagecount, result, max_retries) + _drain_buffer(fd) + time.sleep(0.5) _retry_pjl_query(fd, "@PJL INFO VARIABLES", _parse_variables, result, max_retries) @@ -185,12 +239,29 @@ def query_usb_pjl(max_retries: int = 2) -> USBResult: fd: int | None = None try: - fd = os.open(dev_path, os.O_RDWR) - fcntl.fcntl(fd, fcntl.F_GETFL) + # O_NONBLOCK or this open() hangs forever on an unwell printer. + fd = os.open(dev_path, os.O_RDWR | os.O_NONBLOCK) _run_pjl_queries(fd, result, max_retries) except OSError as e: - result.error = str(e) + result.error = _describe_open_error(dev_path, e) finally: if fd is not None: os.close(fd) + if not result.status_code and not result.error: + result.error = ( + f"The printer did not answer PJL on {dev_path}. It is usually busy" + " with a job, or wedged - power-cycle it if this persists." + ) return result + + +def _describe_open_error(dev_path: str, exc: OSError) -> str: + """Turn an errno from opening the printer into something actionable.""" + if exc.errno == errno.EBUSY: + return ( + f"{dev_path} is busy: another process (usually the CUPS backend" + " mid-job) holds the printer. Wait for the job to finish." + ) + if exc.errno == errno.EACCES: + return f"Permission denied: {dev_path}. Run with sudo." + return f"Could not read {dev_path}: {exc}"