#!/usr/bin/env python3 """Send eight A4 color-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 color setting varies; the A3 tests showed that the document's own page size does not select the media by itself. The eight jobs differ in how color is requested: 1 no color option baseline / queue default 2 print-color-mode=color standard IPP attribute 3 print-color-mode=auto standard IPP auto value 4 print-color-mode=monochrome negative control (expect mono) 5 ColorModel=RGB generic PPD option name 6 ColorModel=CMYK generic PPD option name 7 ColorModel=Gray control for PPD-name handling 8 print-color-mode=color + ColorModel=Gray conflict precedence The controls matter: if the queue already defaults to color, a plain "Color" on the display proves nothing about tests 5, 6 or 8. Test 4 shows whether the display and pipeline really follow the IPP attribute, and tests 4/7/8 show whether the PPD option name is honoured at all. 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 color mode the display showed, and writes ``results-a4-color.txt`` (plain text, safe to upload). Usage: python3 -I color_option_tests.py python3 -I color_option_tests.py --source a4-cmyk-color-printer-test-page.pdf python3 -I color_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 = "a4-cmyk-color-printer-test-page.pdf" DEFAULT_RESULTS = "results-a4-color.txt" # Pinned so that only the color setting varies between jobs. MEDIA_OPTION = ["-o", "media=A4"] # (title, lp options, expected display if honoured, why this test) TESTS = [ ("baseline - no color option", [], "the queue default", "control: is color already the default?"), ("print-color-mode=color", ["-o", "print-color-mode=color"], "color", "standard IPP attribute"), ("print-color-mode=auto", ["-o", "print-color-mode=auto"], "device/printer decides", "standard IPP auto value"), ("print-color-mode=monochrome", ["-o", "print-color-mode=monochrome"], "monochrome", "negative control: proves the display/pipeline follow the attribute"), ("ColorModel=RGB", ["-o", "ColorModel=RGB"], "color only if the PPD name is honoured, else the default", "generic PPD option name"), ("ColorModel=CMYK", ["-o", "ColorModel=CMYK"], "color only if the PPD name is honoured, else the default", "generic PPD option name"), ("ColorModel=Gray", ["-o", "ColorModel=Gray"], "monochrome only if the PPD name is honoured, else the default", "control for PPD-name handling"), ("print-color-mode=color + ColorModel=Gray", ["-o", "print-color-mode=color", "-o", "ColorModel=Gray"], "color if the IPP name wins, monochrome if the PPD name wins", "conflict precedence"), ] MODE_LABELS = { "y": "color", "yes": "color", "c": "color", "color": "color", "colour": "color", "n": "monochrome", "no": "monochrome", "m": "monochrome", "mono": "monochrome", "monochrome": "monochrome", "gray": "monochrome", "grey": "monochrome", "grayscale": "monochrome", "greyscale": "monochrome", "bw": "monochrome", "b/w": "monochrome", "a": "auto", "auto": "auto", } 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_mode(raw: str) -> tuple[str, str]: """'c note' -> ('color', note); 'm' -> ('monochrome', ''); else unknown.""" text = raw.strip() if not text: return "not checked", "" first, _, rest = text.partition(" ") mode = MODE_LABELS.get(first.lower().rstrip(".,:;")) if mode: return mode, 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, does the display/result show Color?") print(" y Color") print(" n Mono / Grayscale / B&W") print(" a Auto") print(" y as y, plus a note") print(" (c and m are accepted as well)") print(" not checked") print(" kept verbatim as a note, mode 'unknown'\n") for entry in entries: if entry["error"]: continue prompt = f"test {entry['number']} ({entry['title']}) color? [y/n/a/note]: " try: raw = input(prompt) except EOFError: print() entry["mode"] = "not checked" continue mode, note = parse_mode(raw) entry["mode"] = mode entry["note"] = note def write_results(path, entries, source, queue): now = datetime.datetime.now().astimezone().isoformat(timespec="seconds") lines = [ "A4 color 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: y = Color; n = Mono/Grayscale/B&W; a = Auto;", " c/m are also accepted; 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['mode']}") lines.append(f" note: {entry['note'] or '(none)'}") lines.append("") lines.append("summary") for entry in entries: lines.append( f" {entry['number']}. {entry['mode']:<12} {entry['title']}" ) shown_color = [str(e["number"]) for e in entries if e["mode"] == "color"] shown_mono = [str(e["number"]) for e in entries if e["mode"] == "monochrome"] shown_auto = [str(e["number"]) for e in entries if e["mode"] == "auto"] lines.append("") lines.append(f"color shown: {', '.join(shown_color) or '(none)'}") lines.append(f"monochrome shown: {', '.join(shown_mono) or '(none)'}") lines.append(f"auto shown: {', '.join(shown_auto) 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 color-option test jobs to a CUPS queue and " "record which color mode 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="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("A4 color print-option tests") print(f" queue: {args.queue}") print(f" source: {source}") print(f" results: {results}") print(" media=A4 is sent on every job; only the color 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="color-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}", "mode": "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, "mode": "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)