Skip to content

Improve and extend the _read() method availability for its users and related methods. Part 3: Implement _read() in File() #712

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
belono wants to merge 3 commits into
base: master
Choose a base branch
from
Open

Improve and extend the _read() method availability for its users and related methods. Part 3: Implement _read() in File() #712

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
272a094
Implement _read() into the File() connector
belono Oct 5, 2025
a0458c8
Add temporary path fixture
belono Oct 6, 2025
4df8913
Update File printer tests
belono Oct 6, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 19 additions & 2 deletions src/escpos/printer/file.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@
"""

import logging
import os
import time
from typing import IO, Literal, Optional, Union

from ..escpos import Escpos
Expand Down Expand Up @@ -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.
Expand All @@ -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
Expand Down Expand Up @@ -98,12 +102,25 @@ 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:
return
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
6 changes: 6 additions & 0 deletions test/conftest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
69 changes: 50 additions & 19 deletions test/test_printers/test_printer_file.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,46 +49,51 @@ 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()

assert "enabled" in caplog.text
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

fileprinter.open()
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")
Expand All @@ -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")
Expand All @@ -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")
Expand All @@ -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):
Expand Down