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

Split the brother_printer cups_service test modules under the cap

Browse Source 
test_cups_service.py 390 -> 172 (usblp port status to _port),
part2 261 -> 115 (status mapping and discovery out), part3 260 -> 105
(the four healthy-port cases out). part3 was one 235-line class, so it
split by method rather than by class.

The four part-numbered files are renamed for what they cover; three of
them now name _cups_fallback, matching where the code moved earlier.

The first run after the part2 split reported 400, not 402: the moved
classes needed subprocess and get_status_info that my hand-written
import block had dropped. Fixed, and the count is 402 again.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011w2Pj4Ek3ZNMQVYcFHgXDY
This commit is contained in:
Krzysztof Rudnicki 2026-08-16 17:19:55 +02:00
parent 65bc65c7e3
commit 607ba98c85
9 changed files with 538 additions and 420 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
docs/superpowers/contracts/brother-printer-test-cups-service-split-2026-08-16.json Normal file
View File

@ -0,0 +1,19 @@
{
"title": "Split the brother_printer cups_service test modules under the 250-line cap",
"objective": "Bring test_cups_service.py (390 lines), test_cups_service_part2.py (261) and test_cups_service_part3.py (260) under the repo-wide 250-line cap by partitioning them at class and method boundaries, and rename the part-numbered survivors after the concern they cover.",
"acceptance_criteria": [
"Every cups_service and _cups_fallback test module is 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.",
"No MOD constant changes: the source split in the previous commit already put every patched name in the module these files name.",
"No assertion is changed and no test is deleted, skipped or weakened; whole classes and whole methods move verbatim and only the import blocks are adjusted.",
"No suppression is added and no coverage exclusion is introduced."
],
"out_of_scope": [
"Changing what any test asserts.",
"Touching the brother_printer source, which is already under the cap.",
"Splitting test_display.py, test_usb_query.py, test_check_brother_printer.py or the consumables pair, which are separate commits.",
"Introducing a shared conftest.py; these classes build their mocks inline and need no shared fixture builders."
],
"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."
}

50
docs/superpowers/evidence/brother-printer-test-cups-service-split-2026-08-16.json Normal file
View File

@ -0,0 +1,50 @@
{
"intent": "Bring the three over-cap cups_service test modules under the 250-line cap by partitioning them, with no change to what is tested.",
"scope": [
"python_pkg/brother_printer/tests/test_cups_service.py (390 -> 172)",
"python_pkg/brother_printer/tests/test_cups_service_port.py (new)",
"python_pkg/brother_printer/tests/test_cups_service_part2.py -> test_cups_economode.py (261 -> 115)",
"python_pkg/brother_printer/tests/test_cups_status_mapping.py (new)",
"python_pkg/brother_printer/tests/test_cups_service_part3.py -> test_cups_fallback_query.py (260 -> 105)",
"python_pkg/brother_printer/tests/test_cups_fallback_port_ok.py (new)",
"python_pkg/brother_printer/tests/test_cups_service_part4.py -> test_cups_ipp.py (rename only)",
"Non-goal: no assertion changed, no test deleted, skipped or weakened"
],
"changes": [
"test_cups_service.py split at TestPortStatusViaUsblp: the pyusb device info and CUPS service control tests stay, the usblp ioctl port-status tests go to test_cups_service_port.py.",
"test_cups_service_part2.py split at TestCupsReasonsToError: economode and CUPS-state mapping stay, the error-reason mapping, port-status-to-code mapping and printer-name discovery go to test_cups_status_mapping.py.",
"test_cups_service_part3.py was a single 235-line class, so it split by method at test_port_ok_toner_exhausted: the no-port-status and hardware-error cases stay, the four healthy-port cases become TestQueryUsbViaCupsPortOk in test_cups_fallback_port_ok.py.",
"Renamed the four part-numbered files after what they actually cover: test_cups_economode.py, test_cups_fallback_query.py, test_cups_ipp.py, plus the two new names.",
"Import blocks trimmed per file to what each uses. No MOD constant changed; the source split in the previous commits already put every patched name in the module these files name."
],
"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."
},
{
"command": "PYTHONPATH=. python3 -m pytest python_pkg/brother_printer/tests/test_cups_status_mapping.py -q",
"result": "fail then pass",
"evidence": "The first run after the part2 split reported 2 failed, 400 passed: the moved classes needed subprocess (for TimeoutExpired) and get_status_info, which my hand-written import block had dropped. Both added; the suite returned to 402. Recorded because it is exactly the failure the pass-count check exists to catch."
},
{
"command": "bash ~/utils/scripts/check_file_length.sh --all | grep brother_printer",
"result": "pass",
"evidence": "No cups_service or _cups_fallback test module is listed any more; brother_printer is down to five over-cap files, all of them other test modules."
},
{
"command": "pre-commit run --files <the touched test files and artifacts>",
"result": "pass",
"evidence": "ruff, ruff format, mypy, pylint, bandit, name-tests-test and pytest-with-coverage all pass."
}
],
"risks": [
"Hand-trimmed import blocks are the live risk in this kind of split, as the two failures above show. A dropped name fails loudly at collection or call time, so the pass count catches it; the danger would be trimming an import that only a skipped test used.",
"Three of these files are now named after _cups_fallback rather than cups_service, matching where the code moved in an earlier commit."
],
"rollback": [
"git revert the commit; the three originals return at 390, 261 and 260 lines and the three new modules disappear.",
"After rollback, re-run the package suite and confirm it still reports 402 passed."
]
}

