mirror of
https://github.com/kuhyx/testsAndMisc.git
synced 2026-11-10 20:07:04 +01:00
Split the brother_printer cups_queue test modules under the cap
test_cups_queue.py 458 -> 167, partitioned at class boundaries into _actions (the four shell-out primitives) and _backend (health check and error-log scanning). test_cups_queue_part2.py 278 -> 131, with the four _handle_* dispatchers moving to _handlers. No MOD changed: unlike the source splits, every patched name already lives in the module each file names, so this is a pure partition. The two part-numbered survivors are renamed for what they cover. Pass count is 402 before and after. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011w2Pj4Ek3ZNMQVYcFHgXDY
This commit is contained in:
parent
ad3d70f197
commit
65bc65c7e3
@ -0,0 +1,19 @@
|
||||
{
|
||||
"title": "Split the brother_printer cups_queue test modules under the 250-line cap",
|
||||
"objective": "Bring test_cups_queue.py (458 lines) and test_cups_queue_part2.py (278 lines) under the repo-wide 250-line cap by partitioning them along the class boundaries the source split already established, without losing or weakening a single test.",
|
||||
"acceptance_criteria": [
|
||||
"All six cups_queue test modules are 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: every name these tests patch already lives in the module the file names, so the split is a pure partition of test classes.",
|
||||
"No assertion is changed and no test is deleted, skipped or weakened; whole test classes move verbatim and only the import blocks are adjusted to what each file actually uses.",
|
||||
"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 the remaining over-cap test modules (test_cups_service.py, test_display.py, test_usb_query.py and the consumables pair), which are separate commits.",
|
||||
"Introducing a shared conftest.py: these classes need no shared fixture builders, and an unnecessary one would push jscpd toward its duplication threshold."
|
||||
],
|
||||
"verifier": "PYTHONPATH=. python3 -m pytest python_pkg/brother_printer/tests -q (expects exactly 402 passed), plus pre-commit run --files on the touched files, plus bash ~/utils/scripts/check_file_length.sh --all for the cap."
|
||||
}
|
||||
@ -0,0 +1,43 @@
|
||||
{
|
||||
"intent": "Bring the two over-cap cups_queue test modules under the 250-line cap by partitioning them by test class, with no change to what is tested.",
|
||||
"scope": [
|
||||
"python_pkg/brother_printer/tests/test_cups_queue.py (458 -> 167)",
|
||||
"python_pkg/brother_printer/tests/test_cups_queue_actions.py, test_cups_queue_backend.py (new)",
|
||||
"python_pkg/brother_printer/tests/test_cups_queue_part2.py -> test_cups_queue_menu.py (278 -> 131)",
|
||||
"python_pkg/brother_printer/tests/test_cups_queue_handlers.py (new)",
|
||||
"python_pkg/brother_printer/tests/test_cups_queue_part3.py -> test_cups_queue_display.py (rename only)",
|
||||
"Non-goal: no assertion changed, no test deleted, skipped or weakened"
|
||||
],
|
||||
"changes": [
|
||||
"test_cups_queue.py split three ways at class boundaries: the lpstat parsing and get_cups_queue_status tests stay, the four _cups_* action primitives go to test_cups_queue_actions.py, and the health check plus backend-error log scanning go to test_cups_queue_backend.py.",
|
||||
"test_cups_queue_part2.py split at TestHandleDisabledWithJobs: the menu and the five _dwj_* choices stay, the four _handle_* dispatchers go to test_cups_queue_handlers.py.",
|
||||
"No MOD constant changed anywhere in this commit. Unlike the source splits, every patched name already lives in the module each file names, so this is a pure partition of test classes.",
|
||||
"Import blocks trimmed to what each file uses: the original lost eight now-unmoved names and StringIO, the backend file needed subprocess added for its TimeoutExpired cases, and the handlers file needed CUPSJob. test_cups_queue_part2.py dropped its four _handle_* imports but keeps patching them by string, since _offer_queue_fix dispatches to them.",
|
||||
"Renamed the two part-numbered survivors to test_cups_queue_menu.py and test_cups_queue_display.py so the file name says which concern it covers."
|
||||
],
|
||||
"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. This is the check that matters here: a partition that dropped a whole class would still report a green run."
|
||||
},
|
||||
{
|
||||
"command": "wc -l python_pkg/brother_printer/tests/test_cups_queue*.py",
|
||||
"result": "pass",
|
||||
"evidence": "167, 157, 164, 161, 131 and 76 lines - all six modules under the 250-line cap."
|
||||
},
|
||||
{
|
||||
"command": "pre-commit run --files <the touched test files and artifacts>",
|
||||
"result": "pass",
|
||||
"evidence": "ruff, ruff format, mypy, pylint, bandit, name-tests-test and pytest-with-coverage all pass."
|
||||
}
|
||||
],
|
||||
"risks": [
|
||||
"Trimming import blocks by hand can remove a name that is only reachable at runtime. Guarded by the pass count plus ruff's F401 sweep, and by checking whether each dropped name was still patched by string rather than imported.",
|
||||
"No shared conftest.py was introduced; if a later split of the remaining test modules needs the same fixture scaffolding, it belongs in tests/conftest.py rather than a fourth copy, to keep jscpd under its 2% duplication threshold."
|
||||
],
|
||||
"rollback": [
|
||||
"git revert the commit; the two originals return at 458 and 278 lines and the three new modules disappear.",
|
||||
"After rollback, re-run the package suite and confirm it still reports 402 passed."
|
||||
]
|
||||
}
|
||||
@ -2,18 +2,10 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from io import StringIO
|
||||
import subprocess
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from python_pkg.brother_printer.cups_queue import (
|
||||
_check_cups_backend_errors,
|
||||
_cups_cancel_all_jobs,
|
||||
_cups_cancel_job,
|
||||
_cups_enable_printer,
|
||||
_cups_restart_service,
|
||||
_find_backend_error_in_log,
|
||||
_is_cups_printer_healthy,
|
||||
_parse_lpstat_jobs,
|
||||
_parse_lpstat_printer_line,
|
||||
get_cups_queue_status,
|
||||
@ -173,286 +165,3 @@ class TestGetCupsQueueStatus:
|
||||
]
|
||||
result = get_cups_queue_status()
|
||||
assert result.enabled is True # default unchanged
|
||||
|
||||
|
||||
class TestCupsEnablePrinter:
|
||||
@patch(f"{MOD}.shutil.which", return_value=None)
|
||||
def test_no_cupsenable(self, m: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
assert _cups_enable_printer("B") is False
|
||||
|
||||
@patch(f"{MOD}.subprocess.run")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/cupsenable")
|
||||
def test_success(self, w: MagicMock, mock_run: MagicMock) -> None:
|
||||
mock_run.return_value = MagicMock()
|
||||
assert _cups_enable_printer("B") is True
|
||||
|
||||
@patch(f"{MOD}.subprocess.run")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/cupsenable")
|
||||
def test_timeout(self, w: MagicMock, mock_run: MagicMock) -> None:
|
||||
mock_run.side_effect = subprocess.TimeoutExpired("cupsenable", 5)
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
assert _cups_enable_printer("B") is False
|
||||
|
||||
@patch(f"{MOD}.subprocess.run")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/cupsenable")
|
||||
def test_oserror(self, w: MagicMock, mock_run: MagicMock) -> None:
|
||||
mock_run.side_effect = OSError("fail")
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
assert _cups_enable_printer("B") is False
|
||||
|
||||
|
||||
class TestCupsCancelAllJobs:
|
||||
@patch(f"{MOD}.shutil.which", return_value=None)
|
||||
def test_no_cancel(self, m: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
assert _cups_cancel_all_jobs("B") is False
|
||||
|
||||
@patch(f"{MOD}.subprocess.run")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/cancel")
|
||||
def test_success(self, w: MagicMock, mock_run: MagicMock) -> None:
|
||||
mock_run.return_value = MagicMock()
|
||||
assert _cups_cancel_all_jobs("B") is True
|
||||
|
||||
@patch(f"{MOD}.subprocess.run")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/cancel")
|
||||
def test_error(self, w: MagicMock, mock_run: MagicMock) -> None:
|
||||
mock_run.side_effect = subprocess.CalledProcessError(1, "cancel")
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
assert _cups_cancel_all_jobs("B") is False
|
||||
|
||||
|
||||
class TestCupsCancelJob:
|
||||
@patch(f"{MOD}.shutil.which", return_value=None)
|
||||
def test_no_cancel(self, m: MagicMock) -> None:
|
||||
assert _cups_cancel_job("job-1") is False
|
||||
|
||||
@patch(f"{MOD}.subprocess.run")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/cancel")
|
||||
def test_success(self, w: MagicMock, mock_run: MagicMock) -> None:
|
||||
mock_run.return_value = MagicMock()
|
||||
assert _cups_cancel_job("job-1") is True
|
||||
|
||||
@patch(f"{MOD}.subprocess.run")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/cancel")
|
||||
def test_error(self, w: MagicMock, mock_run: MagicMock) -> None:
|
||||
mock_run.side_effect = subprocess.CalledProcessError(1, "cancel")
|
||||
assert _cups_cancel_job("job-1") is False
|
||||
|
||||
|
||||
class TestCupsRestartService:
|
||||
@patch(f"{MOD}.shutil.which", return_value=None)
|
||||
def test_no_systemctl(self, m: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
assert _cups_restart_service() is False
|
||||
|
||||
@patch(f"{MOD}.time.sleep")
|
||||
@patch(f"{MOD}.time.time")
|
||||
@patch(f"{MOD}.subprocess.Popen")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/systemctl")
|
||||
def test_success(
|
||||
self,
|
||||
w: MagicMock,
|
||||
mock_popen: MagicMock,
|
||||
mock_time: MagicMock,
|
||||
s: MagicMock,
|
||||
) -> None:
|
||||
proc = MagicMock()
|
||||
proc.poll.side_effect = [None, 0]
|
||||
proc.returncode = 0
|
||||
mock_popen.return_value = proc
|
||||
mock_time.side_effect = [0.0, 1.0, 2.0]
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
assert _cups_restart_service() is True
|
||||
|
||||
@patch(f"{MOD}.time.sleep")
|
||||
@patch(f"{MOD}.time.time")
|
||||
@patch(f"{MOD}.subprocess.Popen")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/systemctl")
|
||||
def test_timeout(
|
||||
self,
|
||||
w: MagicMock,
|
||||
mock_popen: MagicMock,
|
||||
mock_time: MagicMock,
|
||||
s: MagicMock,
|
||||
) -> None:
|
||||
proc = MagicMock()
|
||||
proc.poll.return_value = None
|
||||
mock_popen.return_value = proc
|
||||
mock_time.side_effect = [0.0, 31.0]
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
assert _cups_restart_service() is False
|
||||
proc.kill.assert_called_once()
|
||||
|
||||
@patch(f"{MOD}.time.sleep")
|
||||
@patch(f"{MOD}.time.time")
|
||||
@patch(f"{MOD}.subprocess.Popen")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/systemctl")
|
||||
def test_nonzero_exit(
|
||||
self,
|
||||
w: MagicMock,
|
||||
mock_popen: MagicMock,
|
||||
mock_time: MagicMock,
|
||||
s: MagicMock,
|
||||
) -> None:
|
||||
proc = MagicMock()
|
||||
proc.poll.side_effect = [None, 1]
|
||||
proc.returncode = 1
|
||||
mock_popen.return_value = proc
|
||||
mock_time.side_effect = [0.0, 1.0, 2.0]
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
assert _cups_restart_service() is False
|
||||
|
||||
@patch(f"{MOD}.subprocess.Popen")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/systemctl")
|
||||
def test_oserror(self, w: MagicMock, mock_popen: MagicMock) -> None:
|
||||
mock_popen.side_effect = OSError("fail")
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
assert _cups_restart_service() is False
|
||||
|
||||
|
||||
class TestIsCupsPrinterHealthy:
|
||||
@patch(f"{MOD}.shutil.which", return_value=None)
|
||||
def test_no_lpstat(self, m: MagicMock) -> None:
|
||||
assert _is_cups_printer_healthy("B") is False
|
||||
|
||||
@patch(f"{MOD}.subprocess.run")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/lpstat")
|
||||
def test_healthy(self, w: MagicMock, mock_run: MagicMock) -> None:
|
||||
mock_run.return_value = MagicMock(
|
||||
stdout="printer BrotherHL1110 is idle. enabled since Mon\n",
|
||||
)
|
||||
assert _is_cups_printer_healthy("BrotherHL1110") is True
|
||||
|
||||
@patch(f"{MOD}.subprocess.run")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/lpstat")
|
||||
def test_not_healthy(self, w: MagicMock, mock_run: MagicMock) -> None:
|
||||
mock_run.return_value = MagicMock(
|
||||
stdout="printer BrotherHL1110 disabled\n",
|
||||
)
|
||||
assert _is_cups_printer_healthy("BrotherHL1110") is False
|
||||
|
||||
@patch(f"{MOD}.subprocess.run")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/lpstat")
|
||||
def test_timeout(self, w: MagicMock, mock_run: MagicMock) -> None:
|
||||
mock_run.side_effect = subprocess.TimeoutExpired("lpstat", 5)
|
||||
assert _is_cups_printer_healthy("B") is False
|
||||
|
||||
|
||||
class TestFindBackendErrorInLog:
|
||||
def test_no_errors(self) -> None:
|
||||
lines = ["[2025-01-01] Completed job\n"]
|
||||
err, _, _ = _find_backend_error_in_log(lines)
|
||||
assert err == ""
|
||||
|
||||
def test_backend_error(self) -> None:
|
||||
lines = [
|
||||
"[2025-01-01] Completed job",
|
||||
"[2025-01-02] backend errors for BrotherHL1110",
|
||||
]
|
||||
err, ts, success_ts = _find_backend_error_in_log(lines)
|
||||
assert "backend errors" in err
|
||||
assert ts == "2025-01-02"
|
||||
assert success_ts == "2025-01-01"
|
||||
|
||||
def test_stopped_with_status(self) -> None:
|
||||
lines = [
|
||||
"[2025-01-02] stopped with status 1",
|
||||
]
|
||||
err, ts, _ = _find_backend_error_in_log(lines)
|
||||
assert "stopped with status" in err
|
||||
assert ts == "2025-01-02"
|
||||
|
||||
def test_error_no_timestamp(self) -> None:
|
||||
lines = ["backend errors no timestamp here"]
|
||||
err, ts, _ = _find_backend_error_in_log(lines)
|
||||
assert "backend errors" in err
|
||||
assert ts == ""
|
||||
|
||||
def test_completed_with_total(self) -> None:
|
||||
lines = [
|
||||
"[2025-01-01] page total 10",
|
||||
"[2025-01-02] backend errors",
|
||||
]
|
||||
_, _, success_ts = _find_backend_error_in_log(lines)
|
||||
assert success_ts == "2025-01-01"
|
||||
|
||||
def test_no_success_after_error(self) -> None:
|
||||
lines = [
|
||||
"[2025-01-02] backend errors",
|
||||
]
|
||||
_, _, success_ts = _find_backend_error_in_log(lines)
|
||||
assert success_ts == ""
|
||||
|
||||
def test_completed_no_timestamp(self) -> None:
|
||||
lines = [
|
||||
"Completed job",
|
||||
"[2025-01-02] backend errors",
|
||||
]
|
||||
_, _, success_ts = _find_backend_error_in_log(lines)
|
||||
assert success_ts == ""
|
||||
|
||||
|
||||
class TestCheckCupsBackendErrors:
|
||||
@patch(f"{MOD}._is_cups_printer_healthy", return_value=True)
|
||||
def test_healthy_printer(self, m: MagicMock) -> None:
|
||||
has_errors, _ = _check_cups_backend_errors("B")
|
||||
assert has_errors is False
|
||||
|
||||
@patch(f"{MOD}._find_backend_error_in_log", return_value=("", "", ""))
|
||||
@patch(f"{MOD}._is_cups_printer_healthy", return_value=False)
|
||||
def test_no_log_file(self, h: MagicMock, f: MagicMock) -> None:
|
||||
with patch(f"{MOD}.Path") as mock_path:
|
||||
mock_log = MagicMock()
|
||||
mock_log.exists.return_value = False
|
||||
mock_path.return_value = mock_log
|
||||
has_errors, _ = _check_cups_backend_errors("B")
|
||||
assert has_errors is False
|
||||
|
||||
@patch(
|
||||
f"{MOD}._find_backend_error_in_log", return_value=("error", "2025-01-02", "")
|
||||
)
|
||||
@patch(f"{MOD}._is_cups_printer_healthy", return_value=False)
|
||||
def test_has_errors(self, h: MagicMock, f: MagicMock) -> None:
|
||||
with patch(f"{MOD}.Path") as mock_path:
|
||||
mock_log = MagicMock()
|
||||
mock_log.exists.return_value = True
|
||||
mock_log.read_text.return_value = "log content"
|
||||
mock_path.return_value = mock_log
|
||||
has_errors, _ = _check_cups_backend_errors("B")
|
||||
assert has_errors is True
|
||||
|
||||
@patch(
|
||||
f"{MOD}._find_backend_error_in_log",
|
||||
return_value=("error", "2025-01-01", "2025-01-02"),
|
||||
)
|
||||
@patch(f"{MOD}._is_cups_printer_healthy", return_value=False)
|
||||
def test_success_after_error(self, h: MagicMock, f: MagicMock) -> None:
|
||||
with patch(f"{MOD}.Path") as mock_path:
|
||||
mock_log = MagicMock()
|
||||
mock_log.exists.return_value = True
|
||||
mock_log.read_text.return_value = "log content"
|
||||
mock_path.return_value = mock_log
|
||||
has_errors, _ = _check_cups_backend_errors("B")
|
||||
assert has_errors is False
|
||||
|
||||
@patch(f"{MOD}._is_cups_printer_healthy", return_value=False)
|
||||
def test_oserror_reading_log(self, h: MagicMock) -> None:
|
||||
with patch(f"{MOD}.Path") as mock_path:
|
||||
mock_log = MagicMock()
|
||||
mock_log.exists.return_value = True
|
||||
mock_log.read_text.side_effect = OSError("fail")
|
||||
mock_path.return_value = mock_log
|
||||
has_errors, _ = _check_cups_backend_errors("B")
|
||||
assert has_errors is False
|
||||
|
||||
@patch(f"{MOD}._find_backend_error_in_log", return_value=("", "", ""))
|
||||
@patch(f"{MOD}._is_cups_printer_healthy", return_value=False)
|
||||
def test_no_backend_error_in_log(self, h: MagicMock, f: MagicMock) -> None:
|
||||
with patch(f"{MOD}.Path") as mock_path:
|
||||
mock_log = MagicMock()
|
||||
mock_log.exists.return_value = True
|
||||
mock_log.read_text.return_value = "clean log"
|
||||
mock_path.return_value = mock_log
|
||||
has_errors, _ = _check_cups_backend_errors("B")
|
||||
assert has_errors is False
|
||||
|
||||
157
python_pkg/brother_printer/tests/test_cups_queue_actions.py
Normal file
157
python_pkg/brother_printer/tests/test_cups_queue_actions.py
Normal file
@ -0,0 +1,157 @@
|
||||
"""Tests for brother_printer.cups_queue - the CUPS fix action primitives.
|
||||
|
||||
Split from test_cups_queue.py under the 250-line cap. These are the four
|
||||
functions that shell out to cupsenable, cancel and systemctl, so every one of
|
||||
them patches shutil.which and subprocess on the module under test.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from io import StringIO
|
||||
import subprocess
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from python_pkg.brother_printer.cups_queue import (
|
||||
_cups_cancel_all_jobs,
|
||||
_cups_cancel_job,
|
||||
_cups_enable_printer,
|
||||
_cups_restart_service,
|
||||
)
|
||||
|
||||
MOD = "python_pkg.brother_printer.cups_queue"
|
||||
|
||||
|
||||
class TestCupsEnablePrinter:
|
||||
@patch(f"{MOD}.shutil.which", return_value=None)
|
||||
def test_no_cupsenable(self, m: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
assert _cups_enable_printer("B") is False
|
||||
|
||||
@patch(f"{MOD}.subprocess.run")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/cupsenable")
|
||||
def test_success(self, w: MagicMock, mock_run: MagicMock) -> None:
|
||||
mock_run.return_value = MagicMock()
|
||||
assert _cups_enable_printer("B") is True
|
||||
|
||||
@patch(f"{MOD}.subprocess.run")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/cupsenable")
|
||||
def test_timeout(self, w: MagicMock, mock_run: MagicMock) -> None:
|
||||
mock_run.side_effect = subprocess.TimeoutExpired("cupsenable", 5)
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
assert _cups_enable_printer("B") is False
|
||||
|
||||
@patch(f"{MOD}.subprocess.run")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/cupsenable")
|
||||
def test_oserror(self, w: MagicMock, mock_run: MagicMock) -> None:
|
||||
mock_run.side_effect = OSError("fail")
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
assert _cups_enable_printer("B") is False
|
||||
|
||||
|
||||
class TestCupsCancelAllJobs:
|
||||
@patch(f"{MOD}.shutil.which", return_value=None)
|
||||
def test_no_cancel(self, m: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
assert _cups_cancel_all_jobs("B") is False
|
||||
|
||||
@patch(f"{MOD}.subprocess.run")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/cancel")
|
||||
def test_success(self, w: MagicMock, mock_run: MagicMock) -> None:
|
||||
mock_run.return_value = MagicMock()
|
||||
assert _cups_cancel_all_jobs("B") is True
|
||||
|
||||
@patch(f"{MOD}.subprocess.run")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/cancel")
|
||||
def test_error(self, w: MagicMock, mock_run: MagicMock) -> None:
|
||||
mock_run.side_effect = subprocess.CalledProcessError(1, "cancel")
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
assert _cups_cancel_all_jobs("B") is False
|
||||
|
||||
|
||||
class TestCupsCancelJob:
|
||||
@patch(f"{MOD}.shutil.which", return_value=None)
|
||||
def test_no_cancel(self, m: MagicMock) -> None:
|
||||
assert _cups_cancel_job("job-1") is False
|
||||
|
||||
@patch(f"{MOD}.subprocess.run")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/cancel")
|
||||
def test_success(self, w: MagicMock, mock_run: MagicMock) -> None:
|
||||
mock_run.return_value = MagicMock()
|
||||
assert _cups_cancel_job("job-1") is True
|
||||
|
||||
@patch(f"{MOD}.subprocess.run")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/cancel")
|
||||
def test_error(self, w: MagicMock, mock_run: MagicMock) -> None:
|
||||
mock_run.side_effect = subprocess.CalledProcessError(1, "cancel")
|
||||
assert _cups_cancel_job("job-1") is False
|
||||
|
||||
|
||||
class TestCupsRestartService:
|
||||
@patch(f"{MOD}.shutil.which", return_value=None)
|
||||
def test_no_systemctl(self, m: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
assert _cups_restart_service() is False
|
||||
|
||||
@patch(f"{MOD}.time.sleep")
|
||||
@patch(f"{MOD}.time.time")
|
||||
@patch(f"{MOD}.subprocess.Popen")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/systemctl")
|
||||
def test_success(
|
||||
self,
|
||||
w: MagicMock,
|
||||
mock_popen: MagicMock,
|
||||
mock_time: MagicMock,
|
||||
s: MagicMock,
|
||||
) -> None:
|
||||
proc = MagicMock()
|
||||
proc.poll.side_effect = [None, 0]
|
||||
proc.returncode = 0
|
||||
mock_popen.return_value = proc
|
||||
mock_time.side_effect = [0.0, 1.0, 2.0]
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
assert _cups_restart_service() is True
|
||||
|
||||
@patch(f"{MOD}.time.sleep")
|
||||
@patch(f"{MOD}.time.time")
|
||||
@patch(f"{MOD}.subprocess.Popen")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/systemctl")
|
||||
def test_timeout(
|
||||
self,
|
||||
w: MagicMock,
|
||||
mock_popen: MagicMock,
|
||||
mock_time: MagicMock,
|
||||
s: MagicMock,
|
||||
) -> None:
|
||||
proc = MagicMock()
|
||||
proc.poll.return_value = None
|
||||
mock_popen.return_value = proc
|
||||
mock_time.side_effect = [0.0, 31.0]
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
assert _cups_restart_service() is False
|
||||
proc.kill.assert_called_once()
|
||||
|
||||
@patch(f"{MOD}.time.sleep")
|
||||
@patch(f"{MOD}.time.time")
|
||||
@patch(f"{MOD}.subprocess.Popen")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/systemctl")
|
||||
def test_nonzero_exit(
|
||||
self,
|
||||
w: MagicMock,
|
||||
mock_popen: MagicMock,
|
||||
mock_time: MagicMock,
|
||||
s: MagicMock,
|
||||
) -> None:
|
||||
proc = MagicMock()
|
||||
proc.poll.side_effect = [None, 1]
|
||||
proc.returncode = 1
|
||||
mock_popen.return_value = proc
|
||||
mock_time.side_effect = [0.0, 1.0, 2.0]
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
assert _cups_restart_service() is False
|
||||
|
||||
@patch(f"{MOD}.subprocess.Popen")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/systemctl")
|
||||
def test_oserror(self, w: MagicMock, mock_popen: MagicMock) -> None:
|
||||
mock_popen.side_effect = OSError("fail")
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
assert _cups_restart_service() is False
|
||||
164
python_pkg/brother_printer/tests/test_cups_queue_backend.py
Normal file
164
python_pkg/brother_printer/tests/test_cups_queue_backend.py
Normal file
@ -0,0 +1,164 @@
|
||||
"""Tests for brother_printer.cups_queue - stale backend error detection.
|
||||
|
||||
Split from test_cups_queue.py under the 250-line cap.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from python_pkg.brother_printer.cups_queue import (
|
||||
_check_cups_backend_errors,
|
||||
_find_backend_error_in_log,
|
||||
_is_cups_printer_healthy,
|
||||
)
|
||||
|
||||
MOD = "python_pkg.brother_printer.cups_queue"
|
||||
|
||||
|
||||
class TestIsCupsPrinterHealthy:
|
||||
@patch(f"{MOD}.shutil.which", return_value=None)
|
||||
def test_no_lpstat(self, m: MagicMock) -> None:
|
||||
assert _is_cups_printer_healthy("B") is False
|
||||
|
||||
@patch(f"{MOD}.subprocess.run")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/lpstat")
|
||||
def test_healthy(self, w: MagicMock, mock_run: MagicMock) -> None:
|
||||
mock_run.return_value = MagicMock(
|
||||
stdout="printer BrotherHL1110 is idle. enabled since Mon\n",
|
||||
)
|
||||
assert _is_cups_printer_healthy("BrotherHL1110") is True
|
||||
|
||||
@patch(f"{MOD}.subprocess.run")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/lpstat")
|
||||
def test_not_healthy(self, w: MagicMock, mock_run: MagicMock) -> None:
|
||||
mock_run.return_value = MagicMock(
|
||||
stdout="printer BrotherHL1110 disabled\n",
|
||||
)
|
||||
assert _is_cups_printer_healthy("BrotherHL1110") is False
|
||||
|
||||
@patch(f"{MOD}.subprocess.run")
|
||||
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/lpstat")
|
||||
def test_timeout(self, w: MagicMock, mock_run: MagicMock) -> None:
|
||||
mock_run.side_effect = subprocess.TimeoutExpired("lpstat", 5)
|
||||
assert _is_cups_printer_healthy("B") is False
|
||||
|
||||
|
||||
class TestFindBackendErrorInLog:
|
||||
def test_no_errors(self) -> None:
|
||||
lines = ["[2025-01-01] Completed job\n"]
|
||||
err, _, _ = _find_backend_error_in_log(lines)
|
||||
assert err == ""
|
||||
|
||||
def test_backend_error(self) -> None:
|
||||
lines = [
|
||||
"[2025-01-01] Completed job",
|
||||
"[2025-01-02] backend errors for BrotherHL1110",
|
||||
]
|
||||
err, ts, success_ts = _find_backend_error_in_log(lines)
|
||||
assert "backend errors" in err
|
||||
assert ts == "2025-01-02"
|
||||
assert success_ts == "2025-01-01"
|
||||
|
||||
def test_stopped_with_status(self) -> None:
|
||||
lines = [
|
||||
"[2025-01-02] stopped with status 1",
|
||||
]
|
||||
err, ts, _ = _find_backend_error_in_log(lines)
|
||||
assert "stopped with status" in err
|
||||
assert ts == "2025-01-02"
|
||||
|
||||
def test_error_no_timestamp(self) -> None:
|
||||
lines = ["backend errors no timestamp here"]
|
||||
err, ts, _ = _find_backend_error_in_log(lines)
|
||||
assert "backend errors" in err
|
||||
assert ts == ""
|
||||
|
||||
def test_completed_with_total(self) -> None:
|
||||
lines = [
|
||||
"[2025-01-01] page total 10",
|
||||
"[2025-01-02] backend errors",
|
||||
]
|
||||
_, _, success_ts = _find_backend_error_in_log(lines)
|
||||
assert success_ts == "2025-01-01"
|
||||
|
||||
def test_no_success_after_error(self) -> None:
|
||||
lines = [
|
||||
"[2025-01-02] backend errors",
|
||||
]
|
||||
_, _, success_ts = _find_backend_error_in_log(lines)
|
||||
assert success_ts == ""
|
||||
|
||||
def test_completed_no_timestamp(self) -> None:
|
||||
lines = [
|
||||
"Completed job",
|
||||
"[2025-01-02] backend errors",
|
||||
]
|
||||
_, _, success_ts = _find_backend_error_in_log(lines)
|
||||
assert success_ts == ""
|
||||
|
||||
|
||||
class TestCheckCupsBackendErrors:
|
||||
@patch(f"{MOD}._is_cups_printer_healthy", return_value=True)
|
||||
def test_healthy_printer(self, m: MagicMock) -> None:
|
||||
has_errors, _ = _check_cups_backend_errors("B")
|
||||
assert has_errors is False
|
||||
|
||||
@patch(f"{MOD}._find_backend_error_in_log", return_value=("", "", ""))
|
||||
@patch(f"{MOD}._is_cups_printer_healthy", return_value=False)
|
||||
def test_no_log_file(self, h: MagicMock, f: MagicMock) -> None:
|
||||
with patch(f"{MOD}.Path") as mock_path:
|
||||
mock_log = MagicMock()
|
||||
mock_log.exists.return_value = False
|
||||
mock_path.return_value = mock_log
|
||||
has_errors, _ = _check_cups_backend_errors("B")
|
||||
assert has_errors is False
|
||||
|
||||
@patch(
|
||||
f"{MOD}._find_backend_error_in_log", return_value=("error", "2025-01-02", "")
|
||||
)
|
||||
@patch(f"{MOD}._is_cups_printer_healthy", return_value=False)
|
||||
def test_has_errors(self, h: MagicMock, f: MagicMock) -> None:
|
||||
with patch(f"{MOD}.Path") as mock_path:
|
||||
mock_log = MagicMock()
|
||||
mock_log.exists.return_value = True
|
||||
mock_log.read_text.return_value = "log content"
|
||||
mock_path.return_value = mock_log
|
||||
has_errors, _ = _check_cups_backend_errors("B")
|
||||
assert has_errors is True
|
||||
|
||||
@patch(
|
||||
f"{MOD}._find_backend_error_in_log",
|
||||
return_value=("error", "2025-01-01", "2025-01-02"),
|
||||
)
|
||||
@patch(f"{MOD}._is_cups_printer_healthy", return_value=False)
|
||||
def test_success_after_error(self, h: MagicMock, f: MagicMock) -> None:
|
||||
with patch(f"{MOD}.Path") as mock_path:
|
||||
mock_log = MagicMock()
|
||||
mock_log.exists.return_value = True
|
||||
mock_log.read_text.return_value = "log content"
|
||||
mock_path.return_value = mock_log
|
||||
has_errors, _ = _check_cups_backend_errors("B")
|
||||
assert has_errors is False
|
||||
|
||||
@patch(f"{MOD}._is_cups_printer_healthy", return_value=False)
|
||||
def test_oserror_reading_log(self, h: MagicMock) -> None:
|
||||
with patch(f"{MOD}.Path") as mock_path:
|
||||
mock_log = MagicMock()
|
||||
mock_log.exists.return_value = True
|
||||
mock_log.read_text.side_effect = OSError("fail")
|
||||
mock_path.return_value = mock_log
|
||||
has_errors, _ = _check_cups_backend_errors("B")
|
||||
assert has_errors is False
|
||||
|
||||
@patch(f"{MOD}._find_backend_error_in_log", return_value=("", "", ""))
|
||||
@patch(f"{MOD}._is_cups_printer_healthy", return_value=False)
|
||||
def test_no_backend_error_in_log(self, h: MagicMock, f: MagicMock) -> None:
|
||||
with patch(f"{MOD}.Path") as mock_path:
|
||||
mock_log = MagicMock()
|
||||
mock_log.exists.return_value = True
|
||||
mock_log.read_text.return_value = "clean log"
|
||||
mock_path.return_value = mock_log
|
||||
has_errors, _ = _check_cups_backend_errors("B")
|
||||
assert has_errors is False
|
||||
@ -1,4 +1,4 @@
|
||||
"""Tests for brother_printer._queue_fix module - part 3 (display status)."""
|
||||
"""Tests for brother_printer._queue_fix - the queue status display."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
@ -1,4 +1,9 @@
|
||||
"""Tests for brother_printer._queue_fix module - part 2 (interactive fix)."""
|
||||
"""Tests for brother_printer._queue_fix - the choice dispatchers.
|
||||
|
||||
Split from test_cups_queue_part2.py under the 250-line cap. Each dispatcher
|
||||
maps a menu choice onto the CUPS action primitives, which are patched on
|
||||
_queue_fix because that is the module whose globals they resolve through.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
@ -6,138 +11,16 @@ from io import StringIO
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from python_pkg.brother_printer._queue_fix import (
|
||||
_dwj_cancel_and_enable,
|
||||
_dwj_cancel_only,
|
||||
_dwj_enable_only,
|
||||
_dwj_restart_and_enable,
|
||||
_dwj_restart_only,
|
||||
_handle_backend_errors_only,
|
||||
_handle_disabled_no_jobs,
|
||||
_handle_disabled_with_jobs,
|
||||
_handle_enabled_with_jobs,
|
||||
_offer_queue_fix,
|
||||
)
|
||||
from python_pkg.brother_printer.data_classes import CUPSJob, CUPSQueueStatus
|
||||
|
||||
MOD = "python_pkg.brother_printer._queue_fix"
|
||||
|
||||
|
||||
# ── _offer_queue_fix ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestOfferQueueFix:
|
||||
"""Tests for _offer_queue_fix menu routing."""
|
||||
|
||||
@patch(f"{MOD}._handle_disabled_with_jobs")
|
||||
@patch(f"{MOD}._prompt", return_value="1")
|
||||
def test_disabled_with_jobs(self, p: MagicMock, mock_handler: MagicMock) -> None:
|
||||
queue = CUPSQueueStatus(
|
||||
printer_name="B",
|
||||
enabled=False,
|
||||
jobs=[CUPSJob("j1", "alice", "1024", "Mon")],
|
||||
)
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_offer_queue_fix(queue)
|
||||
mock_handler.assert_called_once_with(queue, "1")
|
||||
|
||||
@patch(f"{MOD}._handle_disabled_no_jobs")
|
||||
@patch(f"{MOD}._prompt", return_value="2")
|
||||
def test_disabled_no_jobs(self, p: MagicMock, mock_handler: MagicMock) -> None:
|
||||
queue = CUPSQueueStatus(printer_name="B", enabled=False)
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_offer_queue_fix(queue)
|
||||
mock_handler.assert_called_once_with(queue, "2")
|
||||
|
||||
@patch(f"{MOD}._handle_enabled_with_jobs")
|
||||
@patch(f"{MOD}._prompt", return_value="1")
|
||||
def test_enabled_with_jobs(self, p: MagicMock, mock_handler: MagicMock) -> None:
|
||||
queue = CUPSQueueStatus(
|
||||
printer_name="B",
|
||||
enabled=True,
|
||||
jobs=[CUPSJob("j1", "alice", "1024", "Mon")],
|
||||
)
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_offer_queue_fix(queue)
|
||||
mock_handler.assert_called_once_with(queue, "1")
|
||||
|
||||
@patch(f"{MOD}._handle_backend_errors_only")
|
||||
@patch(f"{MOD}._prompt", return_value="1")
|
||||
def test_backend_errors_only(self, p: MagicMock, mock_handler: MagicMock) -> None:
|
||||
queue = CUPSQueueStatus(printer_name="B", enabled=True)
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_offer_queue_fix(queue)
|
||||
mock_handler.assert_called_once_with("1")
|
||||
|
||||
|
||||
# ── _dwj_* action functions ─────────────────────────────────────────
|
||||
|
||||
|
||||
class TestDwjEnableOnly:
|
||||
@patch(f"{MOD}._cups_enable_printer", return_value=True)
|
||||
def test_success(self, m: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_dwj_enable_only("B")
|
||||
|
||||
@patch(f"{MOD}._cups_enable_printer", return_value=False)
|
||||
def test_failure(self, m: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_dwj_enable_only("B")
|
||||
|
||||
|
||||
class TestDwjCancelAndEnable:
|
||||
@patch(f"{MOD}._cups_enable_printer", return_value=True)
|
||||
@patch(f"{MOD}._cups_cancel_all_jobs", return_value=True)
|
||||
def test_success(self, c: MagicMock, e: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_dwj_cancel_and_enable("B")
|
||||
|
||||
@patch(f"{MOD}._cups_enable_printer", return_value=False)
|
||||
@patch(f"{MOD}._cups_cancel_all_jobs", return_value=True)
|
||||
def test_enable_fails(self, c: MagicMock, e: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_dwj_cancel_and_enable("B")
|
||||
|
||||
|
||||
class TestDwjCancelOnly:
|
||||
@patch(f"{MOD}._cups_cancel_all_jobs", return_value=True)
|
||||
def test_success(self, m: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_dwj_cancel_only("B")
|
||||
|
||||
@patch(f"{MOD}._cups_cancel_all_jobs", return_value=False)
|
||||
def test_failure(self, m: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_dwj_cancel_only("B")
|
||||
|
||||
|
||||
class TestDwjRestartOnly:
|
||||
@patch(f"{MOD}._cups_restart_service", return_value=True)
|
||||
def test_success(self, m: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_dwj_restart_only("B")
|
||||
|
||||
@patch(f"{MOD}._cups_restart_service", return_value=False)
|
||||
def test_failure(self, m: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_dwj_restart_only("B")
|
||||
|
||||
|
||||
class TestDwjRestartAndEnable:
|
||||
@patch(f"{MOD}._cups_enable_printer", return_value=True)
|
||||
@patch(f"{MOD}._cups_restart_service", return_value=True)
|
||||
def test_success(self, r: MagicMock, e: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_dwj_restart_and_enable("B")
|
||||
|
||||
@patch(f"{MOD}._cups_restart_service", return_value=False)
|
||||
def test_restart_fails(self, r: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_dwj_restart_and_enable("B")
|
||||
|
||||
|
||||
# ── _handle_disabled_with_jobs ───────────────────────────────────────
|
||||
|
||||
|
||||
class TestHandleDisabledWithJobs:
|
||||
"""Tests for _handle_disabled_with_jobs dispatch."""
|
||||
|
||||
131
python_pkg/brother_printer/tests/test_cups_queue_menu.py
Normal file
131
python_pkg/brother_printer/tests/test_cups_queue_menu.py
Normal file
@ -0,0 +1,131 @@
|
||||
"""Tests for brother_printer._queue_fix - the interactive fix menu."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from io import StringIO
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from python_pkg.brother_printer._queue_fix import (
|
||||
_dwj_cancel_and_enable,
|
||||
_dwj_cancel_only,
|
||||
_dwj_enable_only,
|
||||
_dwj_restart_and_enable,
|
||||
_dwj_restart_only,
|
||||
_offer_queue_fix,
|
||||
)
|
||||
from python_pkg.brother_printer.data_classes import CUPSJob, CUPSQueueStatus
|
||||
|
||||
MOD = "python_pkg.brother_printer._queue_fix"
|
||||
|
||||
|
||||
# ── _offer_queue_fix ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestOfferQueueFix:
|
||||
"""Tests for _offer_queue_fix menu routing."""
|
||||
|
||||
@patch(f"{MOD}._handle_disabled_with_jobs")
|
||||
@patch(f"{MOD}._prompt", return_value="1")
|
||||
def test_disabled_with_jobs(self, p: MagicMock, mock_handler: MagicMock) -> None:
|
||||
queue = CUPSQueueStatus(
|
||||
printer_name="B",
|
||||
enabled=False,
|
||||
jobs=[CUPSJob("j1", "alice", "1024", "Mon")],
|
||||
)
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_offer_queue_fix(queue)
|
||||
mock_handler.assert_called_once_with(queue, "1")
|
||||
|
||||
@patch(f"{MOD}._handle_disabled_no_jobs")
|
||||
@patch(f"{MOD}._prompt", return_value="2")
|
||||
def test_disabled_no_jobs(self, p: MagicMock, mock_handler: MagicMock) -> None:
|
||||
queue = CUPSQueueStatus(printer_name="B", enabled=False)
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_offer_queue_fix(queue)
|
||||
mock_handler.assert_called_once_with(queue, "2")
|
||||
|
||||
@patch(f"{MOD}._handle_enabled_with_jobs")
|
||||
@patch(f"{MOD}._prompt", return_value="1")
|
||||
def test_enabled_with_jobs(self, p: MagicMock, mock_handler: MagicMock) -> None:
|
||||
queue = CUPSQueueStatus(
|
||||
printer_name="B",
|
||||
enabled=True,
|
||||
jobs=[CUPSJob("j1", "alice", "1024", "Mon")],
|
||||
)
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_offer_queue_fix(queue)
|
||||
mock_handler.assert_called_once_with(queue, "1")
|
||||
|
||||
@patch(f"{MOD}._handle_backend_errors_only")
|
||||
@patch(f"{MOD}._prompt", return_value="1")
|
||||
def test_backend_errors_only(self, p: MagicMock, mock_handler: MagicMock) -> None:
|
||||
queue = CUPSQueueStatus(printer_name="B", enabled=True)
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_offer_queue_fix(queue)
|
||||
mock_handler.assert_called_once_with("1")
|
||||
|
||||
|
||||
# ── _dwj_* action functions ─────────────────────────────────────────
|
||||
|
||||
|
||||
class TestDwjEnableOnly:
|
||||
@patch(f"{MOD}._cups_enable_printer", return_value=True)
|
||||
def test_success(self, m: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_dwj_enable_only("B")
|
||||
|
||||
@patch(f"{MOD}._cups_enable_printer", return_value=False)
|
||||
def test_failure(self, m: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_dwj_enable_only("B")
|
||||
|
||||
|
||||
class TestDwjCancelAndEnable:
|
||||
@patch(f"{MOD}._cups_enable_printer", return_value=True)
|
||||
@patch(f"{MOD}._cups_cancel_all_jobs", return_value=True)
|
||||
def test_success(self, c: MagicMock, e: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_dwj_cancel_and_enable("B")
|
||||
|
||||
@patch(f"{MOD}._cups_enable_printer", return_value=False)
|
||||
@patch(f"{MOD}._cups_cancel_all_jobs", return_value=True)
|
||||
def test_enable_fails(self, c: MagicMock, e: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_dwj_cancel_and_enable("B")
|
||||
|
||||
|
||||
class TestDwjCancelOnly:
|
||||
@patch(f"{MOD}._cups_cancel_all_jobs", return_value=True)
|
||||
def test_success(self, m: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_dwj_cancel_only("B")
|
||||
|
||||
@patch(f"{MOD}._cups_cancel_all_jobs", return_value=False)
|
||||
def test_failure(self, m: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_dwj_cancel_only("B")
|
||||
|
||||
|
||||
class TestDwjRestartOnly:
|
||||
@patch(f"{MOD}._cups_restart_service", return_value=True)
|
||||
def test_success(self, m: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_dwj_restart_only("B")
|
||||
|
||||
@patch(f"{MOD}._cups_restart_service", return_value=False)
|
||||
def test_failure(self, m: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_dwj_restart_only("B")
|
||||
|
||||
|
||||
class TestDwjRestartAndEnable:
|
||||
@patch(f"{MOD}._cups_enable_printer", return_value=True)
|
||||
@patch(f"{MOD}._cups_restart_service", return_value=True)
|
||||
def test_success(self, r: MagicMock, e: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_dwj_restart_and_enable("B")
|
||||
|
||||
@patch(f"{MOD}._cups_restart_service", return_value=False)
|
||||
def test_restart_fails(self, r: MagicMock) -> None:
|
||||
with patch("sys.stdout", new_callable=StringIO):
|
||||
_dwj_restart_and_enable("B")
|
||||
Loading…
Reference in New Issue
Block a user