#!/usr/bin/env python3 """Send eight A3 print-option test jobs to a CUPS queue and record the answers. The queue is assumed to use the server's default Follow-Me.ppd and *not* the papercut_relay_backend.py option-mapping wrapper, so whatever `lp` puts on the wire is what the Mobility Print queue receives. Each job requests A3 in a different way (or not at all), so that the physical output shows which option form the queue honours: 1 no media option baseline / queue default 2 media=iso_a3 standard PWG keyword 3 media=iso_a3_297x420mm self-describing PWG keyword 4 media=A3 non-standard keyword 5 media-col={media-size={x-dimension=...}} dimensions (captured form) 6 media-col with media-size-name=iso_a3 dimensions plus the name 7 PageSize=A3 PPD option name 8 PageSize=A3 + media=iso_a3 PPD name should not mask IPP The source PDF is copied to test1.pdf ... test8.pdf so the CUPS job names make the tests distinguishable in lpstat and in the release list. All eight jobs are submitted first. The script then asks, per test, whether the sheet printed on A3 portrait and writes the answers to ``results-a3-portrait.txt`` (plain text, safe to upload). Usage: python3 -I a3_option_tests.py python3 -I a3_option_tests.py --source a3-cmyk-color-printer-test-page.pdf python3 -I a3_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 = "a3-cmyk-color-printer-test-page.pdf" DEFAULT_RESULTS = "results-a3-portrait.txt" A3_COLLECTION = ( "media-col={media-size={x-dimension=29700 y-dimension=42000}}" ) A3_COLLECTION_NAMED = ( "media-col={media-size={x-dimension=29700 y-dimension=42000} " "media-size-name=iso_a3}" ) # (title, lp options, why this test) TESTS = [ ("baseline - no media option", [], "queue/printer default; also shows whether the document's own page size " "already selects A3"), ("media=iso_a3", ["-o", "media=iso_a3"], "standard PWG media keyword"), ("media=iso_a3_297x420mm", ["-o", "media=iso_a3_297x420mm"], "self-describing PWG media keyword"), ("media=A3", ["-o", "media=A3"], "explicitly non-standard keyword"), ("media-col by dimensions", ["-o", A3_COLLECTION], "the media-col form advertised in the capture (29700 x 42000)"), ("media-col with media-size-name", ["-o", A3_COLLECTION_NAMED], "dimensions plus the name, if the server prefers the name"), ("PageSize=A3", ["-o", "PageSize=A3"], "PPD option name; expected to be forwarded literally and ignored"), ("PageSize=A3 + media=iso_a3", ["-o", "PageSize=A3", "-o", "media=iso_a3"], "PPD name present, IPP name should still win"), ] 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_answer(raw: str) -> tuple[str, str]: """'y note' -> ('yes', note); 'n note' -> ('no', note); else unknown.""" text = raw.strip() if not text: return "not checked", "" first, _, rest = text.partition(" ") first = first.lower().rstrip(".,:;") if first in ("y", "yes"): return "yes", rest.strip() if first in ("n", "no"): return "no", rest.strip() return "unknown", text def ask_all(entries): print("\nAll jobs submitted. Release/collect the sheets and check them.") print("For each test answer on one line:") print(" y printed on A3 paper (297 x 420 mm), portrait") print(" n did not print on A3 (wrong size, format or failure)") print(" y as y, plus a note") print(" n as n, plus a note") print(" not checked") print(" kept verbatim as a note, verdict 'unknown'\n") for entry in entries: if entry["error"]: continue prompt = f"test {entry['number']} ({entry['title']}) worked? [y/n/note]: " try: raw = input(prompt) except EOFError: print() entry["verdict"] = "not checked" continue verdict, note = parse_answer(raw) entry["verdict"] = verdict entry["note"] = note def write_results(path, entries, source, queue): now = datetime.datetime.now().astimezone().isoformat(timespec="seconds") lines = [ "A3 portrait print-option test results", "", f"date: {now}", f"queue: {queue}", f"source: {source}", "PPD: the server's default Follow-Me.ppd, no option-mapping backend", "meaning: verdict 'yes' = printed on A3 paper (297 x 420 mm), portrait", "", ] for entry in entries: lines.append(f"test {entry['number']}: {entry['title']}") lines.append(f" why: {entry['why']}") 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" verdict: {entry['verdict']}") lines.append(f" note: {entry['note'] or '(none)'}") lines.append("") lines.append("summary") for entry in entries: lines.append( f" {entry['number']}. {entry['verdict']:<11} {entry['title']}" ) 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 A3 option test jobs to a CUPS queue and " "record which option form printed on A3.", 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="PDF to print (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("A3 portrait print-option tests") print(f" queue: {args.queue}") print(f" source: {source}") print(f" results: {results}") 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, why) in enumerate(TESTS, 1): print(f" {number}. {title}") print(f" {why}") if args.dry_run: print("\nDry run: the commands would be") for number, (_title, options, _why) in enumerate(TESTS, 1): command = ["lp", "-d", args.queue, *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="a3-tests-") entries = [] print(f"\nSubmitting (job copies in {workdir}):\n") try: for number, (title, options, 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, "why": why, "command": shlex.join( ["lp", "-d", args.queue, *options, job_path]), "job": "", "error": f"cannot copy source: {exc}", "verdict": "not submitted", "note": "", }) continue command = ["lp", "-d", args.queue, *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, "why": why, "command": shlex.join(command), "job": job, "error": error, "verdict": "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}") if any(entry["verdict"] == "yes" for entry in entries): winners = [e["number"] for e in entries if e["verdict"] == "yes"] print(f" A3 confirmed for test(s): {', '.join(map(str, winners))}") return 0 if __name__ == "__main__": try: raise SystemExit(main()) except KeyboardInterrupt: print("\nInterrupted.", file=sys.stderr) raise SystemExit(130)