115
python_pkg/brother_printer/tests/test_cups_economode.py Normal file
View File

@ -0,0 +1,115 @@
"""Tests for brother_printer._cups_fallback - economode and state mapping."""
from __future__ import annotations
import subprocess
from unittest.mock import MagicMock, patch
from python_pkg.brother_printer._cups_fallback import (
_get_cups_economode,
_map_cups_to_status_code,
)
from python_pkg.brother_printer.constants import (
_CUPS_REASONS_TO_STATUS,
get_status_info,
)
MOD = "python_pkg.brother_printer._cups_fallback"
# ── _get_cups_economode ──────────────────────────────────────────────
class TestGetCupsEconomode:
"""Tests for _get_cups_economode."""
@patch(f"{MOD}.shutil.which", return_value=None)
def test_no_lpoptions(self, m: MagicMock) -> None:
assert _get_cups_economode("Brother") == ""
@patch("python_pkg.brother_printer._query.subprocess.run")
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/lpoptions")
def test_economode_on(self, w: MagicMock, mock_run: MagicMock) -> None:
mock_run.return_value = MagicMock(
stdout="BREconomode/Toner Save Mode: *True False\n"
)
assert _get_cups_economode("Brother") == "ON"
@patch("python_pkg.brother_printer._query.subprocess.run")
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/lpoptions")
def test_economode_off(self, w: MagicMock, mock_run: MagicMock) -> None:
mock_run.return_value = MagicMock(
stdout="BREconomode/Toner Save Mode: True *False\n"
)
assert _get_cups_economode("Brother") == "OFF"
@patch("python_pkg.brother_printer._query.subprocess.run")
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/lpoptions")
def test_no_economode_line(self, w: MagicMock, mock_run: MagicMock) -> None:
mock_run.return_value = MagicMock(
stdout="Resolution/Output Resolution: 600dpi *1200dpi\n"
)
assert _get_cups_economode("Brother") == ""
@patch("python_pkg.brother_printer._query.subprocess.run")
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/lpoptions")
def test_economode_no_star_match(self, w: MagicMock, mock_run: MagicMock) -> None:
mock_run.return_value = MagicMock(
stdout="BREconomode/Toner Save Mode: True False\n"
)
assert _get_cups_economode("Brother") == ""
@patch("python_pkg.brother_printer._query.subprocess.run")
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/lpoptions")
def test_timeout(self, w: MagicMock, mock_run: MagicMock) -> None:
mock_run.side_effect = subprocess.TimeoutExpired("lpoptions", 5)
assert _get_cups_economode("Brother") == ""
@patch("python_pkg.brother_printer._query.subprocess.run")
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/lpoptions")
def test_oserror(self, w: MagicMock, mock_run: MagicMock) -> None:
mock_run.side_effect = OSError("fail")
assert _get_cups_economode("Brother") == ""
# ── _map_cups_to_status_code ─────────────────────────────────────────
class TestMapCupsToStatusCode:
"""Tests for _map_cups_to_status_code.
Every code here must exist in BROTHER_STATUS_CODES; mapping CUPS onto a
code the table does not know silently degrades to "unknown status".
"""
def test_reason_match(self) -> None:
assert _map_cups_to_status_code("idle", "toner-low-report") == "40038"
def test_state_match(self) -> None:
assert _map_cups_to_status_code("idle", "none") == "10001"
def test_processing_state(self) -> None:
assert _map_cups_to_status_code("processing", "none") == "10023"
def test_stopped_state(self) -> None:
assert _map_cups_to_status_code("stopped", "none") == "40079"
def test_unknown_state(self) -> None:
assert _map_cups_to_status_code("mystery", "none") == "10001"
def test_state_with_parenthetical(self) -> None:
assert _map_cups_to_status_code("idle (on fire)", "none") == "10001"
def test_every_mapped_code_is_resolvable(self) -> None:
"""Guard the class of bug where a mapping points at a deleted code."""
for state in ("idle", "processing", "stopped"):
code = _map_cups_to_status_code(state, "none")
_, text, _ = get_status_info(code)
assert "Unknown" not in text
for reason in _CUPS_REASONS_TO_STATUS:
code = _map_cups_to_status_code("idle", reason)
_, text, _ = get_status_info(code)
assert "Unknown" not in text
# ── _cups_reasons_to_error ───────────────────────────────────────────

