mirror of
https://github.com/kuhyx/testsAndMisc.git
synced 2026-07-04 20:03:12 +02:00
Krzysztof kuhy Rudnicki
ee27d10fef
Fix ruff violations in ~15 source files and ~60+ test files to minimize per-file-ignores in pyproject.toml. Remaining ignores are justified with comments explaining why each suppression is necessary. Source fixes: FBT003 (keyword args), S310 (URL validation), SLF001 (private access), T201 (print→logging), C901 (complexity), E501 (line length), E402 (import order). Test fixes: SIM117 (combined with), FBT (boolean args), PERF203 (try in loop), S310/S607 (URLs/executables), E402/E501 (imports/lines), S108 (tmp paths), PLR0913 (too many args), ARG (unused args), ANN (type annotations), RUF059 (unused unpacked vars), PT019 (fixture naming). Remaining per-file-ignores (with justifications): - Tests: ARG, D, PLC0415, PLR2004, S101, SLF001 - music_gen sources: PLC0415 (heavy ML lazy imports) - moviepy_showcase: PLC0415 (circular dependency) - generate_images: PLR0913 (matplotlib helpers need many params) - praca_magisterska_video: E501, E402 (long paths, mpl.use)
455 lines
16 KiB
Python
455 lines
16 KiB
Python
"""Tests for brother_printer.cups_service module."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from python_pkg.brother_printer.cups_service import (
|
|
_ensure_cups_running,
|
|
_get_cups_total_pages,
|
|
_get_pyusb_device_info,
|
|
_load_consumable_state,
|
|
_query_usb_port_status_raw,
|
|
_save_consumable_state,
|
|
_stop_cups,
|
|
is_cups_scheduler_running,
|
|
reset_consumable,
|
|
start_cups,
|
|
)
|
|
|
|
MOD = "python_pkg.brother_printer.cups_service"
|
|
|
|
|
|
class TestGetPyusbDeviceInfo:
|
|
def test_found(self) -> None:
|
|
import sys as _sys
|
|
|
|
mock_usb = MagicMock()
|
|
mock_dev = MagicMock()
|
|
mock_dev.product = "HL-1110"
|
|
mock_dev.serial_number = "SN123"
|
|
mock_usb.core.find.return_value = mock_dev
|
|
with patch.dict(_sys.modules, {"usb": mock_usb, "usb.core": mock_usb.core}):
|
|
result = _get_pyusb_device_info()
|
|
assert result["product"] == "HL-1110"
|
|
assert result["serial"] == "SN123"
|
|
|
|
def test_import_error(self) -> None:
|
|
import sys as _sys
|
|
|
|
mock_usb = MagicMock()
|
|
mock_usb.core.find.side_effect = ImportError("no usb")
|
|
with patch.dict(_sys.modules, {"usb": mock_usb, "usb.core": mock_usb.core}):
|
|
result = _get_pyusb_device_info()
|
|
assert result == {}
|
|
|
|
def test_not_found(self) -> None:
|
|
import sys as _sys
|
|
|
|
mock_usb = MagicMock()
|
|
mock_usb.core.find.return_value = None
|
|
with patch.dict(_sys.modules, {"usb": mock_usb, "usb.core": mock_usb.core}):
|
|
result = _get_pyusb_device_info()
|
|
assert result == {}
|
|
|
|
def test_none_product_serial(self) -> None:
|
|
import sys as _sys
|
|
|
|
mock_usb = MagicMock()
|
|
mock_dev = MagicMock()
|
|
mock_dev.product = None
|
|
mock_dev.serial_number = None
|
|
mock_usb.core.find.return_value = mock_dev
|
|
with patch.dict(_sys.modules, {"usb": mock_usb, "usb.core": mock_usb.core}):
|
|
result = _get_pyusb_device_info()
|
|
assert result["product"] == ""
|
|
assert result["serial"] == ""
|
|
|
|
def test_oserror(self) -> None:
|
|
import sys as _sys
|
|
|
|
mock_usb = MagicMock()
|
|
mock_usb.core.find.side_effect = OSError("usb fail")
|
|
with patch.dict(_sys.modules, {"usb": mock_usb, "usb.core": mock_usb.core}):
|
|
result = _get_pyusb_device_info()
|
|
assert result == {}
|
|
|
|
def test_value_error(self) -> None:
|
|
import sys as _sys
|
|
|
|
mock_usb = MagicMock()
|
|
mock_usb.core.find.side_effect = ValueError("bad")
|
|
with patch.dict(_sys.modules, {"usb": mock_usb, "usb.core": mock_usb.core}):
|
|
result = _get_pyusb_device_info()
|
|
assert result == {}
|
|
|
|
|
|
class TestStopCups:
|
|
@patch(f"{MOD}.shutil.which", return_value=None)
|
|
def test_no_systemctl(self, m: MagicMock) -> None:
|
|
assert _stop_cups() is False
|
|
|
|
@patch(f"{MOD}.time.sleep")
|
|
@patch(f"{MOD}.subprocess.run")
|
|
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/systemctl")
|
|
def test_success(self, w: MagicMock, mock_run: MagicMock, s: MagicMock) -> None:
|
|
mock_run.return_value = MagicMock()
|
|
assert _stop_cups() is True
|
|
|
|
@patch(f"{MOD}.subprocess.run")
|
|
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/systemctl")
|
|
def test_timeout(self, w: MagicMock, mock_run: MagicMock) -> None:
|
|
mock_run.side_effect = subprocess.TimeoutExpired("systemctl", 15)
|
|
assert _stop_cups() is False
|
|
|
|
@patch(f"{MOD}.subprocess.run")
|
|
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/systemctl")
|
|
def test_called_process_error(self, w: MagicMock, mock_run: MagicMock) -> None:
|
|
mock_run.side_effect = subprocess.CalledProcessError(1, "systemctl")
|
|
assert _stop_cups() is False
|
|
|
|
@patch(f"{MOD}.subprocess.run")
|
|
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/systemctl")
|
|
def test_oserror(self, w: MagicMock, mock_run: MagicMock) -> None:
|
|
mock_run.side_effect = OSError("fail")
|
|
assert _stop_cups() is False
|
|
|
|
|
|
class TestIsCupsSchedulerRunning:
|
|
@patch(f"{MOD}.shutil.which", return_value=None)
|
|
def test_no_lpstat(self, m: MagicMock) -> None:
|
|
assert is_cups_scheduler_running() is False
|
|
|
|
@patch(f"{MOD}.subprocess.run")
|
|
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/lpstat")
|
|
def test_running(self, w: MagicMock, mock_run: MagicMock) -> None:
|
|
mock_run.return_value = MagicMock(stdout="scheduler is running")
|
|
assert is_cups_scheduler_running() is True
|
|
|
|
@patch(f"{MOD}.subprocess.run")
|
|
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/lpstat")
|
|
def test_not_running(self, w: MagicMock, mock_run: MagicMock) -> None:
|
|
mock_run.return_value = MagicMock(stdout="scheduler is not running")
|
|
assert is_cups_scheduler_running() 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", 3)
|
|
assert is_cups_scheduler_running() is False
|
|
|
|
@patch(f"{MOD}.subprocess.run")
|
|
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/lpstat")
|
|
def test_oserror(self, w: MagicMock, mock_run: MagicMock) -> None:
|
|
mock_run.side_effect = OSError("fail")
|
|
assert is_cups_scheduler_running() is False
|
|
|
|
|
|
class TestStartCups:
|
|
@patch(f"{MOD}.shutil.which", return_value=None)
|
|
def test_no_systemctl(self, m: MagicMock) -> None:
|
|
assert start_cups() is False
|
|
|
|
@patch(f"{MOD}.time.sleep")
|
|
@patch(f"{MOD}.is_cups_scheduler_running")
|
|
@patch(f"{MOD}.subprocess.run")
|
|
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/systemctl")
|
|
def test_success(
|
|
self,
|
|
w: MagicMock,
|
|
mock_run: MagicMock,
|
|
mock_is_running: MagicMock,
|
|
s: MagicMock,
|
|
) -> None:
|
|
mock_run.return_value = MagicMock()
|
|
mock_is_running.return_value = True
|
|
assert start_cups() is True
|
|
|
|
@patch(f"{MOD}.subprocess.run")
|
|
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/systemctl")
|
|
def test_timeout(self, w: MagicMock, mock_run: MagicMock) -> None:
|
|
mock_run.side_effect = subprocess.TimeoutExpired("systemctl", 15)
|
|
assert start_cups() is False
|
|
|
|
@patch(f"{MOD}.subprocess.run")
|
|
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/systemctl")
|
|
def test_called_process_error(self, w: MagicMock, mock_run: MagicMock) -> None:
|
|
mock_run.side_effect = subprocess.CalledProcessError(1, "systemctl")
|
|
assert start_cups() is False
|
|
|
|
@patch(f"{MOD}.time.sleep")
|
|
@patch(f"{MOD}.is_cups_scheduler_running", return_value=False)
|
|
@patch(f"{MOD}.subprocess.run")
|
|
@patch(f"{MOD}.shutil.which", return_value="/usr/bin/systemctl")
|
|
def test_never_starts(
|
|
self,
|
|
w: MagicMock,
|
|
mock_run: MagicMock,
|
|
is_running: MagicMock,
|
|
s: MagicMock,
|
|
) -> None:
|
|
mock_run.return_value = MagicMock()
|
|
assert start_cups() is False
|
|
|
|
|
|
class TestEnsureCupsRunning:
|
|
@patch(f"{MOD}.is_cups_scheduler_running", return_value=True)
|
|
def test_already_running(self, m: MagicMock) -> None:
|
|
assert _ensure_cups_running() is True
|
|
|
|
@patch(f"{MOD}.start_cups", return_value=True)
|
|
@patch(f"{MOD}.is_cups_scheduler_running", return_value=False)
|
|
def test_needs_start(self, is_running: MagicMock, st: MagicMock) -> None:
|
|
assert _ensure_cups_running() is True
|
|
|
|
@patch(f"{MOD}.start_cups", return_value=False)
|
|
@patch(f"{MOD}.is_cups_scheduler_running", return_value=False)
|
|
def test_start_fails(self, is_running: MagicMock, st: MagicMock) -> None:
|
|
assert _ensure_cups_running() is False
|
|
|
|
|
|
class TestQueryUsbPortStatusRaw:
|
|
def test_import_error(self) -> None:
|
|
with (
|
|
patch(f"{MOD}._stop_cups"),
|
|
# Simulate ImportError for usb.core
|
|
patch.dict(
|
|
"sys.modules", {"usb": None, "usb.core": None, "usb.util": None}
|
|
),
|
|
):
|
|
result = _query_usb_port_status_raw()
|
|
assert result is None
|
|
|
|
@patch(f"{MOD}.start_cups")
|
|
@patch(f"{MOD}._stop_cups", return_value=False)
|
|
def test_stop_cups_fails(self, st: MagicMock, s: MagicMock) -> None:
|
|
import sys as _sys
|
|
|
|
mock_usb = MagicMock()
|
|
mock_usb.core.find.return_value = MagicMock()
|
|
with patch.dict(
|
|
_sys.modules,
|
|
{"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util},
|
|
):
|
|
result = _query_usb_port_status_raw()
|
|
assert result is None
|
|
|
|
@patch(f"{MOD}.start_cups")
|
|
@patch(f"{MOD}._stop_cups", return_value=True)
|
|
def test_dev_none_after_reset(self, st: MagicMock, s: MagicMock) -> None:
|
|
import sys as _sys
|
|
|
|
mock_usb = MagicMock()
|
|
mock_dev = MagicMock()
|
|
mock_usb.core.find.side_effect = [mock_dev, None]
|
|
with (
|
|
patch.dict(
|
|
_sys.modules,
|
|
{"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util},
|
|
),
|
|
patch(f"{MOD}.time.sleep"),
|
|
):
|
|
result = _query_usb_port_status_raw()
|
|
assert result is None
|
|
|
|
@patch(f"{MOD}.start_cups")
|
|
@patch(f"{MOD}._stop_cups", return_value=True)
|
|
def test_success(self, stop: MagicMock, start: MagicMock) -> None:
|
|
import sys as _sys
|
|
|
|
mock_usb = MagicMock()
|
|
mock_dev = MagicMock()
|
|
mock_dev.is_kernel_driver_active.return_value = True
|
|
mock_dev.ctrl_transfer.return_value = [0x18]
|
|
mock_usb.core.find.return_value = mock_dev
|
|
mock_usb.core.USBError = type("USBError", (Exception,), {})
|
|
with (
|
|
patch.dict(
|
|
_sys.modules,
|
|
{"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util},
|
|
),
|
|
patch(f"{MOD}.time.sleep"),
|
|
):
|
|
result = _query_usb_port_status_raw()
|
|
assert result is not None
|
|
assert result.online is True
|
|
|
|
@patch(f"{MOD}.start_cups")
|
|
@patch(f"{MOD}._stop_cups", return_value=True)
|
|
def test_kernel_driver_not_active(self, stop: MagicMock, start: MagicMock) -> None:
|
|
import sys as _sys
|
|
|
|
mock_usb = MagicMock()
|
|
mock_dev = MagicMock()
|
|
mock_dev.is_kernel_driver_active.return_value = False
|
|
mock_dev.ctrl_transfer.return_value = [0x18]
|
|
mock_usb.core.find.return_value = mock_dev
|
|
mock_usb.core.USBError = type("USBError", (Exception,), {})
|
|
with (
|
|
patch.dict(
|
|
_sys.modules,
|
|
{"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util},
|
|
),
|
|
patch(f"{MOD}.time.sleep"),
|
|
):
|
|
result = _query_usb_port_status_raw()
|
|
assert result is not None
|
|
|
|
@patch(f"{MOD}.start_cups")
|
|
@patch(f"{MOD}._stop_cups", return_value=True)
|
|
def test_kernel_driver_usberror(self, stop: MagicMock, start: MagicMock) -> None:
|
|
import sys as _sys
|
|
|
|
mock_usb = MagicMock()
|
|
mock_dev = MagicMock()
|
|
usb_error_cls = type("USBError", (Exception,), {})
|
|
mock_dev.is_kernel_driver_active.side_effect = usb_error_cls("err")
|
|
mock_dev.ctrl_transfer.return_value = [0x18]
|
|
mock_usb.core.find.return_value = mock_dev
|
|
mock_usb.core.USBError = usb_error_cls
|
|
with (
|
|
patch.dict(
|
|
_sys.modules,
|
|
{"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util},
|
|
),
|
|
patch(f"{MOD}.time.sleep"),
|
|
):
|
|
result = _query_usb_port_status_raw()
|
|
assert result is not None
|
|
|
|
@patch(f"{MOD}.start_cups")
|
|
@patch(f"{MOD}._stop_cups", return_value=True)
|
|
def test_oserror_during_transfer(self, stop: MagicMock, start: MagicMock) -> None:
|
|
import sys as _sys
|
|
|
|
mock_usb = MagicMock()
|
|
mock_dev = MagicMock()
|
|
mock_dev.is_kernel_driver_active.return_value = False
|
|
mock_usb.core.find.return_value = mock_dev
|
|
mock_usb.core.USBError = type("USBError", (Exception,), {})
|
|
mock_usb.util.claim_interface.side_effect = OSError("usb fail")
|
|
with (
|
|
patch.dict(
|
|
_sys.modules,
|
|
{"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util},
|
|
),
|
|
patch(f"{MOD}.time.sleep"),
|
|
):
|
|
result = _query_usb_port_status_raw()
|
|
assert result is None
|
|
|
|
@patch(f"{MOD}.start_cups")
|
|
@patch(f"{MOD}._stop_cups", return_value=True)
|
|
def test_dev_none_initial(self, stop: MagicMock, start: MagicMock) -> None:
|
|
import sys as _sys
|
|
|
|
mock_usb = MagicMock()
|
|
mock_usb.core.find.return_value = None
|
|
with patch.dict(
|
|
_sys.modules,
|
|
{"usb": mock_usb, "usb.core": mock_usb.core, "usb.util": mock_usb.util},
|
|
):
|
|
result = _query_usb_port_status_raw()
|
|
assert result is None
|
|
|
|
|
|
class TestGetCupsTotalPages:
|
|
@patch(f"{MOD}.CUPS_PAGE_LOG")
|
|
def test_no_log(self, mock_log: MagicMock) -> None:
|
|
mock_log.exists.return_value = False
|
|
assert _get_cups_total_pages() == 0
|
|
|
|
@patch(f"{MOD}.CUPS_PAGE_LOG")
|
|
def test_with_entries(self, mock_log: MagicMock) -> None:
|
|
mock_log.exists.return_value = True
|
|
mock_log.read_text.return_value = (
|
|
"printer 1 [2025-01-01] total 5\n"
|
|
"printer 2 [2025-01-01] total 3\n"
|
|
"printer 1 [2025-01-01] total 10\n"
|
|
)
|
|
assert _get_cups_total_pages() == 13 # max(5,10) + 3
|
|
|
|
@patch(f"{MOD}.CUPS_PAGE_LOG")
|
|
def test_oserror(self, mock_log: MagicMock) -> None:
|
|
mock_log.exists.return_value = True
|
|
mock_log.read_text.side_effect = OSError("fail")
|
|
assert _get_cups_total_pages() == 0
|
|
|
|
@patch(f"{MOD}.CUPS_PAGE_LOG")
|
|
def test_no_matching_lines(self, mock_log: MagicMock) -> None:
|
|
mock_log.exists.return_value = True
|
|
mock_log.read_text.return_value = "some garbage\n"
|
|
assert _get_cups_total_pages() == 0
|
|
|
|
|
|
class TestLoadConsumableState:
|
|
@patch(f"{MOD}.CONSUMABLE_STATE_FILE")
|
|
def test_no_file(self, mock_file: MagicMock) -> None:
|
|
mock_file.exists.return_value = False
|
|
result = _load_consumable_state()
|
|
assert result == {"toner_replaced_at": 0, "drum_replaced_at": 0}
|
|
|
|
@patch(f"{MOD}.CONSUMABLE_STATE_FILE")
|
|
def test_valid_file(self, mock_file: MagicMock) -> None:
|
|
mock_file.exists.return_value = True
|
|
mock_file.read_text.return_value = json.dumps(
|
|
{"toner_replaced_at": 100, "drum_replaced_at": 200},
|
|
)
|
|
result = _load_consumable_state()
|
|
assert result["toner_replaced_at"] == 100
|
|
assert result["drum_replaced_at"] == 200
|
|
|
|
@patch(f"{MOD}.CONSUMABLE_STATE_FILE")
|
|
def test_oserror(self, mock_file: MagicMock) -> None:
|
|
mock_file.exists.return_value = True
|
|
mock_file.read_text.side_effect = OSError("fail")
|
|
result = _load_consumable_state()
|
|
assert result["toner_replaced_at"] == 0
|
|
|
|
@patch(f"{MOD}.CONSUMABLE_STATE_FILE")
|
|
def test_bad_json(self, mock_file: MagicMock) -> None:
|
|
mock_file.exists.return_value = True
|
|
mock_file.read_text.return_value = "not json"
|
|
result = _load_consumable_state()
|
|
assert result["toner_replaced_at"] == 0
|
|
|
|
@patch(f"{MOD}.CONSUMABLE_STATE_FILE")
|
|
def test_bad_values(self, mock_file: MagicMock) -> None:
|
|
mock_file.exists.return_value = True
|
|
mock_file.read_text.return_value = json.dumps(
|
|
{"toner_replaced_at": "bad"},
|
|
)
|
|
result = _load_consumable_state()
|
|
assert result["toner_replaced_at"] == 0
|
|
|
|
|
|
class TestSaveConsumableState:
|
|
@patch(f"{MOD}.CONSUMABLE_STATE_FILE")
|
|
def test_saves(self, mock_file: MagicMock) -> None:
|
|
mock_file.parent = MagicMock()
|
|
_save_consumable_state({"toner_replaced_at": 100, "drum_replaced_at": 200})
|
|
mock_file.write_text.assert_called_once()
|
|
written = mock_file.write_text.call_args[0][0]
|
|
data = json.loads(written)
|
|
assert data["toner_replaced_at"] == 100
|
|
|
|
|
|
class TestResetConsumable:
|
|
@patch(f"{MOD}._out")
|
|
@patch(f"{MOD}._save_consumable_state")
|
|
@patch(f"{MOD}._load_consumable_state")
|
|
@patch(f"{MOD}._get_cups_total_pages", return_value=500)
|
|
def test_reset_toner(
|
|
self,
|
|
pages: MagicMock,
|
|
load: MagicMock,
|
|
mock_save: MagicMock,
|
|
out: MagicMock,
|
|
) -> None:
|
|
load.return_value = {"toner_replaced_at": 0, "drum_replaced_at": 0}
|
|
reset_consumable("toner")
|
|
saved_state = mock_save.call_args[0][0]
|
|
assert saved_state["toner_replaced_at"] == 500
|