diff --git a/docs/superpowers/contracts/brother-printer-stuck-job-2026-10-12.json b/docs/superpowers/contracts/brother-printer-stuck-job-2026-10-12.json new file mode 100644 index 00000000..9769df6d --- /dev/null +++ b/docs/superpowers/contracts/brother-printer-stuck-job-2026-10-12.json @@ -0,0 +1,16 @@ +{ + "title": "brother_printer: truthful probe note, pyusb auto-install, stuck-job detection", + "objective": "./run.sh says why the USB probe was skipped instead of falsely blaming pyusb, installs python-pyusb itself, and reports a stalled job or a hung backend of a cancelled job instead of 'Printing / processing job'.", + "acceptance_criteria": [ + "While CUPS prints, the fallback note reads 'USB status probe skipped: CUPS has a job on the printer'", + "run.sh checks /usr/bin/python3 for usb.core and runs pacman -S --needed python-pyusb when it is missing", + "A frozen job-impressions-completed counter for 15 s reports cups:job-stalled", + "A usb://Brother backend whose job is no longer queued reports cups:backend-orphaned and offers a SIGKILL", + "brother_printer tests pass at 100% branch coverage" + ], + "out_of_scope": [ + "Physical paper-feed problems on the HL-1110", + "Changing the queue default resolution" + ], + "verifier": "python3 -m pytest -q -o addopts='--strict-markers --strict-config' --cov=python_pkg.brother_printer --cov-branch python_pkg/brother_printer/tests" +} diff --git a/docs/superpowers/evidence/brother-printer-stuck-job-2026-10-12.json b/docs/superpowers/evidence/brother-printer-stuck-job-2026-10-12.json new file mode 100644 index 00000000..85f43805 --- /dev/null +++ b/docs/superpowers/evidence/brother-printer-stuck-job-2026-10-12.json @@ -0,0 +1,48 @@ +{ + "intent": "The Brother status report stops lying while a job is stuck: it names the real reason the USB probe was skipped, installs pyusb on its own, and flags a stalled job (paper misfeed) or a hung backend left behind by a cancelled job.", + "scope": [ + "python_pkg/brother_printer/{_stuck_job,_cups_fallback,cups_service,constants,data_classes,display}.py, run.sh, tests", + "Non-goal: fixing the printer's paper pick-up, which is mechanical" + ], + "changes": [ + "USBResult.probe_note carries _usb_probe_skip_reason(): CUPS busy / pyusb missing / printer silent; display prints it instead of 'pyusb not available'", + "run.sh ensure_pyusb: probes /usr/bin/python3 (what sudo runs; the user's python3 is a pyenv shim) and pacman-installs python-pyusb", + "_stuck_job.py: scans /proc for usb://Brother backends; orphaned backend of a cancelled job -> cups:backend-orphaned + y/N SIGKILL offer; job-impressions-completed frozen 15 s -> cups:job-stalled", + "41xxx status now reads 'Out of Paper / not feeding' since a misfeed sets the same paper-empty bit" + ], + "verification": [ + { + "command": "./run.sh during stalled job 105 (2026-10-12 16:01)", + "result": "pass", + "evidence": "Note: USB status probe skipped: CUPS has a job on the printer; status obtained via CUPS only." + }, + { + "command": "fake orphan (exec -a usb://Brother/... sleep 40) + display_usb_results with stdin y", + "result": "pass", + "evidence": "Queue blocked by a hung backend of a cancelled job / Kill hung backend PID 960784 ... Backend killed; process gone" + }, + { + "command": "flag_stuck_job with frozen vs moving counter", + "result": "pass", + "evidence": "frozen: cups:job-stalled after 15.0s; moving: cleared after 1.0s" + }, + { + "command": "python3 -m pytest -q --cov=python_pkg.brother_printer --cov-branch python_pkg/brother_printer/tests", + "result": "pass", + "evidence": "431 passed, coverage 100.00%" + }, + { + "command": "lpr Krasnale-sie-rzadza.pdf (jobs 106, 107 pages 52-55), monitored via ipptool", + "result": "pass", + "evidence": "all 55 pages on paper, user-confirmed; printer back at 10001 Ready" + } + ], + "risks": [ + "Stall window is 15 s: a healthy run froze the counter up to ~9 s, so a very slow first page could false-flag", + "A status run during printing now blocks up to 15 s while it samples progress" + ], + "rollback": [ + "git revert the commit", + "Run ./run.sh on an idle printer and confirm Ready 10001" + ] +} diff --git a/python_pkg/brother_printer/_cups_fallback.py b/python_pkg/brother_printer/_cups_fallback.py index 50a3e36c..73675fc8 100644 --- a/python_pkg/brother_printer/_cups_fallback.py +++ b/python_pkg/brother_printer/_cups_fallback.py @@ -17,6 +17,7 @@ from python_pkg.brother_printer._query import ( printer_info_from_cups, run_command_text, ) +from python_pkg.brother_printer._stuck_job import flag_stuck_job from python_pkg.brother_printer.constants import ( _CUPS_REASONS_TO_STATUS, _CUPS_STATE_TO_STATUS, @@ -27,9 +28,11 @@ from python_pkg.brother_printer.constants import ( ) from python_pkg.brother_printer.consumables import estimate_consumable_life from python_pkg.brother_printer.cups_service import ( + _cups_is_busy, _ensure_cups_running, _get_pyusb_device_info, _query_usb_port_status_raw, + _usb_probe_skip_reason, ) from python_pkg.brother_printer.data_classes import ( USBPortStatus, @@ -201,8 +204,19 @@ def query_usb_via_cups() -> USBResult: result.online = "TRUE" return result + _apply_cups_only_status(result, printer_name, ipp) + return result + + +def _apply_cups_only_status( + result: USBResult, printer_name: str, ipp: dict[str, str] +) -> None: + """Fill the status from CUPS alone, when the USB port could not be read.""" + state = ipp.get("printer-state", "") + result.cups.probe_note = _usb_probe_skip_reason(state) + if _cups_is_busy(state) and flag_stuck_job(result, printer_name): + return + reasons = ipp.get("printer-state-reasons", "none") result.status_code = _map_cups_to_status_code(state, reasons) result.display = ipp.get("printer-state-message", "") result.online = "TRUE" if state.lower() in {"idle", "processing"} else "FALSE" - - return result diff --git a/python_pkg/brother_printer/_stuck_job.py b/python_pkg/brother_printer/_stuck_job.py new file mode 100644 index 00000000..845fc02a --- /dev/null +++ b/python_pkg/brother_printer/_stuck_job.py @@ -0,0 +1,174 @@ +"""Detect a print job that CUPS calls "processing" but that is going nowhere. + +Two failure modes seen on the HL-1110 (2026-10-12), neither visible in +``printer-state-reasons``: + +* **Stalled job** - the printer failed to pick up paper. It raises the + paper-empty bit, the ``usb://`` backend blocks, and CUPS keeps saying + "processing". Only the job's page counter freezes. +* **Orphaned backend** - ``cancel`` removes the job but the blocked backend + survives, so the queue stays "now printing -0" forever and even + ``systemctl restart cups`` hangs. Killing the backend is the only cure. + +``job-impressions-completed`` counts pages the filters have *rasterised*, not +sheets out of the printer: it runs ~5 pages ahead of the paper. That is fine +for spotting a freeze, useless as "pages printed". +""" + +from __future__ import annotations + +from dataclasses import dataclass +import os +from pathlib import Path +import re +import shutil +import signal +import time +from typing import TYPE_CHECKING + +from python_pkg.brother_printer._query import run_command_text +from python_pkg.brother_printer.constants import ( + DERIVED_BACKEND_ORPHANED, + DERIVED_JOB_STALLED, + DIM, + GREEN, + RED, + RESET, + _out, + _prompt, +) + +if TYPE_CHECKING: + from python_pkg.brother_printer.data_classes import USBResult + +# A healthy run on 2026-10-12 froze the counter for up to ~9 s between pages +# (sampled every 3 s), so 15 s leaves headroom before calling it stalled. +STALL_WINDOW_SEC = 15.0 +POLL_INTERVAL_SEC = 1.0 +PROC_DIR = Path("/proc") +_BACKEND_PREFIX = "usb://Brother" +_IMPRESSIONS_RE = re.compile(r"job-impressions-completed \(integer\) = (\d+)") + + +@dataclass +class StuckJob: + """A job (or its leftover backend) that is not making progress.""" + + kind: str # "stalled" or "orphaned" + job_id: str + backend_pid: int = 0 + + +def find_backend_pids() -> dict[str, int]: + """Map CUPS job id -> PID of every running Brother ``usb://`` backend.""" + backends: dict[str, int] = {} + for entry in PROC_DIR.iterdir(): + if not entry.name.isdigit(): + continue + try: + argv = (entry / "cmdline").read_bytes().split(b"\0") + except OSError: + continue # process exited mid-scan, or not ours to read + # cmdline is NUL-terminated, so argv[1] is b"" when there is no job id. + if ( + len(argv) > 1 + and argv[1] + and argv[0].decode(errors="replace").startswith(_BACKEND_PREFIX) + ): + backends[argv[1].decode(errors="replace")] = int(entry.name) + return backends + + +def _job_impressions(job_id: str) -> int | None: + """Read a job's rasterised-page counter, or None if CUPS will not say.""" + ipptool_path = shutil.which("ipptool") + if not ipptool_path: + return None + uri = f"ipp://localhost/jobs/{job_id}" + output = run_command_text([ipptool_path, "-tv", uri, "get-job-attributes.test"]) + match = _IMPRESSIONS_RE.search(output) + return int(match.group(1)) if match else None + + +def _job_is_stalled(job_id: str) -> bool: + """Watch the page counter; True if it does not move within the window.""" + start = _job_impressions(job_id) + if start is None: + return False + deadline = time.monotonic() + STALL_WINDOW_SEC + while time.monotonic() < deadline: + time.sleep(POLL_INTERVAL_SEC) + if _job_impressions(job_id) != start: + return False + return True + + +def detect_stuck_job(active_job_ids: list[str]) -> StuckJob | None: + """Find an orphaned backend or a stalled active job. + + Args: + active_job_ids: numeric ids of the jobs CUPS still lists for the queue. + """ + backends = find_backend_pids() + for job_id, pid in sorted(backends.items()): + if job_id not in active_job_ids: + return StuckJob("orphaned", job_id, pid) + if active_job_ids and _job_is_stalled(active_job_ids[0]): + job_id = active_job_ids[0] + return StuckJob("stalled", job_id, backends.get(job_id, 0)) + return None + + +def kill_backend(pid: int) -> bool: + """SIGKILL a hung backend; SIGTERM is what it already ignored.""" + try: + os.kill(pid, signal.SIGKILL) + except OSError: + return False + return True + + +def offer_backend_kill(pid: int) -> None: + """Ask before killing an orphaned backend; nothing else unblocks the queue.""" + _out() + choice = _prompt(f" Kill hung backend PID {pid} to unblock the queue? [y/N]: ") + if choice.strip().lower() != "y": + _out(f" {DIM}No changes made.{RESET}") + elif kill_backend(pid): + _out(f" {GREEN}✓ Backend killed. CUPS will return to idle.{RESET}") + else: + _out(f" {RED}Could not kill PID {pid} (already gone, or not root).{RESET}") + + +def _active_job_ids(printer_name: str) -> list[str]: + """Numeric ids of the jobs CUPS still lists for the queue.""" + lpstat_path = shutil.which("lpstat") + if not lpstat_path: + return [] + output = run_command_text([lpstat_path, "-o", printer_name]) + prefix = f"{printer_name}-" + return [ + line.split()[0].removeprefix(prefix) + for line in output.splitlines() + if line.startswith(prefix) + ] + + +def flag_stuck_job(result: USBResult, printer_name: str) -> bool: + """Replace a misleading "printing" status when the job is going nowhere.""" + stuck = detect_stuck_job(_active_job_ids(printer_name)) + if stuck is None: + return False + result.online = "TRUE" + if stuck.kind == "orphaned": + result.status_code = DERIVED_BACKEND_ORPHANED + result.display = ( + f"backend PID {stuck.backend_pid} of cancelled job {stuck.job_id}" + ' still running (a leftover "filter failed" message is just' + " the cancel killing Ghostscript)" + ) + result.cups.orphaned_backend_pid = stuck.backend_pid + else: + result.status_code = DERIVED_JOB_STALLED + result.display = f"job {stuck.job_id} has not advanced a page in 15 s" + return True diff --git a/python_pkg/brother_printer/constants.py b/python_pkg/brother_printer/constants.py index 6518460a..a8598ddd 100644 --- a/python_pkg/brother_printer/constants.py +++ b/python_pkg/brother_printer/constants.py @@ -125,11 +125,14 @@ BROTHER_STATUS_CODES: dict[int, tuple[str, str, str]] = { # on a single-tray A4 printer). So match the family and ignore the digits. # Each entry is an inclusive low/high bound plus the usual status triple. +_PAPER_ACTION = "Load paper; if the tray is full, fan and reseat the stack." + _STATUS_CODE_RANGES: tuple[tuple[int, int, tuple[str, str, str]], ...] = ( ( 41000, 41999, - ("critical", "Out of Paper", "Load paper into the paper tray."), + # A failed pick-up raises the same paper-empty bit as an empty tray. + ("critical", "Out of Paper / not feeding", _PAPER_ACTION), ), ( 50000, @@ -188,6 +191,8 @@ _ERROR_REASON_MAP: tuple[tuple[tuple[str, ...], str, str], ...] = ( DERIVED_TONER_END = "estimate:toner-end" DERIVED_TONER_LOW = "estimate:toner-low" DERIVED_CUPS_ERROR = "cups:error" +DERIVED_JOB_STALLED = "cups:job-stalled" +DERIVED_BACKEND_ORPHANED = "cups:backend-orphaned" _DERIVED_STATUS_CODES: dict[str, tuple[str, str, str]] = { DERIVED_TONER_END: ( @@ -201,6 +206,16 @@ _DERIVED_STATUS_CODES: dict[str, tuple[str, str, str]] = { "Printer reports an error", "Check the printer display for details.", ), + DERIVED_JOB_STALLED: ( + "critical", + "Print job stalled (no page progress in 15 s)", + "Paper likely failed to feed: reseat it or power-cycle the printer.", + ), + DERIVED_BACKEND_ORPHANED: ( + "critical", + "Queue blocked by a hung backend of a cancelled job", + "Kill the hung backend (offered below); CUPS recovers on its own.", + ), } diff --git a/python_pkg/brother_printer/cups_service.py b/python_pkg/brother_printer/cups_service.py index 244c7524..1f9a2320 100644 --- a/python_pkg/brother_printer/cups_service.py +++ b/python_pkg/brother_printer/cups_service.py @@ -156,6 +156,17 @@ def _cups_is_busy(cups_state: str) -> bool: return "processing" in cups_state.lower() or "printing" in cups_state.lower() +def _usb_probe_skip_reason(cups_state: str) -> str: + """Explain why _query_usb_port_status_raw returned None.""" + if _cups_is_busy(cups_state): + return "USB status probe skipped: CUPS has a job on the printer" + try: + _import_or_raise("usb.core") + except ImportError: + return "pyusb not installed (run.sh installs it)" + return "printer did not answer the USB status probe" + + def _query_usb_port_status_raw(cups_state: str = "") -> USBPortStatus | None: """Query the printer's port status without disturbing it. diff --git a/python_pkg/brother_printer/data_classes.py b/python_pkg/brother_printer/data_classes.py index 516d4388..13bc25b8 100644 --- a/python_pkg/brother_printer/data_classes.py +++ b/python_pkg/brother_printer/data_classes.py @@ -72,6 +72,16 @@ class USBPortStatus: raw_byte: int = 0 +@dataclass +class CupsDiagnosis: + """What the CUPS fallback learned beyond the status code itself.""" + + # Why port_status is None, in words for the report. Empty when it was read. + probe_note: str = "" + # PID of a usb:// backend still running for a cancelled job; 0 if none. + orphaned_backend_pid: int = 0 + + @dataclass class USBResult: """Result from a USB PJL query.""" @@ -90,6 +100,7 @@ class USBResult: # (CUPS fallback mode), which forces the estimate onto the CUPS page log. page_count: str = "" port_status: USBPortStatus | None = None + cups: CupsDiagnosis = field(default_factory=CupsDiagnosis) @dataclass diff --git a/python_pkg/brother_printer/display.py b/python_pkg/brother_printer/display.py index f8056bf8..b9624cf9 100644 --- a/python_pkg/brother_printer/display.py +++ b/python_pkg/brother_printer/display.py @@ -16,6 +16,7 @@ from python_pkg.brother_printer._severity import ( _SEVERITY_ICONS, _SEVERITY_SUMMARIES, ) +from python_pkg.brother_printer._stuck_job import offer_backend_kill from python_pkg.brother_printer._supply import _collect_supply_items from python_pkg.brother_printer.constants import ( BOLD, @@ -120,9 +121,9 @@ def _display_cups_fallback_note(result: USBResult) -> None: ) else: _out( - f" {DIM}Note: pyusb not available; status obtained via" - f" CUPS only. Detailed toner/drum levels are not" - f" available in this mode.{RESET}" + f" {DIM}Note: {result.cups.probe_note or 'USB status probe failed'};" + f" status obtained via CUPS only. Detailed toner/drum levels" + f" are not available in this mode.{RESET}" ) @@ -141,6 +142,8 @@ def display_usb_results(result: USBResult) -> None: if result.device == "cups": _display_cups_fallback_note(result) + if result.cups.orphaned_backend_pid: + offer_backend_kill(result.cups.orphaned_backend_pid) _out() printer_total = int(result.page_count) if result.page_count.isdigit() else 0 diff --git a/python_pkg/brother_printer/run.sh b/python_pkg/brother_printer/run.sh index 38edc57b..0563d26a 100755 --- a/python_pkg/brother_printer/run.sh +++ b/python_pkg/brother_printer/run.sh @@ -3,8 +3,8 @@ set -euo pipefail SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" REPO_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)" -# No pip dependencies — script only uses stdlib (+pyusb for fallback info) -# Requires root for USB hardware queries and CUPS management +# Depends on pyusb (USB port-status probe when CUPS owns the printer) and runs +# as root for USB hardware queries and CUPS management. # # Imports are absolute (python_pkg.brother_printer.*), so the monorepo root must # be importable. sudo's env_reset drops an exported PYTHONPATH, hence passing it @@ -13,7 +13,21 @@ REPO_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)" # Usage: ./run.sh # auto-detect # ./run.sh # network/SNMP mode +# The tool runs under the system python3 (sudo secure_path), so pyusb must come +# from the distro package, not a venv. +ensure_pyusb() { + /usr/bin/python3 -c 'import usb.core' 2>/dev/null && return 0 + if ! command -v pacman >/dev/null; then + echo "pyusb missing and pacman not found; install python-pyusb for system python3." >&2 + exit 1 + fi + echo "Installing python-pyusb..." + sudo pacman -S --needed --noconfirm python-pyusb +} + +ensure_pyusb + if [[ $EUID -ne 0 ]]; then - exec sudo PYTHONPATH="$REPO_ROOT" python3 -m python_pkg.brother_printer "$@" + exec sudo PYTHONPATH="$REPO_ROOT" python3 -m python_pkg.brother_printer "$@" fi exec env PYTHONPATH="$REPO_ROOT" python3 -m python_pkg.brother_printer "$@" diff --git a/python_pkg/brother_printer/tests/test_constants.py b/python_pkg/brother_printer/tests/test_constants.py index bc6df812..6f5bd4c4 100644 --- a/python_pkg/brother_printer/tests/test_constants.py +++ b/python_pkg/brother_printer/tests/test_constants.py @@ -138,7 +138,7 @@ class TestGetStatusInfo: """ severity, text, _ = get_status_info("41213") assert severity == "critical" - assert text == "Out of Paper" + assert text == "Out of Paper / not feeding" def test_derived_codes_are_not_pjl(self) -> None: """States we infer ourselves must not masquerade as printer reports.""" diff --git a/python_pkg/brother_printer/tests/test_cups_fallback_query.py b/python_pkg/brother_printer/tests/test_cups_fallback_query.py index 1084c634..67aa66cd 100644 --- a/python_pkg/brother_printer/tests/test_cups_fallback_query.py +++ b/python_pkg/brother_printer/tests/test_cups_fallback_query.py @@ -100,3 +100,34 @@ class TestQueryUsbViaCups: result = query_usb_via_cups() assert result.status_code == "41000" assert result.online == "FALSE" + + +class TestQueryUsbViaCupsStuckJob: + """A busy queue whose job goes nowhere must not read as "printing".""" + + def _query(self, *, stuck: bool) -> tuple[object, MagicMock]: + with ( + patch(f"{MOD}._ensure_cups_running", return_value=True), + patch(f"{MOD}.find_cups_printer_name", return_value="Brother"), + patch(f"{MOD}._get_pyusb_device_info", return_value={}), + patch(f"{MOD}.printer_info_from_cups", return_value={}), + patch( + f"{MOD}._get_cups_ipp_status", + return_value={"printer-state": "processing"}, + ), + patch(f"{MOD}._get_cups_economode", return_value=""), + patch(f"{MOD}._query_usb_port_status_raw", return_value=None), + patch(f"{MOD}.flag_stuck_job", return_value=stuck) as flag, + ): + return query_usb_via_cups(), flag + + def test_stuck_job_short_circuits(self) -> None: + result, flag = self._query(stuck=True) + flag.assert_called_once() + assert result.status_code == "" # left for flag_stuck_job to set + + def test_progressing_job_reads_as_printing(self) -> None: + result, flag = self._query(stuck=False) + flag.assert_called_once() + assert result.status_code == "10023" + assert "CUPS has a job" in result.cups.probe_note diff --git a/python_pkg/brother_printer/tests/test_display.py b/python_pkg/brother_printer/tests/test_display.py index 5584342e..afd73e5f 100644 --- a/python_pkg/brother_printer/tests/test_display.py +++ b/python_pkg/brother_printer/tests/test_display.py @@ -8,6 +8,7 @@ from unittest.mock import MagicMock, patch import pytest from python_pkg.brother_printer.data_classes import ( + CupsDiagnosis, USBPortStatus, USBResult, ) @@ -157,11 +158,17 @@ class TestDisplayCupsFallbackNote: _display_cups_fallback_note(r) assert "USB port query" in out.getvalue() - def test_without_port_status(self) -> None: + def test_without_port_status_shows_reason(self) -> None: + r = USBResult(cups=CupsDiagnosis(probe_note="pyusb not installed")) + with patch("sys.stdout", new_callable=StringIO) as out: + _display_cups_fallback_note(r) + assert "Note: pyusb not installed;" in out.getvalue() + + def test_without_port_status_or_reason(self) -> None: r = USBResult(port_status=None) with patch("sys.stdout", new_callable=StringIO) as out: _display_cups_fallback_note(r) - assert "pyusb not available" in out.getvalue() + assert "USB status probe failed" in out.getvalue() class TestDisplayUsbResults: @@ -190,10 +197,29 @@ class TestDisplayUsbResults: patch(f"{MOD}.get_cups_queue_status"), patch(f"{MOD}.display_cups_queue_status"), patch(f"{MOD}._display_cups_fallback_note") as mock_fallback, + patch(f"{MOD}.offer_backend_kill") as mock_kill, patch("sys.stdout", new_callable=StringIO), ): display_usb_results(r) mock_fallback.assert_called_once() + mock_kill.assert_not_called() + + def test_cups_device_offers_backend_kill(self) -> None: + r = USBResult(device="cups", cups=CupsDiagnosis(orphaned_backend_pid=4242)) + with ( + patch(f"{MOD}._display_report_header"), + patch(f"{MOD}._display_usb_device_info"), + patch(f"{MOD}._display_pjl_status"), + patch(f"{MOD}._display_page_count_estimate"), + patch(f"{MOD}._display_consumables_reference"), + patch(f"{MOD}.get_cups_queue_status"), + patch(f"{MOD}.display_cups_queue_status"), + patch(f"{MOD}._display_cups_fallback_note"), + patch(f"{MOD}.offer_backend_kill") as mock_kill, + patch("sys.stdout", new_callable=StringIO), + ): + display_usb_results(r) + mock_kill.assert_called_once_with(4242) def test_error(self) -> None: r = USBResult(error="fail") diff --git a/python_pkg/brother_printer/tests/test_stuck_job.py b/python_pkg/brother_printer/tests/test_stuck_job.py new file mode 100644 index 00000000..cad5379d --- /dev/null +++ b/python_pkg/brother_printer/tests/test_stuck_job.py @@ -0,0 +1,196 @@ +"""Tests for brother_printer._stuck_job - stalled jobs and orphaned backends. + +/proc, ipptool, lpstat, time and os.kill are all patched: nothing here reads +the live process table, talks to CUPS, sleeps, or signals a real process. +""" + +from __future__ import annotations + +from io import StringIO +from pathlib import Path +from typing import TYPE_CHECKING +from unittest.mock import MagicMock, patch + +from python_pkg.brother_printer import _stuck_job +from python_pkg.brother_printer._stuck_job import ( + StuckJob, + detect_stuck_job, + find_backend_pids, + flag_stuck_job, + kill_backend, + offer_backend_kill, +) +from python_pkg.brother_printer.constants import ( + DERIVED_BACKEND_ORPHANED, + DERIVED_JOB_STALLED, +) +from python_pkg.brother_printer.data_classes import USBResult + +if TYPE_CHECKING: + import pytest + +MOD = "python_pkg.brother_printer._stuck_job" + + +def _proc(tmp_path: Path, pid: str, cmdline: bytes | None) -> None: + """Create a fake /proc/; cmdline None means it is unreadable.""" + entry = tmp_path / pid + entry.mkdir() + if cmdline is not None: + (entry / "cmdline").write_bytes(cmdline) + + +class TestFindBackendPids: + def test_maps_job_id_to_pid( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + _proc(tmp_path, "101", b"usb://Brother/HL-1110\x00105\x00kuhy\x00") + _proc(tmp_path, "102", b"/usr/bin/gs\x00-dQUIET\x00") + _proc(tmp_path, "103", b"usb://Brother/HL-1110\x00") # no job argument + _proc(tmp_path, "104", None) # exited mid-scan + (tmp_path / "self").mkdir() # non-numeric entry + monkeypatch.setattr(_stuck_job, "PROC_DIR", tmp_path) + assert find_backend_pids() == {"105": 101} + + +class TestJobImpressions: + @patch(f"{MOD}.shutil.which", return_value=None) + def test_no_ipptool(self, w: MagicMock) -> None: + assert _stuck_job._job_impressions("105") is None + + @patch(f"{MOD}.run_command_text") + @patch(f"{MOD}.shutil.which", return_value="/usr/bin/ipptool") + def test_parses_counter(self, w: MagicMock, run: MagicMock) -> None: + run.return_value = " job-impressions-completed (integer) = 7\n" + assert _stuck_job._job_impressions("105") == 7 + assert "ipp://localhost/jobs/105" in run.call_args.args[0] + + @patch(f"{MOD}.run_command_text", return_value="") + @patch(f"{MOD}.shutil.which", return_value="/usr/bin/ipptool") + def test_counter_absent(self, w: MagicMock, r: MagicMock) -> None: + assert _stuck_job._job_impressions("105") is None + + +class TestJobIsStalled: + @patch(f"{MOD}._job_impressions", return_value=None) + def test_unknown_counter_is_not_stalled(self, i: MagicMock) -> None: + assert _stuck_job._job_is_stalled("105") is False + + @patch(f"{MOD}.time") + @patch(f"{MOD}._job_impressions", side_effect=[7, 7, 8]) + def test_progress_clears_it(self, i: MagicMock, t: MagicMock) -> None: + t.monotonic.side_effect = [0.0, 0.0, 1.0] + assert _stuck_job._job_is_stalled("105") is False + + @patch(f"{MOD}.time") + @patch(f"{MOD}._job_impressions", return_value=7) + def test_frozen_counter_is_stalled(self, i: MagicMock, t: MagicMock) -> None: + t.monotonic.side_effect = [0.0, 0.0, 10.0, 15.0] + assert _stuck_job._job_is_stalled("105") is True + assert t.sleep.call_count == 2 + + +class TestDetectStuckJob: + @patch(f"{MOD}._job_is_stalled") + @patch(f"{MOD}.find_backend_pids", return_value={"104": 820758}) + def test_orphaned_backend(self, f: MagicMock, stalled: MagicMock) -> None: + assert detect_stuck_job([]) == StuckJob("orphaned", "104", 820758) + stalled.assert_not_called() + + @patch(f"{MOD}._job_is_stalled", return_value=True) + @patch(f"{MOD}.find_backend_pids", return_value={"105": 884180}) + def test_stalled_job(self, f: MagicMock, s: MagicMock) -> None: + assert detect_stuck_job(["105"]) == StuckJob("stalled", "105", 884180) + + @patch(f"{MOD}._job_is_stalled", return_value=True) + @patch(f"{MOD}.find_backend_pids", return_value={}) + def test_stalled_before_backend_starts(self, f: MagicMock, s: MagicMock) -> None: + assert detect_stuck_job(["105"]) == StuckJob("stalled", "105", 0) + + @patch(f"{MOD}._job_is_stalled", return_value=False) + @patch(f"{MOD}.find_backend_pids", return_value={"105": 884180}) + def test_healthy_job(self, f: MagicMock, s: MagicMock) -> None: + assert detect_stuck_job(["105"]) is None + + @patch(f"{MOD}.find_backend_pids", return_value={}) + def test_nothing_running(self, f: MagicMock) -> None: + assert detect_stuck_job([]) is None + + +class TestKillBackend: + @patch(f"{MOD}.os.kill") + def test_sigkill(self, kill: MagicMock) -> None: + assert kill_backend(42) is True + kill.assert_called_once_with(42, _stuck_job.signal.SIGKILL) + + @patch(f"{MOD}.os.kill", side_effect=ProcessLookupError) + def test_already_gone(self, k: MagicMock) -> None: + assert kill_backend(42) is False + + +class TestOfferBackendKill: + def _offer(self, answer: str, *, killed: bool = True) -> tuple[str, MagicMock]: + with ( + patch(f"{MOD}._prompt", return_value=answer), + patch(f"{MOD}.kill_backend", return_value=killed) as kill, + patch("sys.stdout", new_callable=StringIO) as out, + ): + offer_backend_kill(42) + return out.getvalue(), kill + + def test_declined(self) -> None: + out, kill = self._offer("") + kill.assert_not_called() + assert "No changes made" in out + + def test_accepted(self) -> None: + out, kill = self._offer(" Y ") + kill.assert_called_once_with(42) + assert "Backend killed" in out + + def test_kill_fails(self) -> None: + out, _kill = self._offer("y", killed=False) + assert "Could not kill PID 42" in out + + +class TestActiveJobIds: + @patch(f"{MOD}.shutil.which", return_value=None) + def test_no_lpstat(self, w: MagicMock) -> None: + assert _stuck_job._active_job_ids("Brother") == [] + + @patch(f"{MOD}.run_command_text") + @patch(f"{MOD}.shutil.which", return_value="/usr/bin/lpstat") + def test_parses_ids(self, w: MagicMock, run: MagicMock) -> None: + run.return_value = ( + "Brother-105 kuhy 160768 Mon 12 Oct 2026\nOther-7 kuhy 1 Mon 12 Oct 2026\n" + ) + assert _stuck_job._active_job_ids("Brother") == ["105"] + + +class TestFlagStuckJob: + @patch(f"{MOD}._active_job_ids", return_value=[]) + @patch(f"{MOD}.detect_stuck_job", return_value=None) + def test_nothing_stuck(self, d: MagicMock, a: MagicMock) -> None: + result = USBResult(device="cups") + assert flag_stuck_job(result, "Brother") is False + assert result.status_code == "" + + @patch(f"{MOD}._active_job_ids", return_value=[]) + @patch(f"{MOD}.detect_stuck_job") + def test_orphaned(self, detect: MagicMock, a: MagicMock) -> None: + detect.return_value = StuckJob("orphaned", "104", 820758) + result = USBResult(device="cups") + assert flag_stuck_job(result, "Brother") is True + assert result.status_code == DERIVED_BACKEND_ORPHANED + assert result.cups.orphaned_backend_pid == 820758 + assert "cancelled job 104" in result.display + + @patch(f"{MOD}._active_job_ids", return_value=["105"]) + @patch(f"{MOD}.detect_stuck_job") + def test_stalled(self, detect: MagicMock, a: MagicMock) -> None: + detect.return_value = StuckJob("stalled", "105", 884180) + result = USBResult(device="cups") + assert flag_stuck_job(result, "Brother") is True + assert result.status_code == DERIVED_JOB_STALLED + assert result.cups.orphaned_backend_pid == 0 + assert "job 105" in result.display diff --git a/python_pkg/brother_printer/tests/test_usb_probe_reason.py b/python_pkg/brother_printer/tests/test_usb_probe_reason.py new file mode 100644 index 00000000..83b5592b --- /dev/null +++ b/python_pkg/brother_printer/tests/test_usb_probe_reason.py @@ -0,0 +1,25 @@ +"""Tests for cups_service._usb_probe_skip_reason - the fallback note's reason.""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +from python_pkg.brother_printer.cups_service import _usb_probe_skip_reason + +MOD = "python_pkg.brother_printer.cups_service" + + +class TestUsbProbeSkipReason: + """The note must name the real reason, not blame a pyusb that is there.""" + + def test_busy_cups_wins(self) -> None: + reason = _usb_probe_skip_reason("processing") + assert "CUPS has a job" in reason + + @patch(f"{MOD}._import_or_raise", side_effect=ImportError("usb.core")) + def test_pyusb_missing(self, m: MagicMock) -> None: + assert "pyusb not installed" in _usb_probe_skip_reason("idle") + + @patch(f"{MOD}._import_or_raise", return_value=MagicMock()) + def test_printer_silent(self, m: MagicMock) -> None: + assert "did not answer" in _usb_probe_skip_reason("idle")