diff --git a/src/stack_pr/autoland/__init__.py b/src/stack_pr/autoland/__init__.py index 43e0dd3..ecd490f 100644 --- a/src/stack_pr/autoland/__init__.py +++ b/src/stack_pr/autoland/__init__.py @@ -28,24 +28,31 @@ - ``checks``: evaluating CI checks. - ``plan``: generating, formatting, parsing, and replanning landing plans. - ``display``: rendering a plan's progress. +- ``repo``: the local clone (fetch, ancestry, landed SHAs, rebase + resubmit). - ``waits``: waiting for approval, checks, the merge queue, and workflows. - ``native_stack``: merging a run of land steps as one GitHub stack. -- ``services``: the tools a step uses as it executes (console, GitHub, the - waits, the repo), and their production bindings. +- ``services``: the tools a plan's execution uses (console, GitHub, sleep, + the repo, the waits), and their production bindings. - ``optimize``: execution units, the plan optimization policies that group steps into them (e.g. ``--merge-as-stack``), and the registry of policies. - ``engine``: optimizing and executing a plan. - ``status``: the ``--status`` report. - ``commands``: the subparser, and the fresh / resume / replan flows. -Convention for shared state and patch targets: a module-level singleton -(``runtime.console``, the ``gh.github`` client, ``runtime.HAVE_RICH``, -``display.GLYPHS``) lives in exactly one module, and other modules reach it -through that module -- ``from stack_pr.autoland import runtime``, then -``runtime.console`` -- never ``from stack_pr.autoland.runtime import console``. -The same goes for a function that tests replace to intercept calls made from -another module: ``runtime.run``, ``runtime.resilient_sleep``, the -``waits.wait_for_*`` helpers, ``engine.execute_plan``, the ``discovery`` +Executing a plan (the engine, the steps, the execution units, the waits, the +stack merge, and the repo operations) uses only the ``services.Services`` it +is given, so tests inject fakes there instead of patching modules. + +Convention for the shared state that remains, and its patch targets: a +module-level singleton (``runtime.console``, the ``gh.github`` client, +``runtime.HAVE_RICH``, ``display.GLYPHS``) lives in exactly one module, and +other modules reach it through that module -- ``from stack_pr.autoland import +runtime``, then ``runtime.console`` -- never ``from stack_pr.autoland.runtime +import console``. ``Services.production()`` binds the singletons as they are +when it is called; the command flows' own output, ``--status``, plan parsing, +discovery, and the shell runner use them directly. The same goes for a +function that tests replace to intercept calls made from another module: +``runtime.resilient_sleep``, ``engine.execute_plan``, the ``discovery`` functions, and ``stack_pr.git.get_current_branch_name``. Each therefore has one canonical patch target, e.g. ``mocker.patch("stack_pr.autoland.runtime.console")``, and a patch can't silently miss a stale copy bound in another module. (A name diff --git a/src/stack_pr/autoland/commands.py b/src/stack_pr/autoland/commands.py index 761d342..48036d3 100644 --- a/src/stack_pr/autoland/commands.py +++ b/src/stack_pr/autoland/commands.py @@ -243,6 +243,8 @@ def _land( opts: AutolandOptions, checkpointer: AutolandCheckpointer, worktree: Worktree | None, + *, + services: Services, ) -> None: """Execute the plan, report how the run ended, and exit if it didn't succeed. @@ -263,7 +265,7 @@ def _land( signal.signal(signal.SIGINT, signal.default_int_handler) try: success = engine.execute_plan( - ctx, common, opts, checkpointer, services=Services.production() + ctx, common, opts, checkpointer, services=services ) except KeyboardInterrupt: ctx.abort_reason = "User interrupted (Ctrl+C)" @@ -349,7 +351,16 @@ def _ask_replan_or_overwrite(state_path: Path) -> str | None: return None -def _run_fresh(common: cli.CommonArgs, opts: AutolandOptions) -> None: +def _run_fresh( + common: cli.CommonArgs, + opts: AutolandOptions, + *, + services: Services | None = None, +) -> None: + """Land a new plan. *services* (default: production) is what it is + executed with.""" + if services is None: + services = Services.production() branch = opts.branch or git.get_current_branch_name() state_path = opts.state_file or AutolandCheckpointer.default_path(branch) @@ -386,7 +397,7 @@ def _run_fresh(common: cli.CommonArgs, opts: AutolandOptions) -> None: return replan = choice == "replan" if replan: - _replan(common, opts, state_path) + _replan(common, opts, state_path, services=services) return if opts.branch: @@ -433,13 +444,13 @@ def _run_fresh(common: cli.CommonArgs, opts: AutolandOptions) -> None: ) print_status(ctx) - engine.print_execution_preview(ctx, opts) + engine.print_execution_preview(ctx, opts, ui=services.ui) if opts.dry_run: runtime.console.print("\n[yellow]Dry run — exiting.[/yellow]") return runtime.console.print(f"[dim]State file: {checkpointer.path}[/dim]\n") - _land(ctx, common, opts, checkpointer, worktree) + _land(ctx, common, opts, checkpointer, worktree, services=services) finally: if worktree is not None: worktree.remove_unless_landing() @@ -447,7 +458,16 @@ def _run_fresh(common: cli.CommonArgs, opts: AutolandOptions) -> None: lock.release() -def _run_resume(common: cli.CommonArgs, opts: AutolandOptions) -> None: +def _run_resume( + common: cli.CommonArgs, + opts: AutolandOptions, + *, + services: Services | None = None, +) -> None: + """Resume the run checkpointed in the state file. *services* (default: + production) is what it is executed with.""" + if services is None: + services = Services.production() sf_path = _state_path(opts) if not sf_path.exists(): @@ -502,9 +522,9 @@ def _run_resume(common: cli.CommonArgs, opts: AutolandOptions) -> None: return print_status(ctx) - engine.print_execution_preview(ctx, opts) + engine.print_execution_preview(ctx, opts, ui=services.ui) runtime.console.print(f"[dim]State file: {sf_path}[/dim]\n") - _land(ctx, common, opts, checkpointer, worktree) + _land(ctx, common, opts, checkpointer, worktree, services=services) finally: if worktree is not None: worktree.remove_unless_landing() @@ -634,12 +654,20 @@ def _replacement_plan( sys.exit(1) -def _replan(common: cli.CommonArgs, opts: AutolandOptions, state_path: Path) -> None: +def _replan( + common: cli.CommonArgs, + opts: AutolandOptions, + state_path: Path, + *, + services: Services | None = None, +) -> None: """Replace the plan of the run checkpointed at *state_path*, keep its - progress, and continue. + progress, and continue, executing with *services* (default: production). The caller holds the lock, except on a dry run, which only previews. """ + if services is None: + services = Services.production() try: old_checkpointer, old = AutolandCheckpointer.load(state_path) except (OSError, ValueError, KeyError) as e: @@ -678,7 +706,7 @@ def _replan(common: cli.CommonArgs, opts: AutolandOptions, state_path: Path) -> ctx = LandingContext(stack=stack, plan=plan) print_status(ctx) - engine.print_execution_preview(ctx, opts) + engine.print_execution_preview(ctx, opts, ui=services.ui) if lost: runtime.console.print( "\n[yellow]Done in the previous run, but not carried over (changed, " @@ -719,7 +747,7 @@ def _replan(common: cli.CommonArgs, opts: AutolandOptions, state_path: Path) -> path=state_path, branch=branch, base=common.target, plan_file=plan_file ) runtime.console.print(f"[dim]State file: {checkpointer.path}[/dim]\n") - _land(ctx, common, opts, checkpointer, worktree) + _land(ctx, common, opts, checkpointer, worktree, services=services) finally: if worktree is not None: worktree.remove_unless_landing() diff --git a/src/stack_pr/autoland/engine.py b/src/stack_pr/autoland/engine.py index 071abf8..7c7df73 100644 --- a/src/stack_pr/autoland/engine.py +++ b/src/stack_pr/autoland/engine.py @@ -14,13 +14,12 @@ import functools from stack_pr import cli -from stack_pr.autoland import runtime from stack_pr.autoland.display import _describe_step from stack_pr.autoland.model import LandingContext from stack_pr.autoland.optimize import PolicyRegistry, UnitContext, optimize_plan from stack_pr.autoland.options import AutolandOptions from stack_pr.autoland.plan import landed_prefix_end -from stack_pr.autoland.services import Services +from stack_pr.autoland.services import UI, Services from stack_pr.autoland.state import AutolandCheckpointer from stack_pr.autoland.steps import StepContext, StepResult @@ -98,13 +97,14 @@ def print_execution_preview( ctx: LandingContext, opts: AutolandOptions, *, + ui: UI, policies: PolicyRegistry | None = None, ) -> None: """Announce what the optimized plan does beyond executing its steps one by one. Derived from the same units ``execute_plan`` executes.""" for unit in optimize_plan(ctx, opts, policies): for line in unit.preview(): - runtime.console.print(line) + ui.print(line) def _record_landed_sha(ctx: LandingContext, sha: str) -> None: diff --git a/src/stack_pr/autoland/native_stack.py b/src/stack_pr/autoland/native_stack.py index 64f1ea3..e7b15c5 100644 --- a/src/stack_pr/autoland/native_stack.py +++ b/src/stack_pr/autoland/native_stack.py @@ -8,13 +8,16 @@ import contextlib from dataclasses import dataclass +from typing import TYPE_CHECKING -from stack_pr import cli -from stack_pr.autoland import gh, runtime, waits from stack_pr.autoland.model import LandingContext, PRState, StackEntry from stack_pr.autoland.options import AutolandOptions from stack_pr.autoland.waits import _poll_read +if TYPE_CHECKING: + from stack_pr import cli + from stack_pr.autoland.services import Services + # --------------------------------------------------------------------------- # Merging a run of land steps as one GitHub stack # --------------------------------------------------------------------------- @@ -29,7 +32,7 @@ def _open_native_stack_prs(native_stack: dict) -> list[int]: ] -def _native_stack_for(prs: list[int]) -> tuple[int | None, bool]: +def _native_stack_for(prs: list[int], services: Services) -> tuple[int | None, bool]: """Find or create a GitHub stack whose bottom open PRs are exactly *prs*. Returns ``(stack number, whether it holds just the run)``, with a ``None`` @@ -37,24 +40,23 @@ def _native_stack_for(prs: list[int]) -> tuple[int | None, bool]: run is ours to dissolve if the merge falls through, even when it predates this call: it is most likely one an interrupted run left behind. """ + ui, github = services.ui, services.github try: - existing = gh.github.find_native_stack(prs[0]) + existing = github.find_native_stack(prs[0]) except RuntimeError as e: - runtime.console.print(f"[yellow]Could not look up GitHub stacks: {e}[/yellow]") + ui.print(f"[yellow]Could not look up GitHub stacks: {e}[/yellow]") return None, False if existing is None: try: - return gh.github.create_native_stack(prs)["number"], True + return github.create_native_stack(prs)["number"], True except (RuntimeError, KeyError) as e: # The POST may have created the stack even though it looked failed. with contextlib.suppress(RuntimeError): - existing = gh.github.find_native_stack(prs[0]) + existing = github.find_native_stack(prs[0]) if existing is not None and _open_native_stack_prs(existing) == prs: return existing["number"], True - runtime.console.print( - f"[yellow]Could not create a GitHub stack: {e}[/yellow]" - ) + ui.print(f"[yellow]Could not create a GitHub stack: {e}[/yellow]") return None, False # A stack made earlier — by an interrupted run, or by hand with gh stack — @@ -63,7 +65,7 @@ def _native_stack_for(prs: list[int]) -> tuple[int | None, bool]: open_prs = _open_native_stack_prs(existing) if open_prs[: len(prs)] == prs: return existing["number"], open_prs == prs - runtime.console.print( + ui.print( f"[yellow]PR #{prs[0]} is already in GitHub stack #{existing['number']}, " "which does not match the plan.[/yellow]" ) @@ -83,6 +85,7 @@ def _await_native_stack_merge( uuid: str | None, *, opts: AutolandOptions, + services: Services, ) -> str: """Wait for a requested stack merge to finish. @@ -90,6 +93,7 @@ def _await_native_stack_merge( have merged some of the PRs: GitHub stops a stack merge at the PR that failed, and keeps the ones below it. """ + github = services.github top = entries[-1] awake_elapsed = 0.0 while True: @@ -103,7 +107,7 @@ def _await_native_stack_merge( for entry in entries: if entry.state == PRState.MERGED: continue - state = _poll_read(gh.github.pr_state, entry.pr_number) + state = _poll_read(github.pr_state, entry.pr_number, services) unread = unread or state is None if state == "MERGED": entry.state = PRState.MERGED @@ -118,7 +122,7 @@ def _await_native_stack_merge( status, message = "", "" if uuid: with contextlib.suppress(RuntimeError): - status, message = gh.github.merge_async_status(top.pr_number, uuid) + status, message = github.merge_async_status(top.pr_number, uuid) if status == "failed": return message or "GitHub reported the stack merge as failed" # Every PR enters the queue together, and GitHub drops the PRs above any @@ -129,18 +133,18 @@ def _await_native_stack_merge( status == "enqueued" or (not uuid and awake_elapsed > 0) ) # A failed lookup (None) leaves the merge undecided; keep polling. - if settled and gh.github.in_merge_queue(still_open[0].pr_number) is False: + if settled and github.in_merge_queue(still_open[0].pr_number) is False: return f"PR #{still_open[0].pr_number} was booted from the merge queue" mins = int(awake_elapsed) // 60 for entry in still_open: entry.error_message = f"Merging as a stack ({mins}m elapsed)..." - runtime.console.print( + services.ui.print( f"[dim]Stack of {len(entries)} PRs up to #{top.pr_number}: " f"{status or 'merging'} ({mins}m) — polling in {opts.poll_interval}s" "[/dim]" ) - runtime.resilient_sleep(opts.poll_interval) + services.sleep(opts.poll_interval) awake_elapsed += opts.poll_interval @@ -150,6 +154,7 @@ def land_as_native_stack( ctx: LandingContext, common: cli.CommonArgs, opts: AutolandOptions, + services: Services, ) -> NativeStackMergeResult: """Land a run of consecutive PRs (bottom first) with one stack merge. @@ -164,18 +169,22 @@ def land_as_native_stack( just the run is dissolved first, since GitHub only lets stacked PRs merge through the stack merge API. """ + ui, github, waits, repo = ( + services.ui, + services.github, + services.waits, + services.repo, + ) prs = [e.pr_number for e in entries] top = entries[-1] - runtime.console.print( + ui.print( f"\n{'=' * 60}\n[bold]Landing {len(entries)} PRs as a GitHub stack: " f"{', '.join(f'#{n}' for n in prs)}[/bold]\n{'=' * 60}" ) - native_stack_number, dissolvable = _native_stack_for(prs) + native_stack_number, dissolvable = _native_stack_for(prs, services) if native_stack_number is None: - runtime.console.print( - "[yellow]Landing these PRs one at a time instead.[/yellow]" - ) + ui.print("[yellow]Landing these PRs one at a time instead.[/yellow]") return NativeStackMergeResult() failure = "" @@ -210,7 +219,7 @@ def land_as_native_stack( if entry.state != PRState.MERGED: entry.state = PRState.IN_MERGE_QUEUE entry.error_message = "Merging as a stack..." - runtime.console.print( + ui.print( f"\n[bold cyan]Merging PRs #{bottom.pr_number}-#{top.pr_number} " f"as a stack[/bold cyan]" ) @@ -224,7 +233,7 @@ def land_as_native_stack( above = ctx.stack.index(top) + 1 if dissolvable and above < len(ctx.stack): try: - gh.github.set_base(ctx.stack[above].pr_number, common.target) + github.set_base(ctx.stack[above].pr_number, common.target) except RuntimeError as e: failure = ( f"could not base PR #{ctx.stack[above].pr_number} on " @@ -234,8 +243,8 @@ def land_as_native_stack( try: # Ask GitHub rather than trust the config: the request body must # match whether the branch really has a queue (see merge_async). - has_queue = gh.github.has_merge_queue(common.target) - uuid = gh.github.merge_async( + has_queue = github.has_merge_queue(common.target) + uuid = github.merge_async( top.pr_number, merge_queue=opts.merge_queue if has_queue is None else has_queue, ) @@ -245,22 +254,24 @@ def land_as_native_stack( if "HTTP 409" not in str(e): failure = f"stack merge request failed: {e}" if not failure: - failure = _await_native_stack_merge(entries, uuid, opts=opts) + failure = _await_native_stack_merge( + entries, uuid, opts=opts, services=services + ) landed = all(e.state == PRState.MERGED for e in entries) if not landed and dissolvable: # Leave no stack behind: landing one PR at a time only works on PRs # that aren't in one. Queued PRs stay queued. try: - gh.github.unstack_native_stack(native_stack_number) + github.unstack_native_stack(native_stack_number) except RuntimeError as e: - runtime.console.print( + ui.print( f"[yellow]Could not dissolve GitHub stack #{native_stack_number}: {e}[/yellow]" ) merged = [e for e in entries if e.state == PRState.MERGED] if failure and not abort_reason: - runtime.console.print( + ui.print( f"[yellow]Stack merge did not complete ({failure}). Landing the " "remaining PRs one at a time.[/yellow]" ) @@ -269,15 +280,17 @@ def land_as_native_stack( entry.state = PRState.PENDING entry.error_message = f"Stack merge: {failure}" elif landed: - runtime.console.print( + ui.print( f"\n[bold green]PRs {', '.join(f'#{n}' for n in prs)} merged![/bold green]" ) if merged: - waits._refresh_last_landed_sha(ctx, common, merged[-1].pr_number) + landed_sha = repo.landed_sha(common, merged[-1].pr_number) + if landed_sha: + ctx.last_landed_sha = landed_sha if ctx.stack.index(merged[-1]) < len(ctx.stack) - 1: try: - waits.rebase_and_resubmit(common) + repo.rebase_and_resubmit(common) except Exception as e: # noqa: BLE001 - report any resubmit failure return NativeStackMergeResult( landed=landed, diff --git a/src/stack_pr/autoland/optimize/stack_merge.py b/src/stack_pr/autoland/optimize/stack_merge.py index 97c13cf..6c985a9 100644 --- a/src/stack_pr/autoland/optimize/stack_merge.py +++ b/src/stack_pr/autoland/optimize/stack_merge.py @@ -86,7 +86,11 @@ def execute(self, uc: UnitContext) -> str: entries = stack_merge_run(uc.run, range(index, self.span.stop)) if entries: result = land_as_native_stack( - entries, ctx=uc.run, common=uc.common, opts=uc.opts + entries, + ctx=uc.run, + common=uc.common, + opts=uc.opts, + services=uc.services, ) if result.abort_reason: return result.abort_reason diff --git a/src/stack_pr/autoland/repo.py b/src/stack_pr/autoland/repo.py new file mode 100644 index 0000000..30f5cab --- /dev/null +++ b/src/stack_pr/autoland/repo.py @@ -0,0 +1,114 @@ +"""The local clone: fetching, ancestry, landed SHAs, and rebase + resubmit. + +``GitRepo`` is the production ``services.Repo``. It runs git in the current +working directory (autoland chdirs into a temporary worktree when ``--branch`` +is used) and stack-pr's own submit for the resubmit. +""" + +from __future__ import annotations + +import contextlib +from dataclasses import replace +from typing import TYPE_CHECKING + +from stack_pr import cli +from stack_pr.autoland import runtime +from stack_pr.git import GitError, is_ancestor + +if TYPE_CHECKING: + from stack_pr.autoland import gh + from stack_pr.autoland.services import UI + + +class GitRepo: + """The clone autoland runs in, and the stack-pr operations on it.""" + + def __init__(self, github: gh.GitHub, ui: UI) -> None: + self._github = github + self._ui = ui + + def fetch_target(self, common: cli.CommonArgs) -> None: + """Fetch ``/``. Best-effort: a failure is ignored.""" + with contextlib.suppress(RuntimeError): # fetch is non-critical + runtime.run(["git", "fetch", common.remote, common.target], quiet=True) + + def is_ancestor(self, ancestor: str, descendant: str) -> bool | None: + """Whether ``ancestor`` is an ancestor of ``descendant``, per this clone. + + ``None`` means the local repo cannot answer, typically because either + commit is missing from this clone. That must not be conflated with a + genuine "no" — a workflow run's head commit is routinely absent here, + because ``origin/`` advances while we poll. + """ + try: + return is_ancestor(ancestor, descendant) + except GitError: + return None + + def landed_sha(self, common: cli.CommonArgs, pr_number: int | None = None) -> str: + """The SHA of the most recently landed code, or "" if it can't be found. + + When ``pr_number`` is given, prefer that PR's exact merge commit so the + workflow checkpoint targets the code we actually landed. In a busy + repo, ``origin/`` can advance past our merge commit (bot + commits, other PRs) between the merge and this call, so using its HEAD + would overshoot and make the checkpoint wait for a deploy of code that + was never our own. + """ + self.fetch_target(common) + + if pr_number is not None: + merge_sha = self._github.merge_commit(pr_number) + if merge_sha: + return merge_sha + + try: + result = runtime.run( + ["git", "rev-parse", f"{common.remote}/{common.target}"], quiet=True + ) + except RuntimeError: + return "" # non-critical; will retry when needed + return result.stdout.strip() + + def rebase_and_resubmit(self, common: cli.CommonArgs) -> None: + """After a merge, rebase the local stack on the target and re-submit.""" + self._ui.print( + f"\n[bold]Rebasing stack on {common.remote}/{common.target}...[/bold]" + ) + runtime.run(["git", "fetch", common.remote, common.target], quiet=False) + # Rebase the current branch (don't name it) so this works even when the + # branch is checked out in another worktree. + try: + runtime.run( + ["git", "rebase", f"{common.remote}/{common.target}"], + quiet=False, + retries=0, + ) + except RuntimeError: + # Without --branch this is the user's own working copy; don't leave + # it stuck mid-rebase. Best-effort: `--abort` is harmless (and + # fails quietly) if no rebase is actually in progress. + runtime.run( + ["git", "rebase", "--abort"], check=False, quiet=True, retries=0 + ) + raise + + # Re-deduce the base against the *current* origin/. + # `common.base` was deduced once when autoland started (merge-base with + # the target at that time). After other PRs land in the target and we + # rebase onto it, that cached base is stale: the range base..HEAD would + # then sweep in every commit merged by others in the meantime, and + # submit would try to open bogus PRs for them. Clearing the base forces + # deduce_base to recompute merge-base(HEAD, origin/) from the + # freshly rebased HEAD. + resubmit_common = cli.deduce_base(replace(common, base="")) + + self._ui.print("[bold]Re-submitting stack...[/bold]") + cli.command_submit( + resubmit_common, + draft=False, + reviewer="", + keep_body=True, + keep_title=True, + draft_bitmask=None, + ) diff --git a/src/stack_pr/autoland/services.py b/src/stack_pr/autoland/services.py index fbd48b3..6709954 100644 --- a/src/stack_pr/autoland/services.py +++ b/src/stack_pr/autoland/services.py @@ -1,30 +1,35 @@ """The tools a plan step may use while it executes, injected through ``Services``. -A step reaches the console, GitHub, the wait loops, and the local repo only -through the ``Services`` in its ``StepContext``, never through module globals, -so a step can be run in isolation against fakes. - -``Services.production()`` binds these to today's implementations. Each one -looks its target up through the module attribute *at call time* (``runtime.console``, -``gh.github``, ``waits.wait_for_approval``, ...) rather than holding on to the -object it found at construction, so that tests which patch those attributes -keep intercepting every call a step makes. Once tests inject fake services -directly, the delegation can bind the objects themselves. +A step, an execution unit, and everything they call while landing (the wait +loops, the stack merge, the landed-SHA lookup, rebase + resubmit) reach the +console, GitHub, the clock, and the local repo only through one ``Services``, +never through module globals, so any of them can be run against fakes. + +``Services.production()`` binds the real implementations: the ``gh.github`` +client, the ``runtime.console``, ``runtime.resilient_sleep``, a ``GitRepo``, +and the polling waits of ``autoland.waits``. + +Module singletons remain only for what runs outside a plan's execution: the +command flows' own output and prompts (``commands``), ``--status``, plan +parsing and the discovery of the stack (which ask ``gh.github``), and the +shell runner (``runtime.run``), which echoes each command it runs to +``runtime.console``. """ from __future__ import annotations from collections.abc import Callable from dataclasses import dataclass -from typing import TYPE_CHECKING, Protocol, cast +from typing import TYPE_CHECKING, Any, Protocol from stack_pr.autoland import display, gh, runtime, waits -from stack_pr.autoland.model import LandingContext +from stack_pr.autoland.repo import GitRepo if TYPE_CHECKING: from stack_pr import cli - from stack_pr.autoland.model import StackEntry, WorkflowStep + from stack_pr.autoland.model import LandingContext, StackEntry, WorkflowStep from stack_pr.autoland.options import AutolandOptions + from stack_pr.autoland.waits import MergeableResult class UI(Protocol): @@ -46,6 +51,10 @@ class Waits(Protocol): Each returns whether it succeeded, and records its progress on the entry or step it is given. Ctrl+C while one is waiting raises KeyboardInterrupt out of it, which the run handles at its boundary (``commands._land``). + + The production waits (``PollingWaits``) poll GitHub through the other + services. This is a service of its own so that a step, or a stack merge, + can be tested against waits that just succeed or fail. """ def wait_for_approval( @@ -54,6 +63,10 @@ def wait_for_approval( def wait_for_checks(self, entry: StackEntry, *, opts: AutolandOptions) -> bool: ... + def wait_for_mergeable( + self, entry: StackEntry, *, opts: AutolandOptions + ) -> MergeableResult: ... + def enqueue_and_wait(self, entry: StackEntry, *, opts: AutolandOptions) -> bool: ... def wait_for_workflow( @@ -72,6 +85,13 @@ def wait_for_workflow( class Repo(Protocol): """The local clone and the stack-pr operations on it.""" + def fetch_target(self, common: cli.CommonArgs) -> None: + """Fetch the target branch. Best-effort: never raises.""" + + def is_ancestor(self, ancestor: str, descendant: str) -> bool | None: + """Whether *ancestor* is an ancestor of (or is) *descendant*, per this + clone. ``None`` if the clone can't tell (e.g. a commit is missing).""" + def landed_sha(self, common: cli.CommonArgs, pr_number: int | None = None) -> str: """The commit of the code landed so far: *pr_number*'s merge commit, or else the head of the target branch. "" if it can't be found.""" @@ -83,64 +103,81 @@ def rebase_and_resubmit(self, common: cli.CommonArgs) -> None: @dataclass(frozen=True) class Services: - """Everything a step may use beyond its ``StepContext``'s run state.""" + """Everything a step may use beyond its ``StepContext``'s run state. + + ``waits`` defaults to the real polling waits over these services' GitHub, + UI, sleep, and repo. They always use the ``Services`` they belong to, so + ``dataclasses.replace`` of, say, ``github`` gives waits that poll the new + one. + """ github: gh.GitHub ui: UI - waits: Waits - repo: Repo # Sleep for some seconds, surviving system sleep; returns the time the # machine spent suspended. sleep: Callable[[int], float] + repo: Repo + waits: Waits = None # type: ignore[assignment] # set in __post_init__ + + def __post_init__(self) -> None: + if self.waits is None or ( + isinstance(self.waits, PollingWaits) and self.waits.services is not self + ): + object.__setattr__(self, "waits", PollingWaits(self)) @classmethod def production(cls) -> Services: - """The real console, GitHub client, wait loops, and repo.""" + """The real console, GitHub client, clock, repo, and wait loops.""" + ui = ConsoleUI(runtime.console) return cls( - github=cast("gh.GitHub", _LiveGitHub()), - ui=_ConsoleUI(), - waits=_ModuleWaits(), - repo=_GitRepo(), - sleep=_resilient_sleep, + github=gh.github, + ui=ui, + sleep=runtime.resilient_sleep, + repo=GitRepo(gh.github, ui), ) -# --------------------------------------------------------------------------- -# Production implementations: thin delegates resolved at call time -# --------------------------------------------------------------------------- - +class _Console(Protocol): + def print(self, *objects: Any) -> None: ... # noqa: ANN401 -class _LiveGitHub: - """Forwards every attribute to whatever ``gh.github`` is when it is used.""" + def input(self, prompt: str) -> str: ... - def __getattr__(self, name: str) -> object: - return getattr(gh.github, name) +class ConsoleUI: + """A ``UI`` on a rich (or plain) console.""" -def _resilient_sleep(seconds: int) -> float: - return runtime.resilient_sleep(seconds) + def __init__(self, console: _Console) -> None: + self._console = console - -class _ConsoleUI: def print(self, *objects: object) -> None: - runtime.console.print(*objects) + self._console.print(*objects) def input(self, prompt: str) -> str: - return runtime.console.input(prompt) + return self._console.input(prompt) def escape(self, text: str) -> str: return display._escape_markup(text) -class _ModuleWaits: +class PollingWaits: + """The waits of ``autoland.waits``, polling through *services*.""" + + def __init__(self, services: Services) -> None: + self.services = services + def wait_for_approval(self, entry: StackEntry, *, opts: AutolandOptions) -> bool: - return waits.wait_for_approval(entry, opts=opts) + return waits.wait_for_approval(entry, opts=opts, services=self.services) def wait_for_checks(self, entry: StackEntry, *, opts: AutolandOptions) -> bool: - return waits.wait_for_checks(entry, opts=opts) + return waits.wait_for_checks(entry, opts=opts, services=self.services) + + def wait_for_mergeable( + self, entry: StackEntry, *, opts: AutolandOptions + ) -> MergeableResult: + return waits.wait_for_mergeable(entry, opts=opts, services=self.services) def enqueue_and_wait(self, entry: StackEntry, *, opts: AutolandOptions) -> bool: - return waits.enqueue_and_wait(entry, opts=opts) + return waits.enqueue_and_wait(entry, opts=opts, services=self.services) def wait_for_workflow( self, @@ -152,22 +189,10 @@ def wait_for_workflow( ctx: LandingContext, ) -> bool: return waits.wait_for_workflow( - step, opts=opts, common=common, ctx=ctx, target_sha=target_sha + step, + opts=opts, + common=common, + ctx=ctx, + target_sha=target_sha, + services=self.services, ) - - -class _GitRepo: - def landed_sha(self, common: cli.CommonArgs, pr_number: int | None = None) -> str: - # Goes through waits._refresh_last_landed_sha, which records into a - # context, so that tests patching it keep intercepting the lookup. A - # scratch context keeps the run's own state untouched: the engine - # records what the step reports. - scratch = LandingContext() - if pr_number is None: - waits._refresh_last_landed_sha(scratch, common) - else: - waits._refresh_last_landed_sha(scratch, common, pr_number) - return scratch.last_landed_sha - - def rebase_and_resubmit(self, common: cli.CommonArgs) -> None: - waits.rebase_and_resubmit(common) diff --git a/src/stack_pr/autoland/waits.py b/src/stack_pr/autoland/waits.py index 609fdf1..31cbe71 100644 --- a/src/stack_pr/autoland/waits.py +++ b/src/stack_pr/autoland/waits.py @@ -1,115 +1,32 @@ -"""Waiting for approval, checks, the merge queue, and workflows; rebase + resubmit.""" +"""Waiting for approval, checks, the merge queue, and workflows. + +Every wait polls through the ``Services`` it is given: GitHub, the console, +the clock, and the local repo. ``services.PollingWaits`` binds them as the +production ``Waits``. +""" from __future__ import annotations import contextlib from collections.abc import Callable -from dataclasses import dataclass, replace -from typing import TypeVar +from dataclasses import dataclass +from typing import TYPE_CHECKING, TypeVar -from stack_pr import cli -from stack_pr.autoland import gh, runtime from stack_pr.autoland.checks import CheckStatus, evaluate_checks from stack_pr.autoland.display import _REVIEW_DECISION_DISPLAY from stack_pr.autoland.gh import MergeQueuePollResult, _sha_eq from stack_pr.autoland.model import LandingContext, PRState, StackEntry, WorkflowStep from stack_pr.autoland.options import AutolandOptions -from stack_pr.git import GitError, is_ancestor - -# --------------------------------------------------------------------------- -# Post-merge rebase + resubmit (reuses stack-pr submit) -# --------------------------------------------------------------------------- - - -def rebase_and_resubmit(common: cli.CommonArgs) -> None: - """After a merge, rebase the local stack on the target and re-submit.""" - runtime.console.print( - f"\n[bold]Rebasing stack on {common.remote}/{common.target}...[/bold]" - ) - runtime.run(["git", "fetch", common.remote, common.target], quiet=False) - # Rebase the current branch (don't name it) so this works even when the - # branch is checked out in another worktree. - try: - runtime.run( - ["git", "rebase", f"{common.remote}/{common.target}"], - quiet=False, - retries=0, - ) - except RuntimeError: - # Without --branch this is the user's own working copy; don't leave it - # stuck mid-rebase. Best-effort: `--abort` is harmless (and fails - # quietly) if no rebase is actually in progress. - runtime.run(["git", "rebase", "--abort"], check=False, quiet=True, retries=0) - raise - - # Re-deduce the base against the *current* origin/. `common.base` - # was deduced once when autoland started (merge-base with the target at - # that time). After other PRs land in the target and we rebase onto it, - # that cached base is stale: the range base..HEAD would then sweep in every - # commit merged by others in the meantime, and submit would try to open - # bogus PRs for them. Clearing the base forces deduce_base to recompute - # merge-base(HEAD, origin/) from the freshly rebased HEAD. - resubmit_common = cli.deduce_base(replace(common, base="")) - - runtime.console.print("[bold]Re-submitting stack...[/bold]") - cli.command_submit( - resubmit_common, - draft=False, - reviewer="", - keep_body=True, - keep_title=True, - draft_bitmask=None, - ) +if TYPE_CHECKING: + from stack_pr import cli + from stack_pr.autoland.services import Repo, Services # --------------------------------------------------------------------------- # Workflow checkpoint polling # --------------------------------------------------------------------------- -def _refresh_last_landed_sha( - ctx: LandingContext, common: cli.CommonArgs, pr_number: int | None = None -) -> None: - """Record the SHA of the most recently landed code. - - When ``pr_number`` is given, prefer that PR's exact merge commit so the - workflow checkpoint targets the code we actually landed. In a busy repo, - ``origin/`` can advance past our merge commit (bot commits, other - PRs) between the merge and this call, so using its HEAD would overshoot and - make the checkpoint wait for a deploy of code that was never our own. - """ - with contextlib.suppress(RuntimeError): # fetch is non-critical - runtime.run(["git", "fetch", common.remote, common.target], quiet=True) - - if pr_number is not None: - merge_sha = gh.github.merge_commit(pr_number) - if merge_sha: - ctx.last_landed_sha = merge_sha - return - - try: - result = runtime.run( - ["git", "rev-parse", f"{common.remote}/{common.target}"], quiet=True - ) - ctx.last_landed_sha = result.stdout.strip() - except RuntimeError: - pass # non-critical; will retry when needed - - -def _local_is_ancestor(ancestor: str, descendant: str) -> bool | None: - """Whether ``ancestor`` is an ancestor of ``descendant``, per this clone. - - ``None`` means the local repo cannot answer, typically because either - commit is missing from this clone. That must not be conflated with a - genuine "no" — a workflow run's head commit is routinely absent here, - because ``origin/`` advances while we poll. - """ - try: - return is_ancestor(ancestor, descendant) - except GitError: - return None - - class _Ancestry: """Answers "does this workflow run's commit include the code we landed?". @@ -123,8 +40,10 @@ class _Ancestry: the wait; the poll loop re-examines the same runs every interval. """ - def __init__(self, common: cli.CommonArgs) -> None: + def __init__(self, common: cli.CommonArgs, services: Services) -> None: self._common = common + self._repo: Repo = services.repo + self._github = services.github self._verdicts: dict[tuple[str, str], bool] = {} self._fetched_for: set[str] = set() @@ -139,28 +58,21 @@ def contains(self, ancestor: str, descendant: str) -> bool | None: if key in self._verdicts: return self._verdicts[key] - verdict = _local_is_ancestor(ancestor, descendant) + verdict = self._repo.is_ancestor(ancestor, descendant) if verdict is None and descendant not in self._fetched_for: # A commit is missing locally. Fetch the target branch once for # this commit and retry; fetching again for the same commit would # not tell us anything new. self._fetched_for.add(descendant) - self._fetch_target() - verdict = _local_is_ancestor(ancestor, descendant) + self._repo.fetch_target(self._common) + verdict = self._repo.is_ancestor(ancestor, descendant) if verdict is None: - verdict = gh.github.contains(ancestor, descendant) + verdict = self._github.contains(ancestor, descendant) if verdict is not None: self._verdicts[key] = verdict return verdict - def _fetch_target(self) -> None: - with contextlib.suppress(RuntimeError): # fetch is non-critical - runtime.run( - ["git", "fetch", self._common.remote, self._common.target], - quiet=True, - ) - def wait_for_workflow( step: WorkflowStep, @@ -169,20 +81,22 @@ def wait_for_workflow( common: cli.CommonArgs, ctx: LandingContext, target_sha: str | None = None, + services: Services, ) -> bool: """Wait for a workflow to complete with code at or after *target_sha*. *target_sha* defaults to the run's last landed SHA. """ + ui, github = services.ui, services.github if target_sha is None: target_sha = ctx.last_landed_sha step.state = "waiting" - runtime.console.print( + ui.print( f"\n[bold blue]Waiting for workflow: {step.workflow}[/bold blue]" f"\n[dim]Target SHA: {target_sha[:12]}[/dim]" ) - ancestry = _Ancestry(common) + ancestry = _Ancestry(common, services) awake_elapsed = 0.0 while True: if awake_elapsed > opts.workflow_timeout: @@ -193,12 +107,10 @@ def wait_for_workflow( return False try: - data = gh.github.workflow_runs(step.workflow, common.target) + data = github.workflow_runs(step.workflow, common.target) except RuntimeError as e: - runtime.console.print( - f"[yellow]Warning: could not poll workflow: {e}[/yellow]" - ) - runtime.resilient_sleep(opts.poll_interval) + ui.print(f"[yellow]Warning: could not poll workflow: {e}[/yellow]") + services.sleep(opts.poll_interval) awake_elapsed += opts.poll_interval continue @@ -212,7 +124,7 @@ def wait_for_workflow( continue verdict = ancestry.contains(target_sha, run_sha) if verdict is None: - runtime.console.print( + ui.print( f"[yellow]Warning: could not tell whether run " f"{run_sha[:12]} includes {target_sha[:12]}; " "will retry[/yellow]" @@ -221,7 +133,7 @@ def wait_for_workflow( if verdict: step.state = "succeeded" step.error_message = "" - runtime.console.print( + ui.print( f"\n[bold green]Workflow {step.workflow} completed " f"with SHA {run_sha[:12]}[/bold green]" ) @@ -229,11 +141,11 @@ def wait_for_workflow( mins = int(awake_elapsed) // 60 step.error_message = f"Waiting for workflow ({mins}m elapsed)..." - runtime.console.print( + ui.print( f"[dim]Workflow {step.workflow}: waiting ({mins}m) — " f"polling in {opts.poll_interval}s[/dim]" ) - runtime.resilient_sleep(opts.poll_interval) + services.sleep(opts.poll_interval) awake_elapsed += opts.poll_interval @@ -245,7 +157,9 @@ def wait_for_workflow( _T = TypeVar("_T") -def _poll_read(read: Callable[[int], _T], pr_number: int) -> _T | None: +def _poll_read( + read: Callable[[int], _T], pr_number: int, services: Services +) -> _T | None: """One poll's read of a PR from GitHub, or ``None`` if it failed. A long wait shouldn't die on one bad response from GitHub, whether the @@ -256,27 +170,30 @@ def _poll_read(read: Callable[[int], _T], pr_number: int) -> _T | None: try: return read(pr_number) except RuntimeError as e: # includes GitHubError - runtime.console.print( + services.ui.print( f"[yellow]Warning: could not poll PR #{pr_number}: {e}; will retry[/yellow]" ) return None -def _refresh_review(entry: StackEntry) -> bool: +def _refresh_review(entry: StackEntry, services: Services) -> bool: """Update the entry's review decision, tolerating a transient gh failure. Returns whether the entry is now approved. """ with contextlib.suppress(RuntimeError): - entry.review_decision = gh.github.review_decision(entry.pr_number) + entry.review_decision = services.github.review_decision(entry.pr_number) return entry.is_approved -def wait_for_approval(entry: StackEntry, *, opts: AutolandOptions) -> bool: +def wait_for_approval( + entry: StackEntry, *, opts: AutolandOptions, services: Services +) -> bool: """Wait until the PR has required approvals (or merged). Returns False if the PR was closed.""" + ui, github = services.ui, services.github # An unknown state (a failed read) is settled by the checks wait after this. - pr_state = _poll_read(gh.github.pr_state, entry.pr_number) + pr_state = _poll_read(github.pr_state, entry.pr_number, services) if pr_state == "MERGED": entry.state = PRState.MERGED return True @@ -285,25 +202,25 @@ def wait_for_approval(entry: StackEntry, *, opts: AutolandOptions) -> bool: entry.error_message = "PR was closed" return False - if _refresh_review(entry): + if _refresh_review(entry, services): return True entry.state = PRState.WAITING_FOR_APPROVAL label, _ = _REVIEW_DECISION_DISPLAY.get(entry.review_decision, ("not approved", "")) entry.error_message = label - runtime.console.print( + ui.print( f"[magenta]PR #{entry.pr_number} is not yet approved " f"({entry.review_decision or 'REVIEW_REQUIRED'}). Waiting...[/magenta]" ) while True: - runtime.console.print( + ui.print( f"[dim]PR #{entry.pr_number}: waiting for approval — " f"polling in {opts.poll_interval}s[/dim]" ) - runtime.resilient_sleep(opts.poll_interval) + services.sleep(opts.poll_interval) - pr_state = _poll_read(gh.github.pr_state, entry.pr_number) + pr_state = _poll_read(github.pr_state, entry.pr_number, services) if pr_state is None: continue if pr_state == "MERGED": @@ -314,26 +231,27 @@ def wait_for_approval(entry: StackEntry, *, opts: AutolandOptions) -> bool: entry.error_message = "PR was closed" return False - if _refresh_review(entry): + if _refresh_review(entry, services): entry.error_message = "" - runtime.console.print( - f"[green]PR #{entry.pr_number} is now approved[/green]" - ) + ui.print(f"[green]PR #{entry.pr_number} is now approved[/green]") return True if entry.review_decision == "CHANGES_REQUESTED": entry.error_message = "Changes requested — cannot proceed" - runtime.console.print( + ui.print( f"[red]PR #{entry.pr_number} has changes requested. " "Resolve review comments and re-request review.[/red]" ) -def wait_for_checks(entry: StackEntry, *, opts: AutolandOptions) -> bool: +def wait_for_checks( + entry: StackEntry, *, opts: AutolandOptions, services: Services +) -> bool: """Wait for required checks to pass. Returns True on success.""" + ui, github = services.ui, services.github entry.state = PRState.WAITING_FOR_CHECKS while True: - pr_state = _poll_read(gh.github.pr_state, entry.pr_number) + pr_state = _poll_read(github.pr_state, entry.pr_number, services) if pr_state == "MERGED": entry.state = PRState.MERGED return True @@ -343,10 +261,12 @@ def wait_for_checks(entry: StackEntry, *, opts: AutolandOptions) -> bool: return False checks = ( - None if pr_state is None else _poll_read(gh.github.checks, entry.pr_number) + None + if pr_state is None + else _poll_read(github.checks, entry.pr_number, services) ) if checks is None: - runtime.resilient_sleep(opts.poll_interval) + services.sleep(opts.poll_interval) continue result = evaluate_checks(checks, opts.required_checks) entry.error_message = result.summary @@ -364,18 +284,18 @@ def wait_for_checks(entry: StackEntry, *, opts: AutolandOptions) -> bool: ) return False entry.check_retries += 1 - runtime.console.print( + ui.print( f"[yellow]Rerunning failed checks " f"(attempt {entry.check_retries}/{opts.max_check_retries}): " f"{', '.join(result.failed_names)}[/yellow]" ) - gh.github.rerun_failed(result.failed_runs) + github.rerun_failed(result.failed_runs) - runtime.console.print( + ui.print( f"[dim]PR #{entry.pr_number}: {result.summary} — " f"polling in {opts.poll_interval}s[/dim]" ) - runtime.resilient_sleep(opts.poll_interval) + services.sleep(opts.poll_interval) _MERGEABLE_STATES = {"CLEAN", "UNSTABLE", "HAS_HOOKS"} @@ -388,21 +308,22 @@ class MergeableResult: error: str = "" -def wait_for_mergeable(entry: StackEntry, *, opts: AutolandOptions) -> MergeableResult: +def wait_for_mergeable( + entry: StackEntry, *, opts: AutolandOptions, services: Services +) -> MergeableResult: """Wait until GitHub reports the PR as mergeable.""" + ui, github = services.ui, services.github while True: - data = _poll_read(gh.github.merge_state, entry.pr_number) + data = _poll_read(github.merge_state, entry.pr_number, services) if data is None: - runtime.resilient_sleep(opts.poll_interval) + services.sleep(opts.poll_interval) continue pr_state = data.get("state", "") merge_state = data.get("mergeStateStatus", "UNKNOWN") mergeable = data.get("mergeable", "UNKNOWN") if pr_state == "MERGED": - runtime.console.print( - f"[green]PR #{entry.pr_number} is already merged[/green]" - ) + ui.print(f"[green]PR #{entry.pr_number} is already merged[/green]") return MergeableResult(ready=True, already_merged=True) if pr_state == "CLOSED": entry.state = PRState.FAILED @@ -410,7 +331,7 @@ def wait_for_mergeable(entry: StackEntry, *, opts: AutolandOptions) -> Mergeable return MergeableResult(error="PR was closed") if merge_state in _MERGEABLE_STATES: - runtime.console.print( + ui.print( f"[green]PR #{entry.pr_number} is mergeable " f"(mergeStateStatus={merge_state})[/green]" ) @@ -418,66 +339,68 @@ def wait_for_mergeable(entry: StackEntry, *, opts: AutolandOptions) -> Mergeable if mergeable == "CONFLICTING": entry.error_message = "PR has merge conflicts — waiting for resolution" - runtime.console.print( + ui.print( f"\n[bold red]PR #{entry.pr_number} has merge conflicts! " "Resolve them on the PR branch and push; autoland will " "resume automatically.[/bold red]" ) - runtime.resilient_sleep(opts.poll_interval) + services.sleep(opts.poll_interval) continue # UNKNOWN can also mean "already in the merge queue". - if ( - merge_state == "UNKNOWN" - and gh.github.in_merge_queue(entry.pr_number) is True - ): - runtime.console.print( + if merge_state == "UNKNOWN" and github.in_merge_queue(entry.pr_number) is True: + ui.print( f"[cyan]PR #{entry.pr_number} is already in the merge queue — " "skipping enqueue[/cyan]" ) return MergeableResult(ready=True, already_merged=False) entry.error_message = f"Waiting for mergeable state (currently {merge_state})" - runtime.console.print( + ui.print( f"[dim]PR #{entry.pr_number}: mergeStateStatus={merge_state} — " f"polling in {opts.poll_interval}s[/dim]" ) - runtime.resilient_sleep(opts.poll_interval) + services.sleep(opts.poll_interval) -def _rewait_after_retry(entry: StackEntry, *, opts: AutolandOptions) -> bool: +def _rewait_after_retry( + entry: StackEntry, *, opts: AutolandOptions, services: Services +) -> bool: """Re-verify approval and checks before a queue retry.""" - if not wait_for_approval(entry, opts=opts): + if not wait_for_approval(entry, opts=opts, services=services): return False entry.check_retries = 0 - return wait_for_checks(entry, opts=opts) + return wait_for_checks(entry, opts=opts, services=services) -def enqueue_and_wait(entry: StackEntry, *, opts: AutolandOptions) -> bool: +def enqueue_and_wait( + entry: StackEntry, *, opts: AutolandOptions, services: Services +) -> bool: """Add the PR to the merge queue and wait for it to merge.""" + ui, github = services.ui, services.github while True: - mergeable_result = wait_for_mergeable(entry, opts=opts) + mergeable_result = wait_for_mergeable(entry, opts=opts, services=services) if not mergeable_result.ready: return False if mergeable_result.already_merged: entry.state = PRState.MERGED entry.error_message = "" - runtime.console.print( + ui.print( f"\n[bold green]PR #{entry.pr_number} already merged![/bold green]" ) return True entry.state = PRState.IN_MERGE_QUEUE entry.error_message = "Adding to merge queue..." - runtime.console.print( + ui.print( f"\n[bold cyan]Adding PR #{entry.pr_number} to merge queue[/bold cyan]" ) try: - gh.github.enqueue(entry.pr_number) + github.enqueue(entry.pr_number) except RuntimeError as e: entry.error_message = f"Failed to enqueue: {e}" - runtime.console.print(f"[red]Failed to add to merge queue: {e}[/red]") + ui.print(f"[red]Failed to add to merge queue: {e}[/red]") if entry.queue_retries >= opts.max_queue_retries: entry.state = PRState.FAILED entry.error_message = ( @@ -485,7 +408,7 @@ def enqueue_and_wait(entry: StackEntry, *, opts: AutolandOptions) -> bool: ) return False entry.queue_retries += 1 - if not _rewait_after_retry(entry, opts=opts): + if not _rewait_after_retry(entry, opts=opts, services=services): return False continue @@ -499,22 +422,20 @@ def enqueue_and_wait(entry: StackEntry, *, opts: AutolandOptions) -> bool: # A failed read leaves the merge undecided; poll again. poll = ( - _poll_read(gh.github.poll_merge, entry.pr_number) + _poll_read(github.poll_merge, entry.pr_number, services) or MergeQueuePollResult() ) if poll.merged: entry.state = PRState.MERGED entry.error_message = "" - runtime.console.print( - f"\n[bold green]PR #{entry.pr_number} merged![/bold green]" - ) + ui.print(f"\n[bold green]PR #{entry.pr_number} merged![/bold green]") return True if poll.error: entry.state = PRState.FAILED entry.error_message = poll.error return False if poll.booted: - runtime.console.print( + ui.print( f"\n[yellow]PR #{entry.pr_number} was booted from the " "merge queue[/yellow]" ) @@ -525,15 +446,15 @@ def enqueue_and_wait(entry: StackEntry, *, opts: AutolandOptions) -> bool: ) return False entry.queue_retries += 1 - if not _rewait_after_retry(entry, opts=opts): + if not _rewait_after_retry(entry, opts=opts, services=services): return False break # re-enqueue in outer loop mins = int(awake_elapsed) // 60 entry.error_message = f"In merge queue ({mins}m elapsed)..." - runtime.console.print( + ui.print( f"[dim]PR #{entry.pr_number}: in merge queue ({mins}m) — " f"polling in {opts.poll_interval}s[/dim]" ) - runtime.resilient_sleep(opts.poll_interval) + services.sleep(opts.poll_interval) awake_elapsed += opts.poll_interval diff --git a/tests/autoland_fakes.py b/tests/autoland_fakes.py new file mode 100644 index 0000000..58536df --- /dev/null +++ b/tests/autoland_fakes.py @@ -0,0 +1,485 @@ +"""Fakes for everything autoland's execution reaches through ``Services``. + +Build a ``FakeServices`` with whichever fakes a test cares about, and hand its +``services`` to the code under test (a step, an execution unit, the engine, a +wait loop, a stack merge, or a command flow). Nothing is patched: each fake +records what it was asked to do, so the test reads it back afterwards. + +- ``FakeUI``: records output; answers prompts from a script, then EOF. +- ``FakeWaits``: waits that succeed or fail at once, per outcome function. +- ``FakeRepo``: landed SHAs from a table; counts rebases and fetches; knows + the ancestry it is given, plus what it learns from a fetch. +- ``FakeGitHub``: an in-memory GitHub (PR states, reviews, checks, the merge + queue, workflow runs, GitHub stacks), whose calls can be scripted. +- ``FakeSleep``: records each sleep; can end a wait by raising + ``StopPolling`` after a number of sleeps. + +``FakeServices()`` runs against ``FakeWaits``; ``FakeServices.polling()`` +runs the real wait loops of ``autoland.waits`` over the other fakes. +""" + +from __future__ import annotations + +import functools +from collections.abc import Callable +from dataclasses import dataclass, field +from functools import cached_property +from typing import Any, TypeVar, cast + +from stack_pr import cli +from stack_pr.autoland import gh +from stack_pr.autoland.gh import MergeQueuePollResult +from stack_pr.autoland.model import ( + LandingContext, + PRState, + StackEntry, + WorkflowStep, +) +from stack_pr.autoland.options import AutolandOptions +from stack_pr.autoland.services import Services +from stack_pr.autoland.waits import MergeableResult + +OPTS = AutolandOptions( + merge_queue=True, + required_checks=[], + poll_interval=0, + max_check_retries=0, + max_queue_retries=0, + merge_timeout=0, + workflow_timeout=0, + default_workflow=None, + count=None, + dry_run=False, + branch=None, + interactive=False, + resume=False, + state_file=None, + always_cleanup=False, + merge_as_stack=False, +) + + +def stack(*prs: int) -> list[StackEntry]: + return [ + StackEntry( + pr_url=f"https://github.com/o/r/pull/{n}", + pr_number=n, + branch=f"b{n}", + title=f"Title {n}", + ) + for n in prs + ] + + +# --------------------------------------------------------------------------- +# UI +# --------------------------------------------------------------------------- + + +class FakeUI: + """Records output; answers prompts from a script, then raises EOFError.""" + + def __init__(self, answers: list[str] | None = None) -> None: + self.answers = list(answers or []) + self.lines: list[str] = [] + + def print(self, *objects: object) -> None: + self.lines.append(" ".join(str(o) for o in objects)) + + def input(self, prompt: str) -> str: + self.lines.append(prompt) + if not self.answers: + raise EOFError + return self.answers.pop(0) + + def escape(self, text: str) -> str: + return text.replace("[", r"\[") + + @property + def output(self) -> str: + return "\n".join(self.lines) + + +# --------------------------------------------------------------------------- +# Waits +# --------------------------------------------------------------------------- + +# What a fake wait does to the PR it waits on. +Outcome = Callable[[StackEntry], bool] + + +def ok(_entry: StackEntry) -> bool: + return True + + +def fails(_entry: StackEntry) -> bool: + return False + + +def merges(entry: StackEntry) -> bool: + entry.state = PRState.MERGED + return True + + +@dataclass +class FakeWaits: + approval: Outcome = ok + checks: Outcome = ok + mergeable: Outcome = ok + queue: Outcome = merges + workflow_succeeds: bool = True + # Each wait, in order: "approval", "checks", "mergeable", or "queue". + calls: list[str] = field(default_factory=list) + # The entry of each wait, in the same order as ``calls``. + entries: list[StackEntry] = field(default_factory=list) + workflow_steps: list[WorkflowStep] = field(default_factory=list) + workflow_targets: list[str] = field(default_factory=list) + + def _wait(self, kind: str, entry: StackEntry, outcome: Outcome) -> bool: + self.calls.append(kind) + self.entries.append(entry) + return outcome(entry) + + def waited(self, kind: str) -> list[int]: + """The PRs a *kind* wait was made for, in order.""" + return [e.pr_number for k, e in zip(self.calls, self.entries) if k == kind] + + def wait_for_approval(self, entry: StackEntry, *, opts: AutolandOptions) -> bool: + return self._wait("approval", entry, self.approval) + + def wait_for_checks(self, entry: StackEntry, *, opts: AutolandOptions) -> bool: + return self._wait("checks", entry, self.checks) + + def wait_for_mergeable( + self, entry: StackEntry, *, opts: AutolandOptions + ) -> MergeableResult: + return MergeableResult(ready=self._wait("mergeable", entry, self.mergeable)) + + def enqueue_and_wait(self, entry: StackEntry, *, opts: AutolandOptions) -> bool: + return self._wait("queue", entry, self.queue) + + def wait_for_workflow( + self, + step: WorkflowStep, + target_sha: str, + *, + opts: AutolandOptions, + common: cli.CommonArgs, + ctx: LandingContext, + ) -> bool: + self.workflow_steps.append(step) + self.workflow_targets.append(target_sha) + step.state = "succeeded" if self.workflow_succeeds else "failed" + return self.workflow_succeeds + + +# --------------------------------------------------------------------------- +# Repo +# --------------------------------------------------------------------------- + + +@dataclass +class FakeRepo: + # PR number (None: the target branch head) -> its landed commit. + shas: dict[int | None, str] = field(default_factory=dict) + rebase_error: Exception | None = None + # (ancestor, descendant) -> whether it is one, as far as the clone knows; + # a pair it doesn't know is "can't tell" (None). A fetch teaches it + # ``after_fetch``. + ancestry: dict[tuple[str, str], bool] = field(default_factory=dict) + after_fetch: dict[tuple[str, str], bool] = field(default_factory=dict) + lookups: list[int | None] = field(default_factory=list) + rebases: int = 0 + fetches: int = 0 + + def fetch_target(self, common: cli.CommonArgs) -> None: + self.fetches += 1 + self.ancestry.update(self.after_fetch) + + def is_ancestor(self, ancestor: str, descendant: str) -> bool | None: + return self.ancestry.get((ancestor, descendant)) + + def landed_sha(self, common: cli.CommonArgs, pr_number: int | None = None) -> str: + self.lookups.append(pr_number) + return self.shas.get(pr_number, "") + + def rebase_and_resubmit(self, common: cli.CommonArgs) -> None: + self.rebases += 1 + if self.rebase_error is not None: + raise self.rebase_error + + +# --------------------------------------------------------------------------- +# GitHub +# --------------------------------------------------------------------------- + +_F = TypeVar("_F", bound=Callable[..., Any]) + + +def _scriptable(method: _F) -> _F: + """Log each call, and answer it from the method's script while it lasts.""" + + @functools.wraps(method) + def call(self: FakeGitHub, *args: Any, **kwargs: Any) -> Any: + name = method.__name__ + self.calls.append((name, args)) + replies = self._scripts.get(name) + if replies: + reply = replies.pop(0) + if isinstance(reply, BaseException): + raise reply + return reply(*args, **kwargs) if callable(reply) else reply + return method(self, *args, **kwargs) + + return cast("_F", call) + + +class FakeGitHub: + """GitHub in memory: what autoland's execution asks the ``gh.GitHub`` client. + + A PR is open unless it is in ``merged`` or ``closed``. By default it is + approved, has no checks, is mergeable, is merged as soon as it is + enqueued, and stays in the merge queue while open. Every call is logged + in ``calls`` as ``(method, args)``. + + ``script(method, *replies)`` answers the next calls of *method*, one + reply each, before it goes back to answering from the fake's state. A + reply is returned, or raised if it is an exception, or called with the + call's arguments if it is callable. + """ + + def __init__(self) -> None: + self.calls: list[tuple[str, tuple]] = [] + self._scripts: dict[str, list[Any]] = {} + self.merged: set[int] = set() + self.closed: set[int] = set() + # PR -> review decision (default "APPROVED"), and its checks. + self.reviews: dict[int, str] = {} + self.pr_checks: dict[int, list[dict]] = {} + self.merge_commits: dict[int, str] = {} + self.enqueued: list[int] = [] + self.booted: set[int] = set() + # The latest runs of every workflow, newest first, and GitHub's answer + # to whether (ancestor, descendant) is one (unknown if absent). + self.runs: list[dict] = [] + self.compare: dict[tuple[str, str], bool] = {} + # GitHub stacks: number -> PRs, bottom first. + self.stacks: dict[int, list[int]] = {} + # Every create-a-stack request, and every merge request's PR. + self.stack_requests: list[list[int]] = [] + self.created: list[list[int]] = [] + self.unstacked: list[int] = [] + self.merge_requests: list[int] = [] + # (PR, new base) for each base change, and the base changes made by + # the time of each merge request. + self.base_changes: list[tuple[int, str]] = [] + self.bases_at_merge: list[list[tuple[int, str]]] = [] + self.create_error = "" + self.merge_error = "" + self.set_base_error = "" + # The PRs a merge request lands (default: the whole stack up to the PR + # it was made on), and the status GitHub then reports for it. + self.lands: list[int] | None = None + self.status = "merged" + + def script(self, method: str, *replies: object) -> None: + self._scripts.setdefault(method, []).extend(replies) + + def called(self, method: str) -> list[tuple]: + """The arguments of each call of *method*.""" + return [args for name, args in self.calls if name == method] + + # --- PRs ------------------------------------------------------------ + + @_scriptable + def owner_repo(self) -> tuple[str, str]: + return ("o", "r") + + @_scriptable + def pr_state(self, pr: int) -> str: + if pr in self.merged: + return "MERGED" + return "CLOSED" if pr in self.closed else "OPEN" + + @_scriptable + def review_decision(self, pr: int) -> str: + return self.reviews.get(pr, "APPROVED") + + @_scriptable + def merge_state(self, pr: int) -> dict: + return { + "state": self.pr_state(pr), + "mergeStateStatus": "CLEAN", + "mergeable": "MERGEABLE", + } + + @_scriptable + def checks(self, pr: int) -> list[dict]: + return self.pr_checks.get(pr, []) + + @_scriptable + def rerun_failed(self, run_ids: list[int]) -> None: + pass + + @_scriptable + def merge_commit(self, pr: int) -> str | None: + return self.merge_commits.get(pr) + + # --- the merge queue ------------------------------------------------- + + @_scriptable + def in_merge_queue(self, pr: int) -> bool | None: + return pr not in self.booted + + @_scriptable + def has_merge_queue(self, branch: str) -> bool | None: + return True + + @_scriptable + def enqueue(self, pr: int) -> None: + self.enqueued.append(pr) + self.merged.add(pr) + + @_scriptable + def poll_merge(self, pr: int) -> MergeQueuePollResult: + state = self.pr_state(pr) + if state == "MERGED": + return MergeQueuePollResult(merged=True) + if state == "CLOSED": + return MergeQueuePollResult(error="PR was closed") + if self.in_merge_queue(pr) is False: + return MergeQueuePollResult(booted=True) + return MergeQueuePollResult() + + # --- workflows ------------------------------------------------------- + + @_scriptable + def workflow_runs(self, workflow: str, branch: str) -> list[dict]: + return list(self.runs) + + @_scriptable + def contains(self, ancestor: str, descendant: str) -> bool | None: + return self.compare.get((ancestor, descendant)) + + # --- GitHub stacks --------------------------------------------------- + + @_scriptable + def find_native_stack(self, pr: int) -> dict | None: + for number, prs in self.stacks.items(): + if pr in prs: + return { + "number": number, + "pull_requests": [ + {"number": n, "state": "closed" if n in self.merged else "open"} + for n in prs + ], + } + return None + + @_scriptable + def create_native_stack(self, prs: list[int]) -> dict: + self.stack_requests.append(prs) + if self.create_error: + raise RuntimeError(self.create_error) + self.created.append(prs) + self.stacks[7] = prs + return {"number": 7} + + @_scriptable + def unstack_native_stack(self, number: int) -> None: + self.unstacked.append(number) + self.stacks.pop(number) + + @_scriptable + def set_base(self, pr: int, base: str) -> None: + if self.set_base_error: + raise RuntimeError(self.set_base_error) + self.base_changes.append((pr, base)) + + @_scriptable + def merge_async(self, pr: int, *, merge_queue: bool) -> str: + self.merge_requests.append(pr) + self.bases_at_merge.append(list(self.base_changes)) + if self.merge_error: + raise RuntimeError(self.merge_error) + prs = next(p for p in self.stacks.values() if pr in p) + self.merged.update(self.lands if self.lands is not None else prs) + return "uuid-1" + + @_scriptable + def merge_async_status(self, pr: int, uuid: str) -> tuple[str, str]: + return self.status, "conflict in PR #102" if self.status == "failed" else "" + + +class ForbiddenGitHub: + """A GitHub that fails the test if anything asks it anything.""" + + def __getattr__(self, name: str) -> object: + raise AssertionError(f"GitHub was called directly: {name}") + + +# --------------------------------------------------------------------------- +# Sleep +# --------------------------------------------------------------------------- + + +class StopPolling(Exception): # noqa: N818 - it is a signal, not an error + """Raised by a ``FakeSleep`` out of its limit, to end a wait.""" + + +@dataclass +class FakeSleep: + """Records each sleep, without sleeping. + + With a *limit*, the sleep after the first *limit* ones raises + ``StopPolling`` instead: a wait that is still waiting by then ends with it. + """ + + limit: int | None = None + slept: list[int] = field(default_factory=list) + + def __call__(self, seconds: int) -> float: + if self.limit is not None and len(self.slept) >= self.limit: + raise StopPolling + self.slept.append(seconds) + return 0.0 + + +def forbidden_sleep(_seconds: int) -> float: + raise AssertionError("slept") + + +# --------------------------------------------------------------------------- +# All of them +# --------------------------------------------------------------------------- + + +@dataclass +class FakeServices: + """A set of fakes, and the ``Services`` made of them (``services``). + + ``waits`` is ``None`` for the real wait loops, polling the other fakes + (see ``polling``). ``github`` may be any GitHub client, e.g. a real + ``gh.GitHub()`` over a faked ``gh``. + """ + + github: Any = field(default_factory=FakeGitHub) + ui: FakeUI = field(default_factory=FakeUI) + sleep: Callable[[int], float] = field(default_factory=FakeSleep) + repo: FakeRepo = field(default_factory=FakeRepo) + waits: FakeWaits | None = field(default_factory=FakeWaits) + + @classmethod + def polling(cls, **fakes: Any) -> FakeServices: + """Fakes under the real wait loops.""" + return cls(waits=None, **fakes) + + @cached_property + def services(self) -> Services: + return Services( + github=cast("gh.GitHub", self.github), + ui=self.ui, + sleep=self.sleep, + repo=self.repo, + waits=cast("Any", self.waits), + ) diff --git a/tests/test_autoland.py b/tests/test_autoland.py index 15f3eda..cd31cbe 100644 --- a/tests/test_autoland.py +++ b/tests/test_autoland.py @@ -8,7 +8,9 @@ import sys import tempfile import unicodedata +from collections.abc import Callable from pathlib import Path +from typing import cast from unittest.mock import Mock import pytest @@ -60,8 +62,20 @@ parse_plan, plan_from_file, ) +from stack_pr.autoland.repo import GitRepo from stack_pr.autoland.runtime import _RE_MARKUP, _PlainConsole +from stack_pr.autoland.services import Services from stack_pr.autoland.state import AutolandCheckpointer, AutolandLock +from tests.autoland_fakes import ( + FakeGitHub, + FakeRepo, + FakeServices, + FakeSleep, + FakeUI, + FakeWaits, + StopPolling, + ok, +) from tests.helpers import common_args, git, init_repo @@ -206,26 +220,38 @@ def test_evaluate_checks_empty_required_no_checks() -> None: # --- merge status polling (GitHub.poll_merge) ---------------------------- -def test_poll_merge_merged(mocker) -> None: # noqa: ANN001 - mocker.patch.object(gh.github, "pr_state", return_value="MERGED") - assert gh.github.poll_merge(1).merged is True +class _GitHubWith(gh.GitHub): + """The real client, over a canned PR state and, unless it is to be looked + up, a canned merge-queue entry.""" + + def __init__(self, state: str, *, queued: object = "lookup") -> None: + super().__init__() + self._owner_repo = ("o", "r") + self._state, self._queued = state, queued + + def pr_state(self, pr_number: int) -> str: + return self._state + + def in_merge_queue(self, pr_number: int) -> bool | None: + if self._queued == "lookup": + return super().in_merge_queue(pr_number) + return cast("bool | None", self._queued) -def test_poll_merge_closed(mocker) -> None: # noqa: ANN001 - mocker.patch.object(gh.github, "pr_state", return_value="CLOSED") - assert gh.github.poll_merge(1).error == "PR was closed" +def test_poll_merge_merged() -> None: + assert _GitHubWith("MERGED").poll_merge(1).merged is True -def test_poll_merge_booted(mocker) -> None: # noqa: ANN001 - mocker.patch.object(gh.github, "pr_state", return_value="OPEN") - mocker.patch.object(gh.github, "in_merge_queue", return_value=False) - assert gh.github.poll_merge(1).booted is True +def test_poll_merge_closed() -> None: + assert _GitHubWith("CLOSED").poll_merge(1).error == "PR was closed" -def test_poll_merge_still_queued(mocker) -> None: # noqa: ANN001 - mocker.patch.object(gh.github, "pr_state", return_value="OPEN") - mocker.patch.object(gh.github, "in_merge_queue", return_value=True) - res = gh.github.poll_merge(1) +def test_poll_merge_booted() -> None: + assert _GitHubWith("OPEN", queued=False).poll_merge(1).booted is True + + +def test_poll_merge_still_queued() -> None: + res = _GitHubWith("OPEN", queued=True).poll_merge(1) assert not res.merged assert not res.booted assert not res.error @@ -234,10 +260,8 @@ def test_poll_merge_still_queued(mocker) -> None: # noqa: ANN001 def test_poll_merge_lookup_failure_is_not_booted(mocker) -> None: # noqa: ANN001 # A transient failure looking up the merge-queue entry says nothing about # whether the PR is still queued, so it must not count as being booted. - mocker.patch.object(gh.github, "pr_state", return_value="OPEN") - mocker.patch.object(gh.github, "owner_repo", return_value=("o", "r")) - mocker.patch.object(runtime, "run", side_effect=RuntimeError("HTTP 502")) - res = gh.github.poll_merge(1) + _shell(mocker, lambda _cmd: RuntimeError("HTTP 502")) + res = _GitHubWith("OPEN").poll_merge(1) assert not res.merged assert not res.booted assert not res.error @@ -247,145 +271,104 @@ def test_poll_merge_lookup_failure_is_not_booted(mocker) -> None: # noqa: ANN00 def test_merge_commit_parses_oid(mocker) -> None: # noqa: ANN001 - mocker.patch.object( - gh, "gh_json", return_value={"mergeCommit": {"oid": "deadbeef"}} - ) - assert gh.github.merge_commit(1) == "deadbeef" + _gh_replies(mocker, view=[json.dumps({"mergeCommit": {"oid": "deadbeef"}})]) + assert gh.GitHub().merge_commit(1) == "deadbeef" def test_merge_commit_none_when_unmerged(mocker) -> None: # noqa: ANN001 - mocker.patch.object(gh, "gh_json", return_value={"mergeCommit": None}) - assert gh.github.merge_commit(1) is None + _gh_replies(mocker, view=[json.dumps({"mergeCommit": None})]) + assert gh.GitHub().merge_commit(1) is None + +def _git_repo(github: FakeGitHub | None = None) -> GitRepo: + return GitRepo(cast("gh.GitHub", github or FakeGitHub()), FakeUI()) -def test_refresh_last_landed_sha_prefers_merge_commit(mocker) -> None: # noqa: ANN001 + +def test_landed_sha_prefers_merge_commit(mocker) -> None: # noqa: ANN001 # origin/ HEAD has advanced past our merge commit (bot commits, # other PRs). We must record OUR merge commit, not the moving HEAD. - mocker.patch.object(runtime, "run") # git fetch is a no-op - mocker.patch.object(gh.github, "merge_commit", return_value="mergesha") - ctx = LandingContext(last_landed_sha="") - waits._refresh_last_landed_sha(ctx, common_args(), pr_number=42) # noqa: SLF001 - assert ctx.last_landed_sha == "mergesha" + _shell(mocker, lambda _cmd: "") # git fetch is a no-op + github = FakeGitHub() + github.merge_commits[42] = "mergesha" + assert _git_repo(github).landed_sha(common_args(), pr_number=42) == "mergesha" -def test_refresh_last_landed_sha_falls_back_to_head(mocker) -> None: # noqa: ANN001 +def test_landed_sha_falls_back_to_head(mocker) -> None: # noqa: ANN001 # No PR context (e.g. resume) or the merge commit is unknown: fall back to # origin/ HEAD. - mocker.patch.object( - runtime, - "run", - return_value=argparse.Namespace(stdout="headsha\n", returncode=0), + _shell(mocker, lambda _cmd: "headsha\n") + assert _git_repo().landed_sha(common_args(), pr_number=42) == "headsha" + + +def _wait_for_workflow(fakes: FakeServices, step: WorkflowStep) -> bool: + ctx = LandingContext(last_landed_sha="mergesha") + return waits.wait_for_workflow( + step, opts=_opts(), common=common_args(), ctx=ctx, services=fakes.services ) - mocker.patch.object(gh.github, "merge_commit", return_value=None) - ctx = LandingContext(last_landed_sha="") - waits._refresh_last_landed_sha(ctx, common_args(), pr_number=42) # noqa: SLF001 - assert ctx.last_landed_sha == "headsha" -def test_wait_for_workflow_accepts_run_on_merge_commit(mocker) -> None: # noqa: ANN001 +def _success(sha: str) -> dict: + return {"headSha": sha, "status": "completed", "conclusion": "success"} + + +def test_wait_for_workflow_accepts_run_on_merge_commit() -> None: # Regression: a green deploy run on our exact merge commit must satisfy the # checkpoint even though origin/ has since moved on. - mocker.patch.object( - gh.github, - "workflow_runs", - return_value=[ - {"headSha": "mergesha", "status": "completed", "conclusion": "success"} - ], - ) + fakes = FakeServices.polling() + fakes.github.runs = [_success("mergesha")] step = WorkflowStep(workflow="deploy.yaml") - ctx = LandingContext(last_landed_sha="mergesha") - assert waits.wait_for_workflow(step, opts=_opts(), common=common_args(), ctx=ctx) + assert _wait_for_workflow(fakes, step) assert step.state == "succeeded" -def test_wait_for_workflow_fetches_when_run_sha_is_unknown_locally(mocker) -> None: # noqa: ANN001 +def test_wait_for_workflow_fetches_when_run_sha_is_unknown_locally() -> None: # Regression: origin/ had advanced past our last fetch, so the # completed run's head commit was not in this clone. `git merge-base # --is-ancestor` then exits 128 ("Not a valid commit name"), which used to # read as "not an ancestor" and made the checkpoint poll forever. A fetch # must resolve it. - mocker.patch.object( - gh.github, - "workflow_runs", - return_value=[ - {"headSha": "newersha", "status": "completed", "conclusion": "success"} - ], - ) - calls: list[list] = [] - - def _run(cmd, **_kwargs): # noqa: ANN001, ANN003, ANN202 - calls.append(cmd) - if cmd[:3] == ["git", "merge-base", "--is-ancestor"]: - # Unknown commit before the fetch, a genuine answer after it. - fetched = ["git", "fetch", "origin", "main"] in calls - return argparse.Namespace( - stdout="", stderr=b"", returncode=0 if fetched else 128 - ) - return argparse.Namespace(stdout="", stderr=b"", returncode=0) - - mocker.patch.object(runtime, "run", side_effect=_run) # the fetch - mocker.patch("stack_pr.git.run_shell_command", side_effect=_run) # merge-base - contains = mocker.patch.object(gh.github, "contains") + repo = FakeRepo(after_fetch={("mergesha", "newersha"): True}) + fakes = FakeServices.polling(repo=repo) + fakes.github.runs = [_success("newersha")] step = WorkflowStep(workflow="deploy.yaml") - ctx = LandingContext(last_landed_sha="mergesha") - assert waits.wait_for_workflow(step, opts=_opts(), common=common_args(), ctx=ctx) + assert _wait_for_workflow(fakes, step) assert step.state == "succeeded" - assert ["git", "fetch", "origin", "main"] in calls - contains.assert_not_called() # the fetch answered it; no API call needed + assert repo.fetches == 1 + # the fetch answered it; no API call needed + assert fakes.github.called("contains") == [] -def test_wait_for_workflow_falls_back_to_github_compare(mocker) -> None: # noqa: ANN001 +def test_wait_for_workflow_falls_back_to_github_compare() -> None: # The run's head commit is still missing after the fetch (e.g. a # merge-queue commit that never landed on the target branch): ask GitHub. - mocker.patch.object( - gh.github, - "workflow_runs", - return_value=[ - {"headSha": "queuesha", "status": "completed", "conclusion": "success"} - ], - ) - unknown = argparse.Namespace(stdout="", stderr=b"", returncode=128) - mocker.patch.object(runtime, "run", return_value=unknown) # the fetch - mocker.patch("stack_pr.git.run_shell_command", return_value=unknown) - contains = mocker.patch.object(gh.github, "contains", return_value=True) - step = WorkflowStep(workflow="deploy.yaml") - ctx = LandingContext(last_landed_sha="mergesha") - assert waits.wait_for_workflow(step, opts=_opts(), common=common_args(), ctx=ctx) - contains.assert_called_once_with("mergesha", "queuesha") - + fakes = FakeServices.polling() + fakes.github.runs = [_success("queuesha")] + fakes.github.compare[("mergesha", "queuesha")] = True + assert _wait_for_workflow(fakes, WorkflowStep(workflow="deploy.yaml")) + assert fakes.github.called("contains") == [("mergesha", "queuesha")] -class _StopPollingError(Exception): - """Raised from a wait's sleep to end the wait after one poll.""" - -def test_wait_for_workflow_keeps_waiting_when_ancestry_unknown(mocker) -> None: # noqa: ANN001 +def test_wait_for_workflow_keeps_waiting_when_ancestry_unknown() -> None: # Neither git nor GitHub can answer: warn and keep waiting rather than # treating "unknown" as "the deploy included our code". - def _runs(*_a, **_k) -> list: # noqa: ANN002, ANN003 - return [{"headSha": "mystery", "status": "completed", "conclusion": "success"}] - - mocker.patch.object(gh.github, "workflow_runs", side_effect=_runs) - unknown = argparse.Namespace(stdout="", stderr=b"", returncode=128) - mocker.patch.object(runtime, "run", return_value=unknown) # the fetch - mocker.patch("stack_pr.git.run_shell_command", return_value=unknown) - mocker.patch.object(gh.github, "contains", return_value=None) - mocker.patch.object(runtime, "resilient_sleep", side_effect=_StopPollingError) + fakes = FakeServices.polling(sleep=FakeSleep(limit=0)) + fakes.github.runs = [_success("mystery")] step = WorkflowStep(workflow="deploy.yaml") - ctx = LandingContext(last_landed_sha="mergesha") # Still waiting when the first poll's sleep comes around. - with pytest.raises(_StopPollingError): - waits.wait_for_workflow(step, opts=_opts(), common=common_args(), ctx=ctx) + with pytest.raises(StopPolling): + _wait_for_workflow(fakes, step) assert step.state != "succeeded" -def test_local_is_ancestor_distinguishes_no_from_unknown(mocker) -> None: # noqa: ANN001 +def test_repo_is_ancestor_distinguishes_no_from_unknown(mocker) -> None: # noqa: ANN001 run_mock = mocker.patch("stack_pr.git.run_shell_command") + repo = _git_repo() run_mock.return_value = argparse.Namespace(stderr=b"", returncode=0) - assert waits._local_is_ancestor("a", "b") is True # noqa: SLF001 + assert repo.is_ancestor("a", "b") is True run_mock.return_value = argparse.Namespace(stderr=b"", returncode=1) - assert waits._local_is_ancestor("a", "b") is False # noqa: SLF001 + assert repo.is_ancestor("a", "b") is False run_mock.return_value = argparse.Namespace(stderr=b"", returncode=128) - assert waits._local_is_ancestor("a", "b") is None # noqa: SLF001 + assert repo.is_ancestor("a", "b") is None def test_sha_eq_requires_a_long_enough_prefix() -> None: @@ -398,52 +381,50 @@ def test_sha_eq_requires_a_long_enough_prefix() -> None: def test_github_contains_uses_merge_base(mocker) -> None: # noqa: ANN001 - mocker.patch.object(gh.github, "_owner_repo", ("o", "r")) - run_mock = mocker.patch.object(runtime, "run") - run_mock.return_value = argparse.Namespace(stdout="basesha\n", returncode=0) - assert gh.github.contains("basesha", "headsha") is True - run_mock.return_value = argparse.Namespace(stdout="othersha\n", returncode=0) - assert gh.github.contains("basesha", "headsha") is False - run_mock.return_value = argparse.Namespace(stdout="null\n", returncode=0) - assert gh.github.contains("basesha", "headsha") is None - run_mock.side_effect = RuntimeError("HTTP 404") - assert gh.github.contains("basesha", "headsha") is None - - -def test_ancestry_caches_verdicts(mocker) -> None: # noqa: ANN001 + github = gh.GitHub() + github._owner_repo = ("o", "r") # noqa: SLF001 + replies: list[object] = [ + "basesha\n", + "othersha\n", + "null\n", + RuntimeError("HTTP 404"), + ] + _shell(mocker, lambda _cmd: replies.pop(0)) + assert github.contains("basesha", "headsha") is True + assert github.contains("basesha", "headsha") is False + assert github.contains("basesha", "headsha") is None + assert github.contains("basesha", "headsha") is None + + +def test_ancestry_caches_verdicts() -> None: # The poll loop re-examines the same runs every interval; commits are # immutable, so each pair is decided at most once. - contains = mocker.patch.object(gh.github, "contains", return_value=False) - unknown = argparse.Namespace(stdout="", stderr=b"", returncode=128) - mocker.patch.object(runtime, "run", return_value=unknown) # the fetch - mocker.patch("stack_pr.git.run_shell_command", return_value=unknown) - ancestry = waits._Ancestry(common_args()) # noqa: SLF001 + fakes = FakeServices.polling() + fakes.github.compare.update( + {("landedsha", "runsha1"): False, ("landedsha", "runsha2"): False} + ) + ancestry = waits._Ancestry(common_args(), fakes.services) # noqa: SLF001 assert ancestry.contains("landedsha", "runsha1") is False assert ancestry.contains("landedsha", "runsha1") is False assert ancestry.contains("landedsha", "runsha2") is False # One GitHub request per distinct pair, however often it is asked about. - assert [c.args for c in contains.call_args_list] == [ + assert fakes.github.called("contains") == [ ("landedsha", "runsha1"), ("landedsha", "runsha2"), ] -def test_wait_for_workflow_ignores_failed_and_incomplete(mocker) -> None: # noqa: ANN001 +def test_wait_for_workflow_ignores_failed_and_incomplete() -> None: # A failed run and a still-running run on our SHA must not satisfy the # checkpoint: the wait is still going when its first poll's sleep comes. - mocker.patch.object( - gh.github, - "workflow_runs", - return_value=[ - {"headSha": "mergesha", "status": "completed", "conclusion": "failure"}, - {"headSha": "mergesha", "status": "in_progress", "conclusion": None}, - ], - ) - mocker.patch.object(runtime, "resilient_sleep", side_effect=_StopPollingError) + fakes = FakeServices.polling(sleep=FakeSleep(limit=0)) + fakes.github.runs = [ + {"headSha": "mergesha", "status": "completed", "conclusion": "failure"}, + {"headSha": "mergesha", "status": "in_progress", "conclusion": None}, + ] step = WorkflowStep(workflow="deploy.yaml") - ctx = LandingContext(last_landed_sha="mergesha") - with pytest.raises(_StopPollingError): - waits.wait_for_workflow(step, opts=_opts(), common=common_args(), ctx=ctx) + with pytest.raises(StopPolling): + _wait_for_workflow(fakes, step) assert step.state != "succeeded" @@ -1072,12 +1053,18 @@ def test_rich_status_truncates_titles_but_never_the_failure_reason() -> None: # --- executing a partially-landed plan ----------------------------------- -@pytest.mark.usefixtures("autoland_console") -def test_execute_plan_skips_landed_prefix_and_targets_last_landed_sha(mocker) -> None: # noqa: ANN001 - refresh = mocker.patch("stack_pr.autoland.waits._refresh_last_landed_sha") - wait_for_workflow = mocker.patch( - "stack_pr.autoland.waits.wait_for_workflow", return_value=True - ) +class _NoCheckpoints(AutolandCheckpointer): + """A checkpointer that saves nothing.""" + + def __init__(self) -> None: + super().__init__(path=Path("/dev/null"), branch="feat", base="main") + + def save(self, ctx: LandingContext) -> None: + pass + + +def test_execute_plan_skips_landed_prefix_and_targets_last_landed_sha() -> None: + fakes = FakeServices() stack = _pinned_stack([103]) plan = parse_plan( @@ -1086,66 +1073,50 @@ def test_execute_plan_skips_landed_prefix_and_targets_last_landed_sha(mocker) -> pr_is_merged=lambda pr: pr in (101, 102), ) ctx = LandingContext(stack=stack, plan=plan) - checkpointer = AutolandCheckpointer( - path=Path("/dev/null"), branch="feat", base="main" - ) - mocker.patch.object(checkpointer, "save") - assert engine.execute_plan(ctx, common_args(), _opts(), checkpointer) is True + assert engine.execute_plan( + ctx, common_args(), _opts(), _NoCheckpoints(), services=fakes.services + ) # Only the trailing workflow runs; the skipped one and the confirmation # (which would otherwise block on stdin) are passed over. - (waited_on,) = [c.args[0] for c in wait_for_workflow.call_args_list] + assert fakes.waits is not None + (waited_on,) = fakes.waits.workflow_steps assert waited_on is plan[-1] # The trailing workflow waits for #102's code: only the last PR of the # landed prefix is looked up, not every PR in it. - pinned = [c.args[2] for c in refresh.call_args_list if len(c.args) > 2] + pinned = [pr for pr in fakes.repo.lookups if pr is not None] assert pinned == [102] -def test_confirm_step_banner_is_a_single_line(mocker, autoland_console) -> None: # noqa: ANN001 +def test_confirm_step_banner_is_a_single_line() -> None: # Regression: the banner was once built as two list elements, so the join # split "Step 1/1: Manual confirmation required" across two lines. - autoland_console.input.return_value = "y" - mocker.patch("stack_pr.autoland.waits._refresh_last_landed_sha") + fakes = FakeServices(ui=FakeUI(answers=["y"])) ctx = LandingContext(stack=_pinned_stack([101]), plan=[ConfirmStep(condition="QA")]) - checkpointer = AutolandCheckpointer( - path=Path("/dev/null"), branch="feat", base="main" - ) - mocker.patch.object(checkpointer, "save") - - assert engine.execute_plan(ctx, common_args(), _opts(), checkpointer) is True - printed = "\n".join( - str(c.args[0]) for c in autoland_console.print.call_args_list if c.args + assert engine.execute_plan( + ctx, common_args(), _opts(), _NoCheckpoints(), services=fakes.services ) - assert "Step 1/1: Manual confirmation required" in printed + assert "Step 1/1: Manual confirmation required" in fakes.ui.output -@pytest.mark.usefixtures("autoland_console") -def test_execute_plan_lands_a_pinned_step_that_is_still_open(mocker) -> None: # noqa: ANN001 - mocker.patch("stack_pr.autoland.waits._refresh_last_landed_sha") - approval = mocker.patch( - "stack_pr.autoland.waits.wait_for_approval", return_value=True - ) - checks = mocker.patch("stack_pr.autoland.waits.wait_for_checks", return_value=True) - enqueue = mocker.patch( - "stack_pr.autoland.waits.enqueue_and_wait", return_value=True - ) + +def test_execute_plan_lands_a_pinned_step_that_is_still_open() -> None: + fakes = FakeServices(waits=FakeWaits(queue=ok)) stack = _pinned_stack([103]) plan = parse_plan("l 102\nl 103\n", stack, pr_is_merged=lambda pr: pr == 102) ctx = LandingContext(stack=stack, plan=plan) - checkpointer = AutolandCheckpointer( - path=Path("/dev/null"), branch="feat", base="main" - ) - mocker.patch.object(checkpointer, "save") - assert engine.execute_plan(ctx, common_args(), _opts(), checkpointer) is True + assert engine.execute_plan( + ctx, common_args(), _opts(), _NoCheckpoints(), services=fakes.services + ) - for mock in (approval, checks, enqueue): - assert mock.call_args.args[0] is stack[0] + assert fakes.waits is not None + assert fakes.waits.calls == ["approval", "checks", "queue"] + assert all(entry is stack[0] for entry in fakes.waits.entries) # --- state round-trip ---------------------------------------------------- @@ -1208,7 +1179,6 @@ def test_load_state_version_mismatch(tmp_path) -> None: # noqa: ANN001 # --- rebase + resubmit --------------------------------------------------- -@pytest.mark.usefixtures("autoland_console") def test_rebase_and_resubmit_rededuces_base(mocker) -> None: # noqa: ANN001 # After rebasing onto an advanced target, the base cached at autoland start # is stale; resubmit must re-deduce it (else it sweeps others' commits into @@ -1217,11 +1187,11 @@ def test_rebase_and_resubmit_rededuces_base(mocker) -> None: # noqa: ANN001 stale = dataclasses.replace(common_args(), base="STALE_MERGE_BASE") fresh = dataclasses.replace(stale, base="FRESH_ORIGIN_MASTER") - mocker.patch("stack_pr.autoland.runtime.run") # git fetch / rebase + _shell(mocker, lambda _cmd: "") # git fetch / rebase deduce = mocker.patch("stack_pr.cli.deduce_base", return_value=fresh) submit = mocker.patch("stack_pr.cli.command_submit") - waits.rebase_and_resubmit(stale) + _git_repo().rebase_and_resubmit(stale) # deduce_base is called with the cached base cleared... assert deduce.call_args.args[0].base == "" @@ -1229,7 +1199,6 @@ def test_rebase_and_resubmit_rededuces_base(mocker) -> None: # noqa: ANN001 assert submit.call_args.args[0].base == "FRESH_ORIGIN_MASTER" -@pytest.mark.usefixtures("autoland_console") def test_rebase_and_resubmit_aborts_conflicted_rebase( tmp_path, # noqa: ANN001 monkeypatch, # noqa: ANN001 @@ -1260,7 +1229,7 @@ def test_rebase_and_resubmit_aborts_conflicted_rebase( submit = mocker.patch("stack_pr.cli.command_submit") with pytest.raises(RuntimeError, match="rebase"): - waits.rebase_and_resubmit(common_args()) + _git_repo().rebase_and_resubmit(common_args()) submit.assert_not_called() for state_dir in ("rebase-merge", "rebase-apply"): @@ -1947,25 +1916,30 @@ def test_stop_running_autoland_declined_leaves_it_running( ("CHANGES_REQUESTED", False), ], ) -@pytest.mark.usefixtures("autoland_console") -def test_wait_for_approval_by_review_decision( - mocker, # noqa: ANN001 - decision: str, - approved: bool, -) -> None: - mocker.patch.object(gh.github, "pr_state", return_value="OPEN") - mocker.patch.object(gh.github, "review_decision", return_value=decision) - mocker.patch( - "stack_pr.autoland.runtime.resilient_sleep", side_effect=_StopPollingError - ) +def test_wait_for_approval_by_review_decision(decision: str, approved: bool) -> None: + fakes = FakeServices.polling(sleep=FakeSleep(limit=0)) + fakes.github.reviews[101] = decision + svc = fakes.services entry = _pinned_stack([101])[0] if approved: - assert waits.wait_for_approval(entry, opts=_opts()) + assert svc.waits.wait_for_approval(entry, opts=_opts()) else: # A PR that isn't approved is still waited on at the first poll's sleep. - with pytest.raises(_StopPollingError): - waits.wait_for_approval(entry, opts=_opts()) + with pytest.raises(StopPolling): + svc.waits.wait_for_approval(entry, opts=_opts()) + + +def test_replaced_services_wait_through_their_own_github() -> None: + old, new = FakeGitHub(), FakeGitHub() + new.merged.add(101) + svc = dataclasses.replace(FakeServices.polling(github=old).services, github=new) + + entry = _pinned_stack([101])[0] + assert svc.waits.wait_for_approval(entry, opts=_opts()) + + assert entry.state == PRState.MERGED + assert old.calls == [] # --- merging a run of land steps as a GitHub stack ------------------------- @@ -1982,141 +1956,68 @@ def test_merge_as_stack_defaults_on_and_flag_overrides_config() -> None: assert flagged.merge_as_stack is True -class _FakeNativeStackGitHub: - """The slice of GitHub a stack merge talks to, holding the stacks in memory.""" - - def __init__(self) -> None: - self.stacks: dict[int, list[int]] = {} - self.merged: set[int] = set() - self.created: list[list[int]] = [] - self.unstacked: list[int] = [] - self.merge_requests: list[int] = [] - # (PR, new base) for each base change, and the open PRs' bases at the - # time of each merge request. - self.base_changes: list[tuple[int, str]] = [] - self.bases_at_merge: list[list[tuple[int, str]]] = [] - self.create_error = "" - self.merge_error = "" - self.set_base_error = "" - # The PRs a merge request lands (default: the whole stack up to the PR - # it was made on), and the status GitHub then reports for it. - self.lands: list[int] | None = None - self.status = "merged" - - def find_native_stack(self, pr: int) -> dict | None: - for number, prs in self.stacks.items(): - if pr in prs: - return { - "number": number, - "pull_requests": [ - {"number": n, "state": "closed" if n in self.merged else "open"} - for n in prs - ], - } - return None - - def create_native_stack(self, prs: list[int]) -> dict: - if self.create_error: - raise RuntimeError(self.create_error) - self.created.append(prs) - self.stacks[7] = prs - return {"number": 7} - - def unstack_native_stack(self, number: int) -> None: - self.unstacked.append(number) - self.stacks.pop(number) - - def set_base(self, pr: int, base: str) -> None: - if self.set_base_error: - raise RuntimeError(self.set_base_error) - self.base_changes.append((pr, base)) - - def merge_async(self, pr: int, *, merge_queue: bool) -> str: - self.merge_requests.append(pr) - self.bases_at_merge.append(list(self.base_changes)) - if self.merge_error: - raise RuntimeError(self.merge_error) - prs = next(p for p in self.stacks.values() if pr in p) - self.merged.update(self.lands if self.lands is not None else prs) - return "uuid-1" - - def merge_async_status(self, _pr: int, _uuid: str) -> tuple[str, str]: - return self.status, "conflict in PR #102" if self.status == "failed" else "" - - def pr_state(self, pr: int) -> str: - return "MERGED" if pr in self.merged else "OPEN" - - def in_merge_queue(self, _pr: int) -> bool: - return True - - def has_merge_queue(self, _branch: str) -> bool: - return True - - -def _land_with_fake_github(mocker, plan_text: str, prs: list[int], **opts): # noqa: ANN001, ANN003, ANN202 - """Set up running *plan_text* over a stack of *prs* against a fake GitHub. +class _StackLanding: + """Running a plan over a stack of PRs against a fake GitHub. Approval, checks and mergeability are taken as given, and a PR landed one - at a time merges straight away. Returns the fake, a function that runs the - plan, the rebase mock, and a function listing the PRs landed one at a time. + at a time merges straight away. """ - fake = _FakeNativeStackGitHub() - mocker.patch.object(gh, "github", fake) - mocker.patch("stack_pr.autoland.runtime.console") - mocker.patch("stack_pr.autoland.runtime.resilient_sleep") - mocker.patch("stack_pr.autoland.waits._refresh_last_landed_sha") - mocker.patch("stack_pr.autoland.waits.wait_for_approval", return_value=True) - mocker.patch("stack_pr.autoland.waits.wait_for_checks", return_value=True) - mocker.patch( - "stack_pr.autoland.waits.wait_for_mergeable", - return_value=waits.MergeableResult(ready=True), - ) - rebase = mocker.patch("stack_pr.autoland.waits.rebase_and_resubmit") - def land_one(entry, **_kw) -> bool: # noqa: ANN001, ANN003 - fake.merged.add(entry.pr_number) + def __init__(self, plan_text: str, prs: list[int], **opts: object) -> None: + self.github = FakeGitHub() + self.waits = FakeWaits(queue=self._land_one) + self.fakes = FakeServices(github=self.github, waits=self.waits) + stack = _pinned_stack(prs) + self._ctx = LandingContext(stack=stack, plan=parse_plan(plan_text, stack)) + self._opts = _opts(**opts) + + def _land_one(self, entry: StackEntry) -> bool: + self.github.merged.add(entry.pr_number) entry.state = PRState.MERGED return True - one_at_a_time = mocker.patch( - "stack_pr.autoland.waits.enqueue_and_wait", side_effect=land_one - ) + def execute(self) -> bool: + return engine.execute_plan( + self._ctx, + common_args(), + self._opts, + _NoCheckpoints(), + services=self.fakes.services, + ) - stack = _pinned_stack(prs) - ctx = LandingContext(stack=stack, plan=parse_plan(plan_text, stack)) - checkpointer = AutolandCheckpointer( - path=Path("/dev/null"), branch="feat", base="main" - ) - mocker.patch.object(checkpointer, "save") + @property + def rebases(self) -> int: + return self.fakes.repo.rebases - def execute() -> bool: - return engine.execute_plan(ctx, common_args(), _opts(**opts), checkpointer) + def landed_one_by_one(self) -> list[int]: + return self.waits.waited("queue") - def landed_one_by_one() -> list[int]: - return [c.args[0].pr_number for c in one_at_a_time.call_args_list] - return fake, execute, rebase, landed_one_by_one +def _land_with_fake_github(plan_text: str, prs: list[int], **opts: object): # noqa: ANN202 + """A ``_StackLanding``'s fake GitHub, its execute, the landing (for its + rebases), and its landed_one_by_one.""" + landing = _StackLanding(plan_text, prs, **opts) + return landing.github, landing.execute, landing, landing.landed_one_by_one -def test_consecutive_land_steps_merge_as_one_stack(mocker) -> None: # noqa: ANN001 - fake, execute, rebase, landed_one_by_one = _land_with_fake_github( - mocker, "l\nl\nl\n", [101, 102, 103, 104] - ) +def test_consecutive_land_steps_merge_as_one_stack() -> None: + landing = _StackLanding("l\nl\nl\n", [101, 102, 103, 104]) + fake = landing.github - assert execute() is True + assert landing.execute() is True assert fake.created == [[101, 102, 103]] assert fake.merge_requests == [103] # one request, on the top of the run assert fake.merged == {101, 102, 103} - assert landed_one_by_one() == [] + assert landing.landed_one_by_one() == [] # #104 stays open, so the stack is rebased onto the landed code — once. - rebase.assert_called_once() + assert landing.rebases == 1 -def test_stack_merge_first_bases_the_pr_above_the_run_on_the_target(mocker) -> None: # noqa: ANN001 +def test_stack_merge_first_bases_the_pr_above_the_run_on_the_target() -> None: # Otherwise GitHub closes #104 once the run's branches are deleted. fake, execute, _rebase, _landed = _land_with_fake_github( - mocker, "l\nl\nl\n", [101, 102, 103, 104] + "l\nl\nl\n", [101, 102, 103, 104] ) assert execute() is True @@ -2124,9 +2025,9 @@ def test_stack_merge_first_bases_the_pr_above_the_run_on_the_target(mocker) -> N assert fake.bases_at_merge == [[(104, "main")]] -def test_stack_merge_falls_back_when_the_pr_above_cannot_be_rebased(mocker) -> None: # noqa: ANN001 +def test_stack_merge_falls_back_when_the_pr_above_cannot_be_rebased() -> None: fake, execute, _rebase, landed_one_by_one = _land_with_fake_github( - mocker, "l\nl\n", [101, 102, 103] + "l\nl\n", [101, 102, 103] ) fake.set_base_error = "HTTP 502" @@ -2137,9 +2038,9 @@ def test_stack_merge_falls_back_when_the_pr_above_cannot_be_rebased(mocker) -> N assert landed_one_by_one() == [101, 102] -def test_merge_as_stack_off_lands_one_at_a_time(mocker) -> None: # noqa: ANN001 +def test_merge_as_stack_off_lands_one_at_a_time() -> None: fake, execute, _rebase, landed_one_by_one = _land_with_fake_github( - mocker, "l\nl\n", [101, 102], merge_as_stack=False + "l\nl\n", [101, 102], merge_as_stack=False ) assert execute() is True @@ -2148,9 +2049,9 @@ def test_merge_as_stack_off_lands_one_at_a_time(mocker) -> None: # noqa: ANN001 assert landed_one_by_one() == [101, 102] -def test_native_stack_merge_falls_back_when_the_stack_cannot_be_created(mocker) -> None: # noqa: ANN001 +def test_native_stack_merge_falls_back_when_the_stack_cannot_be_created() -> None: fake, execute, _rebase, landed_one_by_one = _land_with_fake_github( - mocker, "l\nl\nl\n", [101, 102, 103] + "l\nl\nl\n", [101, 102, 103] ) fake.create_error = "HTTP 422: base ref does not match" @@ -2160,9 +2061,9 @@ def test_native_stack_merge_falls_back_when_the_stack_cannot_be_created(mocker) assert landed_one_by_one() == [101, 102, 103] -def test_partly_failed_stack_merge_lands_the_rest_one_at_a_time(mocker) -> None: # noqa: ANN001 +def test_partly_failed_stack_merge_lands_the_rest_one_at_a_time() -> None: fake, execute, rebase, landed_one_by_one = _land_with_fake_github( - mocker, "l\nl\nl\n", [101, 102, 103] + "l\nl\nl\n", [101, 102, 103] ) fake.lands = [101] fake.status = "failed" @@ -2174,17 +2075,16 @@ def test_partly_failed_stack_merge_lands_the_rest_one_at_a_time(mocker) -> None: assert fake.unstacked == [7] assert fake.merge_requests == [103] assert landed_one_by_one() == [102, 103] - assert rebase.call_count == 2 # after the stack merge, and after #102 + assert rebase.rebases == 2 # after the stack merge, and after #102 -def test_native_stack_merge_waits_for_a_request_already_in_flight(mocker) -> None: # noqa: ANN001 +def test_native_stack_merge_waits_for_a_request_already_in_flight() -> None: fake, execute, _rebase, landed_one_by_one = _land_with_fake_github( - mocker, "l\nl\n", [101, 102] + "l\nl\n", [101, 102] ) fake.merge_error = "gh: existing merge request already enqueued (HTTP 409)" # The earlier request merges the PRs while we wait for it. - states = iter(["OPEN", "OPEN", "MERGED", "MERGED"]) - mocker.patch.object(fake, "pr_state", side_effect=lambda _pr: next(states)) + fake.script("pr_state", "OPEN", "OPEN", "MERGED", "MERGED") assert execute() is True @@ -2192,9 +2092,9 @@ def test_native_stack_merge_waits_for_a_request_already_in_flight(mocker) -> Non assert fake.unstacked == [] -def test_stack_merge_keeps_waiting_when_the_queue_lookup_fails(mocker) -> None: # noqa: ANN001 +def test_stack_merge_keeps_waiting_when_the_queue_lookup_fails() -> None: fake, execute, _rebase, landed_one_by_one = _land_with_fake_github( - mocker, "l\nl\n", [101, 102] + "l\nl\n", [101, 102] ) fake.lands = [] # queued, not merged yet fake.status = "enqueued" @@ -2203,7 +2103,7 @@ def lookup_fails(_pr: int) -> None: # The lookup fails while the stack is still queued; it merges meanwhile. fake.merged.update({101, 102}) - mocker.patch.object(fake, "in_merge_queue", side_effect=lookup_fails) + fake.script("in_merge_queue", lookup_fails) assert execute() is True @@ -2212,10 +2112,8 @@ def lookup_fails(_pr: int) -> None: assert landed_one_by_one() == [] -def test_native_stack_merge_reuses_a_stack_the_run_is_at_the_bottom_of(mocker) -> None: # noqa: ANN001 - fake, execute, _rebase, _landed = _land_with_fake_github( - mocker, "l\nl\n", [101, 102, 103] - ) +def test_native_stack_merge_reuses_a_stack_the_run_is_at_the_bottom_of() -> None: + fake, execute, _rebase, _landed = _land_with_fake_github("l\nl\n", [101, 102, 103]) fake.stacks[3] = [101, 102, 103] fake.lands = [101, 102] @@ -2225,9 +2123,9 @@ def test_native_stack_merge_reuses_a_stack_the_run_is_at_the_bottom_of(mocker) - assert fake.merge_requests == [102] -def test_native_stack_merge_leaves_a_mismatched_stack_alone(mocker) -> None: # noqa: ANN001 +def test_native_stack_merge_leaves_a_mismatched_stack_alone() -> None: fake, execute, _rebase, landed_one_by_one = _land_with_fake_github( - mocker, "l\nl\n", [101, 102] + "l\nl\n", [101, 102] ) fake.stacks[3] = [100, 101, 102] # #100 isn't part of this plan @@ -2243,6 +2141,20 @@ def test_native_stack_merge_leaves_a_mismatched_stack_alone(mocker) -> None: # # --------------------------------------------------------------------------- +def _shell(mocker, answer: Callable[[list[str]], object]) -> None: # noqa: ANN001 + """Answer each command run with ``answer(cmd)``: its stdout, or an + exception for the run to raise.""" + + def _run(cmd: list[str], **_kwargs: object) -> subprocess.CompletedProcess: + reply = answer(cmd) + if isinstance(reply, Exception): + raise reply + return subprocess.CompletedProcess(cmd, 0, stdout=reply, stderr="") + + # Both runtime.run and the shared gh JSON helpers run commands through it. + mocker.patch.object(shell_commands, "run_with_retry", side_effect=_run) + + def _gh_replies(mocker, **replies: list) -> None: # noqa: ANN001 """Answer each ``gh `` with the next of *replies[subcommand]*. @@ -2250,23 +2162,22 @@ def _gh_replies(mocker, **replies: list) -> None: # noqa: ANN001 last reply repeats once the others are used up. """ - def _run(cmd: list[str], **_kwargs: object) -> subprocess.CompletedProcess: + def _answer(cmd: list[str]) -> object: queue = replies[cmd[2]] - reply = queue.pop(0) if len(queue) > 1 else queue[0] - if isinstance(reply, Exception): - raise reply - return subprocess.CompletedProcess(cmd, 0, stdout=reply, stderr="") + return queue.pop(0) if len(queue) > 1 else queue[0] - # Both runtime.run and the shared gh JSON helpers run commands through it. - mocker.patch.object(shell_commands, "run_with_retry", side_effect=_run) - mocker.patch.object(runtime, "resilient_sleep", return_value=0.0) + _shell(mocker, _answer) + + +def _real_github() -> Services: + """The real wait loops and GitHub client, over whatever ``gh`` answers.""" + return FakeServices.polling(github=gh.GitHub()).services _OPEN = json.dumps({"state": "OPEN"}) _PASSING = json.dumps([{"name": "ci", "bucket": "pass"}]) -@pytest.mark.usefixtures("autoland_console") @pytest.mark.parametrize( "bad_reply", [ @@ -2282,18 +2193,16 @@ def test_wait_for_checks_polls_again_after_a_failed_state_read( _gh_replies(mocker, view=[bad_reply, _OPEN], checks=[_PASSING]) entry = _pinned_stack([101])[0] - assert waits.wait_for_checks(entry, opts=_opts()) + assert _real_github().waits.wait_for_checks(entry, opts=_opts()) -@pytest.mark.usefixtures("autoland_console") def test_wait_for_checks_polls_again_after_a_malformed_checks_read(mocker) -> None: # noqa: ANN001 _gh_replies(mocker, view=[_OPEN], checks=['{"oops": 1}', "[1]", _PASSING]) entry = _pinned_stack([101])[0] - assert waits.wait_for_checks(entry, opts=_opts()) + assert _real_github().waits.wait_for_checks(entry, opts=_opts()) -@pytest.mark.usefixtures("autoland_console") def test_wait_for_approval_polls_again_after_a_failed_state_read(mocker) -> None: # noqa: ANN001 _gh_replies( mocker, @@ -2307,66 +2216,51 @@ def test_wait_for_approval_polls_again_after_a_failed_state_read(mocker) -> None ) entry = _pinned_stack([101])[0] - assert waits.wait_for_approval(entry, opts=_opts()) + assert _real_github().waits.wait_for_approval(entry, opts=_opts()) assert entry.state == PRState.MERGED -@pytest.mark.usefixtures("autoland_console") +_CLEAN = json.dumps({"state": "OPEN", "mergeStateStatus": "CLEAN"}) + + def test_wait_for_mergeable_polls_again_after_a_malformed_read(mocker) -> None: # noqa: ANN001 - clean = json.dumps({"state": "OPEN", "mergeStateStatus": "CLEAN"}) - _gh_replies(mocker, view=["", RuntimeError("timed out"), clean]) + _gh_replies(mocker, view=["", RuntimeError("timed out"), _CLEAN]) entry = _pinned_stack([101])[0] - result = waits.wait_for_mergeable(entry, opts=_opts()) + result = _real_github().waits.wait_for_mergeable(entry, opts=_opts()) assert result.ready -def _enqueued(mocker) -> None: # noqa: ANN001 - """Skip straight to waiting in the merge queue.""" - mocker.patch("stack_pr.autoland.runtime.console") - mocker.patch.object( - waits, - "wait_for_mergeable", - return_value=waits.MergeableResult(ready=True), - ) - mocker.patch.object(gh.github, "enqueue") - - +@pytest.mark.usefixtures("autoland_console") # the enqueue echoes its command def test_merge_queue_wait_polls_again_after_a_failed_read(mocker) -> None: # noqa: ANN001 - _enqueued(mocker) + # Mergeable, enqueued, then the reads of whether it merged. _gh_replies( - mocker, view=["{truncated", RuntimeError("HTTP 502"), '{"state": "MERGED"}'] + mocker, + view=[_CLEAN, "{truncated", RuntimeError("HTTP 502"), '{"state": "MERGED"}'], + merge=[""], ) entry = _pinned_stack([101])[0] - assert waits.enqueue_and_wait(entry, opts=_opts()) + assert _real_github().waits.enqueue_and_wait(entry, opts=_opts()) assert entry.state == PRState.MERGED +@pytest.mark.usefixtures("autoland_console") # the enqueue echoes its command def test_merge_queue_wait_still_times_out_while_reads_fail(mocker) -> None: # noqa: ANN001 - _enqueued(mocker) - _gh_replies(mocker, view=[RuntimeError("HTTP 502")]) + _gh_replies(mocker, view=[_CLEAN, RuntimeError("HTTP 502")], merge=[""]) entry = _pinned_stack([101])[0] opts = _opts(poll_interval=60, merge_timeout=300) - assert not waits.enqueue_and_wait(entry, opts=opts) + assert not _real_github().waits.enqueue_and_wait(entry, opts=opts) assert entry.error_message == "Timed out waiting for merge queue" -def test_stack_merge_wait_polls_again_after_a_failed_read(mocker) -> None: # noqa: ANN001 +def test_stack_merge_wait_polls_again_after_a_failed_read() -> None: fake, execute, _rebase, landed_one_by_one = _land_with_fake_github( - mocker, "l\nl\n", [101, 102] + "l\nl\n", [101, 102] ) - real_pr_state = fake.pr_state - failures = [RuntimeError("HTTP 502: Bad Gateway")] - - def flaky_pr_state(pr: int) -> str: - if failures: - raise failures.pop() - return real_pr_state(pr) - - mocker.patch.object(fake, "pr_state", side_effect=flaky_pr_state) + fake.script("pr_state", RuntimeError("HTTP 502: Bad Gateway")) assert execute() is True @@ -2401,10 +2295,11 @@ def test_enqueue_is_not_resent_after_a_transient_looking_failure(mocker) -> None def test_merge_async_is_not_resent_after_a_transient_looking_failure(mocker) -> None: # noqa: ANN001 - mocker.patch.object(gh.github, "_owner_repo", ("o", "r")) + github = gh.GitHub() + github._owner_repo = ("o", "r") # noqa: SLF001 fake = _fake_subprocess(mocker, stderr="HTTP 502: Bad Gateway") with pytest.raises(RuntimeError): - gh.github.merge_async(12, merge_queue=True) + github.merge_async(12, merge_queue=True) assert fake.call_count == 1 diff --git a/tests/test_autoland_interrupt.py b/tests/test_autoland_interrupt.py index 3f6d7d5..672635b 100644 --- a/tests/test_autoland_interrupt.py +++ b/tests/test_autoland_interrupt.py @@ -28,15 +28,9 @@ from stack_pr.autoland.state import AutolandCheckpointer, AutolandLock from stack_pr.autoland.steps import AutolandStep from stack_pr.autoland.worktree import Worktree +from tests.autoland_fakes import OPTS, FakeRepo, FakeUI, FakeWaits, stack from tests.helpers import common_args -from tests.test_autoland_step_execution import ( - OPTS, - FakeRepo, - FakeUI, - FakeWaits, - services, - stack, -) +from tests.test_autoland_step_execution import services INTERRUPTED = "User interrupted (Ctrl+C)" REPO_ROOT = Path(__file__).resolve().parents[1] @@ -76,12 +70,11 @@ def input(self, prompt: str) -> str: @pytest.fixture -def resume_io(mocker: MockerFixture) -> Mock: - """Stub the git side of a resume; return the Services.production mock.""" +def resume_io(mocker: MockerFixture) -> None: + """Stub the git side of a resume, and the refresh of its PRs.""" mocker.patch("stack_pr.git.get_current_branch_name", return_value="feat") mocker.patch("stack_pr.cli.deduce_base", side_effect=lambda c: c) mocker.patch("stack_pr.autoland.discovery.enrich_stack") - return mocker.patch.object(Services, "production") @pytest.fixture @@ -103,8 +96,12 @@ def announce_preserved(self) -> None: return events -def _resume(sf: Path, **overrides: object) -> None: - _run_resume(common_args(), replace(OPTS, resume=True, state_file=sf, **overrides)) +def _resume(sf: Path, svc: Services, **overrides: object) -> None: + _run_resume( + common_args(), + replace(OPTS, resume=True, state_file=sf, **overrides), + services=svc, + ) def _printed(console: Mock) -> str: @@ -128,10 +125,10 @@ def _assert_interrupted(sf: Path, console: Mock, *, at_step: int) -> LandingCont [(False, ["preserved"]), (True, ["removed"])], ids=["keeps-worktree", "always-cleanup"], ) +@pytest.mark.usefixtures("resume_io") def test_ctrl_c_during_a_wait_checkpoints_and_exits_130( tmp_path: Path, autoland_console: Mock, - resume_io: Mock, *, worktree_events: list[str], always_cleanup: bool, @@ -139,12 +136,10 @@ def test_ctrl_c_during_a_wait_checkpoints_and_exits_130( ) -> None: sf = tmp_path / "state.json" _save(sf, _land_both()) - resume_io.return_value = services( - waits=_waits([], interrupt_at=102), repo=FakeRepo() - ) + svc = services(waits=_waits([], interrupt_at=102), repo=FakeRepo()) with pytest.raises(SystemExit) as exc: - _resume(sf, branch="feat", always_cleanup=always_cleanup) + _resume(sf, svc, branch="feat", always_cleanup=always_cleanup) assert exc.value.code == 130 saved = _assert_interrupted(sf, autoland_console, at_step=1) @@ -153,15 +148,15 @@ def test_ctrl_c_during_a_wait_checkpoints_and_exits_130( assert worktree_events == worktree +@pytest.mark.usefixtures("resume_io") def test_ctrl_c_at_a_confirm_prompt_checkpoints_and_exits_130( - tmp_path: Path, autoland_console: Mock, resume_io: Mock + tmp_path: Path, autoland_console: Mock ) -> None: sf = tmp_path / "state.json" _save(sf, [LandStep(entry_index=0, pr_number=101), ConfirmStep()]) - resume_io.return_value = services(InterruptedUI(), repo=FakeRepo()) with pytest.raises(SystemExit) as exc: - _resume(sf) + _resume(sf, services(InterruptedUI(), repo=FakeRepo())) assert exc.value.code == 130 saved = _assert_interrupted(sf, autoland_console, at_step=1) @@ -170,22 +165,18 @@ def test_ctrl_c_at_a_confirm_prompt_checkpoints_and_exits_130( assert not confirm.confirmed -@pytest.mark.usefixtures("autoland_console") +@pytest.mark.usefixtures("autoland_console", "resume_io") def test_resume_after_ctrl_c_continues_from_the_interrupted_step( - tmp_path: Path, resume_io: Mock + tmp_path: Path, ) -> None: sf = tmp_path / "state.json" _save(sf, _land_both()) approved: list[int] = [] - resume_io.return_value = services( - waits=_waits(approved, interrupt_at=102), repo=FakeRepo() - ) with pytest.raises(SystemExit): - _resume(sf) + _resume(sf, services(waits=_waits(approved, interrupt_at=102), repo=FakeRepo())) assert approved == [101] - resume_io.return_value = services(waits=_waits(approved), repo=FakeRepo()) - _resume(sf) + _resume(sf, services(waits=_waits(approved), repo=FakeRepo())) # #101 isn't landed again; #102 is landed from the start of its step. assert approved == [101, 102] @@ -203,9 +194,9 @@ def test_resume_after_ctrl_c_continues_from_the_interrupted_step( signal.signal(signal.SIGINT, signal.SIG_IGN) from stack_pr.autoland import commands -from stack_pr.autoland.services import Services +from tests.autoland_fakes import OPTS, FakeRepo, FakeWaits from tests.helpers import common_args -from tests.test_autoland_step_execution import OPTS, FakeRepo, FakeWaits, services +from tests.test_autoland_step_execution import services def block(entry): print("waiting", flush=True) @@ -215,11 +206,12 @@ def block(entry): sf = Path({path!r}) with mock.patch("stack_pr.git.get_current_branch_name", return_value="feat"), \\ mock.patch("stack_pr.cli.deduce_base", side_effect=lambda c: c), \\ - mock.patch("stack_pr.autoland.discovery.enrich_stack"), \\ - mock.patch.object(Services, "production", - return_value=services(waits=FakeWaits(approval=block), - repo=FakeRepo())): - commands._run_resume(common_args(), replace(OPTS, resume=True, state_file=sf)) + mock.patch("stack_pr.autoland.discovery.enrich_stack"): + commands._run_resume( + common_args(), + replace(OPTS, resume=True, state_file=sf), + services=services(waits=FakeWaits(approval=block), repo=FakeRepo()), + ) """ diff --git a/tests/test_autoland_optimize.py b/tests/test_autoland_optimize.py index 47bca31..fa9329d 100644 --- a/tests/test_autoland_optimize.py +++ b/tests/test_autoland_optimize.py @@ -17,7 +17,6 @@ PRState, WorkflowStep, ) -from stack_pr.autoland.native_stack import NativeStackMergeResult from stack_pr.autoland.optimize import ( ExecutionUnit, PolicyRegistry, @@ -29,8 +28,8 @@ from stack_pr.autoland.options import AutolandOptions from stack_pr.autoland.state import AutolandCheckpointer from stack_pr.autoland.steps import AutolandStep, StepResult +from tests.autoland_fakes import OPTS, FakeRepo, FakeServices, FakeUI, stack from tests.helpers import common_args -from tests.test_autoland_step_execution import OPTS, FakeUI, services, stack STACKED = dataclasses.replace(OPTS, merge_as_stack=True) @@ -167,14 +166,14 @@ def test_merge_as_stack_off_leaves_every_step_on_its_own() -> None: ] -def test_preview_announces_each_stack_merge(mocker) -> None: # noqa: ANN001 - console = mocker.patch("stack_pr.autoland.runtime.console") +def test_preview_announces_each_stack_merge() -> None: + ui = FakeUI() plan = [*land(0, 1), WorkflowStep(workflow="deploy.yaml"), *land(2, 3)] run = LandingContext(stack=stack(101, 102, 103, 104), plan=plan) - engine.print_execution_preview(run, STACKED) + engine.print_execution_preview(run, STACKED, ui=ui) - assert [c.args[0] for c in console.print.call_args_list] == [ + assert ui.lines == [ "[dim]Steps 1-2 (#101, #102) will merge together as a GitHub stack.[/dim]", "[dim]Steps 4-5 (#103, #104) will merge together as a GitHub stack.[/dim]", ] @@ -186,10 +185,15 @@ def test_preview_announces_each_stack_merge(mocker) -> None: # noqa: ANN001 class _Recorder: - """A UnitContext's callbacks, recording what the unit does in order.""" + """A UnitContext's callbacks, recording what the unit does in order. - def __init__(self, run: LandingContext) -> None: + The unit's stack merges go to ``fakes.github``, where each PR is approved, + its checks pass, and a merge request merges the whole stack. + """ + + def __init__(self, run: LandingContext, fakes: FakeServices | None = None) -> None: self.run = run + self.fakes = fakes or FakeServices() self.events: list[object] = [] self.abort_at: int | None = None @@ -198,7 +202,7 @@ def context(self) -> UnitContext: run=self.run, opts=STACKED, common=common_args(), - services=services(), + services=self.fakes.services, enter=lambda i: self.events.append(("enter", i)), execute_step=self.execute_step, checkpoint=lambda: self.events.append("checkpoint"), @@ -211,16 +215,17 @@ def execute_step(self, index: int) -> StepResult: return StepResult.done() -def _stack_merge(mocker, results: list[NativeStackMergeResult]): # noqa: ANN001, ANN202 - """Patch the stack merge to return *results* in turn; return the mock.""" - return mocker.patch( - "stack_pr.autoland.optimize.stack_merge.land_as_native_stack", - side_effect=results, - ) +def _falls_through() -> FakeServices: + """Fakes on which every stack merge falls through: the GitHub stack can't + be created, so the PRs are to be landed one at a time.""" + fakes = FakeServices() + fakes.github.create_error = "HTTP 422" + return fakes -def _merged_prs(merge) -> list[list[int]]: # noqa: ANN001 - return [[e.pr_number for e in c.args[0]] for c in merge.call_args_list] +def _merged_prs(fakes: FakeServices) -> list[list[int]]: + """The PRs of each stack merge attempted, bottom first.""" + return fakes.github.stack_requests def _unit(run: LandingContext) -> StackMergeUnit: @@ -229,23 +234,14 @@ def _unit(run: LandingContext) -> StackMergeUnit: return unit -def test_stack_merge_then_each_step_executes(mocker) -> None: # noqa: ANN001 +def test_stack_merge_then_each_step_executes() -> None: run = LandingContext(stack=stack(101, 102, 103), plan=land(0, 1, 2)) - - def lands(entries, **_kw) -> NativeStackMergeResult: # noqa: ANN001, ANN003 - for e in entries: - e.state = PRState.MERGED - return NativeStackMergeResult(landed=True) - - merge = mocker.patch( - "stack_pr.autoland.optimize.stack_merge.land_as_native_stack", - side_effect=lands, - ) recorder = _Recorder(run) assert _unit(run).execute(recorder.context()) == "" - assert _merged_prs(merge) == [[101, 102, 103]] + assert _merged_prs(recorder.fakes) == [[101, 102, 103]] + assert all(e.state == PRState.MERGED for e in run.stack) # Each step is entered (checkpointed as current) before any of its work. assert recorder.events == [ ("enter", 0), @@ -258,17 +254,14 @@ def lands(entries, **_kw) -> NativeStackMergeResult: # noqa: ANN001, ANN003 ] -def test_a_stack_merge_that_falls_through_lands_the_rest_one_at_a_time( - mocker, # noqa: ANN001 -) -> None: +def test_a_stack_merge_that_falls_through_lands_the_rest_one_at_a_time() -> None: run = LandingContext(stack=stack(101, 102, 103), plan=land(0, 1, 2)) - merge = _stack_merge(mocker, [NativeStackMergeResult(landed=False)]) - recorder = _Recorder(run) + recorder = _Recorder(run, _falls_through()) assert _unit(run).execute(recorder.context()) == "" # The merge is not retried for the steps it covered. - assert _merged_prs(merge) == [[101, 102, 103]] + assert _merged_prs(recorder.fakes) == [[101, 102, 103]] assert [e for e in recorder.events if e != "checkpoint"] == [ ("enter", 0), ("step", 0), @@ -279,13 +272,11 @@ def test_a_stack_merge_that_falls_through_lands_the_rest_one_at_a_time( ] -def test_a_stack_merge_abort_stops_before_the_step(mocker) -> None: # noqa: ANN001 - run = LandingContext(stack=stack(101, 102), plan=land(0, 1)) - _stack_merge( - mocker, - [NativeStackMergeResult(abort_reason="Rebase failed after merging #102: boom")], - ) - recorder = _Recorder(run) +def test_a_stack_merge_abort_stops_before_the_step() -> None: + # #103 stays open above the run, so the stack is rebased after the merge. + run = LandingContext(stack=stack(101, 102, 103), plan=land(0, 1)) + fakes = FakeServices(repo=FakeRepo(rebase_error=RuntimeError("boom"))) + recorder = _Recorder(run, fakes) assert ( _unit(run).execute(recorder.context()) @@ -294,28 +285,26 @@ def test_a_stack_merge_abort_stops_before_the_step(mocker) -> None: # noqa: ANN assert recorder.events == [("enter", 0)] -def test_a_step_abort_stops_the_unit(mocker) -> None: # noqa: ANN001 +def test_a_step_abort_stops_the_unit() -> None: run = LandingContext(stack=stack(101, 102, 103), plan=land(0, 1, 2)) - _stack_merge(mocker, [NativeStackMergeResult(landed=False)]) - recorder = _Recorder(run) + recorder = _Recorder(run, _falls_through()) recorder.abort_at = 1 assert _unit(run).execute(recorder.context()) == "step 1 failed" assert ("enter", 2) not in recorder.events -def test_the_run_is_revalidated_when_the_unit_executes(mocker) -> None: # noqa: ANN001 +def test_the_run_is_revalidated_when_the_unit_executes() -> None: run = LandingContext(stack=stack(101, 102, 103), plan=land(0, 1, 2)) unit = _unit(run) # Merged elsewhere after the plan was optimized. run.stack[0].state = PRState.MERGED - merge = _stack_merge(mocker, [NativeStackMergeResult(landed=True)]) recorder = _Recorder(run) assert unit.execute(recorder.context()) == "" # Step 0 now only skips; the stack merge starts above it. - assert _merged_prs(merge) == [[102, 103]] + assert _merged_prs(recorder.fakes) == [[102, 103]] assert recorder.events[:4] == [ ("enter", 0), ("step", 0), @@ -324,16 +313,17 @@ def test_the_run_is_revalidated_when_the_unit_executes(mocker) -> None: # noqa: ] -def test_a_run_shortened_to_one_pr_is_not_stack_merged(mocker) -> None: # noqa: ANN001 +def test_a_run_shortened_to_one_pr_is_not_stack_merged() -> None: run = LandingContext(stack=stack(101, 102), plan=land(0, 1)) unit = _unit(run) run.stack[1].state = PRState.MERGED - merge = _stack_merge(mocker, []) recorder = _Recorder(run) assert unit.execute(recorder.context()) == "" - merge.assert_not_called() + # No stack merge, so nothing asked GitHub or waited on a PR. + assert recorder.fakes.github.calls == [] + assert recorder.fakes.ui.lines == [] assert recorder.events == [("enter", 0), ("step", 0), ("enter", 1), ("step", 1)] @@ -396,19 +386,19 @@ def optimize( path=tmp_path / "state.json", branch="feat", base="main" ) saves = mocker.spy(checkpointer, "save") - console = mocker.patch("stack_pr.autoland.runtime.console") + preview, fakes = FakeUI(), FakeServices() - engine.print_execution_preview(run, OPTS, policies=registry) + engine.print_execution_preview(run, OPTS, ui=preview, policies=registry) assert engine.execute_plan( run, common_args(), OPTS, checkpointer, - services=services(FakeUI()), + services=fakes.services, policies=registry, ) - assert [c.args[0] for c in console.print.call_args_list] == ["pair 0"] + assert preview.lines == ["pair 0"] assert log == ["pair 0"] assert run.current_step == len(plan) # One checkpoint as each step is entered, and one at the end. diff --git a/tests/test_autoland_step_execution.py b/tests/test_autoland_step_execution.py index b8073bb..6ef16e3 100644 --- a/tests/test_autoland_step_execution.py +++ b/tests/test_autoland_step_execution.py @@ -7,15 +7,11 @@ from __future__ import annotations -from collections.abc import Callable -from dataclasses import dataclass, field from pathlib import Path -from typing import cast import pytest -from stack_pr import cli -from stack_pr.autoland import engine, gh +from stack_pr.autoland import engine from stack_pr.autoland.model import ( ConfirmStep, LandingContext, @@ -24,134 +20,22 @@ StackEntry, WorkflowStep, ) -from stack_pr.autoland.options import AutolandOptions from stack_pr.autoland.services import Services from stack_pr.autoland.state import AutolandCheckpointer from stack_pr.autoland.steps import AutolandStep, StepContext, StepResult -from tests.helpers import common_args - -OPTS = AutolandOptions( - merge_queue=True, - required_checks=[], - poll_interval=0, - max_check_retries=0, - max_queue_retries=0, - merge_timeout=0, - workflow_timeout=0, - default_workflow=None, - count=None, - dry_run=False, - branch=None, - interactive=False, - resume=False, - state_file=None, - always_cleanup=False, - merge_as_stack=False, +from tests.autoland_fakes import ( + OPTS, + FakeRepo, + FakeServices, + FakeUI, + FakeWaits, + ForbiddenGitHub, + fails, + forbidden_sleep, + merges, + stack, ) - - -# --------------------------------------------------------------------------- -# Fakes -# --------------------------------------------------------------------------- - - -class FakeUI: - """Records output; answers prompts from a script, then raises EOFError.""" - - def __init__(self, answers: list[str] | None = None) -> None: - self.answers = list(answers or []) - self.lines: list[str] = [] - - def print(self, *objects: object) -> None: - self.lines.append(" ".join(str(o) for o in objects)) - - def input(self, prompt: str) -> str: - self.lines.append(prompt) - if not self.answers: - raise EOFError - return self.answers.pop(0) - - def escape(self, text: str) -> str: - return text.replace("[", r"\[") - - @property - def output(self) -> str: - return "\n".join(self.lines) - - -# What a fake wait does to the PR it waits on. -Outcome = Callable[[StackEntry], bool] - - -def ok(_entry: StackEntry) -> bool: - return True - - -def fails(_entry: StackEntry) -> bool: - return False - - -def merges(entry: StackEntry) -> bool: - entry.state = PRState.MERGED - return True - - -@dataclass -class FakeWaits: - approval: Outcome = ok - checks: Outcome = ok - queue: Outcome = merges - workflow_succeeds: bool = True - calls: list[str] = field(default_factory=list) - workflow_targets: list[str] = field(default_factory=list) - - def wait_for_approval(self, entry: StackEntry, *, opts: AutolandOptions) -> bool: - self.calls.append("approval") - return self.approval(entry) - - def wait_for_checks(self, entry: StackEntry, *, opts: AutolandOptions) -> bool: - self.calls.append("checks") - return self.checks(entry) - - def enqueue_and_wait(self, entry: StackEntry, *, opts: AutolandOptions) -> bool: - self.calls.append("queue") - return self.queue(entry) - - def wait_for_workflow( - self, - step: WorkflowStep, - target_sha: str, - *, - opts: AutolandOptions, - common: cli.CommonArgs, - ctx: LandingContext, - ) -> bool: - self.workflow_targets.append(target_sha) - step.state = "succeeded" if self.workflow_succeeds else "failed" - return self.workflow_succeeds - - -@dataclass -class FakeRepo: - # PR number (None: the target branch head) -> its landed commit. - shas: dict[int | None, str] = field(default_factory=dict) - rebase_error: Exception | None = None - lookups: list[int | None] = field(default_factory=list) - rebases: int = 0 - - def landed_sha(self, common: cli.CommonArgs, pr_number: int | None = None) -> str: - self.lookups.append(pr_number) - return self.shas.get(pr_number, "") - - def rebase_and_resubmit(self, common: cli.CommonArgs) -> None: - self.rebases += 1 - if self.rebase_error is not None: - raise self.rebase_error - - -class _NoGitHub: - def __getattr__(self, name: str) -> object: - raise AssertionError(f"a step called GitHub directly: {name}") +from tests.helpers import common_args def services( @@ -159,28 +43,15 @@ def services( waits: FakeWaits | None = None, repo: FakeRepo | None = None, ) -> Services: - def no_sleep(_seconds: int) -> float: - raise AssertionError("a step slept") - - return Services( - github=cast("gh.GitHub", _NoGitHub()), + """Fake services for a step, which must not call GitHub or sleep itself: + it waits only through ``waits``.""" + return FakeServices( + github=ForbiddenGitHub(), ui=ui or FakeUI(), - waits=waits or FakeWaits(), + sleep=forbidden_sleep, repo=repo or FakeRepo(), - sleep=no_sleep, - ) - - -def stack(*prs: int) -> list[StackEntry]: - return [ - StackEntry( - pr_url=f"https://github.com/o/r/pull/{n}", - pr_number=n, - branch=f"b{n}", - title=f"Title {n}", - ) - for n in prs - ] + waits=waits or FakeWaits(), + ).services def run_step(