"""Drive the Gradio app's button handlers end-to-end through the real podman runner. This calls the exact functions the Estimate / Launch buttons are wired to, so it exercises the full app code path (validation -> LocalPodmanRunner -> podman job -> output) without a browser. Requires CC_REPACKAGE_EXECUTOR=podman and the podman env vars to be set (see verify_local.sh). """ from __future__ import annotations import os import sys sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from src import config assert config.EXECUTOR == "podman", "set CC_REPACKAGE_EXECUTOR=podman" import app # noqa: E402 # Exercises the content_languages filter while still matching both fixture records: # example.com is "eng", blog.example.org is "deu,eng" (primary deu). FORM = dict( domains_text="example.com", hostnames_text="blog.example.org", languages=["eng", "deu"], crawls=["CC-MAIN-2026-25"], target_bucket="local/out", target_path="run1", name="repackaged", flavor="cpu-upgrade", ) def _drive(gen): last = None for out in gen: last = out status = out[0].splitlines()[0] if out[0] else "" print(f" … {status}") return last def main() -> int: args = list(FORM.values()) print("[1/2] Estimate button →") md, logs, _btn, n_records, _active, _cancel = _drive(app.estimate_handler(*args)) print("\n--- estimate result ---\n" + md + "\n") if "Cost estimate" not in md: print("FAIL: no cost estimate produced") print(logs[-2000:]) return 1 print(f" (n_records={n_records} -> fetch --processes sized accordingly)") print("[2/2] Launch button →") md2, logs2, _active2, _cancel2 = _drive(app.launch_handler(*args, n_records)) print("\n--- launch result ---\n" + md2 + "\n") if "complete" not in md2.lower(): print("FAIL: repackage did not complete") print(logs2[-3000:]) return 1 print("OK: app drove estimate + launch through podman successfully") return 0 if __name__ == "__main__": sys.exit(main())