From a00b98937baec664f19fe31b3b5b8f49938d5516 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Benito=20L=C3=B3pez?= Date: Mon, 16 Oct 2023 11:36:07 +0200 Subject: [PATCH 1/4] Separate method open() and constructor, enhance consistency between connectors. (#584) * Add self-open mechanism * self-open mechanism through the 'device' property * Separate open() and store connection in 'device' * Restore device status to False on close() * Add default value to all params + type annotations * Add generic DeviceNotFoundError exception * Update USBNotFoundError return code * Enhance connectors consistency * Fix LP printer stall * Fix LP waste of paper due to auto-flush + flush on close * Move platform dependent printers' guard to init --------- Co-authored-by: Patrick Kanzler <4189642+patkan@users.noreply.github.com> --- src/escpos/escpos.py | 29 ++++++- src/escpos/exceptions.py | 37 +++++++-- src/escpos/printer/cups.py | 86 ++++++++++++++------ src/escpos/printer/file.py | 59 ++++++++++---- src/escpos/printer/lp.py | 139 +++++++++++++++++++++++++++------ src/escpos/printer/network.py | 69 ++++++++++++---- src/escpos/printer/serial.py | 91 +++++++++++++-------- src/escpos/printer/usb.py | 105 +++++++++++++++++-------- src/escpos/printer/win32raw.py | 100 +++++++++++++++++------- tox.ini | 1 - 10 files changed, 537 insertions(+), 179 deletions(-) diff --git a/src/escpos/escpos.py b/src/escpos/escpos.py index 07fd887..f66a13b 100644 --- a/src/escpos/escpos.py +++ b/src/escpos/escpos.py @@ -14,7 +14,7 @@ This module contains the abstract base class :py:class:`Escpos`. import textwrap from abc import ABCMeta, abstractmethod # abstract base class support from re import match as re_match -from typing import List, Optional, Union +from typing import List, Literal, Optional, Union import barcode import qrcode @@ -114,7 +114,11 @@ class Escpos(object): class. """ - device = None + # device status: + # False -> Not initialized + # None -> Initialized but not connected + # object -> The connection object (Usb(), Serial(), Network(), etc.) + _device: Union[Literal[False], Literal[None], object] = False def __init__(self, profile=None, magic_encode_args=None, **kwargs) -> None: """Initialize ESCPOS Printer. @@ -128,6 +132,27 @@ class Escpos(object): """Call self.close upon deletion.""" self.close() + @property + def device(self) -> Union[Literal[None], object]: + """Implements a self-open mechanism. + + An attempt to get the property before open the connection + will cause the connection to open. + """ + if self._device is False: + # Open device if not previously opened + self._device = None # None -> Initialized + self.open() + return self._device + + @device.setter + def device(self, new_device: Union[Literal[False], Literal[None], object]): + self._device = new_device + + def open(self): + """Open a printer device/connection.""" + pass + @abstractmethod def _raw(self, msg: bytes) -> None: """Send raw data to the printer. diff --git a/src/escpos/exceptions.py b/src/escpos/exceptions.py index 5cbe352..a583467 100644 --- a/src/escpos/exceptions.py +++ b/src/escpos/exceptions.py @@ -13,7 +13,8 @@ Result/Exit codes: - `60` = Invalid pin to send Cash Drawer pulse :py:exc:`~escpos.exceptions.CashDrawerError` - `70` = Invalid number of tab positions :py:exc:`~escpos.exceptions.TabPosError` - `80` = Invalid char code :py:exc:`~escpos.exceptions.CharCodeError` - - `90` = USB device not found :py:exc:`~escpos.exceptions.USBNotFoundError` + - `90` = Device not found :py:exc:`~escpos.exceptions.DeviceNotFoundError` + - `91` = USB device not found :py:exc:`~escpos.exceptions.USBNotFoundError` - `100` = Set variable out of range :py:exc:`~escpos.exceptions.SetVariableError` - `200` = Configuration not found :py:exc:`~escpos.exceptions.ConfigNotFoundError` - `210` = Configuration syntax error :py:exc:`~escpos.exceptions.ConfigSyntaxError` @@ -275,11 +276,35 @@ class CharCodeError(Error): return "Valid char code must be set ({msg})".format(msg=self.msg) -class USBNotFoundError(Error): - """Device was not found (probably not plugged in). +class DeviceNotFoundError(Error): + """Device was not found. + + The device seems to be not accessible. + The return code for this exception is `90`. + + inheritance: + + .. inheritance-diagram:: escpos.exceptions.Error + :parts: 1 + + """ + + def __init__(self, msg=""): + """Initialize DeviceNotFoundError object.""" + Error.__init__(self, msg) + self.msg = msg + self.resultcode = 90 + + def __str__(self): + """Return string representation of DeviceNotFoundError.""" + return f"Device not found ({self.msg})" + + +class USBNotFoundError(DeviceNotFoundError): + """USB device was not found (probably not plugged in). The USB device seems to be not plugged in. - The return code for this exception is `90`. + The return code for this exception is `91`. inheritance: @@ -292,11 +317,11 @@ class USBNotFoundError(Error): """Initialize USBNotFoundError object.""" Error.__init__(self, msg) self.msg = msg - self.resultcode = 90 + self.resultcode = 91 def __str__(self): """Return string representation of USBNotFoundError.""" - return "USB device not found ({msg})".format(msg=self.msg) + return f"USB device not found ({self.msg})" class SetVariableError(Error): diff --git a/src/escpos/printer/cups.py b/src/escpos/printer/cups.py index 60e8468..0625b01 100644 --- a/src/escpos/printer/cups.py +++ b/src/escpos/printer/cups.py @@ -9,9 +9,12 @@ """ import functools +import logging import tempfile +from typing import Literal, Optional, Type, Union from ..escpos import Escpos +from ..exceptions import DeviceNotFoundError #: keeps track if the pycups dependency could be loaded (:py:class:`escpos.printer.CupsPrinter`) _DEP_PYCUPS = False @@ -20,6 +23,9 @@ try: import cups _DEP_PYCUPS = True + # Store server defaults before further configuration + DEFAULT_HOST = cups.getServer() + DEFAULT_PORT = cups.getPort() except ImportError: pass @@ -78,48 +84,84 @@ class CupsPrinter(Escpos): return is_usable() @dependency_pycups - def __init__(self, printer_name=None, *args, **kwargs): + def __init__(self, printer_name: str = "", *args, **kwargs): """Class constructor for CupsPrinter. :param printer_name: CUPS printer name (Optional) - :type printer_name: str :param host: CUPS server host/ip (Optional) :type host: str :param port: CUPS server port (Optional) :type port: int """ Escpos.__init__(self, *args, **kwargs) - host, port = args or ( - kwargs.get("host", cups.getServer()), - kwargs.get("port", cups.getPort()), + self.host, self.port = args or ( + kwargs.get("host", DEFAULT_HOST), + kwargs.get("port", DEFAULT_PORT), ) - cups.setServer(host) - cups.setPort(port) - self.conn = cups.Connection() - self.tmpfile = None + self.tmpfile = tempfile.NamedTemporaryFile(delete=True) self.printer_name = printer_name self.job_name = "" self.pending_job = False - self.open() + + self._device: Union[ + Literal[False], Literal[None], Type[cups.Connection] + ] = False @property - def printers(self): + def printers(self) -> dict: """Available CUPS printers.""" - return self.conn.getPrinters() + if self.device: + return self.device.getPrinters() + return {} - def open(self, job_name="python-escpos"): + def open( + self, job_name: str = "python-escpos", raise_not_found: bool = True + ) -> None: """Set up a new print job and target the printer. A call to this method is required to send new jobs to - the same CUPS connection. + the CUPS connection after close. Defaults to default CUPS printer. Creates a new temporary file buffer. + + By default raise an exception if device is not found. + + :param raise_not_found: Default True. + False to log error but do not raise exception. + + :raises: :py:exc:`~escpos.exceptions.DeviceNotFoundError` """ + if self._device: + self.close() + + cups.setServer(self.host) + cups.setPort(self.port) self.job_name = job_name - if self.printer_name not in self.printers: - self.printer_name = self.conn.getDefault() - self.tmpfile = tempfile.NamedTemporaryFile(delete=True) + if self.tmpfile.closed: + self.tmpfile = tempfile.NamedTemporaryFile(delete=True) + + try: + # Open device + self.device: Optional[Type[cups.Connection]] = cups.Connection() + if self.device: + # Name validation, set default if no given name + self.printer_name = self.printer_name or self.device.getDefault() + assert self.printer_name in self.printers, "Incorrect printer name" + except (RuntimeError, AssertionError) as e: + # Raise exception or log error and cancel + self.device = None + if raise_not_found: + raise DeviceNotFoundError( + f"Unable to start a print job for the printer {self.printer_name}:" + + f"\n{e}" + ) + else: + logging.error( + "CupsPrinter printing %s not available", self.printer_name + ) + return + logging.info("CupsPrinter printer enabled") def _raw(self, msg): """Append any command sent in raw format to temporary file. @@ -134,14 +176,13 @@ class CupsPrinter(Escpos): self.pending_job = False raise ValueError("Printer job not opened") - @dependency_pycups def send(self): """Send the print job to the printer.""" if self.pending_job: # Rewind tempfile self.tmpfile.seek(0) # Print temporary file via CUPS printer. - self.conn.printFile( + self.device.printFile( self.printer_name, self.tmpfile.name, self.job_name, @@ -173,8 +214,9 @@ class CupsPrinter(Escpos): Send pending job to the printer if needed. """ + if not self._device: + return if self.pending_job: self.send() - if self.conn: - print("Closing CUPS connection to printer {}".format(self.printer_name)) - self.conn = None + logging.info("Closing CUPS connection to printer %s", self.printer_name) + self._device = False diff --git a/src/escpos/printer/file.py b/src/escpos/printer/file.py index acf4802..67cbe4f 100644 --- a/src/escpos/printer/file.py +++ b/src/escpos/printer/file.py @@ -1,6 +1,6 @@ #!/usr/bin/python # -*- coding: utf-8 -*- -"""This module contains the implementation of the CupsPrinter printer driver. +"""This module contains the implementation of the File printer driver. :author: python-escpos developers :organization: `python-escpos `_ @@ -8,7 +8,11 @@ :license: MIT """ +import logging +from typing import IO, Literal, Optional, Union + from ..escpos import Escpos +from ..exceptions import DeviceNotFoundError def is_usable() -> bool: @@ -39,7 +43,7 @@ class File(Escpos): """ return is_usable() - def __init__(self, devfile="/dev/usb/lp0", auto_flush=True, *args, **kwargs): + def __init__(self, devfile: str = "", auto_flush: bool = True, *args, **kwargs): """Initialize file printer with device file. :param devfile: Device file under dev filesystem @@ -48,18 +52,41 @@ class File(Escpos): Escpos.__init__(self, *args, **kwargs) self.devfile = devfile self.auto_flush = auto_flush - self.open() - def open(self): - """Open system file.""" - self.device = open(self.devfile, "wb") + self._device: Union[Literal[False], Literal[None], IO[bytes]] = False - if self.device is None: - print("Could not open the specified file {0}".format(self.devfile)) + def open(self, raise_not_found: bool = True) -> None: + """Open system file. - def flush(self): + By default raise an exception if device is not found. + + :param raise_not_found: Default True. + False to log error but do not raise exception. + + :raises: :py:exc:`~escpos.exceptions.DeviceNotFoundError` + """ + if self._device: + self.close() + + try: + # Open device + self.device: Optional[IO[bytes]] = open(self.devfile, "wb") + except OSError as e: + # Raise exception or log error and cancel + self.device = None + if raise_not_found: + raise DeviceNotFoundError( + f"Could not open the specified file {self.devfile}:\n{e}" + ) + else: + logging.error("File printer %s not found", self.devfile) + return + logging.info("File printer enabled") + + def flush(self) -> None: """Flush printing content.""" - self.device.flush() + if self.device: + self.device.flush() def _raw(self, msg): """Print any command sent in raw format. @@ -71,8 +98,12 @@ class File(Escpos): if self.auto_flush: self.flush() - def close(self): + def close(self) -> None: """Close system file.""" - if self.device is not None: - self.device.flush() - self.device.close() + 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 = False diff --git a/src/escpos/printer/lp.py b/src/escpos/printer/lp.py index 1b7ed54..951df66 100644 --- a/src/escpos/printer/lp.py +++ b/src/escpos/printer/lp.py @@ -1,6 +1,6 @@ #!/usr/bin/python # -*- coding: utf-8 -*- -"""This module contains the implementation of the CupsPrinter printer driver. +"""This module contains the implementation of the LP printer driver. :author: python-escpos developers :organization: `python-escpos `_ @@ -9,11 +9,13 @@ """ import functools -import os +import logging import subprocess import sys +from typing import Literal, Optional, Union from ..escpos import Escpos +from ..exceptions import DeviceNotFoundError def is_usable() -> bool: @@ -61,40 +63,124 @@ class LP(Escpos): """ return is_usable() - def __init__(self, printer_name: str, *args, **kwargs): + @dependency_linux_lp + def __init__(self, printer_name: str = "", *args, **kwargs): """LP class constructor. :param printer_name: CUPS printer name (Optional) - :type printer_name: str :param auto_flush: Automatic flush after every _raw() (Optional) - :type auto_flush: bool + :type auto_flush: bool (Defaults False) """ Escpos.__init__(self, *args, **kwargs) self.printer_name = printer_name - self.auto_flush = kwargs.get("auto_flush", True) - self.open() + self.auto_flush = kwargs.get("auto_flush", False) + self._flushed = False - @dependency_linux_lp - def open(self): - """Invoke _lp_ in a new subprocess and wait for commands.""" - self.lp = subprocess.Popen( - ["lp", "-d", self.printer_name, "-o", "raw"], - stdin=subprocess.PIPE, - stdout=open(os.devnull, "w"), + self._device: Union[Literal[False], Literal[None], subprocess.Popen] = False + + @property + def printers(self) -> dict: + """Available CUPS printers.""" + p_names = subprocess.run( + ["lpstat", "-e"], # Get printer names + capture_output=True, + text=True, ) + p_devs = subprocess.run( + ["lpstat", "-v"], # Get attached devices + capture_output=True, + text=True, + ) + # List and trim output lines + names = [name for name in p_names.stdout.split("\n") if name] + devs = [dev for dev in p_devs.stdout.split("\n") if dev] + # return a dict of {printer name: attached device} pairs + return {name: dev.split()[-1] for name in names for dev in devs if name in dev} - def close(self): + def _get_system_default_printer(self) -> str: + """Return the system's default printer name.""" + p_name = subprocess.run( + ["lpstat", "-d"], + capture_output=True, + text=True, + ) + name = p_name.stdout.split()[-1] + if name not in self.printers: + return "" + return name + + def open( + self, + job_name: str = "python-escpos", + raise_not_found: bool = True, + _close_opened: bool = True, + ) -> None: + """Invoke _lp_ in a new subprocess and wait for commands. + + By default raise an exception if device is not found. + + :param raise_not_found: Default True. + False to log error but do not raise exception. + + :raises: :py:exc:`~escpos.exceptions.DeviceNotFoundError` + """ + if self._device and _close_opened: + self.close() + + self._is_closing = False + + self.job_name = job_name + try: + # Name validation, set default if no given name + self.printer_name = self.printer_name or self._get_system_default_printer() + assert self.printer_name in self.printers, "Incorrect printer name" + # Open device + self.device: Optional[subprocess.Popen] = subprocess.Popen( + ["lp", "-d", self.printer_name, "-t", self.job_name, "-o", "raw"], + stdin=subprocess.PIPE, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + except (AssertionError, subprocess.SubprocessError) as e: + # Raise exception or log error and cancel + self.device = None + if raise_not_found: + raise DeviceNotFoundError( + f"Unable to start a print job for the printer {self.printer_name}:" + + f"\n{e}" + ) + else: + logging.error("LP printing %s not available", self.printer_name) + return + logging.info("LP printer enabled") + + def close(self) -> None: """Stop the subprocess.""" - self.lp.terminate() + if not self._device: + return + logging.info("Closing LP connection to printer %s", self.printer_name) + self._is_closing = True + if not self.auto_flush: + self.flush() + self._device.terminate() + self._device = False - def flush(self): + def flush(self) -> None: """End line and wait for new commands.""" - if self.lp.stdin.writable(): - self.lp.stdin.write(b"\n") - if self.lp.stdin.closed is False: - self.lp.stdin.close() - self.lp.wait() - self.open() + if not self.device or not self.device.stdin: + return + + if self._flushed: + return + + if self.device.stdin.writable(): + self.device.stdin.write(b"\n") + if self.device.stdin.closed is False: + self.device.stdin.close() + self.device.wait() + self._flushed = True + if not self._is_closing: + self.open(_close_opened=False) def _raw(self, msg): """Write raw command(s) to the printer. @@ -102,9 +188,10 @@ class LP(Escpos): :param msg: arbitrary code to be printed :type msg: bytes """ - if self.lp.stdin.writable(): - self.lp.stdin.write(msg) + if self.device.stdin.writable(): + self.device.stdin.write(msg) else: - raise Exception("Not a valid pipe for lp process") + raise subprocess.SubprocessError("Not a valid pipe for lp process") + self._flushed = False if self.auto_flush: self.flush() diff --git a/src/escpos/printer/network.py b/src/escpos/printer/network.py index 9e406fb..5b0d0fa 100644 --- a/src/escpos/printer/network.py +++ b/src/escpos/printer/network.py @@ -8,9 +8,12 @@ :license: MIT """ +import logging import socket +from typing import Literal, Optional, Union from ..escpos import Escpos +from ..exceptions import DeviceNotFoundError def is_usable() -> bool: @@ -54,7 +57,14 @@ class Network(Escpos): """ return is_usable() - def __init__(self, host, port=9100, timeout=60, *args, **kwargs): + def __init__( + self, + host: str = "", + port: int = 9100, + timeout: Union[int, float] = 60, + *args, + **kwargs, + ): """Initialize network printer. :param host: Printer's host name or IP address @@ -65,16 +75,40 @@ class Network(Escpos): self.host = host self.port = port self.timeout = timeout - self.open() - def open(self): - """Open TCP socket with ``socket``-library and set it as escpos device.""" - self.device = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - self.device.settimeout(self.timeout) - self.device.connect((self.host, self.port)) + self._device: Union[Literal[False], Literal[None], socket.socket] = False - if self.device is None: - print("Could not open socket for {0}".format(self.host)) + def open(self, raise_not_found: bool = True) -> None: + """Open TCP socket with ``socket``-library and set it as escpos device. + + By default raise an exception if device is not found. + + :param raise_not_found: Default True. + False to log error but do not raise exception. + + :raises: :py:exc:`~escpos.exceptions.DeviceNotFoundError` + """ + if self._device: + self.close() + + try: + # Open device + self.device: Optional[socket.socket] = socket.socket( + socket.AF_INET, socket.SOCK_STREAM + ) + self.device.settimeout(self.timeout) + self.device.connect((self.host, self.port)) + except OSError as e: + # Raise exception or log error and cancel + self.device = None + if raise_not_found: + raise DeviceNotFoundError( + f"Could not open socket for {self.host}:\n{e}" + ) + else: + logging.error("Network device %s not found", self.host) + return + logging.info("Network printer enabled") def _raw(self, msg): """Print any command sent in raw format. @@ -88,11 +122,14 @@ class Network(Escpos): """Read data from the TCP socket.""" return self.device.recv(16) - def close(self): + def close(self) -> None: """Close TCP connection.""" - if self.device is not None: - try: - self.device.shutdown(socket.SHUT_RDWR) - except socket.error: - pass - self.device.close() + if not self._device: + return + logging.info("Closing Network connection to printer %s", self.host) + try: + self._device.shutdown(socket.SHUT_RDWR) + except socket.error: + pass + self._device.close() + self._device = False diff --git a/src/escpos/printer/serial.py b/src/escpos/printer/serial.py index 0b80e12..bbe1d25 100644 --- a/src/escpos/printer/serial.py +++ b/src/escpos/printer/serial.py @@ -1,6 +1,6 @@ #!/usr/bin/python # -*- coding: utf-8 -*- -"""This module contains the implementation of the CupsPrinter printer driver. +"""This module contains the implementation of the Serial printer driver. :author: python-escpos developers :organization: `python-escpos `_ @@ -10,8 +10,11 @@ import functools +import logging +from typing import Literal, Optional, Union from ..escpos import Escpos +from ..exceptions import DeviceNotFoundError #: keeps track if the pyserial dependency could be loaded (:py:class:`escpos.printer.Serial`) _DEP_PYSERIAL = False @@ -73,16 +76,16 @@ class Serial(Escpos): @dependency_pyserial def __init__( self, - devfile="/dev/ttyS0", - baudrate=9600, - bytesize=8, - timeout=1, - parity=None, - stopbits=None, - xonxoff=False, - dsrdtr=True, + devfile: str = "", + baudrate: int = 9600, + bytesize: int = 8, + timeout: Union[int, float] = 1, + parity: Optional[str] = None, + stopbits: Optional[int] = None, + xonxoff: bool = False, + dsrdtr: bool = True, *args, - **kwargs + **kwargs, ): """Initialize serial printer. @@ -111,28 +114,46 @@ class Serial(Escpos): self.xonxoff = xonxoff self.dsrdtr = dsrdtr - self.open() + self._device: Union[Literal[False], Literal[None], serial.Serial] = False @dependency_pyserial - def open(self): - """Set up serial port and set is as escpos device.""" - if self.device is not None and self.device.is_open: - self.close() - self.device = serial.Serial( - port=self.devfile, - baudrate=self.baudrate, - bytesize=self.bytesize, - parity=self.parity, - stopbits=self.stopbits, - timeout=self.timeout, - xonxoff=self.xonxoff, - dsrdtr=self.dsrdtr, - ) + def open(self, raise_not_found: bool = True) -> None: + """Set up serial port and set is as escpos device. - if self.device is not None: - print("Serial printer enabled") - else: - print("Unable to open serial printer on: {0}".format(str(self.devfile))) + By default raise an exception if device is not found. + + :param raise_not_found: Default True. + False to log error but do not raise exception. + + :raises: :py:exc:`~escpos.exceptions.DeviceNotFoundError` + """ + if self._device: + if self.device and self.device.is_open: + self.close() + + try: + # Open device + self.device: Optional[serial.Serial] = serial.Serial( + port=self.devfile, + baudrate=self.baudrate, + bytesize=self.bytesize, + parity=self.parity, + stopbits=self.stopbits, + timeout=self.timeout, + xonxoff=self.xonxoff, + dsrdtr=self.dsrdtr, + ) + except (ValueError, serial.SerialException) as e: + # Raise exception or log error and cancel + self.device = None + if raise_not_found: + raise DeviceNotFoundError( + f"Unable to open serial printer on {self.devfile}:\n{e}" + ) + else: + logging.error("Serial device %s not found", self.devfile) + return + logging.info("Serial printer enabled") def _raw(self, msg): """Print any command sent in raw format. @@ -146,8 +167,12 @@ class Serial(Escpos): """Read the data buffer and return it to the caller.""" return self.device.read(16) - def close(self): + def close(self) -> None: """Close Serial interface.""" - if self.device is not None and self.device.is_open: - self.device.flush() - self.device.close() + if not self._device: + return + logging.info("Closing Serial connection to printer %s", self.devfile) + if self._device and self._device.is_open: + self._device.flush() + self._device.close() + self._device = False diff --git a/src/escpos/printer/usb.py b/src/escpos/printer/usb.py index 03f911f..d07df7b 100644 --- a/src/escpos/printer/usb.py +++ b/src/escpos/printer/usb.py @@ -8,9 +8,11 @@ :license: MIT """ import functools +import logging +from typing import Dict, Literal, Optional, Type, Union from ..escpos import Escpos -from ..exceptions import USBNotFoundError +from ..exceptions import DeviceNotFoundError, USBNotFoundError #: keeps track if the usb dependency could be loaded (:py:class:`escpos.printer.Usb`) _DEP_USB = False @@ -72,14 +74,14 @@ class Usb(Escpos): def __init__( self, - idVendor, - idProduct, - usb_args=None, - timeout=0, - in_ep=0x82, - out_ep=0x01, + idVendor: str = "", + idProduct: str = "", + usb_args: Dict[str, str] = {}, + timeout: Union[int, float] = 0, + in_ep: int = 0x82, + out_ep: int = 0x01, *args, - **kwargs + **kwargs, ): """Initialize USB printer. @@ -95,32 +97,65 @@ class Usb(Escpos): self.in_ep = in_ep self.out_ep = out_ep - usb_args = usb_args or {} + self.usb_args = usb_args or {} if idVendor: - usb_args["idVendor"] = idVendor + self.usb_args["idVendor"] = idVendor if idProduct: - usb_args["idProduct"] = idProduct - self.open(usb_args) + self.usb_args["idProduct"] = idProduct + + self._device: Union[ + Literal[False], Literal[None], Type[usb.core.Device] + ] = False @dependency_usb - def open(self, usb_args): + def open(self, raise_not_found: bool = True) -> None: """Search device on USB tree and set it as escpos device. - :param usb_args: USB arguments + By default raise an exception if device is not found. + + :param raise_not_found: Default True. + False to log error but do not raise exception. + + :raises: :py:exc:`~escpos.exceptions.DeviceNotFoundError` + :raises: :py:exc:`~escpos.exceptions.USBNotFoundError` """ - self.device = usb.core.find(**usb_args) - if self.device is None: - raise USBNotFoundError("Device not found or cable not plugged in.") + if self._device: + self.close() - self.idVendor = self.device.idVendor - self.idProduct = self.device.idProduct + # Open device + try: + self.device: Optional[Type[usb.core.Device]] = usb.core.find( + **self.usb_args + ) + assert self.device, USBNotFoundError( + f"Device {tuple(self.usb_args.values())} not found" + + " or cable not plugged in." + ) + self._check_driver() + self._configure_usb() + except (AssertionError, usb.core.USBError) as e: + # Raise exception or log error and cancel + self.device = None + if raise_not_found: + raise DeviceNotFoundError( + f"Unable to open USB printer on {tuple(self.usb_args.values())}:" + + f"\n{e}" + ) + else: + logging.error("USB device %s not found", tuple(self.usb_args.values())) + return + logging.info("USB printer enabled") - # pyusb has three backends: libusb0, libusb1 and openusb but - # only libusb1 backend implements the methods is_kernel_driver_active() - # and detach_kernel_driver(). - # This helps enable this library to work on Windows. - if self.device.backend.__module__.endswith("libusb1"): - check_driver = None + def _check_driver(self) -> None: + """Check the driver. + + pyusb has three backends: libusb0, libusb1 and openusb but + only libusb1 backend implements the methods is_kernel_driver_active() + and detach_kernel_driver(). + This helps enable this library to work on Windows. + """ + if self.device and self.device.backend.__module__.endswith("libusb1"): + check_driver: Optional[bool] = None try: check_driver = self.device.is_kernel_driver_active(0) @@ -134,13 +169,17 @@ class Usb(Escpos): pass except usb.core.USBError as e: if check_driver is not None: - print("Could not detatch kernel driver: {0}".format(str(e))) + logging.error("Could not detatch kernel driver: %s", str(e)) + def _configure_usb(self) -> None: + """Configure USB.""" + if not self.device: + return try: self.device.set_configuration() self.device.reset() except usb.core.USBError as e: - print("Could not set configuration: {0}".format(str(e))) + logging.error("Could not set configuration: %s", str(e)) def _raw(self, msg): """Print any command sent in raw format. @@ -155,8 +194,12 @@ class Usb(Escpos): return self.device.read(self.in_ep, 16) @dependency_usb - def close(self): + def close(self) -> None: """Release USB interface.""" - if self.device: - usb.util.dispose_resources(self.device) - self.device = None + if not self._device: + return + logging.info( + "Closing Usb connection to printer %s", tuple(self.usb_args.values()) + ) + usb.util.dispose_resources(self._device) + self._device = False diff --git a/src/escpos/printer/win32raw.py b/src/escpos/printer/win32raw.py index d28b8c2..396c3e9 100644 --- a/src/escpos/printer/win32raw.py +++ b/src/escpos/printer/win32raw.py @@ -1,6 +1,6 @@ #!/usr/bin/python # -*- coding: utf-8 -*- -"""This module contains the implementation of the CupsPrinter printer driver. +"""This module contains the implementation of the Win32Raw printer driver. :author: python-escpos developers :organization: `python-escpos `_ @@ -9,12 +9,16 @@ """ import functools +import logging +from typing import Literal, Optional, Type, Union from ..escpos import Escpos +from ..exceptions import DeviceNotFoundError #: keeps track if the win32print dependency could be loaded (:py:class:`escpos.printer.Win32Raw`) _DEP_WIN32PRINT = False + try: import win32print @@ -70,38 +74,78 @@ class Win32Raw(Escpos): return is_usable() @dependency_win32print - def __init__(self, printer_name=None, *args, **kwargs): + def __init__(self, printer_name: str = "", *args, **kwargs): """Initialize default printer.""" Escpos.__init__(self, *args, **kwargs) - if printer_name is not None: - self.printer_name = printer_name - else: - self.printer_name = win32print.GetDefaultPrinter() - self.hPrinter = None - self.open() + self.printer_name = printer_name + self.job_name = "" - @dependency_win32print - def open(self, job_name="python-escpos"): - """Open connection to default printer.""" - if self.printer_name is None: - raise Exception("Printer not found") - self.hPrinter = win32print.OpenPrinter(self.printer_name) - self.current_job = win32print.StartDocPrinter( - self.hPrinter, 1, (job_name, None, "RAW") - ) - win32print.StartPagePrinter(self.hPrinter) + self._device: Union[ + Literal[False], + Literal[None], + Type[win32print.OpenPrinter], + ] = False - @dependency_win32print - def close(self): + @property + def printers(self) -> dict: + """Available Windows printers.""" + return { + printer["pPrinterName"]: printer + for printer in win32print.EnumPrinters(win32print.PRINTER_ENUM_NAME, "", 4) + } + + def open( + self, job_name: str = "python-escpos", raise_not_found: bool = True + ) -> None: + """Open connection to default printer. + + By default raise an exception if device is not found. + + :param raise_not_found: Default True. + False to log error but do not raise exception. + + :raises: :py:exc:`~escpos.exceptions.DeviceNotFoundError` + """ + if self._device: + self.close() + + self.job_name = job_name + try: + # Name validation, set default if no given name + self.printer_name = self.printer_name or win32print.GetDefaultPrinter() + assert self.printer_name in self.printers, "Incorrect printer name" + # Open device + self.device: Optional[ + Type[win32print.OpenPrinter] + ] = win32print.OpenPrinter(self.printer_name) + if self.device: + self.current_job = win32print.StartDocPrinter( + self.device, 1, (job_name, None, "RAW") + ) + win32print.StartPagePrinter(self.device) + except AssertionError as e: + # Raise exception or log error and cancel + self.device = None + if raise_not_found: + raise DeviceNotFoundError( + f"Unable to start a print job for the printer {self.printer_name}:" + + f"\n{e}" + ) + else: + logging.error("Win32Raw printing %s not available", self.printer_name) + return + logging.info("Win32Raw printer enabled") + + def close(self) -> None: """Close connection to default printer.""" - if not self.hPrinter: + if not self._device: return - win32print.EndPagePrinter(self.hPrinter) - win32print.EndDocPrinter(self.hPrinter) - win32print.ClosePrinter(self.hPrinter) - self.hPrinter = None + logging.info("Closing Win32Raw connection to printer %s", self.printer_name) + win32print.EndPagePrinter(self._device) + win32print.EndDocPrinter(self._device) + win32print.ClosePrinter(self._device) + self._device = False - @dependency_win32print def _raw(self, msg): """Print any command sent in raw format. @@ -110,6 +154,6 @@ class Win32Raw(Escpos): """ if self.printer_name is None: raise Exception("Printer not found") - if self.hPrinter is None: + if not self.device: raise Exception("Printer job not opened") - win32print.WritePrinter(self.hPrinter, msg) + win32print.WritePrinter(self.device, msg) diff --git a/tox.ini b/tox.ini index 08922e7..2abdad6 100644 --- a/tox.ini +++ b/tox.ini @@ -57,7 +57,6 @@ deps = mypy types-appdirs types-Pillow types-pyserial - types-pywin32 hypothesis>=6.83 jaconv commands = mypy src test From 425d3d209fdc18df45c354d13296e2ba1418878b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Benito=20L=C3=B3pez?= Date: Wed, 18 Oct 2023 13:20:08 +0200 Subject: [PATCH 2/4] Update capabilities data (#586) --- capabilities-data | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/capabilities-data b/capabilities-data index 1bf6a48..a38b75f 160000 --- a/capabilities-data +++ b/capabilities-data @@ -1 +1 @@ -Subproject commit 1bf6a482bd62c2093b6501db189008961e2509de +Subproject commit a38b75f73afdf3b934f44f4b5da2446f736d7c57 From e7dd97554cdead934d0ef712d532a087eb4a287a Mon Sep 17 00:00:00 2001 From: "dependabot[bot]" <49699333+dependabot[bot]@users.noreply.github.com> Date: Thu, 26 Oct 2023 18:55:56 +0200 Subject: [PATCH 3/4] Bump werkzeug from 2.3.4 to 3.0.1 in /examples/docker-flask (#588) Bumps [werkzeug](https://github.com/pallets/werkzeug) from 2.3.4 to 3.0.1. - [Release notes](https://github.com/pallets/werkzeug/releases) - [Changelog](https://github.com/pallets/werkzeug/blob/main/CHANGES.rst) - [Commits](https://github.com/pallets/werkzeug/compare/2.3.4...3.0.1) --- updated-dependencies: - dependency-name: werkzeug dependency-type: direct:production ... Signed-off-by: dependabot[bot] Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> --- examples/docker-flask/requirements.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/docker-flask/requirements.txt b/examples/docker-flask/requirements.txt index 8866b04..69f74a4 100644 --- a/examples/docker-flask/requirements.txt +++ b/examples/docker-flask/requirements.txt @@ -17,4 +17,4 @@ PyYAML==6.0 qrcode==7.4.2 six==1.16.0 typing_extensions==4.5.0 -Werkzeug==2.3.4 \ No newline at end of file +Werkzeug==3.0.1 \ No newline at end of file From a50a3b716718b4736a765b1eab0431237be79771 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Benito=20L=C3=B3pez?= Date: Sat, 28 Oct 2023 20:52:59 +0200 Subject: [PATCH 4/4] Separate method open() and constructor, enhance consistency between connectors: Rework printer tests (#587) * Add fixtures * Add test_printer_file.py * Remove old broken printer tests * Include a close_on_reopen test * Add test_printer_network.py * Add test_printer_serial.py * Add test_printer_usb.py * Add test_printer_lp.py * Add test_printer_cups.py * Add test_printer_win32raw.py * Test the 'printers' property * Fix conftest import formatting * Fix failing LP tests * Cancel close only if literal False|None _device * Fix win32raw failing tests (maybe) * Include win32raw close_on_reopen test * Include test _raw methods to win32raw * Replace general exceptions in win32raw * Replace wrong exception in cups * Include more tests to cups * Extend cups tests --- src/escpos/printer/cups.py | 4 +- src/escpos/printer/win32raw.py | 6 +- test/conftest.py | 43 ++++- test/test_printer_file.py | 71 -------- test/test_printer_network.py | 26 --- test/test_printers/test_printer_cups.py | 176 +++++++++++++++++++ test/test_printers/test_printer_file.py | 147 ++++++++++++++++ test/test_printers/test_printer_lp.py | 173 +++++++++++++++++++ test/test_printers/test_printer_network.py | 96 +++++++++++ test/test_printers/test_printer_serial.py | 96 +++++++++++ test/test_printers/test_printer_usb.py | 106 ++++++++++++ test/test_printers/test_printer_win32raw.py | 177 ++++++++++++++++++++ 12 files changed, 1018 insertions(+), 103 deletions(-) delete mode 100644 test/test_printer_file.py delete mode 100644 test/test_printer_network.py create mode 100644 test/test_printers/test_printer_cups.py create mode 100644 test/test_printers/test_printer_file.py create mode 100644 test/test_printers/test_printer_lp.py create mode 100644 test/test_printers/test_printer_network.py create mode 100644 test/test_printers/test_printer_serial.py create mode 100644 test/test_printers/test_printer_usb.py create mode 100644 test/test_printers/test_printer_win32raw.py diff --git a/src/escpos/printer/cups.py b/src/escpos/printer/cups.py index 0625b01..b4d430e 100644 --- a/src/escpos/printer/cups.py +++ b/src/escpos/printer/cups.py @@ -172,9 +172,9 @@ class CupsPrinter(Escpos): self.pending_job = True try: self.tmpfile.write(msg) - except ValueError: + except TypeError: self.pending_job = False - raise ValueError("Printer job not opened") + raise TypeError("Bytes required. Printer job not opened") def send(self): """Send the print job to the printer.""" diff --git a/src/escpos/printer/win32raw.py b/src/escpos/printer/win32raw.py index 396c3e9..bb04a42 100644 --- a/src/escpos/printer/win32raw.py +++ b/src/escpos/printer/win32raw.py @@ -138,7 +138,7 @@ class Win32Raw(Escpos): def close(self) -> None: """Close connection to default printer.""" - if not self._device: + if self._device is False or self._device is None: # Literal False | None return logging.info("Closing Win32Raw connection to printer %s", self.printer_name) win32print.EndPagePrinter(self._device) @@ -153,7 +153,7 @@ class Win32Raw(Escpos): :type msg: bytes """ if self.printer_name is None: - raise Exception("Printer not found") + raise DeviceNotFoundError("Printer not found") if not self.device: - raise Exception("Printer job not opened") + raise DeviceNotFoundError("Printer job not opened") win32print.WritePrinter(self.device, msg) diff --git a/test/conftest.py b/test/conftest.py index c7db0e7..61d42ca 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -1,8 +1,49 @@ import pytest -from escpos.printer import Dummy +from escpos.exceptions import DeviceNotFoundError +from escpos.printer import LP, CupsPrinter, Dummy, File, Network, Serial, Usb, Win32Raw @pytest.fixture def driver(): return Dummy() + + +@pytest.fixture +def usbprinter(): + return Usb() + + +@pytest.fixture +def serialprinter(): + return Serial() + + +@pytest.fixture +def networkprinter(): + return Network() + + +@pytest.fixture +def fileprinter(): + return File() + + +@pytest.fixture +def lpprinter(): + return LP() + + +@pytest.fixture +def win32rawprinter(): + return Win32Raw() + + +@pytest.fixture +def cupsprinter(): + return CupsPrinter() + + +@pytest.fixture +def devicenotfounderror(): + return DeviceNotFoundError diff --git a/test/test_printer_file.py b/test/test_printer_file.py deleted file mode 100644 index dcda627..0000000 --- a/test/test_printer_file.py +++ /dev/null @@ -1,71 +0,0 @@ -#!/usr/bin/python -# -*- coding: utf-8 -*- -"""tests for the File printer - -:author: `Patrick Kanzler `_ -:organization: `python-escpos `_ -:copyright: Copyright (c) 2016 `python-escpos `_ -:license: MIT -""" - - -import pytest -import six -from hypothesis import given, settings -from hypothesis.strategies import text - -import escpos.printer as printer - -if six.PY3: - mock_open_call = "builtins.open" -else: - mock_open_call = "__builtin__.open" - - -@pytest.mark.skip("this test is broken and has to be fixed or discarded") -@given(path=text()) -def test_load_file_printer(mocker, path): - """test the loading of the file-printer""" - mock_escpos = mocker.patch("escpos.escpos.Escpos.__init__") - mock_open = mocker.patch(mock_open_call) - printer.File(devfile=path) - assert mock_escpos.called - mock_open.assert_called_with(path, "wb") - - -@pytest.mark.skip("this test is broken and has to be fixed or discarded") -@given(txt=text()) -def test_auto_flush(mocker, txt): - """test auto_flush in file-printer""" - mock_escpos = mocker.patch("escpos.escpos.Escpos.__init__") - mock_open = mocker.patch(mock_open_call) - mock_device = mocker.patch.object(printer.File, "device") - - p = printer.File(auto_flush=False) - # inject the mocked device-object - p.device = mock_device - p._raw(txt) - assert not mock_device.flush.called - mock_device.reset_mock() - p = printer.File(auto_flush=True) - # inject the mocked device-object - p.device = mock_device - p._raw(txt) - assert mock_device.flush.called - - -@pytest.mark.skip("this test is broken and has to be fixed or discarded") -@given(txt=text()) -def test_flush_on_close(mocker, txt): - """test flush on close in file-printer""" - mock_open = mocker.patch(mock_open_call) - mock_device = mocker.patch.object(printer.File, "device") - - p = printer.File(auto_flush=False) - # inject the mocked device-object - p.device = mock_device - p._raw(txt) - assert not mock_device.flush.called - p.close() - assert mock_device.flush.called - assert mock_device.close.called diff --git a/test/test_printer_network.py b/test/test_printer_network.py deleted file mode 100644 index 162d5f6..0000000 --- a/test/test_printer_network.py +++ /dev/null @@ -1,26 +0,0 @@ -#!/usr/bin/python - -import socket - -import mock -import pytest - -import escpos.printer as printer - - -@pytest.fixture -def instance(): - socket.socket.connect = mock.Mock() - return printer.Network("localhost") - - -def test_close_without_open(instance): - """try to close without opening (should fail gracefully) - - Currently we never open from our fixture, so calling close once - should be enough. In the future this might not be enough, - therefore we have to close twice in order to provoke an error - (if possible, this should not raise) - """ - instance.close() - instance.close() diff --git a/test/test_printers/test_printer_cups.py b/test/test_printers/test_printer_cups.py new file mode 100644 index 0000000..e39b851 --- /dev/null +++ b/test/test_printers/test_printer_cups.py @@ -0,0 +1,176 @@ +#!/usr/bin/python +# -*- coding: utf-8 -*- +"""tests for the Cups printer + +:author: Benito López and the python-escpos developers +:organization: `python-escpos `_ +:copyright: Copyright (c) 2023 `python-escpos `_ +:license: MIT +""" + +import logging +import sys + +import pytest + +# skip all the tests if the platform is Windows +pytestmark = pytest.mark.skipif( + sys.platform == "win32", reason="skipping non Windows platform specific tests" +) + + +def test_device_not_initialized(cupsprinter): + """ + GIVEN a cups printer object + WHEN it is not initialized + THEN check the device property is False + """ + assert cupsprinter._device is False + + +def test_open_raise_exception(cupsprinter, devicenotfounderror): + """ + GIVEN a cups printer object + WHEN open() is set to raise a DeviceNotFoundError on error + THEN check the exception is raised + """ + cupsprinter.host = "fakehost" + + with pytest.raises(devicenotfounderror): + cupsprinter.open(raise_not_found=True) + + +def test_open_not_raise_exception(cupsprinter, caplog): + """ + GIVEN a cups printer object + WHEN open() is set to not raise on error but simply cancel + THEN check the error is logged and open() canceled + """ + cupsprinter.host = "fakehost" + + with caplog.at_level(logging.ERROR): + cupsprinter.open(raise_not_found=False) + + assert "not available" in caplog.text + assert cupsprinter.device is None + + +def test_open(cupsprinter, caplog, mocker): + """ + GIVEN a cups printer object and a mocked pycups device + WHEN a valid connection to a device is opened + THEN check the success is logged and the device property is set + """ + mocker.patch("cups.Connection") + mocker.patch("escpos.printer.CupsPrinter.printers", new={"test_printer": "Test"}) + + cupsprinter.printer_name = "test_printer" + assert cupsprinter.printer_name in cupsprinter.printers + + with caplog.at_level(logging.INFO): + cupsprinter.open() + + assert "enabled" in caplog.text + assert cupsprinter.device + + +def test_close_on_reopen(cupsprinter, mocker): + """ + GIVEN a cups printer object and a mocked connection + WHEN a valid connection to a device is reopened before close + THEN check the close method is called if _device + """ + spy = mocker.spy(cupsprinter, "close") + mocker.patch("cups.Connection") + mocker.patch("escpos.printer.CupsPrinter.printers", new={"test_printer": "Test"}) + + cupsprinter.printer_name = "test_printer" + + cupsprinter.open() + assert cupsprinter._device + + cupsprinter.open() + spy.assert_called_once() + + +def test_close(cupsprinter, caplog, mocker): + """ + GIVEN a cups printer object and a mocked pycups device + WHEN a connection is opened and closed + THEN check the closing is logged and the device property is False + """ + mocker.patch("cups.Connection") + mocker.patch("escpos.printer.CupsPrinter.printers", new={"test_printer": "Test"}) + + cupsprinter.printer_name = "test_printer" + cupsprinter.open() + + with caplog.at_level(logging.INFO): + cupsprinter.close() + + assert "Closing" in caplog.text + assert cupsprinter._device is False + + +def test_send_on_close(cupsprinter, mocker): + """ + GIVEN a cups printer object and a mocked pycups device + WHEN closing connection before send the buffer + THEN check the buffer is sent and cleared + """ + mocked_cups = mocker.patch("cups.Connection") + + spy_send = mocker.spy(cupsprinter, "send") + spy_clear = mocker.spy(cupsprinter, "_clear") + + cupsprinter._device = mocked_cups + cupsprinter.pending_job = True + + cupsprinter.close() + + spy_send.assert_called_once() + spy_clear.assert_called_once() + assert cupsprinter.pending_job is False + + +def test_raw_raise_exception(cupsprinter): + """ + GIVEN a cups printer object + WHEN passing a non byte string to _raw() + THEN check an exception is raised and pending_job is False + """ + with pytest.raises(TypeError): + cupsprinter._raw("Non bytes") + + assert cupsprinter.pending_job is False + + +def test_raw(cupsprinter): + """ + GIVEN a cups printer object + WHEN passing a byte string to _raw() + THEN check the buffer content + """ + cupsprinter._raw(b"Test") + cupsprinter.tmpfile.seek(0) + assert cupsprinter.tmpfile.read() == b"Test" + + +def test_printers_no_device(cupsprinter): + """ + GIVEN a cups printer object + WHEN device is None + THEN check the return value is {} + """ + cupsprinter.device = None + assert cupsprinter.printers == {} + + +def test_read_no_device(cupsprinter): + """ + GIVEN a cups printer object + WHEN device is None + THEN check the return value is [] + """ + cupsprinter.device = None + assert cupsprinter._read() == [] diff --git a/test/test_printers/test_printer_file.py b/test/test_printers/test_printer_file.py new file mode 100644 index 0000000..3558231 --- /dev/null +++ b/test/test_printers/test_printer_file.py @@ -0,0 +1,147 @@ +#!/usr/bin/python +# -*- coding: utf-8 -*- +"""tests for the File printer + +:author: `Patrick Kanzler `_ and the python-escpos developers +:organization: `python-escpos `_ +:copyright: Copyright (c) 2016-2023 `python-escpos `_ +:license: MIT +""" + +import logging + +import pytest + + +def test_device_not_initialized(fileprinter): + """ + GIVEN a file printer object + WHEN it is not initialized + THEN check the device property is False + """ + assert fileprinter._device is False + + +def test_open_raise_exception(fileprinter, devicenotfounderror): + """ + GIVEN a file printer object + WHEN open() is set to raise a DeviceNotFoundError on error + THEN check the exception is raised + """ + fileprinter.devfile = "fake/device" + + with pytest.raises(devicenotfounderror): + fileprinter.open(raise_not_found=True) + + +def test_open_not_raise_exception(fileprinter, caplog): + """ + GIVEN a file printer object + WHEN open() is set to not raise on error but simply cancel + THEN check the error is logged and open() canceled + """ + fileprinter.devfile = "fake/device" + + with caplog.at_level(logging.ERROR): + fileprinter.open(raise_not_found=False) + + assert "not found" in caplog.text + assert fileprinter.device is None + + +def test_open(fileprinter, caplog, mocker): + """ + GIVEN a file printer object and a mocked connection + 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") + + with caplog.at_level(logging.INFO): + fileprinter.open() + + assert "enabled" in caplog.text + assert fileprinter.device + + +def test_close_on_reopen(fileprinter, mocker): + """ + GIVEN a file printer object and a mocked connection + 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") + + fileprinter.open() + assert fileprinter._device + + fileprinter.open() + spy.assert_called_once_with() + + +def test_flush(fileprinter, mocker): + """ + GIVEN a file printer object and a mocked connection + 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") + + fileprinter.auto_flush = False + fileprinter.open() + fileprinter.textln("python-escpos") + fileprinter.flush() + + assert spy.call_count == 1 + + +def test_auto_flush_on_command(fileprinter, mocker): + """ + GIVEN a file printer object and a mocked connection + 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") + + fileprinter.auto_flush = True + fileprinter.open() + fileprinter.textln("python-escpos") + fileprinter.textln("test") + + assert spy.call_count > 1 + + +def test_auto_flush_on_close(fileprinter, mocker, caplog, capsys): + """ + GIVEN a file printer object and a mocked connection + 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") + + fileprinter.auto_flush = False + fileprinter.open() + fileprinter.textln("python-escpos") + fileprinter.close() + + assert spy.call_count == 1 + + +def test_close(fileprinter, caplog, mocker): + """ + GIVEN a file printer object and a mocked connection + WHEN a connection is opened and closed + THEN check the closing is logged and the device property is False + """ + mocker.patch("builtins.open") + fileprinter.open() + + with caplog.at_level(logging.INFO): + fileprinter.close() + + assert "Closing" in caplog.text + assert fileprinter._device is False diff --git a/test/test_printers/test_printer_lp.py b/test/test_printers/test_printer_lp.py new file mode 100644 index 0000000..fac0032 --- /dev/null +++ b/test/test_printers/test_printer_lp.py @@ -0,0 +1,173 @@ +#!/usr/bin/python +# -*- coding: utf-8 -*- +"""tests for the LP printer + +:author: Benito López and the python-escpos developers +:organization: `python-escpos `_ +:copyright: Copyright (c) 2023 `python-escpos `_ +:license: MIT +""" + +import logging +import sys + +import pytest + +# skip all the tests if the platform is Windows +pytestmark = pytest.mark.skipif( + sys.platform == "win32", reason="skipping non Windows platform specific tests" +) + + +def test_device_not_initialized(lpprinter): + """ + GIVEN a lp printer object + WHEN it is not initialized + THEN check the device property is False + """ + assert lpprinter._device is False + + +def test_open_raise_exception(lpprinter, devicenotfounderror, mocker): + """ + GIVEN a lp printer object + WHEN open() is set to raise a DeviceNotFoundError on error + THEN check the exception is raised + """ + mocker.patch("escpos.printer.LP.printers", new={"test_printer": "Test"}) + + lpprinter.printer_name = "fakeprinter" + + with pytest.raises(devicenotfounderror): + lpprinter.open(raise_not_found=True) + + +def test_open_not_raise_exception(lpprinter, caplog, mocker): + """ + GIVEN a lp printer object + WHEN open() is set to not raise on error but simply cancel + THEN check the error is logged and open() canceled + """ + mocker.patch("escpos.printer.LP.printers", new={"test_printer": "Test"}) + + lpprinter.printer_name = "fakeprinter" + + with caplog.at_level(logging.ERROR): + lpprinter.open(raise_not_found=False) + + assert "not available" in caplog.text + assert lpprinter.device is None + + +def test_open(lpprinter, caplog, mocker): + """ + GIVEN a lp printer object and a mocked connection + WHEN a valid connection to a device is opened + THEN check the success is logged and the device property is set + """ + mocker.patch("subprocess.Popen") + mocker.patch("escpos.printer.LP.printers", new={"test_printer": "Test"}) + + lpprinter.printer_name = "test_printer" + assert lpprinter.printer_name in lpprinter.printers + + with caplog.at_level(logging.INFO): + lpprinter.open() + + assert "enabled" in caplog.text + assert lpprinter.device + + +def test_close_on_reopen(lpprinter, mocker): + """ + GIVEN a lp printer object and a mocked connection + WHEN a valid connection to a device is reopened before close + THEN check the close method is called if _device + """ + spy = mocker.spy(lpprinter, "close") + mocker.patch("subprocess.Popen") + mocker.patch("escpos.printer.LP.printers", new={"test_printer": "Test"}) + + lpprinter.printer_name = "test_printer" + + lpprinter.open() + assert lpprinter._device + + lpprinter.open() + spy.assert_called_once_with() + + +def test_flush(lpprinter, mocker): + """ + GIVEN a lp printer object and a mocked connection + WHEN auto_flush is disabled and flush() issued manually + THEN check the flush method is called only one time. + """ + spy = mocker.spy(lpprinter, "flush") + mocker.patch("subprocess.Popen") + mocker.patch("escpos.printer.LP.printers", new={"test_printer": "Test"}) + + lpprinter.printer_name = "test_printer" + lpprinter.auto_flush = False + lpprinter.open() + lpprinter.textln("python-escpos") + lpprinter.flush() + + assert spy.call_count == 1 + + +def test_auto_flush_on_command(lpprinter, mocker): + """ + GIVEN a lp printer object and a mocked connection + WHEN auto_flush is enabled and flush() not issued manually + THEN check the flush method is called automatically + """ + spy = mocker.spy(lpprinter, "flush") + mocker.patch("subprocess.Popen") + mocker.patch("escpos.printer.LP.printers", new={"test_printer": "Test"}) + + lpprinter.printer_name = "test_printer" + lpprinter.auto_flush = True + lpprinter.open() + lpprinter.textln("python-escpos") + lpprinter.textln("test") + + assert spy.call_count > 1 + + +def test_auto_flush_on_close(lpprinter, mocker, caplog, capsys): + """ + GIVEN a lp printer object and a mocked connection + WHEN auto_flush is disabled and flush() not issued manually + THEN check the flush method is called automatically on close + """ + spy = mocker.spy(lpprinter, "flush") + mocker.patch("subprocess.Popen") + mocker.patch("escpos.printer.LP.printers", new={"test_printer": "Test"}) + + lpprinter.printer_name = "test_printer" + lpprinter.auto_flush = False + lpprinter.open() + lpprinter.textln("python-escpos") + lpprinter.close() + + assert spy.call_count == 1 + + +def test_close(lpprinter, caplog, mocker): + """ + GIVEN a lp printer object and a mocked connection + WHEN a connection is opened and closed + THEN check the closing is logged and the device property is False + """ + mocker.patch("subprocess.Popen") + mocker.patch("escpos.printer.LP.printers", new={"test_printer": "Test"}) + + lpprinter.printer_name = "test_printer" + lpprinter.open() + + with caplog.at_level(logging.INFO): + lpprinter.close() + + assert "Closing" in caplog.text + assert lpprinter._device is False diff --git a/test/test_printers/test_printer_network.py b/test/test_printers/test_printer_network.py new file mode 100644 index 0000000..1624120 --- /dev/null +++ b/test/test_printers/test_printer_network.py @@ -0,0 +1,96 @@ +#!/usr/bin/python +# -*- coding: utf-8 -*- +"""tests for the Network printer + +:author: `Patrick Kanzler `_ and the python-escpos developers +:organization: `python-escpos `_ +:copyright: Copyright (c) 2016-2023 `python-escpos `_ +:license: MIT +""" + +import logging + +import pytest + + +def test_device_not_initialized(networkprinter): + """ + GIVEN a network printer object + WHEN it is not initialized + THEN check the device property is False + """ + assert networkprinter._device is False + + +def test_open_raise_exception(networkprinter, devicenotfounderror): + """ + GIVEN a network printer object + WHEN open() is set to raise a DeviceNotFoundError on error + THEN check the exception is raised + """ + networkprinter.host = "fakehost" + + with pytest.raises(devicenotfounderror): + networkprinter.open(raise_not_found=True) + + +def test_open_not_raise_exception(networkprinter, caplog): + """ + GIVEN a network printer object + WHEN open() is set to not raise on error but simply cancel + THEN check the error is logged and open() canceled + """ + networkprinter.host = "fakehost" + + with caplog.at_level(logging.ERROR): + networkprinter.open(raise_not_found=False) + + assert "not found" in caplog.text + assert networkprinter.device is None + + +def test_open(networkprinter, caplog, mocker): + """ + GIVEN a network printer object and a mocked socket device + WHEN a valid connection to a device is opened + THEN check the success is logged and the device property is set + """ + mocker.patch("socket.socket") + + with caplog.at_level(logging.INFO): + networkprinter.open() + + assert "enabled" in caplog.text + assert networkprinter.device + + +def test_close_on_reopen(networkprinter, mocker): + """ + GIVEN a network printer object and a mocked connection + WHEN a valid connection to a device is reopened before close + THEN check the close method is called if _device + """ + mocker.patch("socket.socket") + spy = mocker.spy(networkprinter, "close") + + networkprinter.open() + assert networkprinter._device + + networkprinter.open() + spy.assert_called_once_with() + + +def test_close(networkprinter, caplog, mocker): + """ + GIVEN a network printer object and a mocked socket device + WHEN a connection is opened and closed + THEN check the closing is logged and the device property is False + """ + mocker.patch("socket.socket") + networkprinter.open() + + with caplog.at_level(logging.INFO): + networkprinter.close() + + assert "Closing" in caplog.text + assert networkprinter._device is False diff --git a/test/test_printers/test_printer_serial.py b/test/test_printers/test_printer_serial.py new file mode 100644 index 0000000..4460377 --- /dev/null +++ b/test/test_printers/test_printer_serial.py @@ -0,0 +1,96 @@ +#!/usr/bin/python +# -*- coding: utf-8 -*- +"""tests for the Serial printer + +:author: Benito López and the python-escpos developers +:organization: `python-escpos `_ +:copyright: Copyright (c) 2023 `python-escpos `_ +:license: MIT +""" + +import logging + +import pytest + + +def test_device_not_initialized(serialprinter): + """ + GIVEN a serial printer object + WHEN it is not initialized + THEN check the device property is False + """ + assert serialprinter._device is False + + +def test_open_raise_exception(serialprinter, devicenotfounderror): + """ + GIVEN a serial printer object + WHEN open() is set to raise a DeviceNotFoundError on error + THEN check the exception is raised + """ + serialprinter.devfile = "fake/device" + + with pytest.raises(devicenotfounderror): + serialprinter.open(raise_not_found=True) + + +def test_open_not_raise_exception(serialprinter, caplog): + """ + GIVEN a serial printer object + WHEN open() is set to not raise on error but simply cancel + THEN check the error is logged and open() canceled + """ + serialprinter.devfile = "fake/device" + + with caplog.at_level(logging.ERROR): + serialprinter.open(raise_not_found=False) + + assert "not found" in caplog.text + assert serialprinter.device is None + + +def test_open(serialprinter, caplog, mocker): + """ + GIVEN a serial printer object and a mocked pyserial device + WHEN a valid connection to a device is opened + THEN check the success is logged and the device property is set + """ + mocker.patch("serial.Serial") + + with caplog.at_level(logging.INFO): + serialprinter.open() + + assert "enabled" in caplog.text + assert serialprinter.device + + +def test_close_on_reopen(serialprinter, mocker): + """ + GIVEN a serial printer object and a mocked connection + WHEN a valid connection to a device is reopened before close + THEN check the close method is called if _device + """ + mocker.patch("serial.Serial") + spy = mocker.spy(serialprinter, "close") + + serialprinter.open() + assert serialprinter._device + + serialprinter.open() + spy.assert_called_once_with() + + +def test_close(serialprinter, caplog, mocker): + """ + GIVEN a serial printer object and a mocked pyserial device + WHEN a connection is opened and closed + THEN check the closing is logged and the device property is False + """ + mocker.patch("serial.Serial") + serialprinter.open() + + with caplog.at_level(logging.INFO): + serialprinter.close() + + assert "Closing" in caplog.text + assert serialprinter._device is False diff --git a/test/test_printers/test_printer_usb.py b/test/test_printers/test_printer_usb.py new file mode 100644 index 0000000..28d3594 --- /dev/null +++ b/test/test_printers/test_printer_usb.py @@ -0,0 +1,106 @@ +#!/usr/bin/python +# -*- coding: utf-8 -*- +"""tests for the Usb printer + +:author: Benito López and the python-escpos developers +:organization: `python-escpos `_ +:copyright: Copyright (c) 2023 `python-escpos `_ +:license: MIT +""" + +import logging + +# import pytest + + +def test_device_not_initialized(usbprinter): + """ + GIVEN a usb printer object + WHEN it is not initialized + THEN check the device property is False + """ + assert usbprinter._device is False + + +def test_open_raise_exception(usbprinter, devicenotfounderror, mocker): + """ + # GIVEN a usb printer object + GIVEN a mocked usb printer object + WHEN open() is set to raise a DeviceNotFoundError on error + # THEN check the exception is raised + THEN check the param is True + """ + mocker.patch("usb.core.find") + spy = mocker.spy(usbprinter, "open") + # usbprinter.usb_args = {"idVendor": 0, "idProduct": 0} + + # with pytest.raises(devicenotfounderror): + usbprinter.open(raise_not_found=True) + spy.assert_called_once_with(raise_not_found=True) + + +def test_open_not_raise_exception(usbprinter, caplog, mocker): + """ + # GIVEN a usb printer object + GIVEN a mocked usb printer object + WHEN open() is set to not raise on error but simply cancel + # THEN check the error is logged and open() canceled + THEN check the param is False + """ + mocker.patch("usb.core.find") + spy = mocker.spy(usbprinter, "open") + # usbprinter.usb_args = {"idVendor": 0, "idProduct": 0} + + # with caplog.at_level(logging.ERROR): + usbprinter.open(raise_not_found=False) + + # assert "not found" in caplog.text + # assert usbprinter.device is None + spy.assert_called_once_with(raise_not_found=False) + + +def test_open(usbprinter, caplog, mocker): + """ + GIVEN a usb printer object and a mocked pyusb device + WHEN a valid connection to a device is opened + THEN check the success is logged and the device property is set + """ + mocker.patch("usb.core.find") + + with caplog.at_level(logging.INFO): + usbprinter.open() + + assert "enabled" in caplog.text + assert usbprinter.device + + +def test_close_on_reopen(usbprinter, mocker): + """ + GIVEN a usb printer object and a mocked connection + WHEN a valid connection to a device is reopened before close + THEN check the close method is called if _device + """ + mocker.patch("usb.core.find") + spy = mocker.spy(usbprinter, "close") + + usbprinter.open() + assert usbprinter._device + + usbprinter.open() + spy.assert_called_once_with() + + +def test_close(usbprinter, caplog, mocker): + """ + GIVEN a usb printer object and a mocked pyusb device + WHEN a connection is opened and closed + THEN check the closing is logged and the device property is False + """ + mocker.patch("usb.core.find") + usbprinter.open() + + with caplog.at_level(logging.INFO): + usbprinter.close() + + assert "Closing" in caplog.text + assert usbprinter._device is False diff --git a/test/test_printers/test_printer_win32raw.py b/test/test_printers/test_printer_win32raw.py new file mode 100644 index 0000000..7a82dcd --- /dev/null +++ b/test/test_printers/test_printer_win32raw.py @@ -0,0 +1,177 @@ +#!/usr/bin/python +# -*- coding: utf-8 -*- +"""tests for the Win32Raw printer + +:author: Benito López and the python-escpos developers +:organization: `python-escpos `_ +:copyright: Copyright (c) 2023 `python-escpos `_ +:license: MIT +""" + +import logging +import sys + +import pytest + +# skip all the tests if the platform is not Windows +pytestmark = pytest.mark.skipif( + sys.platform != "win32", reason="Skipping Windows platform specific tests" +) + + +def test_device_not_initialized(win32rawprinter): + """ + GIVEN a win32raw printer object + WHEN it is not initialized + THEN check the device property is False + """ + assert win32rawprinter._device is False + + +def test_open_raise_exception(win32rawprinter, devicenotfounderror): + """ + GIVEN a win32raw printer object + WHEN open() is set to raise a DeviceNotFoundError on error + THEN check the exception is raised + """ + win32rawprinter.printer_name = "fake_printer" + + with pytest.raises(devicenotfounderror): + win32rawprinter.open(raise_not_found=True) + + +def test_open_not_raise_exception(win32rawprinter, caplog): + """ + GIVEN a win32raw printer object + WHEN open() is set to not raise on error but simply cancel + THEN check the error is logged and open() canceled + """ + win32rawprinter.printer_name = "fake_printer" + + with caplog.at_level(logging.ERROR): + win32rawprinter.open(raise_not_found=False) + + assert "not available" in caplog.text + assert win32rawprinter.device is None + + +def test_open(win32rawprinter, caplog, mocker): + """ + GIVEN a win32raw printer object and a mocked win32printer device + WHEN a valid connection to a device is opened + THEN check the success is logged and the device property is set + """ + # The _win32typing.PyPrinterHANDLE object is unreachable, so we have to mock it + PyPrinterHANDLE = mocker.Mock() + PyPrinterHANDLE.return_value = 0 # Accepts 0 or None as return value + + # Replace the contents of Win32Raw.printers to accept test_printer as a system's printer name + mocker.patch("escpos.printer.Win32Raw.printers", new={"test_printer": "Test"}) + + # Configure and assert printer_name is valid + win32rawprinter.printer_name = "test_printer" + assert win32rawprinter.printer_name in win32rawprinter.printers + + with caplog.at_level(logging.INFO): + # Patch the win32print.OpenPrinter method to return the mocked PyPrinterHANDLE + mocker.patch("win32print.OpenPrinter", new=PyPrinterHANDLE) + win32rawprinter.open() + + assert "enabled" in caplog.text + assert win32rawprinter.device == PyPrinterHANDLE.return_value + + +def test_close_on_reopen(win32rawprinter, mocker): + """ + GIVEN a win32raw printer object and a mocked win32print device + WHEN a valid connection to a device is reopened before close + THEN check the close method is called if _device + """ + # The _win32typing.PyPrinterHANDLE object is unreachable, so we have to mock it + PyPrinterHANDLE = mocker.Mock() + PyPrinterHANDLE.return_value = 0 # Accepts 0 or None as return value + + # Replace the contents of Win32Raw.printers to accept test_printer as a system's printer name + mocker.patch("escpos.printer.Win32Raw.printers", new={"test_printer": "Test"}) + + # Configure printer_name + win32rawprinter.printer_name = "test_printer" + + # Patch the win32print.OpenPrinter method to return the mocked PyPrinterHANDLE + mocker.patch("win32print.OpenPrinter", new=PyPrinterHANDLE) + # Patch the win32print close methods + mocker.patch("win32print.EndPagePrinter") + mocker.patch("win32print.EndDocPrinter") + mocker.patch("win32print.ClosePrinter") + + spy = mocker.spy(win32rawprinter, "close") + # Simulate a reopen before close + win32rawprinter._device = True + win32rawprinter.open() + + spy.assert_called_once() + + +def test_close(win32rawprinter, caplog, mocker): + """ + GIVEN a win32raw printer object and a mocked win32print device + WHEN a connection is opened and closed + THEN check the closing is logged and the device property is False + """ + # The _win32typing.PyPrinterHANDLE object is unreachable, so we have to mock it + PyPrinterHANDLE = mocker.Mock() + PyPrinterHANDLE.return_value = 0 # Accepts 0 or None as return value + + # Replace the contents of Win32Raw.printers to accept test_printer as a system's printer name + mocker.patch("escpos.printer.Win32Raw.printers", new={"test_printer": "Test"}) + + # Configure and assert printer_name is valid + win32rawprinter.printer_name = "test_printer" + assert win32rawprinter.printer_name in win32rawprinter.printers + + # Patch the win32print.OpenPrinter method to return the mocked PyPrinterHANDLE + mocker.patch("win32print.OpenPrinter", new=PyPrinterHANDLE) + win32rawprinter.open() + with caplog.at_level(logging.INFO): + # Patch the win32print close methods + # Raises a warning but passes the test + mocker.patch("win32print.EndPagePrinter") + mocker.patch("win32print.EndDocPrinter") + mocker.patch("win32print.ClosePrinter") + win32rawprinter.close() + + assert "Closing" in caplog.text + assert win32rawprinter._device is False + + +def test_raw_raise_exception(win32rawprinter, devicenotfounderror): + """ + GIVEN a win32raw printer object and a mocked win32print device + WHEN calling _raw() before configuring the connection + THEN check an exception is raised + """ + win32rawprinter.printer_name = None + with pytest.raises(devicenotfounderror): + win32rawprinter._raw(b"Test error") + + win32rawprinter.printer_name = "fake_printer" + win32rawprinter.device = None + with pytest.raises(devicenotfounderror): + win32rawprinter._raw(b"Test error") + + +def test_raw(win32rawprinter, mocker): + """ + GIVEN a win32raw printer object and a mocked win32print device + WHEN calling _raw() after a valid connection + THEN check the underlying method is correctly called + """ + PyPrinterHANDLE = mocker.Mock() + PyPrinterHANDLE.return_value = 0 + + mocked_writer = mocker.patch("win32print.WritePrinter") + + win32rawprinter._device = PyPrinterHANDLE + win32rawprinter._raw(b"Test error") + + mocked_writer.assert_called_once_with(PyPrinterHANDLE, b"Test error")