diff --git a/doc/conf.py b/doc/conf.py index 89e1809f..d3b45329 100644 --- a/doc/conf.py +++ b/doc/conf.py @@ -34,7 +34,6 @@ # ones. extensions = [ "sphinx.ext.autodoc", - "sphinx_autodoc_typehints", "sphinx.ext.doctest", "sphinx.ext.todo", "sphinx.ext.coverage", @@ -44,6 +43,7 @@ "sphinx.ext.inheritance_diagram", "sphinx.ext.imgconverter", "sphinxarg.ext", + "sphinx_autodoc_typehints", "sphinxcontrib.datatemplates", "sphinxcontrib.spelling", ] diff --git a/src/escpos/constants.py b/src/escpos/constants.py index 34684840..976d620f 100644 --- a/src/escpos/constants.py +++ b/src/escpos/constants.py @@ -299,7 +299,8 @@ RT_STATUS: bytes = DLE + EOT RT_STATUS_ONLINE: bytes = RT_STATUS + b"\x01" RT_STATUS_PAPER: bytes = RT_STATUS + b"\x04" -RT_MASK_ONLINE: int = 8 +RT_MASK_ONLINE: int = 18 +RT_MASK_OFFLINE: int = 26 RT_MASK_PAPER: int = 18 RT_MASK_LOWPAPER: int = 30 RT_MASK_NOPAPER: int = 114 diff --git a/src/escpos/escpos.py b/src/escpos/escpos.py index fc00e713..f28bd3f9 100644 --- a/src/escpos/escpos.py +++ b/src/escpos/escpos.py @@ -72,7 +72,7 @@ QR_MODEL_2, RT_MASK_LOWPAPER, RT_MASK_NOPAPER, - RT_MASK_ONLINE, + RT_MASK_OFFLINE, RT_MASK_PAPER, RT_STATUS_ONLINE, RT_STATUS_PAPER, @@ -1442,7 +1442,7 @@ def is_online(self) -> bool: status = self.query_status(RT_STATUS_ONLINE) if len(status) == 0: return False - return not (status[0] & RT_MASK_ONLINE) + return not (status[0] & RT_MASK_OFFLINE == RT_MASK_OFFLINE) def paper_status(self) -> int: # could be IntEnum """Query the paper status of the printer. diff --git a/src/escpos/printer/cups.py b/src/escpos/printer/cups.py index bdd53c31..b26dfb9d 100644 --- a/src/escpos/printer/cups.py +++ b/src/escpos/printer/cups.py @@ -13,6 +13,7 @@ import tempfile from typing import Literal, Optional, Type, Union +from ..constants import RT_MASK_OFFLINE, RT_MASK_ONLINE from ..escpos import Escpos from ..exceptions import DeviceNotFoundError @@ -199,15 +200,17 @@ def _clear(self) -> None: self.pending_job = False def _read(self) -> bytes: - """Return a single-item array with the accepting state of the print queue. + """Return the byte corresponding to the RT status response. + + Respond on/offline given the accepting state of the print queue. states: idle = [3], printing a job = [4], stopped = [5] """ printer = self.printers.get(self.printer_name, {}) state = printer.get("printer-state") if not state or state in [4, 5]: - return b"8" # offline - return b"0" # online + return bytes([RT_MASK_OFFLINE]) + return bytes([RT_MASK_ONLINE]) def close(self) -> None: """Close CUPS connection. diff --git a/src/escpos/printer/lp.py b/src/escpos/printer/lp.py index fafab262..cfc7ec8c 100644 --- a/src/escpos/printer/lp.py +++ b/src/escpos/printer/lp.py @@ -14,6 +14,7 @@ import sys from typing import Literal, Optional, Union +from ..constants import RT_MASK_OFFLINE, RT_MASK_ONLINE from ..escpos import Escpos from ..exceptions import DeviceNotFoundError @@ -154,6 +155,20 @@ def open( return logging.info("LP printer enabled") + def _read(self) -> bytes: + """Return the byte corresponding to the RT status response. + + Respond on/offline given the accepting state of the print queue. + """ + p_name = subprocess.run( + ["lpstat", "-a", self.printer_name], + capture_output=True, + text=True, + ) + if p_name.returncode > 0 or "Rejecting Jobs" in p_name.stdout: + return bytes([RT_MASK_OFFLINE]) + return bytes([RT_MASK_ONLINE]) + def close(self) -> None: """Stop the subprocess.""" if not self._device: diff --git a/test/test_printers/test_printer_cups.py b/test/test_printers/test_printer_cups.py index 85edfe32..e7d76835 100644 --- a/test/test_printers/test_printer_cups.py +++ b/test/test_printers/test_printer_cups.py @@ -169,8 +169,33 @@ def test_printers_no_device(cupsprinter) -> None: def test_read_no_device(cupsprinter) -> None: """ GIVEN a cups printer object - WHEN device is None - THEN check the return value is b'8' + WHEN device is None while querying for printer status + THEN check the return value is the byte '\x1a' """ cupsprinter.device = None - assert cupsprinter._read() == b"8" + assert cupsprinter._read() == bytes([26]) + + +@pytest.mark.parametrize( + "state,expected", + [ + pytest.param(3, bytes([18])), + pytest.param(4, bytes([26])), + pytest.param(5, bytes([26])), + ], +) +def test_read(cupsprinter, mocker, state, expected) -> None: + """ + GIVEN a cups printer object and a mocked pycups device + WHEN querying for printer status + THEN check the return value is the expected + """ + mocker.patch("cups.Connection") + mocker.patch( + "escpos.printer.CupsPrinter.printers", + new={"test_printer": {"printer-state": state}}, + ) + + cupsprinter.printer_name = "test_printer" + + assert cupsprinter._read() == expected diff --git a/test/test_printers/test_printer_lp.py b/test/test_printers/test_printer_lp.py index fac00328..11da04d1 100644 --- a/test/test_printers/test_printer_lp.py +++ b/test/test_printers/test_printer_lp.py @@ -154,6 +154,36 @@ def test_auto_flush_on_close(lpprinter, mocker, caplog, capsys): assert spy.call_count == 1 +@pytest.mark.parametrize( + "stdout,returncode,expected", + [ + pytest.param("A success message...", 0, bytes([18])), + pytest.param("... Rejecting Jobs", 0, bytes([26])), + pytest.param("An error message...", 1, bytes([26])), + ], +) +def test_read(lpprinter, mocker, stdout, returncode, expected) -> None: + """ + GIVEN a lp printer object and a mocked connection + WHEN querying for printer status + THEN check the return value is the expected + """ + import subprocess + + mocker.patch("escpos.printer.LP.printers", new={"test_printer": "Test"}) + mocker.patch.object( + subprocess, + "run", + return_value=subprocess.CompletedProcess( + [], returncode=returncode, stdout=stdout + ), + ) + + lpprinter.printer_name = "test_printer" + + assert lpprinter._read() == expected + + def test_close(lpprinter, caplog, mocker): """ GIVEN a lp printer object and a mocked connection