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

Split the brother_printer display and usb_query test modules

Browse Source 
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) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011w2Pj4Ek3ZNMQVYcFHgXDY
This commit is contained in:
Krzysztof Rudnicki 2026-08-16 17:23:47 +02:00
parent 607ba98c85
commit 74eee2b57f
8 changed files with 473 additions and 389 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -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."
}

53
docs/superpowers/evidence/brother-printer-test-display-usb-split-2026-08-16.json Normal file
View File

@ -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 <the touched files and artifacts>",
"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."
]
}

24
python_pkg/brother_printer/display.py
View File

@ -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:

159
python_pkg/brother_printer/tests/test_display.py
View File

@ -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

0
python_pkg/brother_printer/tests/test_usb_query_part2.py → python_pkg/brother_printer/tests/test_pjl_io.py
View File

174
python_pkg/brother_printer/tests/test_supply_levels.py Normal file
View File

@ -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

209
python_pkg/brother_printer/tests/test_usb_query.py
View File

@ -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

224
python_pkg/brother_printer/tests/test_usb_query_sequence.py Normal file
View File

@ -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