#!/usr/bin/env python3 """Send eight A4 duplex-option test jobs to a CUPS queue and record the display. The queue is assumed to use the server's default Follow-Me.ppd and *not* the papercut_relay_backend.py option-mapping wrapper. ``media=A4`` is sent on every job so that only the duplex setting varies, and the source PDF has two pages so that a two-sided job is meaningful. The eight jobs differ in how duplex is requested: 1 no option baseline / queue default 2 sides=two-sided-long-edge standard IPP, normal duplex 3 sides=two-sided-short-edge standard IPP, other binding 4 Duplex=DuplexNoTumble generic PPD name (long edge) 5 Duplex=DuplexTumble generic PPD name (short edge) 6 Duplex=None generic PPD name (one-sided) 7 KMDuplex=2Sided Konica Minolta vendor PPD name 8 sides=two-sided-long-edge + Duplex=DuplexTumble conflict precedence The baseline is the one-sided control. Test 3 and the short-edge PPD names distinguish "the option was honoured" from "the option was ignored and the queue default applied": if the display shows one-sided, the binding was not applied; if it shows short edge, the value was honoured. The source PDF is copied to test1.pdf ... test8.pdf so the CUPS job names make the tests distinguishable in lpstat and on the release-device display. All eight jobs are submitted first. The script then asks, per test, which binding the display showed, and writes ``results-duplex-a4.txt`` (plain text, safe to upload). Usage: python3 -I duplex_option_tests.py python3 -I duplex_option_tests.py --source 2324-D.antwWIHM.pdf python3 -I duplex_option_tests.py --dry-run Only `lp`, `lpstat` and the source PDF are used. No relay, JWT or other credential is involved. """ from __future__ import annotations import argparse import datetime import os import re import shlex import shutil import subprocess import sys import tempfile import time DEFAULT_QUEUE = "Papercut-Default" DEFAULT_SOURCE = "2324-D.antwWIHM.pdf" DEFAULT_RESULTS = "results-duplex-a4.txt" # Pinned so that only the duplex setting varies between jobs. MEDIA_OPTION = ["-o", "media=A4"] # (title, lp options, expected display if honoured, why this test) TESTS = [ ("baseline - no duplex option", [], "one-sided (queue default)", "control: is the queue default simplex?"), ("sides=two-sided-long-edge", ["-o", "sides=two-sided-long-edge"], "long edge (normal duplex)", "standard IPP attribute"), ("sides=two-sided-short-edge", ["-o", "sides=two-sided-short-edge"], "short edge", "standard IPP attribute, other binding"), ("Duplex=DuplexNoTumble", ["-o", "Duplex=DuplexNoTumble"], "long edge only if the PPD name is honoured, else the default", "generic PPD option name"), ("Duplex=DuplexTumble", ["-o", "Duplex=DuplexTumble"], "short edge only if the PPD name is honoured, else the default", "generic PPD option name"), ("Duplex=None", ["-o", "Duplex=None"], "one-sided", "PPD name, one-sided control"), ("KMDuplex=2Sided", ["-o", "KMDuplex=2Sided"], "long edge only if the vendor PPD name is honoured, else the default", "Konica Minolta PPD option name"), ("sides=two-sided-long-edge + Duplex=DuplexTumble", ["-o", "sides=two-sided-long-edge", "-o", "Duplex=DuplexTumble"], "long edge if the IPP name wins, short edge if the PPD name wins", "conflict precedence"), ] BINDING_LABELS = { "l": "long-edge", "long": "long-edge", "long-edge": "long-edge", "longedge": "long-edge", "2l": "long-edge", "two-sided-long-edge": "long-edge", "s": "short-edge", "short": "short-edge", "short-edge": "short-edge", "shortedge": "short-edge", "2s": "short-edge", "two-sided-short-edge": "short-edge", "1": "one-sided", "one": "one-sided", "one-sided": "one-sided", "simplex": "one-sided", "single": "one-sided", } JOB_RE = re.compile(r"request id is (\S+)") def one_line(text: str, limit: int = 300) -> str: text = " ".join(text.split()) return text if len(text) <= limit else text[:limit] + "..." def validate_source(path: str): if not os.path.isfile(path): return f"source PDF not found: {path}" try: with open(path, "rb") as fh: head = fh.read(5) except OSError as exc: return f"cannot read {path}: {exc}" if head != b"%PDF-": return f"{path} does not start with the %PDF- signature" if os.path.getsize(path) == 0: return f"{path} is empty" return None def check_queue(queue: str): try: proc = subprocess.run( ["lpstat", "-p", queue], capture_output=True, text=True, timeout=30, ) except (OSError, subprocess.TimeoutExpired) as exc: return f"cannot run lpstat: {exc}" if proc.returncode != 0: detail = one_line((proc.stderr or "") + " " + (proc.stdout or "")) return f"lpstat -p {queue} failed: {detail or 'queue not found'}" return None def submit(command) -> tuple[str, str]: """Run lp; return (job description, error description).""" try: proc = subprocess.run( command, capture_output=True, text=True, timeout=120, ) except (OSError, subprocess.TimeoutExpired) as exc: return "", f"{type(exc).__name__}: {exc}" output = one_line((proc.stdout or "") + " " + (proc.stderr or "")) if proc.returncode != 0: return "", output or f"lp exited with status {proc.returncode}" match = JOB_RE.search(output) return (match.group(1) if match else output or "submitted"), "" def parse_binding(raw: str) -> tuple[str, str]: """'l note' -> ('long-edge', note); 's'/'1' likewise; else unknown.""" text = raw.strip() if not text: return "not checked", "" first, _, rest = text.partition(" ") binding = BINDING_LABELS.get(first.lower().rstrip(".,:;")) if binding: return binding, rest.strip() return "unknown", text def ask_all(entries): print("\nAll jobs submitted. Release/collect the sheets and check the") print("release-device display (or the printed result) for each job.") print("For each test answer which binding the display shows:") print(" l 2-sided, long edge (normal duplex)") print(" s 2-sided, short edge") print(" 1 1-sided") print(" l as l, plus a note") print(" not checked") print(" kept verbatim as a note, binding 'unknown'\n") for entry in entries: if entry["error"]: continue prompt = ( f"test {entry['number']} ({entry['title']}) binding? " "[l/s/1/note]: " ) try: raw = input(prompt) except EOFError: print() entry["binding"] = "not checked" continue binding, note = parse_binding(raw) entry["binding"] = binding entry["note"] = note def write_results(path, entries, source, queue): now = datetime.datetime.now().astimezone().isoformat(timespec="seconds") lines = [ "A4 duplex print-option test results", "", f"date: {now}", f"queue: {queue}", f"source: {source}", "media: media=A4 pinned on every job", "PPD: the server's default Follow-Me.ppd, no option-mapping backend", "answers: l = 2-sided long edge; s = 2-sided short edge;", " 1 = 1-sided; other text = unknown; empty = not checked", "", ] for entry in entries: lines.append(f"test {entry['number']}: {entry['title']}") lines.append(f" why: {entry['why']}") lines.append(f" expect: {entry['expect']}") lines.append(f" lp: {entry['command']}") if entry["error"]: lines.append(f" job: FAILED: {entry['error']}") else: lines.append(f" job: {entry['job']}") lines.append(f" display: {entry['binding']}") lines.append(f" note: {entry['note'] or '(none)'}") lines.append("") lines.append("summary") for entry in entries: lines.append( f" {entry['number']}. {entry['binding']:<12} {entry['title']}" ) shown_long = [str(e["number"]) for e in entries if e["binding"] == "long-edge"] shown_short = [str(e["number"]) for e in entries if e["binding"] == "short-edge"] shown_one = [str(e["number"]) for e in entries if e["binding"] == "one-sided"] lines.append("") lines.append(f"long edge shown: {', '.join(shown_long) or '(none)'}") lines.append(f"short edge shown: {', '.join(shown_short) or '(none)'}") lines.append(f"one-sided shown: {', '.join(shown_one) or '(none)'}") lines.append("") try: with open(path, "w", encoding="utf-8") as fh: fh.write("\n".join(lines)) except OSError as exc: print(f"\n[FAIL] cannot write {path}: {exc}", file=sys.stderr) return False return True def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( description="Send eight A4 duplex-option test jobs to a CUPS queue and " "record which binding the display showed.", formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) p.add_argument("--queue", default=DEFAULT_QUEUE, help="CUPS queue name for lp -d") p.add_argument("--source", default=DEFAULT_SOURCE, help="two-page PDF (copied to test1.pdf ... test8.pdf)") p.add_argument("--results", default=DEFAULT_RESULTS, help="answers are written here") p.add_argument("--delay", type=float, default=1.0, help="seconds between submissions") p.add_argument("--force", action="store_true", help="submit even when the lpstat queue preflight fails") p.add_argument("--dry-run", action="store_true", help="show the plan and the lp commands without printing") return p def main(argv=None) -> int: args = build_parser().parse_args(argv) source = os.path.abspath(args.source) results = os.path.abspath(args.results) try: sys.stdout.reconfigure(line_buffering=True) except (AttributeError, ValueError): pass print("A4 duplex print-option tests (normal duplex = 2-sided long edge)") print(f" queue: {args.queue}") print(f" source: {source}") print(f" results: {results}") print(" media=A4 is sent on every job; only the duplex setting varies") print(" assumes the queue uses the server's default Follow-Me.ppd") print(" (no papercut_relay_backend.py option-mapping wrapper)") print() for number, (title, options, expect, why) in enumerate(TESTS, 1): print(f" {number}. {title}") print(f" {why}; expected: {expect}") if args.dry_run: print("\nDry run: the commands would be") for number, (_title, options, _expect, _why) in enumerate(TESTS, 1): command = ["lp", "-d", args.queue, *MEDIA_OPTION, *options, f"test{number}.pdf"] print(f" {shlex.join(command)}") print("\n(the job files are copied to test1.pdf ... test8.pdf in a " "temporary directory)") return 0 problem = validate_source(source) if problem: print(f"\n[FAIL] {problem}", file=sys.stderr) return 1 if not args.force: problem = check_queue(args.queue) if problem: print(f"\n[FAIL] {problem}", file=sys.stderr) print("Fix the queue name or pass --force to submit anyway.", file=sys.stderr) return 1 workdir = tempfile.mkdtemp(prefix="duplex-tests-") entries = [] print(f"\nSubmitting (job copies in {workdir}):\n") try: for number, (title, options, expect, why) in enumerate(TESTS, 1): job_path = os.path.join(workdir, f"test{number}.pdf") try: shutil.copyfile(source, job_path) except OSError as exc: print(f"[{number}/{len(TESTS)}] {title}") print(f" cannot copy source: {exc}") entries.append({ "number": number, "title": title, "expect": expect, "why": why, "command": shlex.join( ["lp", "-d", args.queue, *MEDIA_OPTION, *options, job_path]), "job": "", "error": f"cannot copy source: {exc}", "binding": "not submitted", "note": "", }) continue command = ["lp", "-d", args.queue, *MEDIA_OPTION, *options, job_path] print(f"[{number}/{len(TESTS)}] {title}") print(f" {shlex.join(command)}") job, error = submit(command) if error: print(f" FAILED: {error}") else: print(f" {job}") entries.append({ "number": number, "title": title, "expect": expect, "why": why, "command": shlex.join(command), "job": job, "error": error, "binding": "not submitted" if error else "not checked", "note": "", }) if number < len(TESTS) and args.delay > 0: time.sleep(args.delay) except KeyboardInterrupt: print("\nInterrupted during submission; keeping what was sent.") submitted = [entry for entry in entries if not entry["error"]] if submitted: try: ask_all(submitted) except KeyboardInterrupt: print("\nInterrupted; saving the answers collected so far.") if not write_results(results, entries, source, args.queue): return 1 print(f"\nAnswers written to {results}") return 0 if __name__ == "__main__": try: raise SystemExit(main()) except KeyboardInterrupt: print("\nInterrupted.", file=sys.stderr) raise SystemExit(130)