scripts/drive_app.py

"""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())