From 69242e4bb3f2f706fde7eee5e4b46adcf7cf5b0d Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Fri, 17 Jul 2026 15:06:25 +0200 Subject: [PATCH] fix(brother-printer): split oversized files, fix order-dependent entry test a12e696 turned CI red two ways, both missed locally because the suite was run for the brother_printer package alone while CI runs everything in a random order. The repo caps Python files at 500 lines and cups_service.py had reached 659. Rather than cut it arbitrarily, extract consumable life tracking - page counters, replacement baselines, the scale migration, dropped-page detection - into consumables.py. The seam is real: none of it depends on CUPS service control, and cups_service needs only estimate_consumable_life for its fallback query, so the dependency runs one way. The test files are split along the same seams. test_main_entry passed or failed depending on execution order: runpy warns when the target module is already in sys.modules, a sibling test imports it, and pytest is configured with filterwarnings=error. Drop the module before running it so the test no longer depends on who ran first. Structural only; no behaviour change. 692 tests, 100% branch coverage. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01RG3eEn8rwZbcXaJrKJxdv8 --- ...-printer-file-length-split-2026-07-17.json | 16 + ...-printer-file-length-split-2026-07-17.json | 52 +++ .../brother_printer/check_brother_printer.py | 2 +- python_pkg/brother_printer/consumables.py | 279 ++++++++++++ python_pkg/brother_printer/cups_service.py | 263 +----------- python_pkg/brother_printer/display.py | 8 +- .../brother_printer/tests/test_consumables.py | 272 ++++++++++++ .../tests/test_consumables_part2.py | 268 ++++++++++++ .../tests/test_cups_service.py | 111 ----- .../tests/test_cups_service_part4.py | 406 +----------------- .../brother_printer/tests/test_display.py | 146 ------- .../tests/test_display_part3.py | 159 +++++++ .../brother_printer/tests/test_main_entry.py | 4 + .../brother_printer/tests/test_usb_query.py | 192 --------- .../tests/test_usb_query_part2.py | 207 +++++++++ 15 files changed, 1268 insertions(+), 1117 deletions(-) create mode 100644 docs/superpowers/contracts/brother-printer-file-length-split-2026-07-17.json create mode 100644 docs/superpowers/evidence/brother-printer-file-length-split-2026-07-17.json create mode 100644 python_pkg/brother_printer/consumables.py create mode 100644 python_pkg/brother_printer/tests/test_consumables.py create mode 100644 python_pkg/brother_printer/tests/test_consumables_part2.py create mode 100644 python_pkg/brother_printer/tests/test_display_part3.py create mode 100644 python_pkg/brother_printer/tests/test_usb_query_part2.py diff --git a/docs/superpowers/contracts/brother-printer-file-length-split-2026-07-17.json b/docs/superpowers/contracts/brother-printer-file-length-split-2026-07-17.json new file mode 100644 index 00000000..cf233068 --- /dev/null +++ b/docs/superpowers/contracts/brother-printer-file-length-split-2026-07-17.json @@ -0,0 +1,16 @@ +{ + "title": "brother_printer: satisfy the 500-line limit and make the entry-point test order-independent", + "objective": "The previous commit (a12e696) turned CI red: four brother_printer files grew past the repo's 500-line ceiling, and the __main__ entry-point test only passed when it ran before the test that imports the module. Both were missed locally because the suite was run for the brother_printer package alone, while the file-length test scans the whole repo and CI randomises test order. Split the oversized files along real seams rather than arbitrary cuts, and make the entry-point test independent of ordering. Success: the full suite passes in any order with 100% coverage, and no file this change owns exceeds 500 lines.", + "acceptance_criteria": [ + "No file owned by this change exceeds 500 lines", + "Consumable life tracking lives in its own module, split at a real responsibility boundary and with no import cycle", + "test_main_entry passes regardless of test execution order, with pytest's filterwarnings=error in force", + "The full python_pkg suite passes with 100% statement and branch coverage", + "No behaviour change: this is a structural split plus one test fix" + ], + "out_of_scope": [ + "linux_configuration/.../focus_mode_daemon.py (509 lines), which breached the limit in commit 22e22c1 and is not owned by this change", + "Any change to the reporting behaviour verified against the printer in a12e696" + ], + "verifier": "python3 -m pytest python_pkg/ (full suite, random order, 100% coverage); wc -l over tracked *.py; ./run.sh against the real HL-1110; pre-commit run --files " +} diff --git a/docs/superpowers/evidence/brother-printer-file-length-split-2026-07-17.json b/docs/superpowers/evidence/brother-printer-file-length-split-2026-07-17.json new file mode 100644 index 00000000..8936e8fb --- /dev/null +++ b/docs/superpowers/evidence/brother-printer-file-length-split-2026-07-17.json @@ -0,0 +1,52 @@ +{ + "intent": "Fix the CI failure introduced by a12e696: bring files back under the repo's 500-line limit and make the entry-point test order-independent, with no behaviour change.", + "scope": [ + "python_pkg/brother_printer/consumables.py: new module holding consumable life tracking", + "python_pkg/brother_printer/cups_service.py: 659 -> 414 lines", + "python_pkg/brother_printer/tests/: split test_consumables, test_display_part3, test_usb_query_part2", + "python_pkg/brother_printer/tests/test_main_entry.py: order-independence fix", + "Non-goal: any behaviour change; a12e696's verified reporting is untouched", + "Non-goal: focus_mode_daemon.py (509 lines, breached in 22e22c1, different package)" + ], + "changes": [ + "Extract page counters, replacement baselines, the scale migration, and dropped-page detection from cups_service.py into consumables.py. The seam is real: none of it depends on CUPS service control, and cups_service only needs estimate_consumable_life for its fallback query, so the dependency runs one way with no cycle. cups_service.py 659 -> 414 lines, consumables.py 281.", + "Split the test files along the same seams: test_consumables.py (state/estimates/resets) and test_consumables_part2.py (counter scales and dropped pages); move page-count and dropped-page display tests to test_display_part3.py; move the PJL wire-protocol tests to test_usb_query_part2.py.", + "test_main_entry: pop python_pkg.brother_printer.__main__ from sys.modules before runpy.run_module. runpy emits a RuntimeWarning when the module is already imported, pytest is configured with filterwarnings=error, and the suite runs in random order - so the test passed or failed depending on whether the sibling import test ran first." + ], + "verification": [ + { + "command": "python3 -m pytest python_pkg/ (full suite, as CI runs it)", + "result": "pass", + "evidence": "692 passed; Required test coverage of 100.0% reached. Total coverage: 100.00%. The only local failure is test_file_length hitting a UnicodeDecodeError on an untracked .ci-mirror-venv/ file absent from CI's checkout." + }, + { + "command": "pytest test_main_entry ordered import-first (the order CI hit)", + "result": "pass", + "evidence": "Previously reproduced the CI failure exactly: RuntimeWarning: '...__main__' found in sys.modules after import of package. Now passes in either order." + }, + { + "command": "wc -l over tracked *.py", + "result": "pass", + "evidence": "No file owned by this change exceeds 500 lines. Only linux_configuration/.../focus_mode_daemon.py (509) remains, from commit 22e22c1." + }, + { + "command": "./run.sh against the real Brother HL-1110", + "result": "pass", + "evidence": "Unchanged after the split: '. Sleep (standby)' CODE=40000, total 2087 pages, toner 68%, drum 80%." + }, + { + "command": "pre-commit run --files ", + "result": "pass", + "evidence": "All hooks pass (ruff, ruff format, mypy, pylint, bandit, coverage enforcement)." + } + ], + "risks": [ + "Import churn: display.py and check_brother_printer.py now import consumables directly. Verified no cycle, and the full suite plus a real run exercise both paths.", + "The split is structural only; no logic moved between functions. Coverage remained at 100% throughout, so no branch lost its test.", + "CI will stay red until focus_mode_daemon.py (509 lines) is addressed. That breach predates this change and belongs to another package; it lay latent because the Python-tests workflow only triggers on python_pkg/** changes." + ], + "rollback": [ + "git revert restores the single-module layout; a12e696's behaviour is unaffected either way.", + "After rollback, validate with ./run.sh and expect the 500-line test to fail again on cups_service.py." + ] +} diff --git a/python_pkg/brother_printer/check_brother_printer.py b/python_pkg/brother_printer/check_brother_printer.py index ae8ad516..1404f510 100644 --- a/python_pkg/brother_printer/check_brother_printer.py +++ b/python_pkg/brother_printer/check_brother_printer.py @@ -35,7 +35,7 @@ import sys from python_pkg.brother_printer._query import run_command_text from python_pkg.brother_printer.constants import CYAN, RED, RESET, _out -from python_pkg.brother_printer.cups_service import reset_consumable +from python_pkg.brother_printer.consumables import reset_consumable from python_pkg.brother_printer.display import ( display_network_results, display_usb_results, diff --git a/python_pkg/brother_printer/consumables.py b/python_pkg/brother_printer/consumables.py new file mode 100644 index 00000000..1b0d60e5 --- /dev/null +++ b/python_pkg/brother_printer/consumables.py @@ -0,0 +1,279 @@ +"""Consumable life tracking: page counters, replacement baselines, dropped pages. + +Everything here hangs off one hard-won fact: CUPS cannot be trusted to say what +the printer actually did. Its page log only sees jobs it printed itself, and it +counts pages the printer silently discarded - it logged "total 50" for a job +that produced 3 sheets. Only the printer's own @PJL INFO PAGECOUNT is +authoritative, so replacement baselines are kept on that scale and the gap +between the two counters is what exposes dropped pages. +""" + +from __future__ import annotations + +import json +import logging +from pathlib import Path +import re + +from python_pkg.brother_printer.constants import ( + CONSUMABLE_STATE_DIR, + CUPS_PAGE_LOG_PATH, + DRUM_RATED_PAGES, + GREEN, + PAGE_DROP_WARN_THRESHOLD, + RESET, + TONER_RATED_PAGES, + _out, +) +from python_pkg.brother_printer.data_classes import ( + PageCountEstimate, + PageDeliveryCheck, +) + +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 + + +# ── Consumable state management ────────────────────────────────────── + + +def _get_cups_total_pages() -> int: + """Parse CUPS page_log to get total pages printed (deduplicated by job).""" + if not CUPS_PAGE_LOG.exists(): + return 0 + try: + text = CUPS_PAGE_LOG.read_text(encoding="utf-8", errors="replace") + except OSError: + return 0 + jobs: dict[str, int] = {} + for line in text.splitlines(): + match = re.search(r"\s(\d+)\s+\[.*?\]\s+total\s+(\d+)", line) + if match: + job_id = match.group(1) + pages = int(match.group(2)) + jobs[job_id] = max(jobs.get(job_id, 0), pages) + return sum(jobs.values()) + + +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, + "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: + data = json.loads( + CONSUMABLE_STATE_FILE.read_text(encoding="utf-8"), + ) + 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) + CONSUMABLE_STATE_FILE.write_text( + json.dumps(state, indent=2) + "\n", + encoding="utf-8", + ) + + +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 _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() + 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)) + drum_pct = max(0, 100 - (drum_pages * 100 // DRUM_RATED_PAGES)) + return PageCountEstimate( + total_pages=total, + toner_pages=toner_pages, + 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, + ) diff --git a/python_pkg/brother_printer/cups_service.py b/python_pkg/brother_printer/cups_service.py index 68ae5efd..4e405f84 100644 --- a/python_pkg/brother_printer/cups_service.py +++ b/python_pkg/brother_printer/cups_service.py @@ -1,11 +1,13 @@ -"""CUPS service management, USB fallback, and consumable state tracking.""" +"""CUPS service management, USB port status, and the CUPS-based fallback query. + +Consumable life tracking lives in consumables.py. +""" from __future__ import annotations import contextlib import fcntl import importlib -import json import logging import os from pathlib import Path @@ -24,21 +26,12 @@ from python_pkg.brother_printer.constants import ( _CUPS_STATE_TO_STATUS, _ERROR_REASON_MAP, 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.consumables import estimate_consumable_life from python_pkg.brother_printer.data_classes import ( - PageCountEstimate, - PageDeliveryCheck, USBPortStatus, USBResult, ) @@ -48,15 +41,6 @@ if TYPE_CHECKING: 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" @@ -250,243 +234,6 @@ def _query_usb_port_status_raw(cups_state: str = "") -> USBPortStatus | None: return None -# ── Consumable state management ────────────────────────────────────── - - -def _get_cups_total_pages() -> int: - """Parse CUPS page_log to get total pages printed (deduplicated by job).""" - if not CUPS_PAGE_LOG.exists(): - return 0 - try: - text = CUPS_PAGE_LOG.read_text(encoding="utf-8", errors="replace") - except OSError: - return 0 - jobs: dict[str, int] = {} - for line in text.splitlines(): - match = re.search(r"\s(\d+)\s+\[.*?\]\s+total\s+(\d+)", line) - if match: - job_id = match.group(1) - pages = int(match.group(2)) - jobs[job_id] = max(jobs.get(job_id, 0), pages) - return sum(jobs.values()) - - -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, - "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: - data = json.loads( - CONSUMABLE_STATE_FILE.read_text(encoding="utf-8"), - ) - 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) - CONSUMABLE_STATE_FILE.write_text( - json.dumps(state, indent=2) + "\n", - encoding="utf-8", - ) - - -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 _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() - 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)) - drum_pct = max(0, 100 - (drum_pages * 100 // DRUM_RATED_PAGES)) - return PageCountEstimate( - total_pages=total, - toner_pages=toner_pages, - 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, - ) - - # ── IPP / CUPS attribute queries ──────────────────────────────────── diff --git a/python_pkg/brother_printer/display.py b/python_pkg/brother_printer/display.py index 20dd2edd..bfdd3319 100644 --- a/python_pkg/brother_printer/display.py +++ b/python_pkg/brother_printer/display.py @@ -22,14 +22,14 @@ from python_pkg.brother_printer.constants import ( _out, get_status_info, ) +from python_pkg.brother_printer.consumables import ( + check_page_delivery, + estimate_consumable_life, +) from python_pkg.brother_printer.cups_queue import ( display_cups_queue_status, get_cups_queue_status, ) -from python_pkg.brother_printer.cups_service import ( - check_page_delivery, - estimate_consumable_life, -) from python_pkg.brother_printer.data_classes import ( NetworkResult, SupplyStatus, diff --git a/python_pkg/brother_printer/tests/test_consumables.py b/python_pkg/brother_printer/tests/test_consumables.py new file mode 100644 index 00000000..ea79b44f --- /dev/null +++ b/python_pkg/brother_printer/tests/test_consumables.py @@ -0,0 +1,272 @@ +"""Tests for brother_printer.consumables - state, estimates, resets.""" + +from __future__ import annotations + +import json +from unittest.mock import MagicMock, patch + +from python_pkg.brother_printer.consumables import ( + STATE_SCHEMA_CUPS_SCALE, + STATE_SCHEMA_PRINTER_SCALE, + _get_cups_total_pages, + _load_consumable_state, + _save_consumable_state, + estimate_consumable_life, + reset_consumable, +) + +MOD = "python_pkg.brother_printer.consumables" + + +class TestGetCupsTotalPages: + @patch(f"{MOD}.CUPS_PAGE_LOG") + def test_no_log(self, mock_log: MagicMock) -> None: + mock_log.exists.return_value = False + assert _get_cups_total_pages() == 0 + + @patch(f"{MOD}.CUPS_PAGE_LOG") + def test_with_entries(self, mock_log: MagicMock) -> None: + mock_log.exists.return_value = True + mock_log.read_text.return_value = ( + "printer 1 [2025-01-01] total 5\n" + "printer 2 [2025-01-01] total 3\n" + "printer 1 [2025-01-01] total 10\n" + ) + assert _get_cups_total_pages() == 13 # max(5,10) + 3 + + @patch(f"{MOD}.CUPS_PAGE_LOG") + def test_oserror(self, mock_log: MagicMock) -> None: + mock_log.exists.return_value = True + mock_log.read_text.side_effect = OSError("fail") + assert _get_cups_total_pages() == 0 + + @patch(f"{MOD}.CUPS_PAGE_LOG") + def test_no_matching_lines(self, mock_log: MagicMock) -> None: + mock_log.exists.return_value = True + mock_log.read_text.return_value = "some garbage\n" + assert _get_cups_total_pages() == 0 + + +class TestLoadConsumableState: + @patch(f"{MOD}.CONSUMABLE_STATE_FILE") + 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, + "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: + mock_file.exists.return_value = True + mock_file.read_text.return_value = json.dumps( + {"toner_replaced_at": 100, "drum_replaced_at": 200}, + ) + result = _load_consumable_state() + assert result["toner_replaced_at"] == 100 + assert result["drum_replaced_at"] == 200 + + @patch(f"{MOD}.CONSUMABLE_STATE_FILE") + def test_oserror(self, mock_file: MagicMock) -> None: + mock_file.exists.return_value = True + mock_file.read_text.side_effect = OSError("fail") + result = _load_consumable_state() + assert result["toner_replaced_at"] == 0 + + @patch(f"{MOD}.CONSUMABLE_STATE_FILE") + def test_bad_json(self, mock_file: MagicMock) -> None: + mock_file.exists.return_value = True + mock_file.read_text.return_value = "not json" + result = _load_consumable_state() + assert result["toner_replaced_at"] == 0 + + @patch(f"{MOD}.CONSUMABLE_STATE_FILE") + def test_bad_values(self, mock_file: MagicMock) -> None: + mock_file.exists.return_value = True + mock_file.read_text.return_value = json.dumps( + {"toner_replaced_at": "bad"}, + ) + result = _load_consumable_state() + assert result["toner_replaced_at"] == 0 + + +class TestSaveConsumableState: + @patch(f"{MOD}.CONSUMABLE_STATE_FILE") + def test_saves(self, mock_file: MagicMock) -> None: + mock_file.parent = MagicMock() + _save_consumable_state({"toner_replaced_at": 100, "drum_replaced_at": 200}) + mock_file.write_text.assert_called_once() + written = mock_file.write_text.call_args[0][0] + data = json.loads(written) + assert data["toner_replaced_at"] == 100 + + +class TestResetConsumable: + @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=500) + def test_reset_toner( + self, + pages: MagicMock, + load: MagicMock, + mock_save: MagicMock, + out: MagicMock, + ) -> None: + load.return_value = {"toner_replaced_at": 0, "drum_replaced_at": 0} + reset_consumable("toner") + saved_state = mock_save.call_args[0][0] + assert saved_state["toner_replaced_at"] == 500 + + +class TestEstimateConsumableLife: + @patch(f"{MOD}._load_consumable_state") + @patch(f"{MOD}._get_cups_total_pages", return_value=0) + def test_no_pages(self, p: MagicMock, mock_load: MagicMock) -> None: + result = estimate_consumable_life() + assert result.total_pages == 0 + + @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, + "last_printer_count": 0, + "last_cups_total": 0, + } + result = estimate_consumable_life() + assert result.total_pages == 500 + assert result.toner_pct_remaining == 50 + assert result.toner_exhausted is False + assert result.toner_low is False + + @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, + "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, + "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, + "last_printer_count": 0, + "last_cups_total": 0, + } + result = estimate_consumable_life() + assert result.drum_near_end is True + + +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 diff --git a/python_pkg/brother_printer/tests/test_consumables_part2.py b/python_pkg/brother_printer/tests/test_consumables_part2.py new file mode 100644 index 00000000..31caac88 --- /dev/null +++ b/python_pkg/brother_printer/tests/test_consumables_part2.py @@ -0,0 +1,268 @@ +"""Tests for brother_printer.consumables - counter scales and dropped pages. + +The printer's counter and the CUPS page log measure different things. These +tests pin the rebasing between them and the dropped-page detection that the +gap makes possible. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +from python_pkg.brother_printer.consumables import ( + STATE_SCHEMA_CUPS_SCALE, + STATE_SCHEMA_PRINTER_SCALE, + _cups_total_on_printer_scale, + _migrate_state_to_printer_scale, + _snapshot_counters, + check_page_delivery, +) + +MOD = "python_pkg.brother_printer.consumables" + + +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 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_cups_service.py b/python_pkg/brother_printer/tests/test_cups_service.py index 1d165c3e..25395de0 100644 --- a/python_pkg/brother_printer/tests/test_cups_service.py +++ b/python_pkg/brother_printer/tests/test_cups_service.py @@ -2,23 +2,17 @@ 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, is_cups_scheduler_running, - reset_consumable, start_cups, ) from python_pkg.brother_printer.data_classes import USBPortStatus @@ -377,111 +371,6 @@ class TestCupsIsBusy: assert _cups_is_busy("") is False -class TestGetCupsTotalPages: - @patch(f"{MOD}.CUPS_PAGE_LOG") - def test_no_log(self, mock_log: MagicMock) -> None: - mock_log.exists.return_value = False - assert _get_cups_total_pages() == 0 - - @patch(f"{MOD}.CUPS_PAGE_LOG") - def test_with_entries(self, mock_log: MagicMock) -> None: - mock_log.exists.return_value = True - mock_log.read_text.return_value = ( - "printer 1 [2025-01-01] total 5\n" - "printer 2 [2025-01-01] total 3\n" - "printer 1 [2025-01-01] total 10\n" - ) - assert _get_cups_total_pages() == 13 # max(5,10) + 3 - - @patch(f"{MOD}.CUPS_PAGE_LOG") - def test_oserror(self, mock_log: MagicMock) -> None: - mock_log.exists.return_value = True - mock_log.read_text.side_effect = OSError("fail") - assert _get_cups_total_pages() == 0 - - @patch(f"{MOD}.CUPS_PAGE_LOG") - def test_no_matching_lines(self, mock_log: MagicMock) -> None: - mock_log.exists.return_value = True - mock_log.read_text.return_value = "some garbage\n" - assert _get_cups_total_pages() == 0 - - -class TestLoadConsumableState: - @patch(f"{MOD}.CONSUMABLE_STATE_FILE") - 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, - "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: - mock_file.exists.return_value = True - mock_file.read_text.return_value = json.dumps( - {"toner_replaced_at": 100, "drum_replaced_at": 200}, - ) - result = _load_consumable_state() - assert result["toner_replaced_at"] == 100 - assert result["drum_replaced_at"] == 200 - - @patch(f"{MOD}.CONSUMABLE_STATE_FILE") - def test_oserror(self, mock_file: MagicMock) -> None: - mock_file.exists.return_value = True - mock_file.read_text.side_effect = OSError("fail") - result = _load_consumable_state() - assert result["toner_replaced_at"] == 0 - - @patch(f"{MOD}.CONSUMABLE_STATE_FILE") - def test_bad_json(self, mock_file: MagicMock) -> None: - mock_file.exists.return_value = True - mock_file.read_text.return_value = "not json" - result = _load_consumable_state() - assert result["toner_replaced_at"] == 0 - - @patch(f"{MOD}.CONSUMABLE_STATE_FILE") - def test_bad_values(self, mock_file: MagicMock) -> None: - mock_file.exists.return_value = True - mock_file.read_text.return_value = json.dumps( - {"toner_replaced_at": "bad"}, - ) - result = _load_consumable_state() - assert result["toner_replaced_at"] == 0 - - -class TestSaveConsumableState: - @patch(f"{MOD}.CONSUMABLE_STATE_FILE") - def test_saves(self, mock_file: MagicMock) -> None: - mock_file.parent = MagicMock() - _save_consumable_state({"toner_replaced_at": 100, "drum_replaced_at": 200}) - mock_file.write_text.assert_called_once() - written = mock_file.write_text.call_args[0][0] - data = json.loads(written) - assert data["toner_replaced_at"] == 100 - - -class TestResetConsumable: - @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=500) - def test_reset_toner( - self, - pages: MagicMock, - load: MagicMock, - mock_save: MagicMock, - out: MagicMock, - ) -> None: - load.return_value = {"toner_replaced_at": 0, "drum_replaced_at": 0} - 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")) 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 ad3f4f89..38add8d0 100644 --- a/python_pkg/brother_printer/tests/test_cups_service_part4.py +++ b/python_pkg/brother_printer/tests/test_cups_service_part4.py @@ -1,4 +1,4 @@ -"""Tests for brother_printer.cups_service module - part 4 (consumable life, IPP).""" +"""Tests for brother_printer.cups_service module - part 4 (IPP queries).""" from __future__ import annotations @@ -6,80 +6,13 @@ 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" -class TestEstimateConsumableLife: - @patch(f"{MOD}._load_consumable_state") - @patch(f"{MOD}._get_cups_total_pages", return_value=0) - def test_no_pages(self, p: MagicMock, mock_load: MagicMock) -> None: - result = estimate_consumable_life() - assert result.total_pages == 0 - - @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, - "last_printer_count": 0, - "last_cups_total": 0, - } - result = estimate_consumable_life() - assert result.total_pages == 500 - assert result.toner_pct_remaining == 50 - assert result.toner_exhausted is False - assert result.toner_low is False - - @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, - "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, - "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, - "last_printer_count": 0, - "last_cups_total": 0, - } - result = estimate_consumable_life() - assert result.drum_near_end is True - - class TestParseIppAttributes: def test_parse(self) -> None: output = " printer-state (enum) = idle\n printer-name (name) = Brother\n" @@ -111,340 +44,3 @@ 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 02f040e3..9deec4f8 100644 --- a/python_pkg/brother_printer/tests/test_display.py +++ b/python_pkg/brother_printer/tests/test_display.py @@ -9,8 +9,6 @@ import pytest from python_pkg.brother_printer.data_classes import ( NetworkResult, - PageCountEstimate, - PageDeliveryCheck, SupplyReadings, USBPortStatus, USBResult, @@ -19,10 +17,7 @@ from python_pkg.brother_printer.display import ( _classify_percentage_level, _classify_supply_level, _collect_supply_items, - _display_consumables_reference, _display_cups_fallback_note, - _display_page_count_estimate, - _display_page_delivery_warning, _display_pjl_status, _display_report_header, _display_supply_levels, @@ -45,113 +40,6 @@ class TestDisplayReportHeader: assert "Brother Laser Printer" in out.getvalue() -class TestDisplayPageCountEstimate: - @patch(f"{MOD}.estimate_consumable_life") - def test_no_pages(self, mock_est: MagicMock) -> None: - mock_est.return_value = PageCountEstimate(total_pages=0) - with patch("sys.stdout", new_callable=StringIO) as out: - _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( - total_pages=100, - toner_pages=100, - drum_pages=100, - toner_pct_remaining=90, - drum_pct_remaining=99, - ) - with patch("sys.stdout", new_callable=StringIO) as out: - _display_page_count_estimate() - assert "Total pages" in out.getvalue() - - @patch(f"{MOD}.estimate_consumable_life") - def test_toner_exhausted(self, mock_est: MagicMock) -> None: - mock_est.return_value = PageCountEstimate( - total_pages=1000, - toner_pages=1000, - drum_pages=100, - toner_pct_remaining=0, - drum_pct_remaining=99, - toner_exhausted=True, - toner_low=True, - ) - with patch("sys.stdout", new_callable=StringIO) as out: - _display_page_count_estimate() - assert "REPLACE NOW" in out.getvalue() - - @patch(f"{MOD}.estimate_consumable_life") - def test_toner_low(self, mock_est: MagicMock) -> None: - mock_est.return_value = PageCountEstimate( - total_pages=800, - toner_pages=800, - drum_pages=100, - toner_pct_remaining=20, - drum_pct_remaining=99, - toner_low=True, - ) - with patch("sys.stdout", new_callable=StringIO) as out: - _display_page_count_estimate() - assert "order soon" in out.getvalue() - - @patch(f"{MOD}.estimate_consumable_life") - def test_drum_near_end(self, mock_est: MagicMock) -> None: - mock_est.return_value = PageCountEstimate( - total_pages=9000, - toner_pages=100, - drum_pages=9000, - toner_pct_remaining=90, - drum_pct_remaining=10, - drum_near_end=True, - ) - with patch("sys.stdout", new_callable=StringIO) as out: - _display_page_count_estimate() - assert "nearing end" in out.getvalue() - - -class TestDisplayConsumablesReference: - def test_prints(self) -> None: - with patch("sys.stdout", new_callable=StringIO) as out: - _display_consumables_reference() - assert "TN-1050" in out.getvalue() - - class TestDisplayUsbDeviceInfo: def test_full_info(self) -> None: r = USBResult( @@ -473,37 +361,3 @@ 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_display_part3.py b/python_pkg/brother_printer/tests/test_display_part3.py new file mode 100644 index 00000000..a46d5512 --- /dev/null +++ b/python_pkg/brother_printer/tests/test_display_part3.py @@ -0,0 +1,159 @@ +"""Tests for brother_printer.display - page counts and dropped-page warning.""" + +from __future__ import annotations + +from io import StringIO +from unittest.mock import MagicMock, patch + +from python_pkg.brother_printer.data_classes import ( + PageCountEstimate, + PageDeliveryCheck, +) +from python_pkg.brother_printer.display import ( + _display_consumables_reference, + _display_page_count_estimate, + _display_page_delivery_warning, +) + +MOD = "python_pkg.brother_printer.display" + + +class TestDisplayPageCountEstimate: + @patch(f"{MOD}.estimate_consumable_life") + def test_no_pages(self, mock_est: MagicMock) -> None: + mock_est.return_value = PageCountEstimate(total_pages=0) + with patch("sys.stdout", new_callable=StringIO) as out: + _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( + total_pages=100, + toner_pages=100, + drum_pages=100, + toner_pct_remaining=90, + drum_pct_remaining=99, + ) + with patch("sys.stdout", new_callable=StringIO) as out: + _display_page_count_estimate() + assert "Total pages" in out.getvalue() + + @patch(f"{MOD}.estimate_consumable_life") + def test_toner_exhausted(self, mock_est: MagicMock) -> None: + mock_est.return_value = PageCountEstimate( + total_pages=1000, + toner_pages=1000, + drum_pages=100, + toner_pct_remaining=0, + drum_pct_remaining=99, + toner_exhausted=True, + toner_low=True, + ) + with patch("sys.stdout", new_callable=StringIO) as out: + _display_page_count_estimate() + assert "REPLACE NOW" in out.getvalue() + + @patch(f"{MOD}.estimate_consumable_life") + def test_toner_low(self, mock_est: MagicMock) -> None: + mock_est.return_value = PageCountEstimate( + total_pages=800, + toner_pages=800, + drum_pages=100, + toner_pct_remaining=20, + drum_pct_remaining=99, + toner_low=True, + ) + with patch("sys.stdout", new_callable=StringIO) as out: + _display_page_count_estimate() + assert "order soon" in out.getvalue() + + @patch(f"{MOD}.estimate_consumable_life") + def test_drum_near_end(self, mock_est: MagicMock) -> None: + mock_est.return_value = PageCountEstimate( + total_pages=9000, + toner_pages=100, + drum_pages=9000, + toner_pct_remaining=90, + drum_pct_remaining=10, + drum_near_end=True, + ) + with patch("sys.stdout", new_callable=StringIO) as out: + _display_page_count_estimate() + assert "nearing end" in out.getvalue() + + +class TestDisplayConsumablesReference: + def test_prints(self) -> None: + with patch("sys.stdout", new_callable=StringIO) as out: + _display_consumables_reference() + assert "TN-1050" in out.getvalue() + + +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 cb0f6aaa..f74b252f 100644 --- a/python_pkg/brother_printer/tests/test_main_entry.py +++ b/python_pkg/brother_printer/tests/test_main_entry.py @@ -12,6 +12,10 @@ class TestMain: def test_main_called_when_run_as_module(self) -> None: """python3 -m python_pkg.brother_printer runs main().""" mock_main = MagicMock() + # runpy warns if __main__ is already imported, and the suite runs tests + # in a random order, so another test may have imported it first. The + # warning is an error here, so drop it and let runpy execute it fresh. + sys.modules.pop("python_pkg.brother_printer.__main__", None) with patch( "python_pkg.brother_printer.check_brother_printer.main", mock_main, diff --git a/python_pkg/brother_printer/tests/test_usb_query.py b/python_pkg/brother_printer/tests/test_usb_query.py index c00e40af..9300c287 100644 --- a/python_pkg/brother_printer/tests/test_usb_query.py +++ b/python_pkg/brother_printer/tests/test_usb_query.py @@ -8,19 +8,14 @@ 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, query_usb_pjl, ) @@ -101,193 +96,6 @@ class TestFindUsbPrinterDev: assert result is None -class TestDrainBuffer: - @patch(f"{MOD}.os.read") - 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") - 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() - - -class TestReadNonblocking: - @patch(f"{MOD}.os.read") - def test_reads_chunks(self, mock_read: MagicMock) -> None: - mock_read.side_effect = [b"hello", b"", OSError] - assert _read_nonblocking(42) == b"hello" - - @patch(f"{MOD}.os.read") - def test_oserror_suppressed(self, mock_read: MagicMock) -> None: - mock_read.side_effect = OSError("would block") - 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: - @patch(f"{MOD}._read_nonblocking") - @patch(f"{MOD}.select.select") - @patch(f"{MOD}.time.time") - def test_response_with_equals( - self, - mock_time: MagicMock, - mock_select: MagicMock, - mock_read: MagicMock, - ) -> None: - mock_time.side_effect = [0.0, 0.5, 1.0] - mock_select.return_value = ([42], [], []) - mock_read.return_value = b"CODE=10001" - assert b"CODE=10001" in _wait_for_pjl_response(42, 5.0) - - @patch(f"{MOD}._read_nonblocking") - @patch(f"{MOD}.select.select") - @patch(f"{MOD}.time.time") - def test_response_with_pjl( - self, - mock_time: MagicMock, - mock_select: MagicMock, - mock_read: MagicMock, - ) -> None: - mock_time.side_effect = [0.0, 0.5, 1.0] - mock_select.return_value = ([42], [], []) - mock_read.return_value = b"@PJL INFO" - assert b"@PJL" in _wait_for_pjl_response(42, 5.0) - - @patch(f"{MOD}.select.select") - @patch(f"{MOD}.time.time") - def test_timeout_no_data( - self, - mock_time: MagicMock, - mock_select: MagicMock, - ) -> None: - mock_time.side_effect = [10.0, 11.0] - assert _wait_for_pjl_response(42, 5.0) == b"" - - @patch(f"{MOD}._read_nonblocking") - @patch(f"{MOD}.select.select") - @patch(f"{MOD}.time.time") - def test_not_readable_then_timeout( - self, - mock_time: MagicMock, - mock_select: MagicMock, - mock_read: MagicMock, - ) -> None: - mock_time.side_effect = [0.0, 0.5, 6.0] - mock_select.return_value = ([], [], []) - assert _wait_for_pjl_response(42, 5.0) == b"" - - @patch(f"{MOD}._read_nonblocking") - @patch(f"{MOD}.select.select") - @patch(f"{MOD}.time.time") - def test_remaining_lte_zero( - self, - mock_time: MagicMock, - mock_select: MagicMock, - mock_read: MagicMock, - ) -> None: - """Inner remaining check triggers break.""" - mock_time.side_effect = [0.0, 6.0, 6.0] - assert _wait_for_pjl_response(42, 5.0) == b"" - mock_select.assert_not_called() - - @patch(f"{MOD}._read_nonblocking") - @patch(f"{MOD}.select.select") - @patch(f"{MOD}.time.time") - def test_response_no_eq_or_pjl( - self, - mock_time: MagicMock, - mock_select: MagicMock, - mock_read: MagicMock, - ) -> None: - """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" - assert _wait_for_pjl_response(42, 5.0) == b"garbage" - - -class TestPjlQuery: - @patch(f"{MOD}._wait_for_pjl_response") - @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_write: MagicMock, - mock_wait: MagicMock, - ) -> None: - mock_wait.return_value = b"CODE=10001" - 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: def test_found(self) -> None: result = USBResult() diff --git a/python_pkg/brother_printer/tests/test_usb_query_part2.py b/python_pkg/brother_printer/tests/test_usb_query_part2.py new file mode 100644 index 00000000..959c65b8 --- /dev/null +++ b/python_pkg/brother_printer/tests/test_usb_query_part2.py @@ -0,0 +1,207 @@ +"""Tests for brother_printer.usb_query - the PJL wire protocol. + +The printer's fd is non-blocking throughout: usblp blocks forever on open +and on writes whenever the printer is unwell, which is exactly when a status +report is wanted. These tests pin that every read and write is bounded. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +from python_pkg.brother_printer.usb_query import ( + _drain_buffer, + _read_nonblocking, + _wait_for_pjl_response, + _write_with_deadline, + pjl_query, +) + +MOD = "python_pkg.brother_printer.usb_query" + + +class TestDrainBuffer: + @patch(f"{MOD}.os.read") + 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") + 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() + + +class TestReadNonblocking: + @patch(f"{MOD}.os.read") + def test_reads_chunks(self, mock_read: MagicMock) -> None: + mock_read.side_effect = [b"hello", b"", OSError] + assert _read_nonblocking(42) == b"hello" + + @patch(f"{MOD}.os.read") + def test_oserror_suppressed(self, mock_read: MagicMock) -> None: + mock_read.side_effect = OSError("would block") + 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: + @patch(f"{MOD}._read_nonblocking") + @patch(f"{MOD}.select.select") + @patch(f"{MOD}.time.time") + def test_response_with_equals( + self, + mock_time: MagicMock, + mock_select: MagicMock, + mock_read: MagicMock, + ) -> None: + mock_time.side_effect = [0.0, 0.5, 1.0] + mock_select.return_value = ([42], [], []) + mock_read.return_value = b"CODE=10001" + assert b"CODE=10001" in _wait_for_pjl_response(42, 5.0) + + @patch(f"{MOD}._read_nonblocking") + @patch(f"{MOD}.select.select") + @patch(f"{MOD}.time.time") + def test_response_with_pjl( + self, + mock_time: MagicMock, + mock_select: MagicMock, + mock_read: MagicMock, + ) -> None: + mock_time.side_effect = [0.0, 0.5, 1.0] + mock_select.return_value = ([42], [], []) + mock_read.return_value = b"@PJL INFO" + assert b"@PJL" in _wait_for_pjl_response(42, 5.0) + + @patch(f"{MOD}.select.select") + @patch(f"{MOD}.time.time") + def test_timeout_no_data( + self, + mock_time: MagicMock, + mock_select: MagicMock, + ) -> None: + mock_time.side_effect = [10.0, 11.0] + assert _wait_for_pjl_response(42, 5.0) == b"" + + @patch(f"{MOD}._read_nonblocking") + @patch(f"{MOD}.select.select") + @patch(f"{MOD}.time.time") + def test_not_readable_then_timeout( + self, + mock_time: MagicMock, + mock_select: MagicMock, + mock_read: MagicMock, + ) -> None: + mock_time.side_effect = [0.0, 0.5, 6.0] + mock_select.return_value = ([], [], []) + assert _wait_for_pjl_response(42, 5.0) == b"" + + @patch(f"{MOD}._read_nonblocking") + @patch(f"{MOD}.select.select") + @patch(f"{MOD}.time.time") + def test_remaining_lte_zero( + self, + mock_time: MagicMock, + mock_select: MagicMock, + mock_read: MagicMock, + ) -> None: + """Inner remaining check triggers break.""" + mock_time.side_effect = [0.0, 6.0, 6.0] + assert _wait_for_pjl_response(42, 5.0) == b"" + mock_select.assert_not_called() + + @patch(f"{MOD}._read_nonblocking") + @patch(f"{MOD}.select.select") + @patch(f"{MOD}.time.time") + def test_response_no_eq_or_pjl( + self, + mock_time: MagicMock, + mock_select: MagicMock, + mock_read: MagicMock, + ) -> None: + """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" + assert _wait_for_pjl_response(42, 5.0) == b"garbage" + + +class TestPjlQuery: + @patch(f"{MOD}._wait_for_pjl_response") + @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_write: MagicMock, + mock_wait: MagicMock, + ) -> None: + mock_wait.return_value = b"CODE=10001" + 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()