diff --git a/src/escpos/printer/file.py b/src/escpos/printer/file.py index 44e414e5..46774a13 100644 --- a/src/escpos/printer/file.py +++ b/src/escpos/printer/file.py @@ -9,6 +9,8 @@ """ import logging +import os +import time from typing import IO, Literal, Optional, Union from ..escpos import Escpos @@ -54,6 +56,7 @@ def __init__(self, devfile: str = "", auto_flush: bool = True, *args, **kwargs): self.auto_flush = auto_flush self._device: Union[Literal[False], Literal[None], IO[bytes]] = False + self._fd: Optional[int] = None def open(self, raise_not_found: bool = True) -> None: """Open system file. @@ -70,7 +73,8 @@ def open(self, raise_not_found: bool = True) -> None: try: # Open device - self.device: Optional[IO[bytes]] = open(self.devfile, "wb") + self._fd = os.open(self.devfile, os.O_RDWR) + self.device: Optional[IO[bytes]] = os.fdopen(self._fd, "wb") except OSError as e: # Raise exception or log error and cancel self.device = None @@ -98,6 +102,19 @@ def _raw(self, msg: bytes) -> None: if self.auto_flush: self.flush() + def _read(self) -> bytes: + """Read a data buffer and return it to the caller.""" + assert self.device and not self.device.closed + assert self._fd is not None + if not self.auto_flush: + logging.warning( + "Param 'auto_flush' is disabled. Forcing a flush before attempting to read the data" + ) + self.flush() + time.sleep(0.2) # Give some time to respond + os.lseek(self._fd, -1, os.SEEK_END) # Rewind 1 byte + return os.read(self._fd, 16) + def close(self) -> None: """Close system file.""" if not self._device: @@ -105,5 +122,5 @@ def close(self) -> None: logging.info("Closing File connection to printer %s", self.devfile) if not self.auto_flush: self.flush() - self._device.close() + self._device.close() # This closes also the file descriptor self._device = False diff --git a/test/conftest.py b/test/conftest.py index f4a58ef7..7343cbc1 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -47,3 +47,9 @@ def cupsprinter(): @pytest.fixture def devicenotfounderror(): return DeviceNotFoundError + + +@pytest.fixture(scope="module") +def temp_path(tmp_path_factory): + f = tmp_path_factory.mktemp("tempdir") + return f diff --git a/test/test_printers/test_printer_file.py b/test/test_printers/test_printer_file.py index 3558231b..9b4aa4da 100644 --- a/test/test_printers/test_printer_file.py +++ b/test/test_printers/test_printer_file.py @@ -49,14 +49,15 @@ def test_open_not_raise_exception(fileprinter, caplog): assert fileprinter.device is None -def test_open(fileprinter, caplog, mocker): +def test_open(fileprinter, caplog, temp_path): """ - GIVEN a file printer object and a mocked connection + GIVEN a file printer object bound to an existing file WHEN a valid connection to a device is opened THEN check the success is logged and the device property is set """ - mocker.patch("builtins.open") - + tmpfile = temp_path / "test_open.bin" + tmpfile.touch() + fileprinter.devfile = tmpfile with caplog.at_level(logging.INFO): fileprinter.open() @@ -64,15 +65,17 @@ def test_open(fileprinter, caplog, mocker): assert fileprinter.device -def test_close_on_reopen(fileprinter, mocker): +def test_close_on_reopen(fileprinter, mocker, temp_path): """ - GIVEN a file printer object and a mocked connection + GIVEN a file printer object bound to an existing file WHEN a valid connection to a device is reopened before close THEN check the close method is called if _device """ - mocker.patch("builtins.open") spy = mocker.spy(fileprinter, "close") + tmpfile = temp_path / "test_close_on_reopen.bin" + tmpfile.touch() + fileprinter.devfile = tmpfile fileprinter.open() assert fileprinter._device @@ -80,15 +83,17 @@ def test_close_on_reopen(fileprinter, mocker): spy.assert_called_once_with() -def test_flush(fileprinter, mocker): +def test_flush(fileprinter, mocker, temp_path): """ - GIVEN a file printer object and a mocked connection + GIVEN a file printer object bound to an existing file WHEN auto_flush is disabled and flush() issued manually THEN check the flush method is called only one time. """ spy = mocker.spy(fileprinter, "flush") - mocker.patch("builtins.open") + tmpfile = temp_path / "test_flush.bin" + tmpfile.touch() + fileprinter.devfile = tmpfile fileprinter.auto_flush = False fileprinter.open() fileprinter.textln("python-escpos") @@ -97,15 +102,17 @@ def test_flush(fileprinter, mocker): assert spy.call_count == 1 -def test_auto_flush_on_command(fileprinter, mocker): +def test_auto_flush_on_command(fileprinter, mocker, temp_path): """ - GIVEN a file printer object and a mocked connection + GIVEN a file printer object bound to an existing file WHEN auto_flush is enabled and flush() not issued manually THEN check the flush method is called automatically """ spy = mocker.spy(fileprinter, "flush") - mocker.patch("builtins.open") + tmpfile = temp_path / "test_auto_flush_on_command.bin" + tmpfile.touch() + fileprinter.devfile = tmpfile fileprinter.auto_flush = True fileprinter.open() fileprinter.textln("python-escpos") @@ -114,15 +121,17 @@ def test_auto_flush_on_command(fileprinter, mocker): assert spy.call_count > 1 -def test_auto_flush_on_close(fileprinter, mocker, caplog, capsys): +def test_auto_flush_on_close(fileprinter, mocker, temp_path): """ - GIVEN a file printer object and a mocked connection + GIVEN a file printer object bound to an existing file WHEN auto_flush is disabled and flush() not issued manually THEN check the flush method is called automatically on close """ spy = mocker.spy(fileprinter, "flush") - mocker.patch("builtins.open") + tmpfile = temp_path / "test_autoflush_on_close.bin" + tmpfile.touch() + fileprinter.devfile = tmpfile fileprinter.auto_flush = False fileprinter.open() fileprinter.textln("python-escpos") @@ -131,13 +140,35 @@ def test_auto_flush_on_close(fileprinter, mocker, caplog, capsys): assert spy.call_count == 1 -def test_close(fileprinter, caplog, mocker): +def test_read(fileprinter, caplog, temp_path): + """ + GIVEN a file printer object bound to an existing file + WHEN reading the file buffer even with auto_flush disabled + THEN check a warning is logged for auto_flush and the last byte is read and response is correct. + """ + tmpfile = temp_path / "test_read.bin" + tmpfile.touch() + fileprinter.devfile = tmpfile + fileprinter.open() + fileprinter._raw(b"\x08\x12") + fileprinter.auto_flush = False + + with caplog.at_level(logging.WARNING): + resp = fileprinter._read() + + assert "Param 'auto_flush' is disabled" in caplog.text + assert resp == b"\x12" + + +def test_close(fileprinter, caplog, temp_path): """ - GIVEN a file printer object and a mocked connection + GIVEN a file printer object bound to an existing file WHEN a connection is opened and closed THEN check the closing is logged and the device property is False """ - mocker.patch("builtins.open") + tmpfile = temp_path / "test_close.bin" + tmpfile.touch() + fileprinter.devfile = tmpfile fileprinter.open() with caplog.at_level(logging.INFO):