kuhyx/testsAndMisc
1
0
Fork 0
mirror of https://github.com/kuhyx/testsAndMisc.git synced 2026-11-10 18:47:04 +01:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

fix(brother-printer): split oversized files, fix order-dependent entry test
Some checks are pending
Pre-commit checks / pre-commit (push) Waiting to run
Details
Python tests / test (push) Waiting to run
Details

Browse Source 
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) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RG3eEn8rwZbcXaJrKJxdv8
This commit is contained in:
Krzysztof Rudnicki 2026-07-17 15:06:25 +02:00
parent a12e696494
commit 69242e4bb3
15 changed files with 1268 additions and 1117 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
docs/superpowers/contracts/brother-printer-file-length-split-2026-07-17.json Normal file
View File

@ -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 <changed files>"
}

52
docs/superpowers/evidence/brother-printer-file-length-split-2026-07-17.json Normal file
View File

@ -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 <changed 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 <commit> 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."
]
}

2
python_pkg/brother_printer/check_brother_printer.py
View File

@ -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,

279
python_pkg/brother_printer/consumables.py Normal file
View File

@ -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,
)

263
python_pkg/brother_printer/cups_service.py
View File

@ -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 ────────────────────────────────────

8
python_pkg/brother_printer/display.py
View File

@ -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,

272
python_pkg/brother_printer/tests/test_consumables.py Normal file
View File

@ -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

268
python_pkg/brother_printer/tests/test_consumables_part2.py Normal file
View File

@ -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

111
python_pkg/brother_printer/tests/test_cups_service.py
View File

@ -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"))

406
python_pkg/brother_printer/tests/test_cups_service_part4.py
View File

@ -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

146
python_pkg/brother_printer/tests/test_display.py
View File

@ -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)

159
python_pkg/brother_printer/tests/test_display_part3.py Normal file
View File

@ -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)

4
python_pkg/brother_printer/tests/test_main_entry.py
View File

@ -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,

192
python_pkg/brother_printer/tests/test_usb_query.py
View File

@ -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()

207
python_pkg/brother_printer/tests/test_usb_query_part2.py Normal file
View File

@ -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()