setop
/
python-escpos
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'master' into update-changelog

This commit is contained in:
Patrick Kanzler 2023-10-28 20:53:21 +02:00 committed by GitHub
parent a8d789fe64 a50a3b7167
commit d2ef5159c5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
22 changed files with 1556 additions and 283 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
capabilities-data

@ -1 +1 @@
Subproject commit 1bf6a482bd62c2093b6501db189008961e2509de
Subproject commit a38b75f73afdf3b934f44f4b5da2446f736d7c57

2
examples/docker-flask/requirements.txt
View File

@ -17,4 +17,4 @@ PyYAML==6.0
qrcode==7.4.2
six==1.16.0
typing_extensions==4.5.0
Werkzeug==2.3.4
Werkzeug==3.0.1

29
src/escpos/escpos.py
View File

@ -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.

37
src/escpos/exceptions.py
View File

@ -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):

88
src/escpos/printer/cups.py
View File

@ -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,49 +84,85 @@ 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()
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.
@ -130,18 +172,17 @@ 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")
@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

57
src/escpos/printer/file.py
View File

@ -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 <https://github.com/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,17 +52,40 @@ 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."""
if self.device:
self.device.flush()
def _raw(self, msg):
@ -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

139
src/escpos/printer/lp.py
View File

@ -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 <https://github.com/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()

61
src/escpos/printer/network.py
View File

@ -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: Union[Literal[False], Literal[None], socket.socket] = False
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))
if self.device is None:
print("Could not open socket for {0}".format(self.host))
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:
if not self._device:
return
logging.info("Closing Network connection to printer %s", self.host)
try:
self.device.shutdown(socket.SHUT_RDWR)
self._device.shutdown(socket.SHUT_RDWR)
except socket.error:
pass
self.device.close()
self._device.close()
self._device = False

71
src/escpos/printer/serial.py
View File

@ -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 <https://github.com/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,14 +114,26 @@ 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:
def open(self, raise_not_found: bool = True) -> None:
"""Set up serial port and set is 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:
if self.device and self.device.is_open:
self.close()
self.device = serial.Serial(
try:
# Open device
self.device: Optional[serial.Serial] = serial.Serial(
port=self.devfile,
baudrate=self.baudrate,
bytesize=self.bytesize,
@ -128,11 +143,17 @@ class Serial(Escpos):
xonxoff=self.xonxoff,
dsrdtr=self.dsrdtr,
)
if self.device is not None:
print("Serial printer enabled")
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:
print("Unable to open serial printer on: {0}".format(str(self.devfile)))
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

105
src/escpos/printer/usb.py
View File

@ -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

102
src/escpos/printer/win32raw.py
View File

@ -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 <https://github.com/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.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._device: Union[
Literal[False],
Literal[None],
Type[win32print.OpenPrinter],
] = False
@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.hPrinter, 1, (job_name, None, "RAW")
self.device, 1, (job_name, None, "RAW")
)
win32print.StartPagePrinter(self.hPrinter)
@dependency_win32print
def close(self):
"""Close connection to default printer."""
if not self.hPrinter:
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
win32print.EndPagePrinter(self.hPrinter)
win32print.EndDocPrinter(self.hPrinter)
win32print.ClosePrinter(self.hPrinter)
self.hPrinter = None
logging.info("Win32Raw printer enabled")
def close(self) -> None:
"""Close connection to default printer."""
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)
win32print.EndDocPrinter(self._device)
win32print.ClosePrinter(self._device)
self._device = False
@dependency_win32print
def _raw(self, msg):
"""Print any command sent in raw format.
@ -109,7 +153,7 @@ class Win32Raw(Escpos):
:type msg: bytes
"""
if self.printer_name is None:
raise Exception("Printer not found")
if self.hPrinter is None:
raise Exception("Printer job not opened")
win32print.WritePrinter(self.hPrinter, msg)
raise DeviceNotFoundError("Printer not found")
if not self.device:
raise DeviceNotFoundError("Printer job not opened")
win32print.WritePrinter(self.device, msg)

43
test/conftest.py
View File

@ -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

71
test/test_printer_file.py
View File

@ -1,71 +0,0 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""tests for the File printer
:author: `Patrick Kanzler <dev@pkanzler.de>`_
:organization: `python-escpos <https://github.com/python-escpos>`_
:copyright: Copyright (c) 2016 `python-escpos <https://github.com/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

26
test/test_printer_network.py
View File

@ -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()

176
test/test_printers/test_printer_cups.py Normal file
View File

@ -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 <https://github.com/python-escpos>`_
:copyright: Copyright (c) 2023 `python-escpos <https://github.com/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() == []

147
test/test_printers/test_printer_file.py Normal file
View File

@ -0,0 +1,147 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""tests for the File printer
:author: `Patrick Kanzler <dev@pkanzler.de>`_ and the python-escpos developers
:organization: `python-escpos <https://github.com/python-escpos>`_
:copyright: Copyright (c) 2016-2023 `python-escpos <https://github.com/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

173
test/test_printers/test_printer_lp.py Normal file
View File

@ -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 <https://github.com/python-escpos>`_
:copyright: Copyright (c) 2023 `python-escpos <https://github.com/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

96
test/test_printers/test_printer_network.py Normal file
View File

@ -0,0 +1,96 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""tests for the Network printer
:author: `Patrick Kanzler <dev@pkanzler.de>`_ and the python-escpos developers
:organization: `python-escpos <https://github.com/python-escpos>`_
:copyright: Copyright (c) 2016-2023 `python-escpos <https://github.com/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

96
test/test_printers/test_printer_serial.py Normal file
View File

@ -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 <https://github.com/python-escpos>`_
:copyright: Copyright (c) 2023 `python-escpos <https://github.com/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

106
test/test_printers/test_printer_usb.py Normal file
View File

@ -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 <https://github.com/python-escpos>`_
:copyright: Copyright (c) 2023 `python-escpos <https://github.com/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

177
test/test_printers/test_printer_win32raw.py Normal file
View File

@ -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 <https://github.com/python-escpos>`_
:copyright: Copyright (c) 2023 `python-escpos <https://github.com/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")

1
tox.ini
View File

@ -57,7 +57,6 @@ deps = mypy
types-appdirs
types-Pillow
types-pyserial
types-pywin32
hypothesis>=6.83
jaconv
commands = mypy src test