105
python_pkg/brother_printer/tests/test_cups_service_part3.py → python_pkg/brother_printer/tests/test_cups_fallback_port_ok.py
View File

@ -1,12 +1,15 @@
"""Tests for brother_printer._cups_fallback module - part 3 (query_usb_via_cups)."""
"""Tests for brother_printer._cups_fallback - fallback with a healthy port.
Split from test_cups_service_part3.py under the 250-line cap. When the USB port
reports no hardware fault, the status has to be derived from the page-count
estimate or from the CUPS state, which is what these cover.
"""
from __future__ import annotations
from unittest.mock import MagicMock, patch
from unittest.mock import patch
from python_pkg.brother_printer._cups_fallback import (
query_usb_via_cups,
)
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,
@ -19,96 +22,8 @@ from python_pkg.brother_printer.data_classes import (
MOD = "python_pkg.brother_printer._cups_fallback"
# ── query_usb_via_cups ───────────────────────────────────────────────
class TestQueryUsbViaCups:
"""Tests for query_usb_via_cups."""
@patch(f"{MOD}.find_cups_printer_name", return_value="")
@patch(f"{MOD}._ensure_cups_running", return_value=True)
def test_no_printer(self, e: MagicMock, f: MagicMock) -> None:
result = query_usb_via_cups()
assert result.error != ""
def test_no_port_status_idle(self) -> None:
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={"product": "HL-1110", "serial": "ABC"},
),
patch(
f"{MOD}._get_cups_ipp_status",
return_value={
"printer-state": "idle",
"printer-state-reasons": "none",
"printer-state-message": "Ready",
},
),
patch(f"{MOD}._get_cups_economode", return_value="ON"),
patch(f"{MOD}._query_usb_port_status_raw", return_value=None),
):
result = query_usb_via_cups()
assert result.online == "TRUE"
assert result.product == "HL-1110"
assert result.economode == "ON"
def test_no_port_status_stopped(self) -> None:
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={"product": "", "serial": ""},
),
patch(
f"{MOD}._get_cups_ipp_status",
return_value={
"printer-state": "stopped",
"printer-state-reasons": "none",
},
),
patch(f"{MOD}._get_cups_economode", return_value=""),
patch(f"{MOD}._query_usb_port_status_raw", return_value=None),
):
result = query_usb_via_cups()
assert result.online == "FALSE"
assert result.product == "Brother Laser Printer"
def test_port_status_hw_error(self) -> None:
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={"product": "", "serial": ""},
),
patch(
f"{MOD}._get_cups_ipp_status",
return_value={
"printer-state": "stopped",
"printer-state-reasons": "none",
},
),
patch(f"{MOD}._get_cups_economode", return_value=""),
patch(
f"{MOD}._query_usb_port_status_raw",
return_value=USBPortStatus(
error=True,
paper_empty=True,
online=False,
raw_byte=0x20,
),
),
):
result = query_usb_via_cups()
assert result.status_code == "41000"
assert result.online == "FALSE"
class TestQueryUsbViaCupsPortOk:
"""query_usb_via_cups when the USB port status reports no fault."""
def test_port_ok_toner_exhausted(self) -> None:
with (

102
python_pkg/brother_printer/tests/test_cups_fallback_query.py Normal file
View File

@ -0,0 +1,102 @@
"""Tests for brother_printer._cups_fallback - the CUPS fallback query."""
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.data_classes import USBPortStatus
MOD = "python_pkg.brother_printer._cups_fallback"
# ── query_usb_via_cups ───────────────────────────────────────────────
class TestQueryUsbViaCups:
"""Tests for query_usb_via_cups."""
@patch(f"{MOD}.find_cups_printer_name", return_value="")
@patch(f"{MOD}._ensure_cups_running", return_value=True)
def test_no_printer(self, e: MagicMock, f: MagicMock) -> None:
result = query_usb_via_cups()
assert result.error != ""
def test_no_port_status_idle(self) -> None:
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={"product": "HL-1110", "serial": "ABC"},
),
patch(
f"{MOD}._get_cups_ipp_status",
return_value={
"printer-state": "idle",
"printer-state-reasons": "none",
"printer-state-message": "Ready",
},
),
patch(f"{MOD}._get_cups_economode", return_value="ON"),
patch(f"{MOD}._query_usb_port_status_raw", return_value=None),
):
result = query_usb_via_cups()
assert result.online == "TRUE"
assert result.product == "HL-1110"
assert result.economode == "ON"
def test_no_port_status_stopped(self) -> None:
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={"product": "", "serial": ""},
),
patch(
f"{MOD}._get_cups_ipp_status",
return_value={
"printer-state": "stopped",
"printer-state-reasons": "none",
},
),
patch(f"{MOD}._get_cups_economode", return_value=""),
patch(f"{MOD}._query_usb_port_status_raw", return_value=None),
):
result = query_usb_via_cups()
assert result.online == "FALSE"
assert result.product == "Brother Laser Printer"
def test_port_status_hw_error(self) -> None:
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={"product": "", "serial": ""},
),
patch(
f"{MOD}._get_cups_ipp_status",
return_value={
"printer-state": "stopped",
"printer-state-reasons": "none",
},
),
patch(f"{MOD}._get_cups_economode", return_value=""),
patch(
f"{MOD}._query_usb_port_status_raw",
return_value=USBPortStatus(
error=True,
paper_empty=True,
online=False,
raw_byte=0x20,
),
),
):
result = query_usb_via_cups()
assert result.status_code == "41000"
assert result.online == "FALSE"

