mirror of
https://github.com/kuhyx/testsAndMisc.git
synced 2026-11-10 21:07:03 +01:00
Split brother_printer CUPS fallback query out of cups_service
cups_service.py 406 -> 234. The IPP attribute queries, the CUPS status-code mapping, find_cups_printer_name and query_usb_via_cups move to _cups_fallback.py; cups_service keeps service control and raw USB port status. The seam is not the section header: test_cups_service_part2.py imports _get_cups_economode and find_cups_printer_name together and patches shutil once, so those two had to travel as a pair. test_cups_service.py, the largest module, needs no edit. usb_query.py reaches the fallback through a late import_module, so both that string and the absolute patch target in test_usb_query.py moved with it. Pass count is 402 before and after. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011w2Pj4Ek3ZNMQVYcFHgXDY
This commit is contained in:
parent
cf44e54215
commit
143dd9ff3d
@ -0,0 +1,20 @@
|
||||
{
|
||||
"title": "Split brother_printer/cups_service.py under the 250-line cap",
|
||||
"objective": "Bring python_pkg/brother_printer/cups_service.py (406 lines) under the repo-wide 250-line cap by moving the IPP attribute queries, the CUPS status-code mapping and the CUPS-based USB fallback query into a new _cups_fallback.py, leaving cups_service holding only service control and raw USB port status.",
|
||||
"acceptance_criteria": [
|
||||
"cups_service.py and the new _cups_fallback.py are each at most 250 lines.",
|
||||
"The brother_printer suite still reports exactly 402 passing tests -- the same count as before the split, proving no test was dropped or silently skipped.",
|
||||
"The package keeps 100% branch coverage of the brother_printer source modules.",
|
||||
"test_cups_service.py, the largest of the four test modules, needs no edit at all: every name it imports and every collaborator it patches on MOD (fcntl, os, Path, shutil, subprocess, time) stays defined in cups_service.py.",
|
||||
"_get_cups_economode and find_cups_printer_name move together, because test_cups_service_part2.py imports both and patches shutil once on a single MOD; separating them would need two MOD constants in one test module.",
|
||||
"usb_query.py's late importlib.import_module call and the absolute patch target in test_usb_query.py both name the new module, so the CUPS fallback is still reached and still mocked.",
|
||||
"No assertion is changed and no test is deleted, skipped or weakened; the only test edits are the MOD constant, the import module path, and one absolute patch string.",
|
||||
"No suppression is added and no coverage exclusion is introduced."
|
||||
],
|
||||
"out_of_scope": [
|
||||
"Changing what any test asserts, or any status-code mapping behaviour.",
|
||||
"Splitting the brother_printer test files, which are separate commits -- part2 and part3 remain over the cap after this change.",
|
||||
"Adding a re-export shim in cups_service; importers point at the new module directly to keep the dependency direction one-way."
|
||||
],
|
||||
"verifier": "PYTHONPATH=. python3 -m pytest python_pkg/brother_printer/tests -q (expects exactly 402 passed), plus pre-commit run --files on the touched files, plus bash ~/utils/scripts/check_file_length.sh --all for the cap."
|
||||
}
|
||||
@ -0,0 +1,48 @@
|
||||
{
|
||||
"intent": "Bring brother_printer/cups_service.py under the 250-line cap by moving the IPP queries, status-code mapping and CUPS-based USB fallback into their own module, with no behaviour change.",
|
||||
"scope": [
|
||||
"python_pkg/brother_printer/_cups_fallback.py (new)",
|
||||
"python_pkg/brother_printer/cups_service.py",
|
||||
"python_pkg/brother_printer/cups_queue.py, usb_query.py (import path only)",
|
||||
"python_pkg/brother_printer/tests/test_cups_service_part2.py, part3, part4 (MOD and import path only)",
|
||||
"python_pkg/brother_printer/tests/test_usb_query.py (one absolute patch target)",
|
||||
"Non-goal: no assertion changed, no test deleted, skipped or weakened"
|
||||
],
|
||||
"changes": [
|
||||
"Moved _parse_ipp_attributes, _get_cups_ipp_status, _get_cups_economode, _map_cups_to_status_code, _cups_reasons_to_error, _port_status_to_status_code, find_cups_printer_name and query_usb_via_cups verbatim into python_pkg/brother_printer/_cups_fallback.py.",
|
||||
"The seam is the whole IPP-onwards section rather than the section header at line 237: test_cups_service_part2.py imports both _get_cups_economode and find_cups_printer_name and patches shutil once, so those two had to travel together or that module would need two MOD constants.",
|
||||
"_cups_fallback imports _ensure_cups_running, _get_pyusb_device_info and _query_usb_port_status_raw from cups_service, so test_cups_service_part3.py can patch all five of its targets on one module.",
|
||||
"cups_service.py dropped the imports only the moved code used (re, printer_info_from_cups, run_command_text, estimate_consumable_life, USBResult, the three CUPS mapping tables and the three DERIVED_* codes) and its docstring now names the new module; 406 -> 234 lines.",
|
||||
"cups_queue.py imports find_cups_printer_name from _cups_fallback, and usb_query.py's late importlib.import_module now names _cups_fallback. test_usb_query.py patches that path absolutely, so its one patch string moved with it."
|
||||
],
|
||||
"verification": [
|
||||
{
|
||||
"command": "PYTHONPATH=. python3 -m pytest python_pkg/brother_printer/tests -q",
|
||||
"result": "pass",
|
||||
"evidence": "402 passed before the split and 402 passed after it - identical pass count, not merely a green run."
|
||||
},
|
||||
{
|
||||
"command": "PYTHONPATH=. python3 -c \"from python_pkg.brother_printer import _cups_fallback as q, cups_service as c, cups_queue as cq; q._ensure_cups_running; q._query_usb_port_status_raw; q.find_cups_printer_name; q.shutil; q.subprocess; c.fcntl; c.os; c.Path; c.time; cq.find_cups_printer_name\"",
|
||||
"result": "pass",
|
||||
"evidence": "runtime probe OK - every name the four test modules patch is still a real attribute of the module they name, so ruff's TC001 rewrite trap did not fire."
|
||||
},
|
||||
{
|
||||
"command": "wc -l python_pkg/brother_printer/cups_service.py python_pkg/brother_printer/_cups_fallback.py",
|
||||
"result": "pass",
|
||||
"evidence": "234 and 209 lines, both under the 250-line cap in a single cut."
|
||||
},
|
||||
{
|
||||
"command": "pre-commit run --files <the nine touched files>",
|
||||
"result": "pass",
|
||||
"evidence": "ruff, ruff format, mypy, pylint, bandit and pytest-with-coverage all pass on the touched files."
|
||||
}
|
||||
],
|
||||
"risks": [
|
||||
"usb_query.py reaches the fallback through a late importlib.import_module with the module path as a string, which no import checker validates. Covered by test_usb_query.py exercising that path, and by the pass count staying at 402 - a stale string would have raised ModuleNotFoundError there.",
|
||||
"cups_queue -> _cups_fallback -> cups_service is now a three-module chain; moving anything else out of cups_service must keep that direction to avoid a cycle."
|
||||
],
|
||||
"rollback": [
|
||||
"git revert the commit; cups_service.py returns to 406 lines and _cups_fallback.py disappears.",
|
||||
"After rollback, re-run the package suite and confirm it still reports 402 passed."
|
||||
]
|
||||
}
|
||||
208
python_pkg/brother_printer/_cups_fallback.py
Normal file
208
python_pkg/brother_printer/_cups_fallback.py
Normal file
@ -0,0 +1,208 @@
|
||||
"""IPP attribute queries, CUPS status mapping, and the CUPS-based USB fallback.
|
||||
|
||||
Split out of :mod:`python_pkg.brother_printer.cups_service` to keep it under
|
||||
the 250-line cap. The service-control and raw USB port-status helpers stay in
|
||||
``cups_service``, which patches ``fcntl``, ``os``, ``Path`` and ``time`` on
|
||||
that module; they are imported here so ``tests/test_cups_service_part3.py``
|
||||
can patch them on this one.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
|
||||
from python_pkg.brother_printer._query import (
|
||||
printer_info_from_cups,
|
||||
run_command_text,
|
||||
)
|
||||
from python_pkg.brother_printer.constants import (
|
||||
_CUPS_REASONS_TO_STATUS,
|
||||
_CUPS_STATE_TO_STATUS,
|
||||
_ERROR_REASON_MAP,
|
||||
DERIVED_CUPS_ERROR,
|
||||
DERIVED_TONER_END,
|
||||
DERIVED_TONER_LOW,
|
||||
)
|
||||
from python_pkg.brother_printer.consumables import estimate_consumable_life
|
||||
from python_pkg.brother_printer.cups_service import (
|
||||
_ensure_cups_running,
|
||||
_get_pyusb_device_info,
|
||||
_query_usb_port_status_raw,
|
||||
)
|
||||
from python_pkg.brother_printer.data_classes import (
|
||||
USBPortStatus,
|
||||
USBResult,
|
||||
)
|
||||
|
||||
# ── IPP / CUPS attribute queries ────────────────────────────────────
|
||||
|
||||
|
||||
def _parse_ipp_attributes(output: str) -> dict[str, str]:
|
||||
"""Parse ipptool verbose output into an attribute dict."""
|
||||
attrs: dict[str, str] = {}
|
||||
for line in output.splitlines():
|
||||
match = re.match(r"\s+(\S+)\s+\([^)]+\)\s+=\s+(.*)", line)
|
||||
if match:
|
||||
attrs[match.group(1)] = match.group(2).strip()
|
||||
return attrs
|
||||
|
||||
|
||||
def _get_cups_ipp_status(printer_name: str) -> dict[str, str]:
|
||||
"""Query printer attributes via CUPS IPP using ipptool."""
|
||||
ipptool_path = shutil.which("ipptool")
|
||||
if not ipptool_path:
|
||||
return {}
|
||||
uri = f"ipp://localhost/printers/{printer_name}"
|
||||
try:
|
||||
r = subprocess.run(
|
||||
[ipptool_path, "-tv", uri, "get-printer-attributes.test"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=10,
|
||||
check=False,
|
||||
)
|
||||
return _parse_ipp_attributes(r.stdout)
|
||||
except (subprocess.TimeoutExpired, subprocess.SubprocessError, OSError):
|
||||
return {}
|
||||
|
||||
|
||||
def _get_cups_economode(printer_name: str) -> str:
|
||||
"""Query toner save mode setting via lpoptions."""
|
||||
lpoptions_path = shutil.which("lpoptions")
|
||||
if not lpoptions_path:
|
||||
return ""
|
||||
command = [lpoptions_path, "-p", printer_name, "-l"]
|
||||
for line in run_command_text(command).splitlines():
|
||||
if "conomode" in line.lower():
|
||||
match = re.search(r"\*(\w+)", line)
|
||||
if match:
|
||||
return "ON" if match.group(1).lower() == "true" else "OFF"
|
||||
return ""
|
||||
|
||||
|
||||
# ── Status code mapping ──────────────────────────────────────────────
|
||||
|
||||
|
||||
def _map_cups_to_status_code(state: str, reasons: str) -> str:
|
||||
"""Map CUPS state + reasons to a Brother PJL status code string."""
|
||||
for keyword, code in _CUPS_REASONS_TO_STATUS.items():
|
||||
if keyword in reasons.lower():
|
||||
return str(code)
|
||||
clean_state = re.sub(r"\(.*\)", "", state).strip().lower()
|
||||
return str(_CUPS_STATE_TO_STATUS.get(clean_state, 10001))
|
||||
|
||||
|
||||
def _cups_reasons_to_error(cups_reasons: str) -> tuple[str, str]:
|
||||
"""Map CUPS reason keywords to a (status_code, display) pair."""
|
||||
reasons_lower = cups_reasons.lower()
|
||||
for keywords, code, display in _ERROR_REASON_MAP:
|
||||
if any(kw in reasons_lower for kw in keywords):
|
||||
return code, display
|
||||
# Nothing recognised. Show what CUPS actually said rather than a bare
|
||||
# "Printer Error", which tells the reader nothing they can act on.
|
||||
detail = cups_reasons.strip()
|
||||
if detail and detail.lower() not in {"none", ""}:
|
||||
return DERIVED_CUPS_ERROR, f"Printer Error (CUPS reports: {detail})"
|
||||
return DERIVED_CUPS_ERROR, "Printer Error (CUPS gave no reason)"
|
||||
|
||||
|
||||
def _port_status_to_status_code(
|
||||
ps: USBPortStatus,
|
||||
cups_reasons: str,
|
||||
) -> tuple[str, str]:
|
||||
"""Map USB port status + CUPS reasons to (status_code, display)."""
|
||||
# The port status exposes only paper_empty/error/online bits, so anything
|
||||
# more specific than "out of paper" has to come from the CUPS reasons -
|
||||
# an error bit alone does not tell us the cover is open.
|
||||
if ps.paper_empty:
|
||||
return "41000", "No Paper"
|
||||
if ps.error:
|
||||
return _cups_reasons_to_error(cups_reasons)
|
||||
if not ps.online:
|
||||
return "10002", "Offline / Sleep"
|
||||
return "", ""
|
||||
|
||||
|
||||
# ── CUPS printer name discovery ──────────────────────────────────────
|
||||
|
||||
|
||||
def find_cups_printer_name() -> str:
|
||||
"""Find the CUPS queue name for a Brother printer."""
|
||||
lpstat_path = shutil.which("lpstat")
|
||||
if not lpstat_path:
|
||||
return ""
|
||||
for line in run_command_text([lpstat_path, "-v"]).splitlines():
|
||||
if "brother" in line.lower():
|
||||
match = re.match(r"device for (\S+):", line)
|
||||
if match:
|
||||
return match.group(1)
|
||||
return ""
|
||||
|
||||
|
||||
# ── CUPS-based USB fallback query ────────────────────────────────────
|
||||
|
||||
|
||||
def query_usb_via_cups() -> USBResult:
|
||||
"""Query USB printer status through CUPS when /dev/usb/lp* is unavailable."""
|
||||
_ensure_cups_running()
|
||||
printer_name = find_cups_printer_name()
|
||||
if not printer_name:
|
||||
return USBResult(
|
||||
error="No USB printer device at /dev/usb/lp*"
|
||||
" (usblp module not available)"
|
||||
" and no Brother printer found in CUPS.",
|
||||
)
|
||||
|
||||
pyusb_info = _get_pyusb_device_info()
|
||||
cups_info = printer_info_from_cups()
|
||||
|
||||
result = USBResult(
|
||||
device="cups",
|
||||
product=(
|
||||
pyusb_info.get("product")
|
||||
or cups_info.get("product")
|
||||
or "Brother Laser Printer"
|
||||
),
|
||||
serial=pyusb_info.get("serial") or cups_info.get("serial", ""),
|
||||
)
|
||||
|
||||
ipp = _get_cups_ipp_status(printer_name)
|
||||
state = ipp.get("printer-state", "")
|
||||
reasons = ipp.get("printer-state-reasons", "none")
|
||||
result.economode = _get_cups_economode(printer_name)
|
||||
|
||||
port_status = _query_usb_port_status_raw(state)
|
||||
if port_status is not None:
|
||||
result.port_status = port_status
|
||||
hw_code, hw_display = _port_status_to_status_code(
|
||||
port_status,
|
||||
reasons,
|
||||
)
|
||||
if hw_code:
|
||||
result.status_code = hw_code
|
||||
result.display = hw_display
|
||||
result.online = "TRUE" if port_status.online else "FALSE"
|
||||
return result
|
||||
estimate = estimate_consumable_life()
|
||||
if estimate.toner_exhausted:
|
||||
result.status_code = DERIVED_TONER_END
|
||||
result.display = "Toner End (estimated from page count)"
|
||||
result.online = "TRUE"
|
||||
return result
|
||||
if estimate.toner_low:
|
||||
result.status_code = DERIVED_TONER_LOW
|
||||
result.display = "Toner Low (estimated from page count)"
|
||||
result.online = "TRUE"
|
||||
return result
|
||||
result.status_code = _map_cups_to_status_code(state, reasons)
|
||||
result.display = ipp.get("printer-state-message", "")
|
||||
result.online = "TRUE"
|
||||
return result
|
||||
|
||||
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
|
||||
@ -9,6 +9,7 @@ import subprocess
|
||||
import sys
|
||||
import time
|
||||
|
||||
from python_pkg.brother_printer._cups_fallback import find_cups_printer_name
|
||||
from python_pkg.brother_printer._query import run_command_text
|
||||
from python_pkg.brother_printer.constants import (
|
||||
DIM,
|
||||
@ -17,7 +18,6 @@ from python_pkg.brother_printer.constants import (
|
||||
RESET,
|
||||
_out,
|
||||
)
|
||||
from python_pkg.brother_printer.cups_service import find_cups_printer_name
|
||||
from python_pkg.brother_printer.data_classes import CUPSJob, CUPSQueueStatus
|
||||
|
||||
# ── Queue inspection ─────────────────────────────────────────────────
|
||||
|
||||
@ -1,6 +1,7 @@
|
||||
"""CUPS service management, USB port status, and the CUPS-based fallback query.
|
||||
"""CUPS service management and raw USB port status.
|
||||
|
||||
Consumable life tracking lives in consumables.py.
|
||||
The CUPS-based fallback query lives in _cups_fallback.py, and consumable life
|
||||
tracking in consumables.py.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@ -11,30 +12,13 @@ import importlib
|
||||
import logging
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import time
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from python_pkg.brother_printer._query import (
|
||||
printer_info_from_cups,
|
||||
run_command_text,
|
||||
)
|
||||
from python_pkg.brother_printer.constants import (
|
||||
_CUPS_REASONS_TO_STATUS,
|
||||
_CUPS_STATE_TO_STATUS,
|
||||
_ERROR_REASON_MAP,
|
||||
BROTHER_USB_VENDOR_ID,
|
||||
DERIVED_CUPS_ERROR,
|
||||
DERIVED_TONER_END,
|
||||
DERIVED_TONER_LOW,
|
||||
)
|
||||
from python_pkg.brother_printer.consumables import estimate_consumable_life
|
||||
from python_pkg.brother_printer.data_classes import (
|
||||
USBPortStatus,
|
||||
USBResult,
|
||||
)
|
||||
from python_pkg.brother_printer.constants import BROTHER_USB_VENDOR_ID
|
||||
from python_pkg.brother_printer.data_classes import USBPortStatus
|
||||
|
||||
if TYPE_CHECKING:
|
||||
import types
|
||||
@ -232,175 +216,3 @@ def _query_usb_port_status_raw(cups_state: str = "") -> USBPortStatus | None:
|
||||
except (OSError, ValueError):
|
||||
logger.debug("USB port status query failed", exc_info=True)
|
||||
return None
|
||||
|
||||
|
||||
# ── IPP / CUPS attribute queries ────────────────────────────────────
|
||||
|
||||
|
||||
def _parse_ipp_attributes(output: str) -> dict[str, str]:
|
||||
"""Parse ipptool verbose output into an attribute dict."""
|
||||
attrs: dict[str, str] = {}
|
||||
for line in output.splitlines():
|
||||
match = re.match(r"\s+(\S+)\s+\([^)]+\)\s+=\s+(.*)", line)
|
||||
if match:
|
||||
attrs[match.group(1)] = match.group(2).strip()
|
||||
return attrs
|
||||
|
||||
|
||||
def _get_cups_ipp_status(printer_name: str) -> dict[str, str]:
|
||||
"""Query printer attributes via CUPS IPP using ipptool."""
|
||||
ipptool_path = shutil.which("ipptool")
|
||||
if not ipptool_path:
|
||||
return {}
|
||||
uri = f"ipp://localhost/printers/{printer_name}"
|
||||
try:
|
||||
r = subprocess.run(
|
||||
[ipptool_path, "-tv", uri, "get-printer-attributes.test"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=10,
|
||||
check=False,
|
||||
)
|
||||
return _parse_ipp_attributes(r.stdout)
|
||||
except (subprocess.TimeoutExpired, subprocess.SubprocessError, OSError):
|
||||
return {}
|
||||
|
||||
|
||||
def _get_cups_economode(printer_name: str) -> str:
|
||||
"""Query toner save mode setting via lpoptions."""
|
||||
lpoptions_path = shutil.which("lpoptions")
|
||||
if not lpoptions_path:
|
||||
return ""
|
||||
command = [lpoptions_path, "-p", printer_name, "-l"]
|
||||
for line in run_command_text(command).splitlines():
|
||||
if "conomode" in line.lower():
|
||||
match = re.search(r"\*(\w+)", line)
|
||||
if match:
|
||||
return "ON" if match.group(1).lower() == "true" else "OFF"
|
||||
return ""
|
||||
|
||||
|
||||
# ── Status code mapping ──────────────────────────────────────────────
|
||||
|
||||
|
||||
def _map_cups_to_status_code(state: str, reasons: str) -> str:
|
||||
"""Map CUPS state + reasons to a Brother PJL status code string."""
|
||||
for keyword, code in _CUPS_REASONS_TO_STATUS.items():
|
||||
if keyword in reasons.lower():
|
||||
return str(code)
|
||||
clean_state = re.sub(r"\(.*\)", "", state).strip().lower()
|
||||
return str(_CUPS_STATE_TO_STATUS.get(clean_state, 10001))
|
||||
|
||||
|
||||
def _cups_reasons_to_error(cups_reasons: str) -> tuple[str, str]:
|
||||
"""Map CUPS reason keywords to a (status_code, display) pair."""
|
||||
reasons_lower = cups_reasons.lower()
|
||||
for keywords, code, display in _ERROR_REASON_MAP:
|
||||
if any(kw in reasons_lower for kw in keywords):
|
||||
return code, display
|
||||
# Nothing recognised. Show what CUPS actually said rather than a bare
|
||||
# "Printer Error", which tells the reader nothing they can act on.
|
||||
detail = cups_reasons.strip()
|
||||
if detail and detail.lower() not in {"none", ""}:
|
||||
return DERIVED_CUPS_ERROR, f"Printer Error (CUPS reports: {detail})"
|
||||
return DERIVED_CUPS_ERROR, "Printer Error (CUPS gave no reason)"
|
||||
|
||||
|
||||
def _port_status_to_status_code(
|
||||
ps: USBPortStatus,
|
||||
cups_reasons: str,
|
||||
) -> tuple[str, str]:
|
||||
"""Map USB port status + CUPS reasons to (status_code, display)."""
|
||||
# The port status exposes only paper_empty/error/online bits, so anything
|
||||
# more specific than "out of paper" has to come from the CUPS reasons -
|
||||
# an error bit alone does not tell us the cover is open.
|
||||
if ps.paper_empty:
|
||||
return "41000", "No Paper"
|
||||
if ps.error:
|
||||
return _cups_reasons_to_error(cups_reasons)
|
||||
if not ps.online:
|
||||
return "10002", "Offline / Sleep"
|
||||
return "", ""
|
||||
|
||||
|
||||
# ── CUPS printer name discovery ──────────────────────────────────────
|
||||
|
||||
|
||||
def find_cups_printer_name() -> str:
|
||||
"""Find the CUPS queue name for a Brother printer."""
|
||||
lpstat_path = shutil.which("lpstat")
|
||||
if not lpstat_path:
|
||||
return ""
|
||||
for line in run_command_text([lpstat_path, "-v"]).splitlines():
|
||||
if "brother" in line.lower():
|
||||
match = re.match(r"device for (\S+):", line)
|
||||
if match:
|
||||
return match.group(1)
|
||||
return ""
|
||||
|
||||
|
||||
# ── CUPS-based USB fallback query ────────────────────────────────────
|
||||
|
||||
|
||||
def query_usb_via_cups() -> USBResult:
|
||||
"""Query USB printer status through CUPS when /dev/usb/lp* is unavailable."""
|
||||
_ensure_cups_running()
|
||||
printer_name = find_cups_printer_name()
|
||||
if not printer_name:
|
||||
return USBResult(
|
||||
error="No USB printer device at /dev/usb/lp*"
|
||||
" (usblp module not available)"
|
||||
" and no Brother printer found in CUPS.",
|
||||
)
|
||||
|
||||
pyusb_info = _get_pyusb_device_info()
|
||||
cups_info = printer_info_from_cups()
|
||||
|
||||
result = USBResult(
|
||||
device="cups",
|
||||
product=(
|
||||
pyusb_info.get("product")
|
||||
or cups_info.get("product")
|
||||
or "Brother Laser Printer"
|
||||
),
|
||||
serial=pyusb_info.get("serial") or cups_info.get("serial", ""),
|
||||
)
|
||||
|
||||
ipp = _get_cups_ipp_status(printer_name)
|
||||
state = ipp.get("printer-state", "")
|
||||
reasons = ipp.get("printer-state-reasons", "none")
|
||||
result.economode = _get_cups_economode(printer_name)
|
||||
|
||||
port_status = _query_usb_port_status_raw(state)
|
||||
if port_status is not None:
|
||||
result.port_status = port_status
|
||||
hw_code, hw_display = _port_status_to_status_code(
|
||||
port_status,
|
||||
reasons,
|
||||
)
|
||||
if hw_code:
|
||||
result.status_code = hw_code
|
||||
result.display = hw_display
|
||||
result.online = "TRUE" if port_status.online else "FALSE"
|
||||
return result
|
||||
estimate = estimate_consumable_life()
|
||||
if estimate.toner_exhausted:
|
||||
result.status_code = DERIVED_TONER_END
|
||||
result.display = "Toner End (estimated from page count)"
|
||||
result.online = "TRUE"
|
||||
return result
|
||||
if estimate.toner_low:
|
||||
result.status_code = DERIVED_TONER_LOW
|
||||
result.display = "Toner Low (estimated from page count)"
|
||||
result.online = "TRUE"
|
||||
return result
|
||||
result.status_code = _map_cups_to_status_code(state, reasons)
|
||||
result.display = ipp.get("printer-state-message", "")
|
||||
result.online = "TRUE"
|
||||
return result
|
||||
|
||||
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
|
||||
|
||||
@ -1,27 +1,27 @@
|
||||
"""Tests for brother_printer.cups_service module - part 2."""
|
||||
"""Tests for brother_printer._cups_fallback module - part 2."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from python_pkg.brother_printer.constants import (
|
||||
_CUPS_REASONS_TO_STATUS,
|
||||
DERIVED_CUPS_ERROR,
|
||||
get_status_info,
|
||||
)
|
||||
from python_pkg.brother_printer.cups_service import (
|
||||
from python_pkg.brother_printer._cups_fallback import (
|
||||
_cups_reasons_to_error,
|
||||
_get_cups_economode,
|
||||
_map_cups_to_status_code,
|
||||
_port_status_to_status_code,
|
||||
find_cups_printer_name,
|
||||
)
|
||||
from python_pkg.brother_printer.constants import (
|
||||
_CUPS_REASONS_TO_STATUS,
|
||||
DERIVED_CUPS_ERROR,
|
||||
get_status_info,
|
||||
)
|
||||
from python_pkg.brother_printer.data_classes import (
|
||||
USBPortStatus,
|
||||
)
|
||||
|
||||
MOD = "python_pkg.brother_printer.cups_service"
|
||||
MOD = "python_pkg.brother_printer._cups_fallback"
|
||||
|
||||
|
||||
# ── _get_cups_economode ──────────────────────────────────────────────
|
||||
|
||||
@ -1,22 +1,22 @@
|
||||
"""Tests for brother_printer.cups_service module - part 3 (query_usb_via_cups)."""
|
||||
"""Tests for brother_printer._cups_fallback module - part 3 (query_usb_via_cups)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from python_pkg.brother_printer._cups_fallback import (
|
||||
query_usb_via_cups,
|
||||
)
|
||||
from python_pkg.brother_printer.constants import (
|
||||
DERIVED_TONER_END,
|
||||
DERIVED_TONER_LOW,
|
||||
)
|
||||
from python_pkg.brother_printer.cups_service import (
|
||||
query_usb_via_cups,
|
||||
)
|
||||
from python_pkg.brother_printer.data_classes import (
|
||||
PageCountEstimate,
|
||||
USBPortStatus,
|
||||
)
|
||||
|
||||
MOD = "python_pkg.brother_printer.cups_service"
|
||||
MOD = "python_pkg.brother_printer._cups_fallback"
|
||||
|
||||
|
||||
# ── query_usb_via_cups ───────────────────────────────────────────────
|
||||
|
||||
@ -1,16 +1,16 @@
|
||||
"""Tests for brother_printer.cups_service module - part 4 (IPP queries)."""
|
||||
"""Tests for brother_printer._cups_fallback module - part 4 (IPP queries)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from python_pkg.brother_printer.cups_service import (
|
||||
from python_pkg.brother_printer._cups_fallback import (
|
||||
_get_cups_ipp_status,
|
||||
_parse_ipp_attributes,
|
||||
)
|
||||
|
||||
MOD = "python_pkg.brother_printer.cups_service"
|
||||
MOD = "python_pkg.brother_printer._cups_fallback"
|
||||
|
||||
|
||||
class TestParseIppAttributes:
|
||||
|
||||
@ -251,7 +251,7 @@ class TestQueryUsbPjl:
|
||||
@patch(f"{MOD}.find_usb_printer_dev", return_value=None)
|
||||
def test_no_dev_falls_back_to_cups(self, f: MagicMock) -> None:
|
||||
with patch(
|
||||
"python_pkg.brother_printer.cups_service.query_usb_via_cups",
|
||||
"python_pkg.brother_printer._cups_fallback.query_usb_via_cups",
|
||||
) as mock_cups:
|
||||
mock_cups.return_value = USBResult(device="cups")
|
||||
result = query_usb_pjl()
|
||||
|
||||
@ -227,10 +227,10 @@ def query_usb_pjl(max_retries: int = 2) -> USBResult:
|
||||
"""Query a Brother printer via PJL over /dev/usb/lp*."""
|
||||
dev_path = find_usb_printer_dev()
|
||||
if not dev_path:
|
||||
cups_service = importlib.import_module(
|
||||
"python_pkg.brother_printer.cups_service",
|
||||
cups_fallback = importlib.import_module(
|
||||
"python_pkg.brother_printer._cups_fallback",
|
||||
)
|
||||
return cups_service.query_usb_via_cups()
|
||||
return cups_fallback.query_usb_via_cups()
|
||||
|
||||
result = _init_usb_result(dev_path)
|
||||
if not os.access(dev_path, os.R_OK | os.W_OK):
|
||||
|
||||
Loading…
Reference in New Issue
Block a user