From 74eee2b57f0795194da52f3d7b2242cd8967b42c Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Sun, 16 Aug 2026 17:23:47 +0200 Subject: [PATCH] Split the brother_printer display and usb_query test modules test_display.py 363 -> 204, with the supply maths and its display going to test_supply_levels.py. test_usb_query.py 328 -> 126, with the retry and sequencing tests going to test_usb_query_sequence.py. That removed the last reader of the six _supply names through display, so display.py's __all__ shrinks to the two public entry points and the stale comment about tests reaching them there is gone. Checked by grep across python_pkg and an import-time attribute probe, not by the suite. The first run after the usb_query split reported 393: TestParsePagecount sat between the two moved classes and travelled with them. Fixed. Pass count is 402 before and after. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_011w2Pj4Ek3ZNMQVYcFHgXDY --- ...ter-test-display-usb-split-2026-08-16.json | 19 ++ ...ter-test-display-usb-split-2026-08-16.json | 53 +++++ python_pkg/brother_printer/display.py | 24 +- .../brother_printer/tests/test_display.py | 159 ------------- ...test_usb_query_part2.py => test_pjl_io.py} | 0 .../tests/test_supply_levels.py | 174 ++++++++++++++ .../brother_printer/tests/test_usb_query.py | 209 +--------------- .../tests/test_usb_query_sequence.py | 224 ++++++++++++++++++ 8 files changed, 473 insertions(+), 389 deletions(-) create mode 100644 docs/superpowers/contracts/brother-printer-test-display-usb-split-2026-08-16.json create mode 100644 docs/superpowers/evidence/brother-printer-test-display-usb-split-2026-08-16.json rename python_pkg/brother_printer/tests/{test_usb_query_part2.py => test_pjl_io.py} (100%) create mode 100644 python_pkg/brother_printer/tests/test_supply_levels.py create mode 100644 python_pkg/brother_printer/tests/test_usb_query_sequence.py diff --git a/docs/superpowers/contracts/brother-printer-test-display-usb-split-2026-08-16.json b/docs/superpowers/contracts/brother-printer-test-display-usb-split-2026-08-16.json new file mode 100644 index 00000000..d8da0ed6 --- /dev/null +++ b/docs/superpowers/contracts/brother-printer-test-display-usb-split-2026-08-16.json @@ -0,0 +1,19 @@ +{ + "title": "Split the brother_printer display and usb_query test modules under the 250-line cap", + "objective": "Bring test_display.py (363 lines) and test_usb_query.py (328) under the repo-wide 250-line cap by partitioning them at class boundaries, and retire display.py's now-dead __all__ re-export block once no test reaches the supply helpers through it.", + "acceptance_criteria": [ + "test_display.py, test_supply_levels.py, test_usb_query.py and test_usb_query_sequence.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.", + "display.py keeps every name that test_display.py and test_display_part2.py patch on it as a real module attribute, verified by an import-time probe rather than inferred from a green run.", + "The six _supply names are removed from display.py's __all__ only after confirming by grep that no source module and no test still imports them through display.", + "No assertion is changed and no test is deleted, skipped or weakened; whole classes 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.", + "Moving _display_supply_levels or _display_supply_warnings out of display.py: test_display_part2.py patches _display_supply_levels there.", + "Splitting test_check_brother_printer.py or the consumables pair, which are separate commits." + ], + "verifier": "PYTHONPATH=. python3 -m pytest python_pkg/brother_printer/tests -q (expects exactly 402 passed), plus the display attribute probe, 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-display-usb-split-2026-08-16.json b/docs/superpowers/evidence/brother-printer-test-display-usb-split-2026-08-16.json new file mode 100644 index 00000000..dfa8d148 --- /dev/null +++ b/docs/superpowers/evidence/brother-printer-test-display-usb-split-2026-08-16.json @@ -0,0 +1,53 @@ +{ + "intent": "Bring the display and usb_query test modules under the 250-line cap, and remove the re-export block in display.py that the earlier partial split left behind.", + "scope": [ + "python_pkg/brother_printer/tests/test_display.py (363 -> 204)", + "python_pkg/brother_printer/tests/test_supply_levels.py (new)", + "python_pkg/brother_printer/tests/test_usb_query.py (328 -> 126)", + "python_pkg/brother_printer/tests/test_usb_query_sequence.py (new)", + "python_pkg/brother_printer/tests/test_usb_query_part2.py -> test_pjl_io.py (rename only)", + "python_pkg/brother_printer/display.py (dead re-export removed)", + "Non-goal: no assertion changed, no test deleted, skipped or weakened" + ], + "changes": [ + "test_display.py split at TestClassifyPercentageLevel: the USB report tests stay, the consumable supply maths and its display go to test_supply_levels.py, which imports the six classification helpers from _supply directly and the two printing functions from display.", + "That was the last reader of the six _supply names through display, so display.py's __all__ shrank to the two public entry points, its _supply import narrowed to the one name it actually calls, and the stale comment claiming the tests reach those names through this module was deleted. Confirmed by grep across python_pkg that no source module or test still imports them from display.", + "test_usb_query.py split at TestRetryPjlQuery: discovery and PJL response parsing stay, the retry, sequencing, result init and query_usb_pjl tests go to test_usb_query_sequence.py. Those patch the wire helpers on usb_query rather than _pjl_io, which is correct - that is the module whose globals the sequencing code resolves them through.", + "Renamed test_usb_query_part2.py to test_pjl_io.py, matching the module it has covered since the source split." + ], + "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 -q (first run after the usb_query split)", + "result": "fail then pass", + "evidence": "Reported 9 failed, 393 passed: TestParsePagecount sat between the two moved classes and travelled with them, so the new module needed errno, os, _parse_status and _parse_pagecount that my hand-written import block had not anticipated. Added; the count returned to 402." + }, + { + "command": "PYTHONPATH=. python3 -c \"from python_pkg.brother_printer import display as d; d._display_report_header; d._display_supply_levels; d._display_supply_warnings; d._display_network_device_info; d._display_page_count_estimate; d.display_cups_queue_status; d.get_status_info\"", + "result": "pass", + "evidence": "runtime probe OK after the __all__ removal - every name the two display test modules patch is still a real attribute, so shrinking the re-export did not expose them to ruff's F401/TC001 rewrite." + }, + { + "command": "grep -rn 'display import' python_pkg/", + "result": "pass", + "evidence": "Only check_brother_printer.py, test_display.py, test_display_part2.py and test_supply_levels.py import from display, and none of them names a _supply helper - which is what made the __all__ entries safe to drop rather than merely unused-looking." + }, + { + "command": "pre-commit run --files ", + "result": "pass", + "evidence": "ruff removed two imports the split left unused in test_usb_query.py; clean on re-run, and pytest-with-coverage passes." + } + ], + "risks": [ + "Removing names from __all__ is the one edit here that could break a caller silently at import time rather than loudly. Guarded by the repo-wide grep and the attribute probe, not by the test run alone.", + "test_usb_query_sequence.py patches _drain_buffer and pjl_query on usb_query while test_pjl_io.py patches their definitions on _pjl_io. Both are correct and must stay that way; collapsing them onto one module would unpatch the other caller." + ], + "rollback": [ + "git revert the commit; the two originals return at 363 and 328 lines, display.py's __all__ comes back, and the two new modules disappear.", + "After rollback, re-run the package suite and confirm it still reports 402 passed." + ] +} diff --git a/python_pkg/brother_printer/display.py b/python_pkg/brother_printer/display.py index f9a4af5b..f8056bf8 100644 --- a/python_pkg/brother_printer/display.py +++ b/python_pkg/brother_printer/display.py @@ -5,10 +5,6 @@ from __future__ import annotations import sys from typing import TYPE_CHECKING -# ── Shared display helpers ─────────────────────────────────────────── -# Re-exported through __all__: tests/test_display.py imports these names from -# this module. Only the un-patched supply *maths* moved to _supply -- everything -# the tests reach with mock.patch has to stay resolvable here. from python_pkg.brother_printer._page_count import ( _display_consumables_reference, _display_page_count_estimate, @@ -20,14 +16,7 @@ from python_pkg.brother_printer._severity import ( _SEVERITY_ICONS, _SEVERITY_SUMMARIES, ) -from python_pkg.brother_printer._supply import ( - _classify_percentage_level, - _classify_supply_level, - _collect_supply_items, - _format_supply_bar, - _parse_supply_value, - _process_supply_item, -) +from python_pkg.brother_printer._supply import _collect_supply_items from python_pkg.brother_printer.constants import ( BOLD, CYAN, @@ -47,16 +36,7 @@ if TYPE_CHECKING: USBResult, ) -__all__ = [ - "_classify_percentage_level", - "_classify_supply_level", - "_collect_supply_items", - "_format_supply_bar", - "_parse_supply_value", - "_process_supply_item", - "display_network_results", - "display_usb_results", -] +__all__ = ["display_network_results", "display_usb_results"] def _display_report_header() -> None: diff --git a/python_pkg/brother_printer/tests/test_display.py b/python_pkg/brother_printer/tests/test_display.py index 9deec4f8..5584342e 100644 --- a/python_pkg/brother_printer/tests/test_display.py +++ b/python_pkg/brother_printer/tests/test_display.py @@ -8,25 +8,15 @@ from unittest.mock import MagicMock, patch import pytest from python_pkg.brother_printer.data_classes import ( - NetworkResult, - SupplyReadings, USBPortStatus, USBResult, ) from python_pkg.brother_printer.display import ( - _classify_percentage_level, - _classify_supply_level, - _collect_supply_items, _display_cups_fallback_note, _display_pjl_status, _display_report_header, - _display_supply_levels, - _display_supply_warnings, _display_usb_device_info, _format_status_detail, - _format_supply_bar, - _parse_supply_value, - _process_supply_item, display_usb_results, ) @@ -212,152 +202,3 @@ class TestDisplayUsbResults: pytest.raises(SystemExit), ): display_usb_results(r) - - -class TestClassifyPercentageLevel: - def test_low(self) -> None: - pct, _, _, _, replace = _classify_percentage_level("Toner", 5) - assert pct == 5 - assert replace is True - - def test_warn(self) -> None: - _, _, _, warn, replace = _classify_percentage_level("Toner", 20) - assert replace is False - assert "order soon" in warn - - def test_ok(self) -> None: - _, _, _, warn, replace = _classify_percentage_level("Toner", 80) - assert replace is False - assert warn == "" - - -class TestClassifySupplyLevel: - def test_snmp_ok(self) -> None: - _, text, _, _, replace = _classify_supply_level("Toner", 100, -3) - assert text == "OK" - assert replace is False - - def test_snmp_low(self) -> None: - _, text, _, _, replace = _classify_supply_level("Toner", 100, -2) - assert text == "LOW" - assert replace is True - - def test_empty(self) -> None: - _, text, _, _, replace = _classify_supply_level("Toner", 100, 0) - assert text == "EMPTY" - assert replace is True - - def test_normal_percentage(self) -> None: - pct, _, _, _, replace = _classify_supply_level("Toner", 100, 80) - assert pct == 80 - assert replace is False - - def test_no_max_val(self) -> None: - pct, text, _, _, _ = _classify_supply_level("Toner", 0, 50) - assert pct == -1 - assert text == "" - - def test_over_100_capped(self) -> None: - pct, _, _, _, _ = _classify_supply_level("Toner", 50, 100) - assert pct == 100 - - -class TestFormatSupplyBar: - def test_negative(self) -> None: - assert _format_supply_bar(-1) == "" - - def test_zero(self) -> None: - bar = _format_supply_bar(0) - assert "░" in bar - - def test_full(self) -> None: - bar = _format_supply_bar(100) - assert "█" in bar - - -class TestProcessSupplyItem: - def test_normal(self) -> None: - item = _process_supply_item("Toner", 100, 80) - assert item.status_text == "80%" - - def test_empty(self) -> None: - item = _process_supply_item("Toner", 100, 0) - assert item.needs_replacement is True - - -class TestDisplaySupplyWarnings: - def test_replacement_needed(self) -> None: - with patch("sys.stdout", new_callable=StringIO) as out: - _display_supply_warnings( - needs_replacement=True, - warnings=["Toner low"], - ) - assert "ACTION NEEDED" in out.getvalue() - - def test_warnings_only(self) -> None: - with patch("sys.stdout", new_callable=StringIO) as out: - _display_supply_warnings( - needs_replacement=False, - warnings=["Toner at 20%"], - ) - assert "HEADS UP" in out.getvalue() - - def test_all_healthy(self) -> None: - with patch("sys.stdout", new_callable=StringIO) as out: - _display_supply_warnings( - needs_replacement=False, - warnings=[], - ) - assert "healthy" in out.getvalue() - - -class TestParseSupplyValue: - def test_valid(self) -> None: - assert _parse_supply_value(["10", "20"], 0) == 10 - - def test_index_error(self) -> None: - assert _parse_supply_value([], 0) == 0 - - def test_value_error(self) -> None: - assert _parse_supply_value(["abc"], 0) == 0 - - -class TestCollectSupplyItems: - def test_collect(self) -> None: - result = NetworkResult( - supplies=SupplyReadings( - descriptions=["Toner", "Drum"], - max_values=["100", "200"], - levels=["80", "150"], - ), - ) - items, descs = _collect_supply_items(result) - assert len(items) == 2 - assert descs == ["Toner", "Drum"] - - -class TestDisplaySupplyLevels: - def test_with_items(self) -> None: - result = NetworkResult( - supplies=SupplyReadings( - descriptions=["Toner"], - max_values=["100"], - levels=["80"], - ), - ) - with patch("sys.stdout", new_callable=StringIO) as out: - _display_supply_levels(result) - assert "Toner" in out.getvalue() - - def test_needs_replacement_and_warning(self) -> None: - result = NetworkResult( - supplies=SupplyReadings( - descriptions=["Toner", "Drum"], - max_values=["100", "100"], - levels=["0", "15"], - ), - ) - with patch("sys.stdout", new_callable=StringIO) as out: - _display_supply_levels(result) - text = out.getvalue() - assert "ACTION NEEDED" in text diff --git a/python_pkg/brother_printer/tests/test_usb_query_part2.py b/python_pkg/brother_printer/tests/test_pjl_io.py similarity index 100% rename from python_pkg/brother_printer/tests/test_usb_query_part2.py rename to python_pkg/brother_printer/tests/test_pjl_io.py diff --git a/python_pkg/brother_printer/tests/test_supply_levels.py b/python_pkg/brother_printer/tests/test_supply_levels.py new file mode 100644 index 00000000..f12e0144 --- /dev/null +++ b/python_pkg/brother_printer/tests/test_supply_levels.py @@ -0,0 +1,174 @@ +"""Tests for the consumable supply level maths and its display. + +Split from test_display.py under the 250-line cap. The classification and bar +formatting live in _supply; the two functions that print the section stay in +display, where test_display_part2.py patches them. +""" + +from __future__ import annotations + +from io import StringIO +from unittest.mock import patch + +from python_pkg.brother_printer._supply import ( + _classify_percentage_level, + _classify_supply_level, + _collect_supply_items, + _format_supply_bar, + _parse_supply_value, + _process_supply_item, +) +from python_pkg.brother_printer.data_classes import NetworkResult, SupplyReadings +from python_pkg.brother_printer.display import ( + _display_supply_levels, + _display_supply_warnings, +) + + +class TestClassifyPercentageLevel: + def test_low(self) -> None: + pct, _, _, _, replace = _classify_percentage_level("Toner", 5) + assert pct == 5 + assert replace is True + + def test_warn(self) -> None: + _, _, _, warn, replace = _classify_percentage_level("Toner", 20) + assert replace is False + assert "order soon" in warn + + def test_ok(self) -> None: + _, _, _, warn, replace = _classify_percentage_level("Toner", 80) + assert replace is False + assert warn == "" + + +class TestClassifySupplyLevel: + def test_snmp_ok(self) -> None: + _, text, _, _, replace = _classify_supply_level("Toner", 100, -3) + assert text == "OK" + assert replace is False + + def test_snmp_low(self) -> None: + _, text, _, _, replace = _classify_supply_level("Toner", 100, -2) + assert text == "LOW" + assert replace is True + + def test_empty(self) -> None: + _, text, _, _, replace = _classify_supply_level("Toner", 100, 0) + assert text == "EMPTY" + assert replace is True + + def test_normal_percentage(self) -> None: + pct, _, _, _, replace = _classify_supply_level("Toner", 100, 80) + assert pct == 80 + assert replace is False + + def test_no_max_val(self) -> None: + pct, text, _, _, _ = _classify_supply_level("Toner", 0, 50) + assert pct == -1 + assert text == "" + + def test_over_100_capped(self) -> None: + pct, _, _, _, _ = _classify_supply_level("Toner", 50, 100) + assert pct == 100 + + +class TestFormatSupplyBar: + def test_negative(self) -> None: + assert _format_supply_bar(-1) == "" + + def test_zero(self) -> None: + bar = _format_supply_bar(0) + assert "░" in bar + + def test_full(self) -> None: + bar = _format_supply_bar(100) + assert "█" in bar + + +class TestProcessSupplyItem: + def test_normal(self) -> None: + item = _process_supply_item("Toner", 100, 80) + assert item.status_text == "80%" + + def test_empty(self) -> None: + item = _process_supply_item("Toner", 100, 0) + assert item.needs_replacement is True + + +class TestDisplaySupplyWarnings: + def test_replacement_needed(self) -> None: + with patch("sys.stdout", new_callable=StringIO) as out: + _display_supply_warnings( + needs_replacement=True, + warnings=["Toner low"], + ) + assert "ACTION NEEDED" in out.getvalue() + + def test_warnings_only(self) -> None: + with patch("sys.stdout", new_callable=StringIO) as out: + _display_supply_warnings( + needs_replacement=False, + warnings=["Toner at 20%"], + ) + assert "HEADS UP" in out.getvalue() + + def test_all_healthy(self) -> None: + with patch("sys.stdout", new_callable=StringIO) as out: + _display_supply_warnings( + needs_replacement=False, + warnings=[], + ) + assert "healthy" in out.getvalue() + + +class TestParseSupplyValue: + def test_valid(self) -> None: + assert _parse_supply_value(["10", "20"], 0) == 10 + + def test_index_error(self) -> None: + assert _parse_supply_value([], 0) == 0 + + def test_value_error(self) -> None: + assert _parse_supply_value(["abc"], 0) == 0 + + +class TestCollectSupplyItems: + def test_collect(self) -> None: + result = NetworkResult( + supplies=SupplyReadings( + descriptions=["Toner", "Drum"], + max_values=["100", "200"], + levels=["80", "150"], + ), + ) + items, descs = _collect_supply_items(result) + assert len(items) == 2 + assert descs == ["Toner", "Drum"] + + +class TestDisplaySupplyLevels: + def test_with_items(self) -> None: + result = NetworkResult( + supplies=SupplyReadings( + descriptions=["Toner"], + max_values=["100"], + levels=["80"], + ), + ) + with patch("sys.stdout", new_callable=StringIO) as out: + _display_supply_levels(result) + assert "Toner" in out.getvalue() + + def test_needs_replacement_and_warning(self) -> None: + result = NetworkResult( + supplies=SupplyReadings( + descriptions=["Toner", "Drum"], + max_values=["100", "100"], + levels=["0", "15"], + ), + ) + with patch("sys.stdout", new_callable=StringIO) as out: + _display_supply_levels(result) + text = out.getvalue() + assert "ACTION NEEDED" in text diff --git a/python_pkg/brother_printer/tests/test_usb_query.py b/python_pkg/brother_printer/tests/test_usb_query.py index 1385d2ac..4a189222 100644 --- a/python_pkg/brother_printer/tests/test_usb_query.py +++ b/python_pkg/brother_printer/tests/test_usb_query.py @@ -1,22 +1,15 @@ -"""Tests for brother_printer.usb_query module.""" +"""Tests for brother_printer.usb_query - discovery and PJL response parsing.""" from __future__ import annotations -import errno -import os from unittest.mock import MagicMock, patch from python_pkg.brother_printer.data_classes import USBResult from python_pkg.brother_printer.usb_query import ( - _init_usb_result, - _parse_pagecount, _parse_status, _parse_variables, - _retry_pjl_query, - _run_pjl_queries, find_brother_usb, find_usb_printer_dev, - query_usb_pjl, ) MOD = "python_pkg.brother_printer.usb_query" @@ -126,203 +119,3 @@ class TestParseVariables: def test_not_found(self) -> None: result = USBResult() assert _parse_variables("nothing\n", result) is False - - -class TestRetryPjlQuery: - @patch(f"{MOD}.time.sleep") - @patch(f"{MOD}._drain_buffer") - @patch(f"{MOD}.pjl_query") - def test_success_first_attempt( - self, - mock_pjl: MagicMock, - d: MagicMock, - s: MagicMock, - ) -> None: - result = USBResult() - mock_pjl.return_value = "CODE=10001\n" - _retry_pjl_query(42, "@PJL INFO STATUS", _parse_status, result, 2) - assert result.status_code == "10001" - assert mock_pjl.call_count == 1 - - @patch(f"{MOD}.time.sleep") - @patch(f"{MOD}._drain_buffer") - @patch(f"{MOD}.pjl_query") - def test_retry_then_success( - self, - mock_pjl: MagicMock, - d: MagicMock, - s: MagicMock, - ) -> None: - result = USBResult() - mock_pjl.side_effect = ["garbage\n", "CODE=10001\n"] - _retry_pjl_query(42, "@PJL INFO STATUS", _parse_status, result, 2) - assert result.status_code == "10001" - assert mock_pjl.call_count == 2 - - @patch(f"{MOD}.time.sleep") - @patch(f"{MOD}._drain_buffer") - @patch(f"{MOD}.pjl_query") - def test_all_retries_fail( - self, - mock_pjl: MagicMock, - d: MagicMock, - s: MagicMock, - ) -> None: - result = USBResult() - mock_pjl.return_value = "garbage\n" - _retry_pjl_query(42, "@PJL INFO STATUS", _parse_status, result, 2) - assert result.status_code == "" - assert mock_pjl.call_count == 3 - - -class TestRunPjlQueries: - @patch(f"{MOD}._retry_pjl_query") - @patch(f"{MOD}.time.sleep") - @patch(f"{MOD}._drain_buffer") - @patch(f"{MOD}._write_with_deadline") - def test_runs_status_pagecount_and_variables( - self, - mock_write: MagicMock, - d: MagicMock, - s: MagicMock, - mock_retry: MagicMock, - ) -> None: - result = USBResult() - _run_pjl_queries(42, result, 2) - assert mock_retry.call_count == 3 - queried = [call.args[1] for call in mock_retry.call_args_list] - assert "@PJL INFO PAGECOUNT" in queried - - -class TestParsePagecount: - def test_found(self) -> None: - result = USBResult() - assert _parse_pagecount("PAGECOUNT=2014\n", result) is True - assert result.page_count == "2014" - - def test_not_found(self) -> None: - result = USBResult() - assert _parse_pagecount("nothing\n", result) is False - assert result.page_count == "" - - def test_non_numeric_rejected(self) -> None: - """A garbled reply must not become a page count.""" - result = USBResult() - assert _parse_pagecount("PAGECOUNT=???\n", result) is False - assert result.page_count == "" - - -class TestInitUsbResult: - @patch(f"{MOD}.printer_info_from_cups") - def test_from_cups(self, mock_cups: MagicMock) -> None: - mock_cups.return_value = {"product": "HL-1110", "serial": "SN1"} - result = _init_usb_result("/dev/usb/lp0") - assert result.device == "/dev/usb/lp0" - assert result.product == "HL-1110" - assert result.serial == "SN1" - - @patch(f"{MOD}.printer_info_from_cups") - def test_no_product(self, mock_cups: MagicMock) -> None: - mock_cups.return_value = {"product": "", "serial": ""} - result = _init_usb_result("/dev/usb/lp0") - assert result.product == "Brother Laser Printer" - - -class TestQueryUsbPjl: - def test_success(self) -> None: - with ( - patch(f"{MOD}.find_usb_printer_dev", return_value="/dev/usb/lp0"), - patch(f"{MOD}._init_usb_result") as mock_init, - patch(f"{MOD}.os.access", return_value=True), - patch(f"{MOD}.os.open", return_value=10) as mock_open, - patch(f"{MOD}._run_pjl_queries"), - patch(f"{MOD}.os.close"), - ): - mock_init.return_value = USBResult( - device="/dev/usb/lp0", - status_code="10001", - ) - result = query_usb_pjl() - assert result.device == "/dev/usb/lp0" - assert result.error == "" - # The whole point: a blocking open() hangs on an unwell printer. - assert mock_open.call_args[0][1] & os.O_NONBLOCK - - @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_fallback.query_usb_via_cups", - ) as mock_cups: - mock_cups.return_value = USBResult(device="cups") - result = query_usb_pjl() - assert result.device == "cups" - - @patch(f"{MOD}.os.access", return_value=False) - @patch(f"{MOD}._init_usb_result") - @patch(f"{MOD}.find_usb_printer_dev", return_value="/dev/usb/lp0") - def test_permission_denied( - self, - f: MagicMock, - mock_init: MagicMock, - a: MagicMock, - ) -> None: - mock_init.return_value = USBResult(device="/dev/usb/lp0") - result = query_usb_pjl() - assert "Permission denied" in result.error - - def test_silent_printer_is_reported_not_hidden(self) -> None: - """A printer that answers nothing must say so, not look healthy.""" - with ( - patch(f"{MOD}.find_usb_printer_dev", return_value="/dev/usb/lp0"), - patch(f"{MOD}._init_usb_result") as mock_init, - patch(f"{MOD}.os.access", return_value=True), - patch(f"{MOD}.os.open", return_value=10), - patch(f"{MOD}._run_pjl_queries"), - patch(f"{MOD}.os.close"), - ): - mock_init.return_value = USBResult(device="/dev/usb/lp0") - result = query_usb_pjl() - assert "did not answer" in result.error - - def test_busy_device_explains_why(self) -> None: - """EBUSY means CUPS holds the printer mid-job; say that.""" - exc = OSError("busy") - exc.errno = errno.EBUSY - with ( - patch(f"{MOD}.find_usb_printer_dev", return_value="/dev/usb/lp0"), - patch(f"{MOD}._init_usb_result") as mock_init, - patch(f"{MOD}.os.access", return_value=True), - patch(f"{MOD}.os.open", side_effect=exc), - ): - mock_init.return_value = USBResult(device="/dev/usb/lp0") - result = query_usb_pjl() - assert "busy" in result.error.lower() - - def test_eacces_suggests_sudo(self) -> None: - exc = OSError("denied") - exc.errno = errno.EACCES - with ( - patch(f"{MOD}.find_usb_printer_dev", return_value="/dev/usb/lp0"), - patch(f"{MOD}._init_usb_result") as mock_init, - patch(f"{MOD}.os.access", return_value=True), - patch(f"{MOD}.os.open", side_effect=exc), - ): - mock_init.return_value = USBResult(device="/dev/usb/lp0") - result = query_usb_pjl() - assert "sudo" in result.error - - @patch(f"{MOD}.os.open", side_effect=OSError("no device")) - @patch(f"{MOD}.os.access", return_value=True) - @patch(f"{MOD}._init_usb_result") - @patch(f"{MOD}.find_usb_printer_dev", return_value="/dev/usb/lp0") - def test_oserror_fd_none( - self, - f: MagicMock, - mock_init: MagicMock, - a: MagicMock, - o: MagicMock, - ) -> None: - """os.open raises OSError before fd is set -> fd stays None.""" - mock_init.return_value = USBResult(device="/dev/usb/lp0") - result = query_usb_pjl() - assert "no device" in result.error diff --git a/python_pkg/brother_printer/tests/test_usb_query_sequence.py b/python_pkg/brother_printer/tests/test_usb_query_sequence.py new file mode 100644 index 00000000..852e9bc8 --- /dev/null +++ b/python_pkg/brother_printer/tests/test_usb_query_sequence.py @@ -0,0 +1,224 @@ +"""Tests for brother_printer.usb_query - running the PJL query sequence. + +Split from test_usb_query.py under the 250-line cap. These patch the wire-level +helpers on usb_query rather than on _pjl_io, because that is the module whose +globals the retry and sequencing code resolves them through. +""" + +from __future__ import annotations + +import errno +import os +from unittest.mock import MagicMock, patch + +from python_pkg.brother_printer.data_classes import USBResult +from python_pkg.brother_printer.usb_query import ( + _init_usb_result, + _parse_pagecount, + _parse_status, + _retry_pjl_query, + _run_pjl_queries, + query_usb_pjl, +) + +MOD = "python_pkg.brother_printer.usb_query" + + +class TestRetryPjlQuery: + @patch(f"{MOD}.time.sleep") + @patch(f"{MOD}._drain_buffer") + @patch(f"{MOD}.pjl_query") + def test_success_first_attempt( + self, + mock_pjl: MagicMock, + d: MagicMock, + s: MagicMock, + ) -> None: + result = USBResult() + mock_pjl.return_value = "CODE=10001\n" + _retry_pjl_query(42, "@PJL INFO STATUS", _parse_status, result, 2) + assert result.status_code == "10001" + assert mock_pjl.call_count == 1 + + @patch(f"{MOD}.time.sleep") + @patch(f"{MOD}._drain_buffer") + @patch(f"{MOD}.pjl_query") + def test_retry_then_success( + self, + mock_pjl: MagicMock, + d: MagicMock, + s: MagicMock, + ) -> None: + result = USBResult() + mock_pjl.side_effect = ["garbage\n", "CODE=10001\n"] + _retry_pjl_query(42, "@PJL INFO STATUS", _parse_status, result, 2) + assert result.status_code == "10001" + assert mock_pjl.call_count == 2 + + @patch(f"{MOD}.time.sleep") + @patch(f"{MOD}._drain_buffer") + @patch(f"{MOD}.pjl_query") + def test_all_retries_fail( + self, + mock_pjl: MagicMock, + d: MagicMock, + s: MagicMock, + ) -> None: + result = USBResult() + mock_pjl.return_value = "garbage\n" + _retry_pjl_query(42, "@PJL INFO STATUS", _parse_status, result, 2) + assert result.status_code == "" + assert mock_pjl.call_count == 3 + + +class TestRunPjlQueries: + @patch(f"{MOD}._retry_pjl_query") + @patch(f"{MOD}.time.sleep") + @patch(f"{MOD}._drain_buffer") + @patch(f"{MOD}._write_with_deadline") + def test_runs_status_pagecount_and_variables( + self, + mock_write: MagicMock, + d: MagicMock, + s: MagicMock, + mock_retry: MagicMock, + ) -> None: + result = USBResult() + _run_pjl_queries(42, result, 2) + assert mock_retry.call_count == 3 + queried = [call.args[1] for call in mock_retry.call_args_list] + assert "@PJL INFO PAGECOUNT" in queried + + +class TestParsePagecount: + def test_found(self) -> None: + result = USBResult() + assert _parse_pagecount("PAGECOUNT=2014\n", result) is True + assert result.page_count == "2014" + + def test_not_found(self) -> None: + result = USBResult() + assert _parse_pagecount("nothing\n", result) is False + assert result.page_count == "" + + def test_non_numeric_rejected(self) -> None: + """A garbled reply must not become a page count.""" + result = USBResult() + assert _parse_pagecount("PAGECOUNT=???\n", result) is False + assert result.page_count == "" + + +class TestInitUsbResult: + @patch(f"{MOD}.printer_info_from_cups") + def test_from_cups(self, mock_cups: MagicMock) -> None: + mock_cups.return_value = {"product": "HL-1110", "serial": "SN1"} + result = _init_usb_result("/dev/usb/lp0") + assert result.device == "/dev/usb/lp0" + assert result.product == "HL-1110" + assert result.serial == "SN1" + + @patch(f"{MOD}.printer_info_from_cups") + def test_no_product(self, mock_cups: MagicMock) -> None: + mock_cups.return_value = {"product": "", "serial": ""} + result = _init_usb_result("/dev/usb/lp0") + assert result.product == "Brother Laser Printer" + + +class TestQueryUsbPjl: + def test_success(self) -> None: + with ( + patch(f"{MOD}.find_usb_printer_dev", return_value="/dev/usb/lp0"), + patch(f"{MOD}._init_usb_result") as mock_init, + patch(f"{MOD}.os.access", return_value=True), + patch(f"{MOD}.os.open", return_value=10) as mock_open, + patch(f"{MOD}._run_pjl_queries"), + patch(f"{MOD}.os.close"), + ): + mock_init.return_value = USBResult( + device="/dev/usb/lp0", + status_code="10001", + ) + result = query_usb_pjl() + assert result.device == "/dev/usb/lp0" + assert result.error == "" + # The whole point: a blocking open() hangs on an unwell printer. + assert mock_open.call_args[0][1] & os.O_NONBLOCK + + @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_fallback.query_usb_via_cups", + ) as mock_cups: + mock_cups.return_value = USBResult(device="cups") + result = query_usb_pjl() + assert result.device == "cups" + + @patch(f"{MOD}.os.access", return_value=False) + @patch(f"{MOD}._init_usb_result") + @patch(f"{MOD}.find_usb_printer_dev", return_value="/dev/usb/lp0") + def test_permission_denied( + self, + f: MagicMock, + mock_init: MagicMock, + a: MagicMock, + ) -> None: + mock_init.return_value = USBResult(device="/dev/usb/lp0") + result = query_usb_pjl() + assert "Permission denied" in result.error + + def test_silent_printer_is_reported_not_hidden(self) -> None: + """A printer that answers nothing must say so, not look healthy.""" + with ( + patch(f"{MOD}.find_usb_printer_dev", return_value="/dev/usb/lp0"), + patch(f"{MOD}._init_usb_result") as mock_init, + patch(f"{MOD}.os.access", return_value=True), + patch(f"{MOD}.os.open", return_value=10), + patch(f"{MOD}._run_pjl_queries"), + patch(f"{MOD}.os.close"), + ): + mock_init.return_value = USBResult(device="/dev/usb/lp0") + result = query_usb_pjl() + assert "did not answer" in result.error + + def test_busy_device_explains_why(self) -> None: + """EBUSY means CUPS holds the printer mid-job; say that.""" + exc = OSError("busy") + exc.errno = errno.EBUSY + with ( + patch(f"{MOD}.find_usb_printer_dev", return_value="/dev/usb/lp0"), + patch(f"{MOD}._init_usb_result") as mock_init, + patch(f"{MOD}.os.access", return_value=True), + patch(f"{MOD}.os.open", side_effect=exc), + ): + mock_init.return_value = USBResult(device="/dev/usb/lp0") + result = query_usb_pjl() + assert "busy" in result.error.lower() + + def test_eacces_suggests_sudo(self) -> None: + exc = OSError("denied") + exc.errno = errno.EACCES + with ( + patch(f"{MOD}.find_usb_printer_dev", return_value="/dev/usb/lp0"), + patch(f"{MOD}._init_usb_result") as mock_init, + patch(f"{MOD}.os.access", return_value=True), + patch(f"{MOD}.os.open", side_effect=exc), + ): + mock_init.return_value = USBResult(device="/dev/usb/lp0") + result = query_usb_pjl() + assert "sudo" in result.error + + @patch(f"{MOD}.os.open", side_effect=OSError("no device")) + @patch(f"{MOD}.os.access", return_value=True) + @patch(f"{MOD}._init_usb_result") + @patch(f"{MOD}.find_usb_printer_dev", return_value="/dev/usb/lp0") + def test_oserror_fd_none( + self, + f: MagicMock, + mock_init: MagicMock, + a: MagicMock, + o: MagicMock, + ) -> None: + """os.open raises OSError before fd is set -> fd stays None.""" + mock_init.return_value = USBResult(device="/dev/usb/lp0") + result = query_usb_pjl() + assert "no device" in result.error