diff --git a/README.md b/README.md index 3831e75..4c1f11c 100644 --- a/README.md +++ b/README.md @@ -49,8 +49,8 @@ abench /path/to/model Abench looks for `.yaml` and `.yml` files directly inside `/path/to/model/.abench/` (no recursive search). In a terminal it lists them alphabetically and asks which -experiment to run, then prompts for that experiment's inputs. Enter chooses the -first file; a single file still gets a selection prompt. Only the selected file +experiment to run, then prompts for that experiment's inputs. Use ↑/↓ to move the highlighted selection and Enter to confirm. The first file +is initially selected; a single file still gets a selection prompt. Only the selected file is loaded. `run`, `validate`, and `prepare` all support directory selection. `abench /path/to/model --help` lists available files without prompting or running @@ -196,7 +196,10 @@ vars: ``` In a terminal, run, validate, and prepare prompt for each input in YAML order. -Press Enter to accept a displayed default. Required inputs have no default and +Inputs with `choices` use the same arrow-key menu, with the declared default +initially highlighted (or the first option if required). Enter confirms the +selection and preserves its declared type. Other inputs use text prompts. +Press Enter to accept a displayed default. Required text inputs have no default and must be entered; empty or invalid answers prompt again with an explanation. Ctrl-C cancels before source resolution or downloads. Selected values are printed and saved in `suite.json` as `input_values`. @@ -266,13 +269,119 @@ This experiment file describes **which tests to run**. A model profile such as across suites. Terminal progress identifies the experiment number, image build, warmup, and -measured attempts/retries. Builds and model phases print elapsed time every -15 seconds; model phases also show current and peak cgroup memory when samples -are available. Full console output stays in the printed log paths. These are -status updates, not an estimated completion percentage. +measured attempts/retries. During builds and model phases, an interactive terminal +shows live elapsed time, current and peak cgroup memory (when available), and a +panel with the last 12 lines of the run log. Ctrl-C cancels the command. Full +console output stays in the printed log paths; elapsed time and memory status +are saved every 15 seconds in a sibling `console.progress.log` (or +`build.progress.log` for builds). Redirected output and non-interactive terminals +only print start and finish summaries. These are status updates, not an estimated +completion percentage. + +## Publishing benchmark results to a PR + +A suite can post one results comment after its final comparison report. Add an +explicit target (the repository is always on github.com): + +```yaml +inputs: + pr: + type: integer + required: true + minimum: 1 +publish: + github: + repository: ActivitySim/activitysim + pr: ${pr} + baseline: main +``` + +`baseline` names a run in `runs` and defaults to the first run. The posting target +is independent of source selection: use `pr: ${pr}` in an ActivitySim source +mapping to benchmark that PR's pinned head as well. See +[`examples/sandag-pr.yaml`](examples/sandag-pr.yaml) for a complete example with a +pinned baseline; adjust its model/data paths and baseline commit for your model. + +```bash +abench examples/sandag-pr.yaml --set pr=1110 +# Run benchmarks, but only generate the local comment and charts: +abench examples/sandag-pr.yaml --set pr=1110 --publish-dry-run +# Inspect or publish retained results without rerunning benchmarks: +abench publish /path/to/suite-output --dry-run +abench publish /path/to/suite-output +# Read local status, or verify the comment still exists on GitHub: +abench publish /path/to/suite-output --status +abench publish /path/to/suite-output --verify +# The permanent launcher works from any working directory: +/path/to/suite-output/publication/publish.sh +/path/to/suite-output/publication/publish.sh --dry-run +``` + +Publication requires GitHub CLI 2.99 or newer with `gh pr comment --attach`, an +authenticated account, and repository write access for image uploads. Use +`gh auth login --hostname github.com` (OAuth), or a classic personal access token +through `GH_TOKEN`. Other token types and GitHub Enterprise Server image uploads +are not supported by this first release. `gh` remains optional for ordinary runs +and publication dry runs. Authentication and PR access are checked before asset +preparation or benchmark execution when publication is enabled; validation and +prepare-only commands do not publish. Tokens are never written to experiment +metadata or passed into benchmark containers. + +The comment contains elapsed time and peak memory, percentage changes against the +baseline for valid runs with matching comparison settings, run settings, source +commits, memory traces, and component-runtime charts. Invalid runs have no change +claims and are excluded from the runtime chart; unstarted runs are identified. +PR head/base commits are recorded at startup, and a head change during execution +is noted when posting. These are observations, not statistical significance tests. + +The `publication/` directory retains `comment.md`, `memory.svg`, `runtimes.svg`, +and `state.json` with the comment ID/URL and publication status. `STATUS.md` shows +whether publication is pending, failed, uncertain, or published, with attempt time, +errors, a retry command, and the comment link. It reflects local knowledge; +`--verify` checks GitHub without uploading or editing anything. Verification +failures preserve the previous publication record. A missing previously published +comment is flagged and is not automatically recreated. + +The executable `publish.sh` is permanent: it is safe to call repeatedly and passes +through `--dry-run`, `--status`, or `--verify`. It prefers `abench` on PATH, then +uses `uvx --refresh --from ` when a development checkout was +available when the bundle was prepared. It reports an actionable error if neither +is available and never stores credentials. Output paths with spaces are supported. +Existing suites can create these helpers with `abench publish OUTPUT --status` +without reading measurement data or contacting GitHub. + +The final console message explicitly says whether results were published, gives +the comment URL on success, or gives the launcher command when action is needed. +A connection failure after posting begins is marked as an uncertain outcome; +retrying reconciles the comment marker before attempting another post. + +The exact upload body is retained as `upload.md`. A retry searches for the suite's +unique comment +marker before posting; a recorded successful publication is a no-op. Independent +suite executions get separate comments. Simultaneous publication of the same +output directory is blocked. Partial uploads can leave unused GitHub attachments, +but the local reports are retained and a retry can complete the comment. + +A posting error fails the command and prints the retry command. If the benchmark +also failed, its failure remains primary and the publication error is printed +separately. The full HTML/JSON comparison remains local; this release uploads only +the summary and SVG figures, not raw logs, model data, or the interactive HTML. +Images must fit GitHub's 10 MB attachment limit, and summaries are limited to +60,000 characters. Publication transport tests simulate GitHub; a live attachment +smoke test should use a designated test PR, never a production PR by default. ## Run controls +On macOS, benchmark commands automatically run `/usr/bin/caffeinate -i` to prevent +idle system sleep for the entire invocation, including preparation, builds, +warmup, measured attempts, reporting, and PR publication. The display can still +sleep. The assertion is released on completion or cancellation; it also expires +if the abench process exits unexpectedly. Other platforms are unchanged. +Use `--allow-sleep` to opt out, for example `abench experiments.yaml --allow-sleep`. +Report-only, publish-only, validation, preparation-only, and help commands do not +start caffeinate. This prevents idle sleep; it does not override explicit sleep +or closing a laptop lid. + - `--single-process` (default), or `--multiprocess --processes N`. The count applies to every sliced stage; coordinators are additional processes. - `--sharrow` (default) or `--no-sharrow`. Sharrow enabled requires its source pin. diff --git a/examples/sandag-pr.yaml b/examples/sandag-pr.yaml new file mode 100644 index 0000000..d9c9fdc --- /dev/null +++ b/examples/sandag-pr.yaml @@ -0,0 +1,58 @@ +# Run from any directory: abench /path/to/abench/examples/sandag-pr.yaml +schema_version: 1 +# Publishing is opt-in through this block. Use --publish-dry-run to preview locally. +publish: + github: + repository: ActivitySim/activitysim + pr: ${pr} + baseline: main +inputs: + pr: + type: integer + required: true + minimum: 1 + description: ActivitySim PR to benchmark and comment on + households: + type: integer + default: 28365 + minimum: 0 + description: Households to sample from benchmarking-data; 0 uses all + processes: + type: integer + default: 4 + minimum: 1 + warmup_households: + type: integer + default: 5000 + minimum: 1 +vars: + model: ../../sandag-abm3-example + activitysim_main: 5c6fae24a91a57a2d6dfc2e1dbe062a61d94545a +output_root: ../experiments/sandag-pr${pr}-${timestamp} +defaults: + model_dir: ${model} + profile: sandag + data_dir: ${model}/benchmarking-data + config_overlay: ["${model}/configs_explicit_chunk"] + multiprocess: true + processes: ${processes} + sharrow: true + households: ${households} + # Cache misses in the measured run trigger a retry using the compiled flows. + warmup_households: ${warmup_households} + memory: 80g + shm_size: 8g + platform: linux/arm64 + sources: + - sharrow=ActivitySim/sharrow@fc175b27d8e0c5d202721c67d96b050e6117b235 +runs: + main: + label: SANDAG main — explicit chunking + sources: + - activitysim=ActivitySim/activitysim@${activitysim_main} + candidate: + label: SANDAG PR${pr} — explicit chunking + sources: + - name: activitysim + repository: ActivitySim/activitysim + pr: ${pr} diff --git a/pyproject.toml b/pyproject.toml index 60205b9..4efa0e8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,7 +10,7 @@ readme = "README.md" requires-python = ">=3.10" license = "BSD-3-Clause" license-files = ["LICENSE"] -dependencies = ["PyYAML>=6", "platformdirs>=3", "zstandard>=0.21"] +dependencies = ["PyYAML>=6", "platformdirs>=3", "zstandard>=0.21", "prompt_toolkit>=3.0.36,<4"] keywords = ["activitysim", "benchmark", "transportation", "memory", "sharrow"] classifiers = [ "Development Status :: 3 - Alpha", @@ -40,6 +40,17 @@ abench = "abench.cli:entrypoint" [tool.setuptools.package-data] abench = ["Dockerfile", "profiles/*.yaml"] +# uvx --from must notice source edits, not just metadata changes. +[tool.uv] +cache-keys = [ + { file = "pyproject.toml" }, + { file = "README.md" }, + { file = "MANIFEST.in" }, + { file = "src/abench/**/*.py" }, + { file = "src/abench/Dockerfile" }, + { file = "src/abench/profiles/*.yaml" }, +] + [tool.ruff] line-length = 88 diff --git a/src/abench/__init__.py b/src/abench/__init__.py index eec3576..0150b2a 100644 --- a/src/abench/__init__.py +++ b/src/abench/__init__.py @@ -1,3 +1,3 @@ """Reproducible ActivitySim experiments in Linux containers.""" -__version__ = "0.1.2" +__version__ = "0.2.0" diff --git a/src/abench/awake.py b/src/abench/awake.py new file mode 100644 index 0000000..3bfd04e --- /dev/null +++ b/src/abench/awake.py @@ -0,0 +1,58 @@ +"""Keep macOS hosts awake for the lifetime of a benchmark CLI invocation.""" + +import os +import subprocess +import sys +from contextlib import contextmanager + + +def benchmark_invocation(argv): + """Exclude reporting, validation, data preparation, and informational commands.""" + return not ( + (argv and argv[0] in {"report", "publish", "validate", "prepare"}) + or any( + flag in argv + for flag in ("--help", "-h", "--version", "--report-only", "--allow-sleep") + ) + ) + + +@contextmanager +def keep_awake(enabled=True): + if not enabled or sys.platform != "darwin": + yield + return + # -i prevents idle system sleep without keeping the display on. -w also + # releases the assertion if abench dies without running Python cleanup. + try: + process = subprocess.Popen( + ["/usr/bin/caffeinate", "-i", "-w", str(os.getpid())], + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + except OSError as error: + raise ValueError( + "Cannot prevent macOS idle sleep with caffeinate. " + "Use --allow-sleep to run without sleep prevention." + ) from error + try: + # Detect launch failures before committing to an expensive run. + try: + code = process.wait(timeout=0.05) + except subprocess.TimeoutExpired: + print("macOS idle sleep prevention enabled (caffeinate).", flush=True) + else: + raise ValueError( + f"caffeinate exited unexpectedly (exit {code}); " + "use --allow-sleep to run without sleep prevention." + ) + yield + finally: + if process.poll() is None: + process.terminate() + try: + process.wait(timeout=5) + except subprocess.TimeoutExpired: + process.kill() + process.wait() diff --git a/src/abench/cli.py b/src/abench/cli.py index c2abda0..b4a3dc5 100644 --- a/src/abench/cli.py +++ b/src/abench/cli.py @@ -17,6 +17,7 @@ from . import __version__ from .attempts import measured_attempts +from .awake import benchmark_invocation, keep_awake from .common import read_json, write_json from .failures import BenchmarkFailure, describe_failure from .flow_cache import publish_flows, reuse_flows @@ -60,7 +61,12 @@ def positive(value): def parser(): p = argparse.ArgumentParser( description=__doc__, - epilog="Named experiments: abench experiments.yaml [--set NAME=VALUE]; preflight: abench validate experiments.yaml [--set NAME=VALUE]; data only: abench prepare experiments.yaml", + epilog="Named experiments: abench experiments.yaml [--set NAME=VALUE]; preflight: abench validate experiments.yaml [--set NAME=VALUE]; data only: abench prepare experiments.yaml; publish retained results: abench publish OUTPUT_DIRECTORY [--dry-run]", + ) + p.add_argument( + "--allow-sleep", + action="store_true", + help="disable automatic macOS idle sleep prevention", ) p.add_argument("--version", action="version", version=f"abench {__version__}") p.add_argument("--model-dir", type=Path, default=Path.cwd()) @@ -233,6 +239,35 @@ def container_phase(spec, output, data, image, phase_name): def main(argv=None): p = parser() argv = list(sys.argv[1:] if argv is None else argv) + if argv and argv[0] == "publish": + from .publishing import publish + + publish_parser = argparse.ArgumentParser(prog="abench publish") + publish_parser.add_argument("output_directory", type=Path) + publish_mode = publish_parser.add_mutually_exclusive_group() + publish_mode.add_argument( + "--dry-run", + action="store_true", + help="generate the local bundle without GitHub access", + ) + publish_mode.add_argument( + "--status", + action="store_true", + help="show recorded publication status without contacting GitHub", + ) + publish_mode.add_argument( + "--verify", + action="store_true", + help="check publication status on GitHub without posting", + ) + options = publish_parser.parse_args(argv[1:]) + publish( + options.output_directory, + dry_run=options.dry_run, + status_only=options.status, + verify=options.verify, + ) + return 0 # A file invocation stays separate from model profiles and ordinary flags. candidate = argv[1:] if argv and argv[0] in ("run", "validate", "prepare") else argv if ( @@ -272,6 +307,16 @@ def main(argv=None): action="store_true", help="use defaults and --set values without prompting (required inputs must be supplied)", ) + suite_parser.add_argument( + "--publish-dry-run", + action="store_true", + help="run benchmarks and prepare the comment and charts without publishing", + ) + suite_parser.add_argument( + "--allow-sleep", + action="store_true", + help="disable automatic macOS idle sleep prevention", + ) suite_args = suite_parser.parse_args(candidate) interactive = not suite_args.non_interactive and sys.stdin.isatty() selected = select_experiment(suite_args.experiment_file, interactive) @@ -282,6 +327,7 @@ def main(argv=None): validate_only=argv[0] == "validate", assignments=suite_args.set, interactive=interactive, + **({"publish_dry_run": True} if suite_args.publish_dry_run else {}), ) action = argv.pop(0) if argv and argv[0] in ("run", "report", "validate") else "run" args = p.parse_args(argv) @@ -599,7 +645,9 @@ def publish_cache(): def entrypoint(): """Expose CLI errors without an unnecessary Python traceback.""" try: - sys.exit(main()) + with keep_awake(enabled=benchmark_invocation(sys.argv[1:])): + code = main() + sys.exit(code) except KeyboardInterrupt: print("\nBenchmark cancelled.", file=sys.stderr) sys.exit(130) diff --git a/src/abench/discovery.py b/src/abench/discovery.py index 7b6d381..2057ae0 100644 --- a/src/abench/discovery.py +++ b/src/abench/discovery.py @@ -1,5 +1,7 @@ """Discover and choose experiment instructions stored inside a model directory.""" +from .menus import choose + def instruction_files(directory): """List immediate YAML files deterministically without loading any suite.""" @@ -46,20 +48,7 @@ def select_experiment(path, interactive): + ", ".join(p.name for p in files) + ". Run in a terminal to choose, or pass the experiment YAML path directly." ) - print(f"Experiments in {path / '.abench'}:", flush=True) - for number, file in enumerate(files, 1): - print(f" {number}. {file.name}", flush=True) - while True: - try: - answer = input(f"Choose experiment [1] (1–{len(files)}): ").strip() - except EOFError as error: - raise ValueError( - "Input ended while choosing an experiment; nothing started" - ) from error - if not answer: - answer = "1" - if answer.isascii() and answer.isdigit() and 1 <= int(answer) <= len(files): - chosen = files[int(answer) - 1] - print(f"Selected experiment: {chosen.name}", flush=True) - return chosen - print(f"Enter a number from 1 to {len(files)}.", flush=True) + selected = choose( + f"Experiment in {path / '.abench'}", [file.name for file in files] + ) + return files[selected] diff --git a/src/abench/experiments.py b/src/abench/experiments.py index a02b206..a09f630 100644 --- a/src/abench/experiments.py +++ b/src/abench/experiments.py @@ -8,6 +8,7 @@ import yaml +from . import publishing from .assets import plan_assets, prepare_assets from .common import write_json from .inputs import resolve_inputs @@ -196,6 +197,7 @@ def read_suite(path): "runs", "output_root", "data_assets", + "publish", } if unknown: raise ValueError(f"unknown experiment file fields: {sorted(unknown)}") @@ -218,6 +220,7 @@ def load_suite(path, assignments=(), interactive=False): runs = document.get("runs") if not isinstance(runs, dict) or not runs: raise ValueError("runs must be a nonempty mapping of run names to options") + publishing.configuration(document.get("publish"), runs) output = document.get("output_root") if not isinstance(output, str) or not output: raise ValueError("output_root must be a path") @@ -261,6 +264,7 @@ def run_suite( prepare_only=False, assignments=(), interactive=False, + publish_dry_run=False, ): """Preflight every run, execute serially, and preserve partial failure reports.""" plan = load_suite(path, assignments=assignments, interactive=interactive) @@ -271,6 +275,11 @@ def run_suite( flush=True, ) root = Path(plan["output_root"]) + publish_config = publishing.configuration( + plan["configuration"].get("publish"), [r["name"] for r in plan["runs"]] + ) + if publish_config and not (validate_only or prepare_only or publish_dry_run): + plan["publication_pr"] = publishing.preflight(publish_config) if not validate_only: if plan["data_assets"]: print("Preparing input data (checking shared cache)…", flush=True) @@ -293,6 +302,7 @@ def run_suite( (root / "experiments.yaml").write_text(plan["original_yaml"]) write_json(root / "suite.json", plan) completed = [] + failed = False try: for number, run in enumerate(plan["runs"], 1): print( @@ -306,9 +316,21 @@ def run_suite( if (Path(run["output_dir"]) / "experiment.json").is_file(): completed.append(Path(run["output_dir"])) if code: + failed = True return code + except BaseException: + failed = True + raise finally: if completed: report(completed, root / "comparison.html") print(f"Comparison: {root / 'comparison.html'}", flush=True) + if publish_config: + try: + publishing.publish(root, dry_run=publish_dry_run) + except (ValueError, OSError) as error: + if not failed: + raise + # Keep the model failure primary, but never hide a posting failure. + print(f"Publication failed: {error}", flush=True) return 0 diff --git a/src/abench/inputs.py b/src/abench/inputs.py index 4fd6e08..e087d3a 100644 --- a/src/abench/inputs.py +++ b/src/abench/inputs.py @@ -3,6 +3,8 @@ import math import re +from .menus import choose + TYPES = { "string": (str,), "integer": (int,), @@ -169,6 +171,11 @@ def prompt_value(name, spec): """Keep asking until the user supplies a valid value or accepts a default.""" if spec.get("description"): print(f"{name}: {spec['description']}", flush=True) + if "choices" in spec: + options = spec["choices"] + default_index = options.index(spec["default"]) if "default" in spec else 0 + selected = choose(name, [str(value) for value in options], default_index) + return check_value(name, spec, options[selected]) constraints = ", ".join( f"{key}: {spec[key]}" for key in ("choices", "minimum", "maximum") diff --git a/src/abench/menus.py b/src/abench/menus.py new file mode 100644 index 0000000..2d464e6 --- /dev/null +++ b/src/abench/menus.py @@ -0,0 +1,87 @@ +"""Keyboard-driven choices shared by experiment discovery and typed inputs.""" + +from prompt_toolkit.application import Application +from prompt_toolkit.data_structures import Point +from prompt_toolkit.key_binding import KeyBindings +from prompt_toolkit.layout import HSplit, Layout, Window +from prompt_toolkit.layout.controls import FormattedTextControl +from prompt_toolkit.layout.dimension import Dimension + + +def choose(title, labels, default_index=0): + """Return an index on Enter; preserve defaults and restore the terminal on exit.""" + if not labels or not 0 <= default_index < len(labels): + raise ValueError("choice menu requires options and a valid initial selection") + selected = default_index + bindings = KeyBindings() + + @bindings.add("up") + def up(event): + nonlocal selected + selected = (selected - 1) % len(labels) + + @bindings.add("down") + def down(event): + nonlocal selected + selected = (selected + 1) % len(labels) + + @bindings.add("enter") + def accept(event): + event.app.exit(result=selected) + + @bindings.add("c-c") + def cancel(event): + event.app.exit(exception=KeyboardInterrupt()) + + @bindings.add("c-d") + def eof(event): + event.app.exit(exception=EOFError()) + + def render(): + # Plain formatted-text fragments avoid interpreting labels as markup. + # Reverse video + a marker remain visible without color support. + fragments = [] + for index, label in enumerate(labels): + label = " ".join(str(label).splitlines()) + fragments.append( + ( + "bold reverse" if index == selected else "", + f"{'>' if index == selected else ' '} {label}", + ) + ) + if index < len(labels) - 1: + fragments.append(("", "\n")) + return fragments + + control = FormattedTextControl( + render, + focusable=True, + show_cursor=False, + get_cursor_position=lambda: Point(x=0, y=selected), + ) + app = Application( + layout=Layout( + HSplit( + [ + Window( + FormattedTextControl(title + " (↑/↓ to move, Enter to choose)"), + height=1, + ), + # A bounded viewport scrolls to the selection for long lists. + Window(control, height=Dimension(max=10), wrap_lines=False), + ] + ), + focused_element=control, + ), + key_bindings=bindings, + full_screen=False, + erase_when_done=True, + ) + try: + result = app.run() + except EOFError as error: + raise ValueError( + "Input ended while choosing; experiment not started" + ) from error + print(f"{title}: {labels[result]}", flush=True) + return result diff --git a/src/abench/progress.py b/src/abench/progress.py index 5a5ae3a..2e84058 100644 --- a/src/abench/progress.py +++ b/src/abench/progress.py @@ -1,9 +1,21 @@ """Host-side progress for long commands without changing measured model work.""" +import asyncio import csv +import os import subprocess +import sys import time +from prompt_toolkit.application import Application +from prompt_toolkit.data_structures import Point +from prompt_toolkit.key_binding import KeyBindings +from prompt_toolkit.layout import HSplit, Layout, Window +from prompt_toolkit.layout.controls import FormattedTextControl +from prompt_toolkit.widgets import Frame + +from .failures import tail + def memory_status(path): """Read only the last complete sample; tolerate partial writes and startup.""" @@ -24,24 +36,105 @@ def memory_status(path): return "" +def interactive_terminal(): + return ( + sys.stdin.isatty() and sys.stdout.isatty() and os.environ.get("TERM") != "dumb" + ) + + +def log_tail(path, lines=12): + """Bound reads and remove terminal controls from untrusted subprocess output.""" + text = tail(path, limit=16384) + return ( + "\n".join( + "".join(char if char.isprintable() or char == "\t" else "" for char in line) + for line in text.splitlines()[-lines:] + ) + or "Waiting for log output…" + ) + + +def live_view(status, log): + bindings = KeyBindings() + + @bindings.add("c-c") + @bindings.add("c-d") + def cancel(event): + event.app.exit(exception=KeyboardInterrupt()) + + return Application( + layout=Layout( + HSplit( + [ + Window(FormattedTextControl(status), height=2, wrap_lines=True), + Frame( + Window( + FormattedTextControl( + lambda: log_tail(log), + show_cursor=False, + get_cursor_position=lambda: Point( + x=0, y=len(log_tail(log).splitlines()) - 1 + ), + ), + height=12, + wrap_lines=False, + ), + title=f"Log tail: {log.name}", + ), + ] + ) + ), + key_bindings=bindings, + full_screen=False, + erase_when_done=True, + refresh_interval=0.25, + ) + + def run_logged(args, log, interval=15): - """Keep full logs on disk and print periodic stage status, including failures.""" + """Retain progress on disk and show a live status/log panel on terminals.""" + if interval <= 0: + raise ValueError("progress interval must be positive") label = "Image build" if log.name == "build.log" else log.parent.name + progress_log = log.with_name(f"{log.stem}.progress.log") started = time.monotonic() - print(f"{label}: started; log: {log}", flush=True) - with log.open("w") as stream: + + def status(): + return f"{label}: {time.monotonic() - started:.0f}s elapsed" + memory_status( + log.parent / "memory.csv" + ) + + initial = f"{label}: started; log: {log}; progress: {progress_log}" + print(initial, flush=True) + with log.open("w") as stream, progress_log.open("w", buffering=1) as progress: + progress.write(initial + "\n") process = subprocess.Popen(args, stdout=stream, stderr=subprocess.STDOUT) try: - while True: - try: - code = process.wait(timeout=interval) - break - except subprocess.TimeoutExpired: - detail = memory_status(log.parent / "memory.csv") - print( - f"{label}: {time.monotonic() - started:.0f}s elapsed{detail}", - flush=True, - ) + if interactive_terminal(): + app = live_view(status, log) + + async def monitor(): + try: + next_sample = started + interval + while process.poll() is None: + now = time.monotonic() + if now >= next_sample: + progress.write(status() + "\n") + next_sample = now + interval + await asyncio.sleep(min(0.25, interval)) + except Exception as error: + app.exit(exception=error) + else: + app.exit(result=process.returncode) + + code = app.run(pre_run=lambda: app.create_background_task(monitor())) + else: + while True: + try: + code = process.wait(timeout=interval) + break + except subprocess.TimeoutExpired: + progress.write(status() + "\n") except BaseException: # Let container_phase's finally block remove a cancelled container. # Reap the host Docker client so cancellation never leaves it running. @@ -51,12 +144,14 @@ def run_logged(args, log, interval=15): except subprocess.TimeoutExpired: process.kill() process.wait() + progress.write(f"{status()}; interrupted\n") raise - status = "completed" if code == 0 else f"failed (exit {code})" - print( - f"{label}: {status} after {time.monotonic() - started:.0f}s" - + memory_status(log.parent / "memory.csv"), - flush=True, - ) + outcome = "completed" if code == 0 else f"failed (exit {code})" + final = ( + f"{label}: {outcome} after {time.monotonic() - started:.0f}s" + + memory_status(log.parent / "memory.csv") + ) + progress.write(final + "\n") + print(final, flush=True) if code: raise subprocess.CalledProcessError(code, args) diff --git a/src/abench/publishing.py b/src/abench/publishing.py new file mode 100644 index 0000000..22fed68 --- /dev/null +++ b/src/abench/publishing.py @@ -0,0 +1,558 @@ +"""Optional GitHub publication, isolated from benchmark execution and credentials.""" + +import fcntl +import json +import re +import shlex +import subprocess +import uuid +from datetime import datetime, timezone +from importlib.metadata import PackageNotFoundError, distribution +from pathlib import Path +from urllib.parse import unquote, urlparse +from xml.etree import ElementTree as ET + +from . import __version__ +from .common import read_json, write_json +from .report import COLORS, escape, load_run, memory_chart, runtime_chart + + +def configuration(value, run_names): + if value is None: + return None + if not isinstance(value, dict) or set(value) != {"github"}: + raise ValueError("publish requires a github mapping") + config = value["github"] + if not isinstance(config, dict) or set(config) - {"repository", "pr", "baseline"}: + raise ValueError("invalid publish.github fields") + repository = config.get("repository", "") + if not isinstance(repository, str) or not re.fullmatch( + r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", repository + ): + raise ValueError("publish.github.repository must be OWNER/REPO on github.com") + if type(config.get("pr")) is not int or config["pr"] <= 0: + raise ValueError("publish.github.pr must be a positive integer") + baseline = config.get("baseline", next(iter(run_names))) + if not isinstance(baseline, str) or baseline not in run_names: + raise ValueError("publish.github.baseline must name a suite run") + return {**config, "baseline": baseline} + + +def gh(*args, cwd=None): + try: + return subprocess.run( + ["gh", *args], + check=True, + capture_output=True, + text=True, + timeout=300, + cwd=cwd, + ).stdout.strip() + except FileNotFoundError as error: + raise ValueError( + "GitHub publication requires gh >= 2.99 with --attach" + ) from error + except subprocess.CalledProcessError as error: + raise ValueError( + f"GitHub publication failed: {error.stderr.strip()}" + ) from error + except subprocess.TimeoutExpired as error: + raise ValueError( + "GitHub publication timed out; retry with abench publish" + ) from error + + +def preflight(config): + """Read-only checks before expensive work; never upload in preflight.""" + if "--attach" not in gh("pr", "comment", "--help"): + raise ValueError("GitHub image publication requires gh >= 2.99 with --attach") + repository = config["repository"] + permission = gh( + "api", + f"repos/{repository}", + "--hostname", + "github.com", + "--jq", + ".permissions.push", + ) + if permission != "true": + raise ValueError( + "GitHub image uploads require repository write access; authenticate gh" + ) + return json.loads( + gh( + "pr", + "view", + str(config["pr"]), + "--repo", + f"github.com/{repository}", + "--json", + "number,url,headRefOid,baseRefOid", + ) + ) + + +def cell(value): + # Escape Markdown and suppress mentions from labels or model diagnostics. + text = " ".join(str(value).splitlines()) + return "".join( + f"&#{ord(char)};" if char in "\\`*_{}[]()#+!|@" else escape(char) + for char in text + ) + + +def comparable(a, b): + keys = ( + "households", + "multiprocess", + "processes", + "sharrow", + "profile_name", + "platform", + "memory", + "shm_size", + "config_overlay", + "data_dir", + "model_commit", + "model_git_status", + ) + return all(a["spec"].get(k) == b["spec"].get(k) for k in keys) + + +def delta(run, baseline, value): + if ( + not baseline + or not run["valid"] + or not baseline["valid"] + or not comparable(run, baseline) + ): + return "—" + before, after = value(baseline), value(run) + if before is None or after is None or before <= 0: + return "—" + return f"{(after / before - 1) * 100:+.1f}%" + + +def standalone(svg): + _, _, width, height = ET.fromstring(svg).attrib["viewBox"].split() + svg = svg.replace( + "", f'>', 1 + ) + + +def now(): + return datetime.now(timezone.utc).isoformat() + + +def local_checkout(): + """Find a development checkout, including when running in a uvx environment.""" + candidates = [Path(__file__).resolve().parents[2]] + try: + direct = json.loads(distribution("abench").read_text("direct_url.json") or "{}") + url = urlparse(direct.get("url", "")) + if url.scheme == "file" and url.netloc in ("", "localhost"): + candidates.append(Path(unquote(url.path))) + except (ValueError, OSError, PackageNotFoundError): + pass + return next( + ( + str(p) + for p in candidates + if (p / "pyproject.toml").is_file() and (p / "src/abench").is_dir() + ), + None, + ) + + +def write_launcher(bundle, checkout): + launcher = bundle / "publish.sh" + fallback = "" + if checkout: + quoted = shlex.quote(checkout) + fallback = f"""if command -v uvx >/dev/null 2>&1 && [ -f {quoted}/pyproject.toml ]; then + exec uvx --refresh --from {quoted} abench publish "$suite_dir" "$@" +fi +""" + launcher.write_text( + """#!/bin/sh +# Permanent, safe-to-repeat abench publication launcher. No credentials stored. +set -eu +publication_dir=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd -P) +suite_dir=$(dirname -- "$publication_dir") +if command -v abench >/dev/null 2>&1; then + exec abench publish "$suite_dir" "$@" +fi +""" + + fallback + + """echo 'Cannot find abench. Install a version with PR publishing support, or restore the recorded checkout and install uv.' >&2 +exit 127 +""" + ) + launcher.chmod(0o755) + + +STATUS_LABELS = { + "prepared": "Not published", + "failed": "Publication failed", + "posting": "Publication outcome uncertain", + "uncertain": "Publication outcome uncertain", + "published": "Published", + "missing": "Previously published comment not found", +} + + +def save_status(bundle, state): + """JSON is authoritative; the Markdown file is a readable projection.""" + write_json(bundle / "state.json", state) + label = STATUS_LABELS.get(state["status"], state["status"]) + target = state["target"] + lines = [ + "# Publication status", + "", + f"**{label}**", + "", + f"Target: https://github.com/{target['repository']}/pull/{target['pr']}", + f"Last publication attempt: {state.get('last_attempt_at', 'Never recorded')}", + ] + if state.get("comment_url"): + lines += ["", f"[Open the recorded GitHub comment]({state['comment_url']})"] + if state.get("last_verified_at"): + lines += [f"Last GitHub check: {state['last_verified_at']}"] + if state.get("error"): + lines += ["", f"Last error: {cell(state['error'])}"] + if state.get("verification_error"): + lines += [ + "", + f"GitHub verification failed: {cell(state['verification_error'])}", + ] + if state["status"] in ("posting", "uncertain"): + lines += [ + "", + "GitHub may have accepted the comment. Retry checks for the existing comment before posting again.", + ] + command = shlex.quote(str(bundle / "publish.sh")) + lines += [ + "", + "Publish or retry (safe to repeat):", + "", + "```sh", + command, + "```", + "", + "Preview with `--dry-run`; inspect local status with `--status`; check GitHub with `--verify`.", + "", + "This file records local knowledge; use --verify to check whether the comment still exists.", + ] + (bundle / "STATUS.md").write_text("\n".join(lines) + "\n") + + +def completion(bundle, state): + if state["status"] == "published": + print(f"Results published: {state['comment_url']}") + elif state["status"] in ("posting", "uncertain"): + print("Results saved; publication outcome uncertain.") + else: + print("Results saved but not published.") + print(f"Publication status: {bundle / 'STATUS.md'}") + if state["status"] != "published": + print(f"Publish / retry: {shlex.quote(str(bundle / 'publish.sh'))}") + + +def find_comment(state): + target = state["target"] + pages = json.loads( + gh( + "api", + f"repos/{target['repository']}/issues/{target['pr']}/comments", + "--hostname", + "github.com", + "--paginate", + "--slurp", + ) + ) + marker = f"" + return next( + ( + c + for page in pages + for c in page + if marker in (c.get("body") or "") or c["id"] == state.get("comment_id") + ), + None, + ) + + +def verify_publication(bundle, state): + """Read GitHub only; never recreate or edit a comment during verification.""" + try: + found = find_comment(state) + except (ValueError, OSError) as error: + state["verification_error"] = str(error) + save_status(bundle, state) + raise + state["last_verified_at"] = now() + state.pop("verification_error", None) + if found: + state.update( + status="published", comment_id=found["id"], comment_url=found["html_url"] + ) + state.pop("error", None) + elif state.get("comment_url"): + state["status"] = "missing" + save_status(bundle, state) + + +def prepare_publication(root): + """Create local publication controls without reading benchmark measurements.""" + plan = read_json(root / "suite.json") + if not plan: + raise ValueError(f"No suite.json in {root}") + config = configuration( + plan["configuration"].get("publish"), [r["name"] for r in plan["runs"]] + ) + if not config: + raise ValueError("This suite has no publish.github configuration") + bundle = root / "publication" + bundle.mkdir(exist_ok=True) + state_path = bundle / "state.json" + state = read_json(state_path, {}) + if state and state["target"] != config: + raise ValueError("Publication target differs from the saved publication state") + if not state: + state = {"target": config, "id": str(uuid.uuid4()), "status": "prepared"} + state["checkout"] = local_checkout() + write_json(state_path, state) + if "checkout" not in state: + state["checkout"] = local_checkout() + save_status(bundle, state) + write_launcher(bundle, state.get("checkout")) + return bundle, state + + +def build_bundle(root): + """Rebuild shareable artifacts from retained results, without GitHub access.""" + bundle, state = prepare_publication(root) + plan = read_json(root / "suite.json") + config = state["target"] + marker = f"" + runs = [] + names = [] + for item in plan["runs"]: + path = Path(item["output_dir"]) + if (path / "experiment.json").is_file(): + runs.append(load_run(path)) + names.append(item["name"]) + baseline = next( + (run for name, run in zip(names, runs) if name == config["baseline"]), None + ) + lines = [ + marker, + "## abench results", + "", + f"Auto-generated by **abench {__version__}**.", + "", + f"PR: {config['repository']}#{config['pr']} · Baseline: {cell(config['baseline'])}", + ] + snapshot = plan.get("publication_pr", {}) + if snapshot: + lines += [ + f"PR head at startup: `{snapshot['headRefOid']}`; base: `{snapshot['baseRefOid']}`." + ] + lines += [ + "", + "| Run | Result | Elapsed (s) | Change | Peak (GiB) | Change |", + "|---|---|---:|---:|---:|---:|", + ] + + def elapsed(run): + return run["status"].get("elapsed_seconds") + + for name, run in zip(names, runs): + seconds = elapsed(run) + time_text = f"{seconds:.2f}" if seconds is not None else "—" + peak_text = f"{run['peak'] / 2**30:.3f}" if run["memory"] else "—" + lines += [ + f"| {cell(name)} | {'valid' if run['valid'] else 'FAILED / INVALID'} | {time_text} | {delta(run, baseline, elapsed)} | {peak_text} | {delta(run, baseline, lambda r: r['peak'])} |" + ] + for item in plan["runs"]: + if item["name"] not in names: + lines += [ + f"| {cell(item['name'])} | NOT RUN / NO RESULTS | — | — | — | — |" + ] + lines += [ + "", + "Negative changes mean lower elapsed time or memory. Changes are omitted for invalid runs or differing comparison settings. Single runs do not establish statistical significance.", + "", + "### Configuration and source commits", + ] + for name, run in zip(names, runs): + spec = run["spec"] + lines += [ + "", + f"**{cell(name)}** — {cell(spec.get('label', name))}", + f"Households: {cell(spec.get('households', 'unknown'))}; multiprocess: {cell(spec.get('multiprocess', False))}; processes: {cell(spec.get('processes', 1))}; Sharrow: {cell(spec.get('sharrow', False))}; profile: {cell(spec.get('profile_name', 'unknown'))}; platform: {cell(spec.get('platform', 'unknown'))}.", + ] + for source in spec.get("sources", []): + lines += [ + f"- {cell(source['name'])}: {cell(source['repository'])} at `{cell(source['commit'])}`" + ] + if not run["valid"]: + lines += [ + "- Invalid results are excluded from runtime comparisons and improvement claims." + ] + lines += [ + "", + "### Memory traces", + "", + "![Memory traces](./memory.svg)", + "", + "Blue: total cgroup memory; green dashed: anonymous + shared memory when available. Panels share axes; invalid runs are labeled.", + "", + "### Component runtimes", + "", + "![Component runtimes](./runtimes.svg)", + "", + "Valid runs only. Mean ± population standard deviation across worker executions, not confidence intervals. Parallel component times must not be summed as wall time.", + ] + (bundle / "comment.md").write_text("\n".join(lines) + "\n") + xmax = ( + max((row["elapsed_seconds"] for r in runs for row in r["memory"]), default=1) + or 1 + ) + ymax = ( + max((row["current_bytes"] for r in runs for row in r["memory"]), default=1) or 1 + ) + panels = [] + for i, (name, run) in enumerate(zip(names, runs)): + fragment = memory_chart({**run, "component_windows": []}, xmax, ymax) + svg = fragment[fragment.index("") + 6] + svg = svg.replace("{escape(name)} — {"valid" if run["valid"] else "FAILED / INVALID"}{svg}' + ] + if not panels: + panels = ['No memory results available'] + memory = f'{"".join(panels)}' + valid = [r for r in runs if r["valid"]] + components = sorted({c for r in valid for c in r["components"]}) + chart = runtime_chart(valid, components) + legend = "".join( + f'{escape(r["spec"]["label"])}' + for i, r in enumerate(valid) + ) + height = 30 + 20 * max(1, len(valid)) + chart_height = float(ET.fromstring(chart).attrib["viewBox"].split()[-1]) + chart = chart.replace("No valid runtime results' + runtime = f'{legend}{chart}' + + for filename, svg in (("memory.svg", memory), ("runtimes.svg", runtime)): + (bundle / filename).write_text(standalone(svg)) + return bundle, state + + +def publish(root, *, dry_run=False, status_only=False, verify=False): + """Publish once per suite; recover a successful comment after a lost response.""" + root = Path(root).expanduser().resolve() + if not (root / "suite.json").is_file(): + raise ValueError(f"No suite.json in {root}") + with (root / ".publication.lock").open("w") as lock: + try: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError as error: + raise ValueError("Publication is already running for this suite") from error + if status_only or verify: + bundle, state = prepare_publication(root) + if verify: + verify_publication(bundle, state) + print((bundle / "STATUS.md").read_text()) + return state.get("comment_url") + bundle, state = build_bundle(root) + if dry_run: + print(f"Publication preview: {bundle / 'comment.md'}") + completion(bundle, state) + return None + if state.get("status") == "missing": + raise ValueError( + "The previously published comment was not found on GitHub; not reposting automatically. See publication/STATUS.md." + ) + if state.get("comment_url"): + completion(bundle, state) + return state["comment_url"] + try: + state["last_attempt_at"] = now() + save_status(bundle, state) + config = state["target"] + current = preflight(config) + found = find_comment(state) + if not found: + snapshot = read_json(root / "suite.json").get("publication_pr", {}) + body = (bundle / "comment.md").read_text() + if snapshot and snapshot["headRefOid"] != current["headRefOid"]: + body += f"\nPR head has changed since startup to `{current['headRefOid']}`. Results describe the recorded commits above.\n" + (bundle / "comment.md").write_text(body) + if len(body) > 60000: + raise ValueError( + "Comment exceeds the 60,000 character publication limit" + ) + for name in ("memory.svg", "runtimes.svg"): + if (bundle / name).stat().st_size > 10 * 1024**2: + raise ValueError( + f"{name} exceeds GitHub's 10 MB attachment limit" + ) + # Resolve relative image references from the bundle, including + # output directories containing spaces or Markdown punctuation. + (bundle / "upload.md").write_text(body) + state.update(status="posting", last_attempt_at=now()) + save_status(bundle, state) + url = gh( + "pr", + "comment", + str(config["pr"]), + "--repo", + f"github.com/{config['repository']}", + "--body-file", + str(bundle / "upload.md"), + "--attach", + "./memory.svg", + "--attach", + "./runtimes.svg", + cwd=bundle, + ) + match = re.search(r"https://github\.com/[^\s]+#issuecomment-(\d+)", url) + if not match: + raise ValueError( + "GitHub returned no comment URL; retry to reconcile publication" + ) + found = {"id": int(match[1]), "html_url": match[0]} + state.update( + status="published", + published_at=state.get("published_at") or now(), + comment_id=found["id"], + comment_url=found["html_url"], + ) + state.pop("error", None) + save_status(bundle, state) + except (ValueError, OSError, KeyboardInterrupt) as error: + state.update( + status="uncertain" + if state["status"] in ("posting", "uncertain") + else "failed", + error=str(error) or "Publication interrupted", + last_attempt_at=state.get("last_attempt_at") or now(), + ) + save_status(bundle, state) + completion(bundle, state) + if isinstance(error, KeyboardInterrupt): + raise + raise ValueError( + f"{error}\nResults retained. Retry: {shlex.quote(str(bundle / 'publish.sh'))}" + ) from error + completion(bundle, state) + return state["comment_url"] diff --git a/tests/test_awake.py b/tests/test_awake.py new file mode 100644 index 0000000..84b816e --- /dev/null +++ b/tests/test_awake.py @@ -0,0 +1,114 @@ +"""Sleep inhibition is scoped to benchmark commands and always cleaned up.""" + +import os +import subprocess +from unittest.mock import Mock + +import pytest + +from abench import awake, cli + + +@pytest.mark.parametrize( + "args,expected", + [ + ([], True), + (["."], True), + (["run", "suite.yaml"], True), + (["suite.yaml", "--publish-dry-run"], True), + (["suite.yaml", "--allow-sleep"], False), + (["--help"], False), + (["--version"], False), + (["--report-only"], False), + (["report"], False), + (["publish", "output"], False), + (["validate", "suite.yaml"], False), + (["prepare", "suite.yaml"], False), + ], +) +def test_invocations(args, expected): + assert awake.benchmark_invocation(args) is expected + + +@pytest.mark.parametrize("failure", [None, RuntimeError, KeyboardInterrupt]) +def test_lifetime_and_cleanup(monkeypatch, failure): + monkeypatch.setattr(awake.sys, "platform", "darwin") + process = Mock() + process.poll.return_value = None + process.wait.side_effect = [subprocess.TimeoutExpired("caffeinate", 0.05), 0] + launch = Mock(return_value=process) + monkeypatch.setattr(awake.subprocess, "Popen", launch) + + def run(): + with awake.keep_awake(): + process.terminate.assert_not_called() + if failure: + raise failure() + + if failure: + with pytest.raises(failure): + run() + else: + run() + assert launch.call_args.args[0] == [ + "/usr/bin/caffeinate", + "-i", + "-w", + str(os.getpid()), + ] + process.terminate.assert_called_once() + assert process.wait.call_count == 2 + + +@pytest.mark.parametrize("platform,enabled", [("linux", True), ("darwin", False)]) +def test_noop(monkeypatch, platform, enabled): + monkeypatch.setattr(awake.sys, "platform", platform) + monkeypatch.setattr( + awake.subprocess, "Popen", lambda *a, **kw: pytest.fail("launched caffeinate") + ) + with awake.keep_awake(enabled): + pass + + +def test_failed_launch(monkeypatch): + monkeypatch.setattr(awake.sys, "platform", "darwin") + monkeypatch.setattr( + awake.subprocess, "Popen", Mock(side_effect=FileNotFoundError()) + ) + with pytest.raises(ValueError, match="--allow-sleep"): + with awake.keep_awake(): + pytest.fail("benchmark started") + + +def test_early_exit(monkeypatch): + monkeypatch.setattr(awake.sys, "platform", "darwin") + process = Mock() + process.wait.return_value = 1 + process.poll.return_value = 1 + monkeypatch.setattr(awake.subprocess, "Popen", Mock(return_value=process)) + with pytest.raises(ValueError, match="exited unexpectedly"): + with awake.keep_awake(): + pytest.fail("benchmark started") + + +def test_cli_entrypoint_covers_whole_run(monkeypatch): + from contextlib import contextmanager + + events = [] + + @contextmanager + def inhibit(enabled): + assert enabled + events.append("awake") + try: + yield + finally: + events.append("released") + + monkeypatch.setattr(cli.sys, "argv", ["abench", "suite.yaml"]) + monkeypatch.setattr(cli, "keep_awake", inhibit) + monkeypatch.setattr(cli, "main", lambda: events.append("run") or 7) + with pytest.raises(SystemExit) as error: + cli.entrypoint() + assert error.value.code == 7 + assert events == ["awake", "run", "released"] diff --git a/tests/test_discovery.py b/tests/test_discovery.py index 56891b3..46aae73 100644 --- a/tests/test_discovery.py +++ b/tests/test_discovery.py @@ -15,22 +15,26 @@ def instructions(tmp_path): return folder -def test_selection_order_and_invalid_answers(tmp_path, monkeypatch, capsys): +def test_selection_order(tmp_path, monkeypatch): folder = instructions(tmp_path) - assert [p.name for p in instruction_files(tmp_path)] == ["a.yaml", "b.yml"] - answers = iter(["bad", "0", "3", "2"]) - monkeypatch.setattr("builtins.input", lambda prompt: next(answers)) + + def choose(title, labels): + assert labels == ["a.yaml", "b.yml"] + return 1 + + monkeypatch.setattr("abench.discovery.choose", choose) assert select_experiment(tmp_path, True) == folder / "b.yml" - assert "Enter a number" in capsys.readouterr().out -def test_single_file_still_prompts_and_accepts_enter(tmp_path, monkeypatch): +def test_single_file_still_prompts(tmp_path, monkeypatch): folder = instructions(tmp_path) (folder / "b.yml").unlink() prompts = [] - monkeypatch.setattr("builtins.input", lambda prompt: prompts.append(prompt) or "") + monkeypatch.setattr( + "abench.discovery.choose", lambda title, labels: prompts.append(labels) or 0 + ) assert select_experiment(tmp_path, True) == folder / "a.yaml" - assert len(prompts) == 1 + assert prompts == [["a.yaml"]] assert select_experiment(tmp_path, False) == folder / "a.yaml" @@ -46,17 +50,6 @@ def test_noninteractive_multiple_and_missing(tmp_path): instruction_files(tmp_path) -def test_selection_eof(tmp_path, monkeypatch): - instructions(tmp_path) - - def ended(prompt): - raise EOFError() - - monkeypatch.setattr("builtins.input", ended) - with pytest.raises(ValueError, match="nothing started"): - select_experiment(tmp_path, True) - - def test_directory_help_does_not_prompt_or_parse_suites(tmp_path, monkeypatch, capsys): instructions(tmp_path) @@ -85,7 +78,11 @@ def test_choice_then_inputs_and_file_relative_paths(tmp_path, monkeypatch): households: ${households} """) prompts = [] - answers = iter(["2", "42"]) + answers = iter(["42"]) + monkeypatch.setattr( + "abench.discovery.choose", + lambda title, labels: prompts.append("Choose experiment") or 1, + ) def respond(prompt): prompts.append(prompt) diff --git a/tests/test_menus.py b/tests/test_menus.py new file mode 100644 index 0000000..b0576d6 --- /dev/null +++ b/tests/test_menus.py @@ -0,0 +1,63 @@ +"""Exercise real arrow-key bindings without needing a physical terminal.""" + +import pytest +from prompt_toolkit.application import create_app_session +from prompt_toolkit.input import create_pipe_input +from prompt_toolkit.output import DummyOutput + +from abench.inputs import resolve_inputs +from abench.menus import choose + + +@pytest.mark.parametrize( + "keys,initial,expected", + [ + ("\r", 1, 1), + ("\x1b[B\r", 0, 1), + ("\x1b[A\r", 1, 0), + ("\x1b[A\r", 0, 2), + ("\x1b[B\r", 2, 0), + ], +) +def test_navigation(keys, initial, expected): + with ( + create_pipe_input() as pipe, + create_app_session(input=pipe, output=DummyOutput()), + ): + pipe.send_text(keys) + assert choose("Mode", ["First", "Second", "Third"], initial) == expected + + +@pytest.mark.parametrize( + "key,error", [("\x03", KeyboardInterrupt), ("\x04", ValueError)] +) +def test_cancellation(key, error): + with ( + create_pipe_input() as pipe, + create_app_session(input=pipe, output=DummyOutput()), + ): + pipe.send_text(key) + with pytest.raises(error): + choose("Mode", ["First"]) + + +def test_typed_choice_defaults_and_required(monkeypatch): + calls = [] + + def pick(title, labels, initial): + calls.append((labels, initial)) + return initial + + monkeypatch.setattr("abench.inputs.choose", pick) + values, _ = resolve_inputs( + { + "inputs": { + "count": {"type": "integer", "choices": [2, 4, 8], "default": 4}, + "mode": {"type": "string", "choices": ["a", "b"], "required": True}, + } + }, + [], + interactive=True, + ) + assert calls == [(["2", "4", "8"], 1), (["a", "b"], 0)] + assert values == {"count": 4, "mode": "a"} diff --git a/tests/test_progress.py b/tests/test_progress.py index d4deae5..5a8890f 100644 --- a/tests/test_progress.py +++ b/tests/test_progress.py @@ -25,10 +25,97 @@ def test_logged_command_progress_and_exit(tmp_path, capsys): interval=0.03, ) output = capsys.readouterr().out - assert "elapsed" in output and "completed" in output + assert "elapsed" not in output and "completed" in output + assert "elapsed" in log.with_name("build.progress.log").read_text() assert "build output" not in output assert "build output" in log.read_text() with pytest.raises(subprocess.CalledProcessError) as error: run_logged([sys.executable, "-c", "raise SystemExit(7)"], log) assert error.value.returncode == 7 assert "failed (exit 7)" in capsys.readouterr().out + + +@pytest.mark.parametrize("exit_code", [0, 7]) +def test_live_progress(tmp_path, monkeypatch, exit_code): + from prompt_toolkit.application import create_app_session + from prompt_toolkit.input import create_pipe_input + from prompt_toolkit.output import DummyOutput + + from abench import progress + + log = tmp_path / "console.log" + (tmp_path / "memory.csv").write_text( + "elapsed_seconds,current_bytes,peak_bytes\n1,1073741824,2147483648\n" + ) + monkeypatch.setattr(progress, "interactive_terminal", lambda: True) + frames = [] + original = progress.live_view + + def view(status, path): + app = original(status, path) + app.before_render += lambda _: frames.append( + (status(), progress.log_tail(path)) + ) + return app + + monkeypatch.setattr(progress, "live_view", view) + with ( + create_pipe_input() as pipe, + create_app_session(input=pipe, output=DummyOutput()), + ): + args = [ + sys.executable, + "-u", + "-c", + f"import time; print('model output'); time.sleep(.6); exit({exit_code})", + ] + if exit_code: + with pytest.raises(subprocess.CalledProcessError) as error: + run_logged(args, log, interval=0.03) + assert error.value.returncode == exit_code + else: + run_logged(args, log, interval=0.03) + assert any("model output" in text for _, text in frames) + assert any("memory 1.00 GiB (peak 2.00 GiB)" in status for status, _ in frames) + assert "elapsed" in log.with_name("console.progress.log").read_text() + + +def test_live_cancellation_reaps_process(tmp_path, monkeypatch): + from prompt_toolkit.application import create_app_session + from prompt_toolkit.input import create_pipe_input + from prompt_toolkit.output import DummyOutput + + from abench import progress + + monkeypatch.setattr(progress, "interactive_terminal", lambda: True) + processes = [] + original = subprocess.Popen + + def popen(*args, **kwargs): + process = original(*args, **kwargs) + processes.append(process) + return process + + monkeypatch.setattr(progress.subprocess, "Popen", popen) + with ( + create_pipe_input() as pipe, + create_app_session(input=pipe, output=DummyOutput()), + ): + pipe.send_text("\x03") + with pytest.raises(KeyboardInterrupt): + run_logged( + [sys.executable, "-c", "import time; time.sleep(30)"], + tmp_path / "console.log", + ) + assert processes[0].poll() is not None + assert "interrupted" in (tmp_path / "console.progress.log").read_text() + + +def test_tail_is_bounded_and_plain(tmp_path): + from abench.progress import log_tail + + log = tmp_path / "console.log" + assert log_tail(log) == "Waiting for log output…" + log.write_text("old\n" * 10000 + "new\x1b[2J\x00\nlast\n") + result = log_tail(log, lines=2) + assert result == "new[2J\nlast" diff --git a/tests/test_publishing.py b/tests/test_publishing.py new file mode 100644 index 0000000..b8d11d2 --- /dev/null +++ b/tests/test_publishing.py @@ -0,0 +1,460 @@ +"""Real artifact generation and simulated GitHub transport; never post in tests.""" + +import json +from pathlib import Path +from xml.etree import ElementTree as ET + +import pytest + +from abench import cli, publishing +from abench.common import read_json, write_json + +TARGET = {"repository": "ActivitySim/activitysim", "pr": 12, "baseline": "main"} +SNAPSHOT = {"headRefOid": "b" * 40, "baseRefOid": "a" * 40, "number": 12} +URL = "https://github.com/ActivitySim/activitysim/pull/12#issuecomment-123" + + +def record(path, *, seconds=10, valid=True, households=4): + phase = path / "measured" + phase.mkdir(parents=True) + write_json( + path / "experiment.json", + { + "schema_version": 2, + "label": path.name, + "households": households, + "sources": [ + { + "name": "activitysim", + "repository": TARGET["repository"], + "commit": "a" * 40, + } + ], + }, + ) + write_json( + phase / "status.json", + {"returncode": 0 if valid else 1, "elapsed_seconds": seconds}, + ) + write_json(phase / "docker-state.json", {"ExitCode": 0 if valid else 1}) + write_json(phase / "output-summary.json", {"households": {"rows": households}}) + (phase / "components-1.jsonl").write_text( + json.dumps({"component": "work", "seconds": seconds / 2, "succeeded": True}) + + "\n" + ) + (phase / "memory.csv").write_text( + "elapsed_seconds,current_bytes,peak_bytes\n0,536870912,2147483648\n5,1073741824,2147483648\n10,805306368,2147483648\n" + ) + + +def saved_suite(tmp_path, *, valid=True): + record(tmp_path / "main") + record(tmp_path / "candidate", seconds=8, valid=valid) + write_json( + tmp_path / "suite.json", + { + "configuration": {"publish": {"github": TARGET}}, + "publication_pr": SNAPSHOT, + "runs": [ + {"name": name, "output_dir": str(tmp_path / name)} + for name in ("main", "candidate", "unstarted") + ], + }, + ) + return tmp_path + + +@pytest.mark.parametrize( + "config", + [ + {}, + {"github": None}, + {"github": {**TARGET, "pr": True}}, + {"github": {**TARGET, "pr": 0}}, + {"github": {**TARGET, "repository": "https://github.com/a/b"}}, + {"github": {**TARGET, "baseline": "absent"}}, + {"github": {**TARGET, "baseline": []}}, + {"github": {**TARGET, "token": "secret"}}, + ], +) +def test_invalid_config(config): + with pytest.raises(ValueError): + publishing.configuration(config, ["main"]) + + +def test_bundle_uses_real_results(tmp_path, monkeypatch): + root = saved_suite(tmp_path) + monkeypatch.setattr( + publishing, "gh", lambda *a: pytest.fail("dry-run contacted GitHub") + ) + assert cli.main(["publish", str(root), "--dry-run"]) == 0 + bundle = root / "publication" + body = (bundle / "comment.md").read_text() + assert f"Auto-generated by **abench {publishing.__version__}**." in body + assert "experiment output directory" not in body + assert "-20.0%" in body and "NOT RUN / NO RESULTS" in body + assert SNAPSHOT["headRefOid"] in body and "a" * 40 in body + for name in ("memory.svg", "runtimes.svg"): + svg = ET.parse(bundle / name).getroot() + assert svg.tag == "{http://www.w3.org/2000/svg}svg" + nested = svg.findall(".//{http://www.w3.org/2000/svg}svg") + assert all("height" in element.attrib for element in nested) + before = read_json(bundle / "state.json")["id"] + publishing.publish(root, dry_run=True) + assert read_json(bundle / "state.json")["id"] == before + + +def test_invalid_runs_excluded(tmp_path): + root = saved_suite(tmp_path, valid=False) + bundle, _ = publishing.build_bundle(root) + body = (bundle / "comment.md").read_text() + assert "FAILED / INVALID" in body + assert "-20.0%" not in body + assert "candidate" not in (bundle / "runtimes.svg").read_text() + assert "candidate" in (bundle / "memory.svg").read_text() + + +def test_different_settings_suppress_delta(tmp_path): + root = saved_suite(tmp_path) + path = root / "candidate/experiment.json" + spec = read_json(path) + spec["multiprocess"] = True + write_json(path, spec) + bundle, _ = publishing.build_bundle(root) + assert "-20.0%" not in (bundle / "comment.md").read_text() + + +def fake_github(monkeypatch, *, comments=None, post_error=False, changed=False): + calls = [] + monkeypatch.setattr( + publishing, + "preflight", + lambda config: {**SNAPSHOT, "headRefOid": "c" * 40} if changed else SNAPSHOT, + ) + + def gh(*args, **kwargs): + calls.append(args) + if args[0] == "api": + return json.dumps([comments or []]) + assert args[:2] == ("pr", "comment") + if post_error: + raise ValueError("upload unavailable") + return URL + + monkeypatch.setattr(publishing, "gh", gh) + return calls + + +def test_post_and_repeat_are_idempotent(tmp_path, monkeypatch): + root = saved_suite(tmp_path) + calls = fake_github(monkeypatch, changed=True) + assert publishing.publish(root) == URL + assert publishing.publish(root) == URL + posts = [c for c in calls if c[0] == "pr"] + assert len(posts) == 1 + assert posts[0].count("--attach") == 2 + upload = Path(posts[0][posts[0].index("--body-file") + 1]).read_text() + assert f"Auto-generated by **abench {publishing.__version__}**." in upload + assert "PR head has changed" in upload + assert "./memory.svg" in upload + state = read_json(root / "publication/state.json") + assert state["comment_id"] == 123 and state["status"] == "published" + + +def test_retry_recovers_lost_response(tmp_path, monkeypatch): + root = saved_suite(tmp_path) + fake_github(monkeypatch, post_error=True) + with pytest.raises(ValueError, match="Results retained"): + publishing.publish(root) + state = read_json(root / "publication/state.json") + assert state["status"] == "uncertain" + calls = fake_github( + monkeypatch, + comments=[ + {"id": 123, "html_url": URL, "body": f""} + ], + ) + assert publishing.publish(root) == URL + assert all(c[0] == "api" for c in calls) + + +def test_retry_after_upload_failure(tmp_path, monkeypatch): + root = saved_suite(tmp_path) + fake_github(monkeypatch, post_error=True) + with pytest.raises(ValueError): + publishing.publish(root) + fake_github(monkeypatch) + assert publishing.publish(root) == URL + + +@pytest.mark.parametrize("attach,permission", [(False, "true"), (True, "false")]) +def test_preflight_rejects_unsupported_or_unauthorized(monkeypatch, attach, permission): + def gh(*args): + return ( + ("--attach" if attach else "old gh") if args[-1] == "--help" else permission + ) + + monkeypatch.setattr(publishing, "gh", gh) + with pytest.raises(ValueError): + publishing.preflight(TARGET) + + +def test_empty_suite_bundle(tmp_path): + root = saved_suite(tmp_path) + plan = read_json(root / "suite.json") + plan["runs"] = [{"name": "main", "output_dir": str(tmp_path / "missing")}] + write_json(root / "suite.json", plan) + bundle, _ = publishing.build_bundle(root) + assert "No valid runtime results" in (bundle / "runtimes.svg").read_text() + assert "NOT RUN" in (bundle / "comment.md").read_text() + + +def test_suite_preflight_precedes_assets_and_execution(tmp_path, monkeypatch): + from test_experiments import suite + + from abench import experiments + + path = suite(tmp_path, publish={"github": TARGET}) + monkeypatch.setattr( + publishing, + "preflight", + lambda _: (_ for _ in ()).throw(ValueError("auth failed")), + ) + monkeypatch.setattr( + experiments, + "prepare_assets", + lambda _: pytest.fail("prepared assets before preflight"), + ) + with pytest.raises(ValueError, match="auth failed"): + experiments.run_suite(path, lambda _: pytest.fail("ran without authentication")) + + +@pytest.mark.parametrize("dry_run,code", [(False, 0), (True, 0), (False, 7)]) +def test_suite_finalizes_before_publishing(tmp_path, monkeypatch, dry_run, code): + from test_experiments import suite + + from abench import experiments + + path = suite(tmp_path, publish={"github": TARGET}) + events = [] + monkeypatch.setattr( + publishing, "preflight", lambda _: events.append("preflight") or SNAPSHOT + ) + monkeypatch.setattr(experiments, "report", lambda *_: events.append("report")) + + def invoke(args): + if args[0] == "run": + output = Path(args[args.index("--output-dir") + 1]) + record(output, valid=not code) + events.append("run") + return code + return 0 + + def post(root, **kwargs): + assert events[-1] == "report" + assert kwargs["dry_run"] == dry_run + events.append("publish") + assert read_json(root / "suite.json").get("publication_pr") == ( + None if dry_run else SNAPSHOT + ) + + monkeypatch.setattr(publishing, "publish", post) + assert experiments.run_suite(path, invoke, publish_dry_run=dry_run) == code + assert ("preflight" in events) != dry_run + assert events[-1] == "publish" + + +@pytest.mark.parametrize("code", [0, 7]) +def test_posting_failure_preserves_benchmark_failure( + tmp_path, monkeypatch, capsys, code +): + from test_experiments import suite + + from abench import experiments + + path = suite(tmp_path, publish={"github": TARGET}) + monkeypatch.setattr(publishing, "preflight", lambda _: SNAPSHOT) + monkeypatch.setattr( + publishing, + "publish", + lambda *a, **kw: (_ for _ in ()).throw(ValueError("upload failed")), + ) + if code: + assert ( + experiments.run_suite(path, lambda args: code if args[0] == "run" else 0) + == code + ) + assert "Publication failed" in capsys.readouterr().out + else: + with pytest.raises(ValueError, match="upload failed"): + experiments.run_suite(path, lambda _: 0) + + +def test_transport_uses_argv_and_bundle_directory(tmp_path, monkeypatch): + from types import SimpleNamespace + + calls = [] + monkeypatch.setattr( + publishing.subprocess, + "run", + lambda args, **kwargs: ( + calls.append((args, kwargs)) or SimpleNamespace(stdout=URL) + ), + ) + assert publishing.gh("pr", "comment", "12", cwd=tmp_path) == URL + args, kwargs = calls[0] + assert args == ["gh", "pr", "comment", "12"] + assert kwargs["cwd"] == tmp_path and kwargs["check"] + assert not kwargs.get("shell") + + +def test_concurrent_publication_is_blocked(tmp_path, monkeypatch): + import fcntl + + root = saved_suite(tmp_path) + monkeypatch.setattr( + publishing, + "gh", + lambda *a: pytest.fail("concurrent publisher contacted GitHub"), + ) + with (root / ".publication.lock").open("w") as lock: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + with pytest.raises(ValueError, match="already running"): + publishing.publish(root) + + +def test_comment_labels_cannot_inject_markdown(): + text = publishing.cell( + "![unexpected](https://example.test/img) @someone |