feat: add printer test utils

docs/index.md

@@ -74,6 +74,12 @@ The `code_graph` CLI (console script from the `test_utils` package, not a Frappe
 
 ---
 
+### Printer mocks
+
+[`printers.md`](printers.md) documents the IPP codec, raw/IPP mock servers, sync wrappers for pytest, and the `mock-printer` CLI for manual CUPS testing.
+
+---
+
 ## Test Fixtures
 
 Test Utils ships reference fixture data for bootstrapping Frappe/ERPNext test environments:

docs/printers.md

@@ -0,0 +1,67 @@
+# Printer mocks
+
+`test_utils.printers` provides a stdlib-only IPP codec and mock raw/IPP print servers for integration tests and manual development.
+
+## Install
+
+```bash
+pip install "git+https://github.com/agritheory/test_utils.git@v1.26.0"
+```
+
+## CLI
+
+```bash
+mock-printer raw --save-dir /tmp/prints
+mock-printer ipp --save-dir /tmp/prints --name PDF --port 8631
+mock-printer both --save-dir /tmp/prints
+```
+
+The CLI prints ready-to-copy URIs:
+
+```
+Raw:  socket://127.0.0.1:44895
+IPP:  ipp://127.0.0.1:8631/ipp/print
+```
+
+## Programmatic use
+
+Async (pytest-asyncio, Kenaf):
+
+```python
+from test_utils.printers import raw_printer, ipp_printer
+
+async with raw_printer(save_dir="/tmp/prints") as printer:
+    print(printer.device_uri)
+
+async with ipp_printer(name="PDF", save_dir="/tmp/prints") as printer:
+    print(printer.uri)
+```
+
+Sync (Frappe / BEAM pytest):
+
+```python
+from test_utils.printers import raw_printer_sync, ipp_printer_sync
+
+with raw_printer_sync(save_dir="/tmp/prints") as printer:
+    payload = printer.wait_for_payload()
+```
+
+## CUPS registration
+
+Point CUPS queues at the mock URIs:
+
+```bash
+lpadmin -p BEAM_TEST_RAW -E -v socket://127.0.0.1:9100 -m raw
+lpadmin -p BEAM_TEST_IPP -E -v ipp://127.0.0.1:8631/ipp/print -m everywhere
+cupsenable BEAM_TEST_RAW && cupsaccept BEAM_TEST_RAW
+```
+
+## IPP codec
+
+The codec in `test_utils.printers.ipp.codec` is general-purpose (not test-only). Kenaf re-exports it from `src/ipp_codec.py`.
+
+## Kenaf codec duplication
+
+Default: Kenaf imports the shared codec from `test_utils` via a thin re-export.
+
+If the Kenaf appliance build cannot depend on `test_utils` at runtime, keep a vendored copy of `ipp/codec.py` in Kenaf and sync it in CI. Mocks can remain a dev dependency.

pyproject.toml

@@ -29,6 +29,7 @@ check_code_duplication = "test_utils.pre_commit.check_code_duplication:main"
 static_analysis = "test_utils.pre_commit.static_analysis:main"
 semgrep_fmt = "test_utils.pre_commit.semgrep_fmt:main"
 code_graph = "test_utils.utils.graph.cli:main"
+mock-printer = "test_utils.printers.cli:main"
 
 [build-system]
 requires = ["poetry-core>=2.0.0,<3.0.0"]
@@ -58,6 +59,9 @@ pygithub = "^2.9.1"
 [tool.poetry.group.dev.dependencies]
 coverage = "^7.13.4"
 matplotlib = "^3.10.8"
+pytest = "^8.3.3"
+pytest-asyncio = "^0.24.0"
+httpx = "^0.27.2"
 
 [tool.poetry.group.graph]
 optional = true

pytest.ini

@@ -0,0 +1,3 @@
+[tool.pytest.ini_options]
+asyncio_mode = "auto"
+testpaths = ["tests"]

test_utils/printers/__init__.py

@@ -0,0 +1,47 @@
+from test_utils.printers.ipp import (
+	IppDecoder,
+	IppEncoder,
+	IppJobState,
+	IppOperation,
+	IppPrinterState,
+	IppStatus,
+	IppTag,
+	build_error_response,
+	build_success_response,
+)
+from test_utils.printers.mocks import (
+	IppJob,
+	IppPrinterMock,
+	MockIppJob,
+	MockIppPrinter,
+	MockRawPrinter,
+	RawPrinterMock,
+	ReceivedJob,
+	ipp_printer,
+	ipp_printer_sync,
+	raw_printer,
+	raw_printer_sync,
+)
+
+__all__ = [
+	"IppDecoder",
+	"IppEncoder",
+	"IppJob",
+	"IppJobState",
+	"IppOperation",
+	"IppPrinterMock",
+	"IppPrinterState",
+	"IppStatus",
+	"IppTag",
+	"MockIppJob",
+	"MockIppPrinter",
+	"MockRawPrinter",
+	"RawPrinterMock",
+	"ReceivedJob",
+	"build_error_response",
+	"build_success_response",
+	"ipp_printer",
+	"ipp_printer_sync",
+	"raw_printer",
+	"raw_printer_sync",
+]

test_utils/printers/cli.py

@@ -0,0 +1,121 @@
+import argparse
+import asyncio
+import signal
+from pathlib import Path
+
+from test_utils.printers.mocks.ipp import IppPrinterMock
+from test_utils.printers.mocks.raw import RawPrinterMock
+
+
+async def run_raw(host: str, port: int, save_dir: Path | None) -> RawPrinterMock:
+	mock = RawPrinterMock(host=host, port=port, save_dir=save_dir)
+	await mock.start()
+	print(f"Raw:  {mock.device_uri}")
+	if save_dir:
+		print(f"Saving raw jobs to {save_dir}")
+	return mock
+
+
+async def run_ipp(
+	host: str,
+	port: int,
+	save_dir: Path | None,
+	name: str,
+	printer_path: str,
+) -> IppPrinterMock:
+	mock = IppPrinterMock(
+		host=host,
+		port=port,
+		save_dir=save_dir,
+		name=name,
+		printer_path=printer_path,
+	)
+	await mock.start()
+	print(f"IPP:  {mock.uri}")
+	if save_dir:
+		print(f"Saving IPP jobs to {save_dir}")
+	return mock
+
+
+async def serve(args: argparse.Namespace) -> None:
+	save_dir = Path(args.save_dir).expanduser() if args.save_dir else None
+	mocks: list[RawPrinterMock | IppPrinterMock] = []
+
+	if args.command in ("raw", "both"):
+		mocks.append(
+			await run_raw(args.host, args.port if args.command == "raw" else 0, save_dir)
+		)
+	if args.command in ("ipp", "both"):
+		ipp_port = args.port if args.command == "ipp" else 0
+		printer_path = args.printer_path
+		if args.command == "both" and args.name == "Mock IPP Printer":
+			args.name = "Mock IPP"
+		mocks.append(
+			await run_ipp(
+				args.host,
+				ipp_port,
+				save_dir,
+				args.name,
+				printer_path,
+			)
+		)
+
+	stop_event = asyncio.Event()
+
+	def request_stop(*_signum) -> None:
+		stop_event.set()
+
+	for sig in (signal.SIGINT, signal.SIGTERM):
+		signal.signal(sig, request_stop)
+
+	print("Press Ctrl+C to stop.")
+	await stop_event.wait()
+
+	for mock in mocks:
+		await mock.stop()
+
+
+def build_parser() -> argparse.ArgumentParser:
+	parser = argparse.ArgumentParser(
+		description="Run mock raw and/or IPP printers for manual testing."
+	)
+	subparsers = parser.add_subparsers(dest="command", required=True)
+
+	def add_common(subparser: argparse.ArgumentParser) -> None:
+		subparser.add_argument("--host", default="127.0.0.1")
+		subparser.add_argument(
+			"--port", type=int, default=0, help="Listen port (0 = OS-assigned)."
+		)
+		subparser.add_argument(
+			"--save-dir",
+			help="Directory where received print jobs are written.",
+		)
+
+	raw_parser = subparsers.add_parser("raw", help="Run a raw socket printer mock.")
+	add_common(raw_parser)
+	raw_parser.set_defaults(port=9100)
+
+	ipp_parser = subparsers.add_parser("ipp", help="Run an IPP printer mock.")
+	add_common(ipp_parser)
+	ipp_parser.add_argument("--name", default="Mock IPP Printer")
+	ipp_parser.add_argument("--printer-path", default="/ipp/print")
+	ipp_parser.set_defaults(port=8631)
+
+	both_parser = subparsers.add_parser(
+		"both", help="Run raw and IPP printer mocks together."
+	)
+	add_common(both_parser)
+	both_parser.add_argument("--name", default="Mock IPP")
+	both_parser.add_argument("--printer-path", default="/ipp/print")
+
+	return parser
+
+
+def main() -> None:
+	parser = build_parser()
+	args = parser.parse_args()
+	asyncio.run(serve(args))
+
+
+if __name__ == "__main__":
+	main()

test_utils/printers/ipp/__init__.py

@@ -0,0 +1,23 @@
+from test_utils.printers.ipp.codec import (
+	IppDecoder,
+	IppEncoder,
+	IppJobState,
+	IppOperation,
+	IppPrinterState,
+	IppStatus,
+	IppTag,
+	build_error_response,
+	build_success_response,
+)
+
+__all__ = [
+	"IppDecoder",
+	"IppEncoder",
+	"IppJobState",
+	"IppOperation",
+	"IppPrinterState",
+	"IppStatus",
+	"IppTag",
+	"build_error_response",
+	"build_success_response",
+]