2
python_pkg/brother_printer/tests/test_cups_service_part4.py → python_pkg/brother_printer/tests/test_cups_ipp.py
View File

@ -1,4 +1,4 @@
"""Tests for brother_printer._cups_fallback module - part 4 (IPP queries)."""
"""Tests for brother_printer._cups_fallback - IPP attribute queries."""
from __future__ import annotations

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

@ -1,21 +1,16 @@
"""Tests for brother_printer.cups_service module."""
"""Tests for brother_printer.cups_service - pyusb info and service control."""
from __future__ import annotations
import os
import subprocess
from unittest.mock import MagicMock, patch
from python_pkg.brother_printer.cups_service import (
_cups_is_busy,
_ensure_cups_running,
_get_pyusb_device_info,
_port_status_via_usblp,
_query_usb_port_status_raw,
is_cups_scheduler_running,
start_cups,
)
from python_pkg.brother_printer.data_classes import USBPortStatus
MOD = "python_pkg.brother_printer.cups_service"
@ -175,216 +170,3 @@ class TestEnsureCupsRunning:
@patch(f"{MOD}.is_cups_scheduler_running", return_value=False)
def test_start_fails(self, is_running: MagicMock, st: MagicMock) -> None:
assert _ensure_cups_running() is False
class TestPortStatusViaUsblp:
"""The zero-side-effect path: read the status byte from /dev/usb/lp*."""
@patch(f"{MOD}.Path")
def test_no_dev_dir(self, mock_path: MagicMock) -> None:
mock_path.return_value.is_dir.return_value = False
assert _port_status_via_usblp() is None
@patch(f"{MOD}.Path")
def test_no_devices(self, mock_path: MagicMock) -> None:
mock_path.return_value.is_dir.return_value = True
mock_path.return_value.glob.return_value = []
assert _port_status_via_usblp() is None
@patch(f"{MOD}.os.close")
@patch(f"{MOD}.fcntl.ioctl")
@patch(f"{MOD}.os.open", return_value=7)
@patch(f"{MOD}.Path")
def test_success(
self,
mock_path: MagicMock,
mock_open: MagicMock,
mock_ioctl: MagicMock,
mock_close: MagicMock,
) -> None:
mock_path.return_value.is_dir.return_value = True
mock_path.return_value.glob.return_value = ["/dev/usb/lp0"]
def fill(fd: int, req: int, buf: bytearray) -> None:
buf[0] = 0x18
mock_ioctl.side_effect = fill
result = _port_status_via_usblp()
assert result is not None
assert result.online is True
assert result.error is False
assert result.paper_empty is False
# Must never block: the printer may be mid-job.
assert mock_open.call_args[0][1] & os.O_NONBLOCK
mock_close.assert_called_once_with(7)
@patch(f"{MOD}.os.open", side_effect=OSError("busy"))
@patch(f"{MOD}.Path")
def test_open_fails(self, mock_path: MagicMock, mock_open: MagicMock) -> None:
mock_path.return_value.is_dir.return_value = True
mock_path.return_value.glob.return_value = ["/dev/usb/lp0"]
assert _port_status_via_usblp() is None
class TestQueryUsbPortStatusRaw:
@patch(f"{MOD}._port_status_via_usblp", return_value=USBPortStatus(raw_byte=0x18))
def test_prefers_usblp(self, m: MagicMock) -> None:
"""The free path wins; pyusb is never touched."""
result = _query_usb_port_status_raw("idle")
assert result is not None
assert result.raw_byte == 0x18
@patch(f"{MOD}._port_status_via_usblp", return_value=None)
def test_never_probes_while_printing(self, m: MagicMock) -> None:
"""Regression: probing mid-job used to stop CUPS and kill the print."""
import sys as _sys
mock_usb = MagicMock()
with patch.dict(
_sys.modules,
{"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util},
):
assert _query_usb_port_status_raw("processing") is None
mock_usb.core.find.assert_not_called()
@patch(f"{MOD}._port_status_via_usblp", return_value=None)
def test_import_error(self, m: MagicMock) -> None:
with patch.dict(
"sys.modules", {"usb": None, "usb.core": None, "usb.util": None}
):
assert _query_usb_port_status_raw("idle") is None
@patch(f"{MOD}._port_status_via_usblp", return_value=None)
def test_dev_none(self, m: MagicMock) -> None:
import sys as _sys
mock_usb = MagicMock()
mock_usb.core.find.return_value = None
with patch.dict(
_sys.modules,
{"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util},
):
assert _query_usb_port_status_raw("idle") is None
@patch(f"{MOD}._port_status_via_usblp", return_value=None)
def test_success_reattaches_driver(self, m: MagicMock) -> None:
"""usblp must get its device back, or every later run hits the fallback."""
import sys as _sys
mock_usb = MagicMock()
mock_dev = MagicMock()
mock_dev.is_kernel_driver_active.return_value = True
mock_dev.ctrl_transfer.return_value = [0x18]
mock_usb.core.find.return_value = mock_dev
mock_usb.core.USBError = type("USBError", (Exception,), {})
with patch.dict(
_sys.modules,
{"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util},
):
result = _query_usb_port_status_raw("idle")
assert result is not None
assert result.online is True
mock_dev.reset.assert_not_called()
mock_dev.attach_kernel_driver.assert_called_once_with(0)
@patch(f"{MOD}._port_status_via_usblp", return_value=None)
def test_kernel_driver_not_active(self, m: MagicMock) -> None:
import sys as _sys
mock_usb = MagicMock()
mock_dev = MagicMock()
mock_dev.is_kernel_driver_active.return_value = False
mock_dev.ctrl_transfer.return_value = [0x18]
mock_usb.core.find.return_value = mock_dev
mock_usb.core.USBError = type("USBError", (Exception,), {})
with patch.dict(
_sys.modules,
{"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util},
):
result = _query_usb_port_status_raw("idle")
assert result is not None
# Nothing was detached, so nothing should be re-attached.
mock_dev.attach_kernel_driver.assert_not_called()
@patch(f"{MOD}._port_status_via_usblp", return_value=None)
def test_kernel_driver_usberror(self, m: MagicMock) -> None:
import sys as _sys
mock_usb = MagicMock()
mock_dev = MagicMock()
usb_error_cls = type("USBError", (Exception,), {})
mock_dev.is_kernel_driver_active.side_effect = usb_error_cls("err")
mock_dev.ctrl_transfer.return_value = [0x18]
mock_usb.core.find.return_value = mock_dev
mock_usb.core.USBError = usb_error_cls
with patch.dict(
_sys.modules,
{"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util},
):
assert _query_usb_port_status_raw("idle") is not None
@patch(f"{MOD}._port_status_via_usblp", return_value=None)
def test_oserror_during_transfer(self, m: MagicMock) -> None:
import sys as _sys
mock_usb = MagicMock()
mock_dev = MagicMock()
mock_dev.is_kernel_driver_active.return_value = False
mock_usb.core.find.return_value = mock_dev
mock_usb.core.USBError = type("USBError", (Exception,), {})
mock_usb.util.claim_interface.side_effect = OSError("usb fail")
with patch.dict(
_sys.modules,
{"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util},
):
assert _query_usb_port_status_raw("idle") is None
@patch(f"{MOD}._port_status_via_usblp", return_value=None)
def test_attach_failure_is_survivable(self, m: MagicMock) -> None:
import sys as _sys
mock_usb = MagicMock()
mock_dev = MagicMock()
mock_dev.is_kernel_driver_active.return_value = True
mock_dev.ctrl_transfer.return_value = [0x18]
mock_dev.attach_kernel_driver.side_effect = OSError("cannot reattach")
mock_usb.core.find.return_value = mock_dev
mock_usb.core.USBError = type("USBError", (Exception,), {})
with patch.dict(
_sys.modules,
{"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util},
):
assert _query_usb_port_status_raw("idle") is not None
class TestCupsIsBusy:
def test_processing(self) -> None:
assert _cups_is_busy("processing") is True
def test_printing_text(self) -> None:
assert _cups_is_busy("now printing Brother-98") is True
def test_idle(self) -> None:
assert _cups_is_busy("idle") is False
def test_empty(self) -> None:
assert _cups_is_busy("") is False
class TestPortStatusViaUsblpIoctlFailure:
@patch(f"{MOD}.os.close")
@patch(f"{MOD}.fcntl.ioctl", side_effect=OSError("not supported"))
@patch(f"{MOD}.os.open", return_value=7)
@patch(f"{MOD}.Path")
def test_ioctl_failure_still_closes_fd(
self,
mock_path: MagicMock,
mock_open: MagicMock,
mock_ioctl: MagicMock,
mock_close: MagicMock,
) -> None:
"""A failed ioctl must not leak the fd, or usblp stays busy."""
mock_path.return_value.is_dir.return_value = True
mock_path.return_value.glob.return_value = ["/dev/usb/lp0"]
assert _port_status_via_usblp() is None
mock_close.assert_called_once_with(7)

233
python_pkg/brother_printer/tests/test_cups_service_port.py Normal file
View File

@ -0,0 +1,233 @@
"""Tests for brother_printer.cups_service - USB port status via usblp.
Split from test_cups_service.py under the 250-line cap. These tests read the
printer's IEEE 1284 status byte through an ioctl on /dev/usb/lp*, so they patch
fcntl, os and Path on the module under test.
"""
from __future__ import annotations
import os
from unittest.mock import MagicMock, patch
from python_pkg.brother_printer.cups_service import (
_cups_is_busy,
_port_status_via_usblp,
_query_usb_port_status_raw,
)
from python_pkg.brother_printer.data_classes import USBPortStatus
MOD = "python_pkg.brother_printer.cups_service"
class TestPortStatusViaUsblp:
"""The zero-side-effect path: read the status byte from /dev/usb/lp*."""
@patch(f"{MOD}.Path")
def test_no_dev_dir(self, mock_path: MagicMock) -> None:
mock_path.return_value.is_dir.return_value = False
assert _port_status_via_usblp() is None
@patch(f"{MOD}.Path")
def test_no_devices(self, mock_path: MagicMock) -> None:
mock_path.return_value.is_dir.return_value = True
mock_path.return_value.glob.return_value = []
assert _port_status_via_usblp() is None
@patch(f"{MOD}.os.close")
@patch(f"{MOD}.fcntl.ioctl")
@patch(f"{MOD}.os.open", return_value=7)
@patch(f"{MOD}.Path")
def test_success(
self,
mock_path: MagicMock,
mock_open: MagicMock,
mock_ioctl: MagicMock,
mock_close: MagicMock,
) -> None:
mock_path.return_value.is_dir.return_value = True
mock_path.return_value.glob.return_value = ["/dev/usb/lp0"]
def fill(fd: int, req: int, buf: bytearray) -> None:
buf[0] = 0x18
mock_ioctl.side_effect = fill
result = _port_status_via_usblp()
assert result is not None
assert result.online is True
assert result.error is False
assert result.paper_empty is False
# Must never block: the printer may be mid-job.
assert mock_open.call_args[0][1] & os.O_NONBLOCK
mock_close.assert_called_once_with(7)
@patch(f"{MOD}.os.open", side_effect=OSError("busy"))
@patch(f"{MOD}.Path")
def test_open_fails(self, mock_path: MagicMock, mock_open: MagicMock) -> None:
mock_path.return_value.is_dir.return_value = True
mock_path.return_value.glob.return_value = ["/dev/usb/lp0"]
assert _port_status_via_usblp() is None
class TestQueryUsbPortStatusRaw:
@patch(f"{MOD}._port_status_via_usblp", return_value=USBPortStatus(raw_byte=0x18))
def test_prefers_usblp(self, m: MagicMock) -> None:
"""The free path wins; pyusb is never touched."""
result = _query_usb_port_status_raw("idle")
assert result is not None
assert result.raw_byte == 0x18
@patch(f"{MOD}._port_status_via_usblp", return_value=None)
def test_never_probes_while_printing(self, m: MagicMock) -> None:
"""Regression: probing mid-job used to stop CUPS and kill the print."""
import sys as _sys
mock_usb = MagicMock()
with patch.dict(
_sys.modules,
{"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util},
):
assert _query_usb_port_status_raw("processing") is None
mock_usb.core.find.assert_not_called()
@patch(f"{MOD}._port_status_via_usblp", return_value=None)
def test_import_error(self, m: MagicMock) -> None:
with patch.dict(
"sys.modules", {"usb": None, "usb.core": None, "usb.util": None}
):
assert _query_usb_port_status_raw("idle") is None
@patch(f"{MOD}._port_status_via_usblp", return_value=None)
def test_dev_none(self, m: MagicMock) -> None:
import sys as _sys
mock_usb = MagicMock()
mock_usb.core.find.return_value = None
with patch.dict(
_sys.modules,
{"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util},
):
assert _query_usb_port_status_raw("idle") is None
@patch(f"{MOD}._port_status_via_usblp", return_value=None)
def test_success_reattaches_driver(self, m: MagicMock) -> None:
"""usblp must get its device back, or every later run hits the fallback."""
import sys as _sys
mock_usb = MagicMock()
mock_dev = MagicMock()
mock_dev.is_kernel_driver_active.return_value = True
mock_dev.ctrl_transfer.return_value = [0x18]
mock_usb.core.find.return_value = mock_dev
mock_usb.core.USBError = type("USBError", (Exception,), {})
with patch.dict(
_sys.modules,
{"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util},
):
result = _query_usb_port_status_raw("idle")
assert result is not None
assert result.online is True
mock_dev.reset.assert_not_called()
mock_dev.attach_kernel_driver.assert_called_once_with(0)
@patch(f"{MOD}._port_status_via_usblp", return_value=None)
def test_kernel_driver_not_active(self, m: MagicMock) -> None:
import sys as _sys
mock_usb = MagicMock()
mock_dev = MagicMock()
mock_dev.is_kernel_driver_active.return_value = False
mock_dev.ctrl_transfer.return_value = [0x18]
mock_usb.core.find.return_value = mock_dev
mock_usb.core.USBError = type("USBError", (Exception,), {})
with patch.dict(
_sys.modules,
{"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util},
):
result = _query_usb_port_status_raw("idle")
assert result is not None
# Nothing was detached, so nothing should be re-attached.
mock_dev.attach_kernel_driver.assert_not_called()
@patch(f"{MOD}._port_status_via_usblp", return_value=None)
def test_kernel_driver_usberror(self, m: MagicMock) -> None:
import sys as _sys
mock_usb = MagicMock()
mock_dev = MagicMock()
usb_error_cls = type("USBError", (Exception,), {})
mock_dev.is_kernel_driver_active.side_effect = usb_error_cls("err")
mock_dev.ctrl_transfer.return_value = [0x18]
mock_usb.core.find.return_value = mock_dev
mock_usb.core.USBError = usb_error_cls
with patch.dict(
_sys.modules,
{"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util},
):
assert _query_usb_port_status_raw("idle") is not None
@patch(f"{MOD}._port_status_via_usblp", return_value=None)
def test_oserror_during_transfer(self, m: MagicMock) -> None:
import sys as _sys
mock_usb = MagicMock()
mock_dev = MagicMock()
mock_dev.is_kernel_driver_active.return_value = False
mock_usb.core.find.return_value = mock_dev
mock_usb.core.USBError = type("USBError", (Exception,), {})
mock_usb.util.claim_interface.side_effect = OSError("usb fail")
with patch.dict(
_sys.modules,
{"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util},
):
assert _query_usb_port_status_raw("idle") is None
@patch(f"{MOD}._port_status_via_usblp", return_value=None)
def test_attach_failure_is_survivable(self, m: MagicMock) -> None:
import sys as _sys
mock_usb = MagicMock()
mock_dev = MagicMock()
mock_dev.is_kernel_driver_active.return_value = True
mock_dev.ctrl_transfer.return_value = [0x18]
mock_dev.attach_kernel_driver.side_effect = OSError("cannot reattach")
mock_usb.core.find.return_value = mock_dev
mock_usb.core.USBError = type("USBError", (Exception,), {})
with patch.dict(
_sys.modules,
{"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util},
):
assert _query_usb_port_status_raw("idle") is not None
class TestCupsIsBusy:
def test_processing(self) -> None:
assert _cups_is_busy("processing") is True
def test_printing_text(self) -> None:
assert _cups_is_busy("now printing Brother-98") is True
def test_idle(self) -> None:
assert _cups_is_busy("idle") is False
def test_empty(self) -> None:
assert _cups_is_busy("") is False
class TestPortStatusViaUsblpIoctlFailure:
@patch(f"{MOD}.os.close")
@patch(f"{MOD}.fcntl.ioctl", side_effect=OSError("not supported"))
@patch(f"{MOD}.os.open", return_value=7)
@patch(f"{MOD}.Path")
def test_ioctl_failure_still_closes_fd(
self,
mock_path: MagicMock,
mock_open: MagicMock,
mock_ioctl: MagicMock,
mock_close: MagicMock,
) -> None:
"""A failed ioctl must not leak the fd, or usblp stays busy."""
mock_path.return_value.is_dir.return_value = True
mock_path.return_value.glob.return_value = ["/dev/usb/lp0"]
assert _port_status_via_usblp() is None
mock_close.assert_called_once_with(7)

112
python_pkg/brother_printer/tests/test_cups_service_part2.py → python_pkg/brother_printer/tests/test_cups_status_mapping.py
View File

@ -1,4 +1,9 @@
"""Tests for brother_printer._cups_fallback module - part 2."""
"""Tests for brother_printer._cups_fallback - status mapping and discovery.
Split from test_cups_service_part2.py under the 250-line cap. These cover the
mapping from what CUPS says went wrong onto the Brother PJL status codes the
report displays, plus finding the Brother queue name.
"""
from __future__ import annotations
@ -7,121 +12,18 @@ from unittest.mock import MagicMock, patch
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,
)
from python_pkg.brother_printer.data_classes import USBPortStatus
MOD = "python_pkg.brother_printer._cups_fallback"
# ── _get_cups_economode ──────────────────────────────────────────────
class TestGetCupsEconomode:
"""Tests for _get_cups_economode."""
@patch(f"{MOD}.shutil.which", return_value=None)
def test_no_lpoptions(self, m: MagicMock) -> None:
assert _get_cups_economode("Brother") == ""
@patch("python_pkg.brother_printer._query.subprocess.run")
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/lpoptions")
def test_economode_on(self, w: MagicMock, mock_run: MagicMock) -> None:
mock_run.return_value = MagicMock(
stdout="BREconomode/Toner Save Mode: *True False\n"
)
assert _get_cups_economode("Brother") == "ON"
@patch("python_pkg.brother_printer._query.subprocess.run")
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/lpoptions")
def test_economode_off(self, w: MagicMock, mock_run: MagicMock) -> None:
mock_run.return_value = MagicMock(
stdout="BREconomode/Toner Save Mode: True *False\n"
)
assert _get_cups_economode("Brother") == "OFF"
@patch("python_pkg.brother_printer._query.subprocess.run")
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/lpoptions")
def test_no_economode_line(self, w: MagicMock, mock_run: MagicMock) -> None:
mock_run.return_value = MagicMock(
stdout="Resolution/Output Resolution: 600dpi *1200dpi\n"
)
assert _get_cups_economode("Brother") == ""
@patch("python_pkg.brother_printer._query.subprocess.run")
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/lpoptions")
def test_economode_no_star_match(self, w: MagicMock, mock_run: MagicMock) -> None:
mock_run.return_value = MagicMock(
stdout="BREconomode/Toner Save Mode: True False\n"
)
assert _get_cups_economode("Brother") == ""
@patch("python_pkg.brother_printer._query.subprocess.run")
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/lpoptions")
def test_timeout(self, w: MagicMock, mock_run: MagicMock) -> None:
mock_run.side_effect = subprocess.TimeoutExpired("lpoptions", 5)
assert _get_cups_economode("Brother") == ""
@patch("python_pkg.brother_printer._query.subprocess.run")
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/lpoptions")
def test_oserror(self, w: MagicMock, mock_run: MagicMock) -> None:
mock_run.side_effect = OSError("fail")
assert _get_cups_economode("Brother") == ""
# ── _map_cups_to_status_code ─────────────────────────────────────────
class TestMapCupsToStatusCode:
"""Tests for _map_cups_to_status_code.
Every code here must exist in BROTHER_STATUS_CODES; mapping CUPS onto a
code the table does not know silently degrades to "unknown status".
"""
def test_reason_match(self) -> None:
assert _map_cups_to_status_code("idle", "toner-low-report") == "40038"
def test_state_match(self) -> None:
assert _map_cups_to_status_code("idle", "none") == "10001"
def test_processing_state(self) -> None:
assert _map_cups_to_status_code("processing", "none") == "10023"
def test_stopped_state(self) -> None:
assert _map_cups_to_status_code("stopped", "none") == "40079"
def test_unknown_state(self) -> None:
assert _map_cups_to_status_code("mystery", "none") == "10001"
def test_state_with_parenthetical(self) -> None:
assert _map_cups_to_status_code("idle (on fire)", "none") == "10001"
def test_every_mapped_code_is_resolvable(self) -> None:
"""Guard the class of bug where a mapping points at a deleted code."""
for state in ("idle", "processing", "stopped"):
code = _map_cups_to_status_code(state, "none")
_, text, _ = get_status_info(code)
assert "Unknown" not in text
for reason in _CUPS_REASONS_TO_STATUS:
code = _map_cups_to_status_code("idle", reason)
_, text, _ = get_status_info(code)
assert "Unknown" not in text
# ── _cups_reasons_to_error ───────────────────────────────────────────
class TestCupsReasonsToError:
"""Tests for _cups_reasons_to_error."""