diff --git a/docs/superpowers/contracts/brother-printer-test-cups-service-split-2026-08-16.json b/docs/superpowers/contracts/brother-printer-test-cups-service-split-2026-08-16.json new file mode 100644 index 00000000..b7b8acc9 --- /dev/null +++ b/docs/superpowers/contracts/brother-printer-test-cups-service-split-2026-08-16.json @@ -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." +} diff --git a/docs/superpowers/evidence/brother-printer-test-cups-service-split-2026-08-16.json b/docs/superpowers/evidence/brother-printer-test-cups-service-split-2026-08-16.json new file mode 100644 index 00000000..828f60e6 --- /dev/null +++ b/docs/superpowers/evidence/brother-printer-test-cups-service-split-2026-08-16.json @@ -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 ", + "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." + ] +} diff --git a/python_pkg/brother_printer/tests/test_cups_economode.py b/python_pkg/brother_printer/tests/test_cups_economode.py new file mode 100644 index 00000000..7d4e6550 --- /dev/null +++ b/python_pkg/brother_printer/tests/test_cups_economode.py @@ -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 ─────────────────────────────────────────── diff --git a/python_pkg/brother_printer/tests/test_cups_service_part3.py b/python_pkg/brother_printer/tests/test_cups_fallback_port_ok.py similarity index 60% rename from python_pkg/brother_printer/tests/test_cups_service_part3.py rename to python_pkg/brother_printer/tests/test_cups_fallback_port_ok.py index 65fb076f..44d86072 100644 --- a/python_pkg/brother_printer/tests/test_cups_service_part3.py +++ b/python_pkg/brother_printer/tests/test_cups_fallback_port_ok.py @@ -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 ( diff --git a/python_pkg/brother_printer/tests/test_cups_fallback_query.py b/python_pkg/brother_printer/tests/test_cups_fallback_query.py new file mode 100644 index 00000000..1084c634 --- /dev/null +++ b/python_pkg/brother_printer/tests/test_cups_fallback_query.py @@ -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" diff --git a/python_pkg/brother_printer/tests/test_cups_service_part4.py b/python_pkg/brother_printer/tests/test_cups_ipp.py similarity index 95% rename from python_pkg/brother_printer/tests/test_cups_service_part4.py rename to python_pkg/brother_printer/tests/test_cups_ipp.py index cee20902..972ee335 100644 --- a/python_pkg/brother_printer/tests/test_cups_service_part4.py +++ b/python_pkg/brother_printer/tests/test_cups_ipp.py @@ -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 diff --git a/python_pkg/brother_printer/tests/test_cups_service.py b/python_pkg/brother_printer/tests/test_cups_service.py index 25395de0..3723ad4f 100644 --- a/python_pkg/brother_printer/tests/test_cups_service.py +++ b/python_pkg/brother_printer/tests/test_cups_service.py @@ -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) diff --git a/python_pkg/brother_printer/tests/test_cups_service_port.py b/python_pkg/brother_printer/tests/test_cups_service_port.py new file mode 100644 index 00000000..4093f4a3 --- /dev/null +++ b/python_pkg/brother_printer/tests/test_cups_service_port.py @@ -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) diff --git a/python_pkg/brother_printer/tests/test_cups_service_part2.py b/python_pkg/brother_printer/tests/test_cups_status_mapping.py similarity index 57% rename from python_pkg/brother_printer/tests/test_cups_service_part2.py rename to python_pkg/brother_printer/tests/test_cups_status_mapping.py index ac129105..48de4980 100644 --- a/python_pkg/brother_printer/tests/test_cups_service_part2.py +++ b/python_pkg/brother_printer/tests/test_cups_status_mapping.py @@ -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."""