diff --git a/src/stack_pr/autoland/__init__.py b/src/stack_pr/autoland/__init__.py index 002d164..0f553c4 100644 --- a/src/stack_pr/autoland/__init__.py +++ b/src/stack_pr/autoland/__init__.py @@ -17,7 +17,8 @@ - ``options``: AutolandOptions, resolved from flags, config, and defaults. - ``runtime``: the console, sleep/wake resilience, and the retrying ``run``. - ``steps``: the AutolandStep base class, the registry of step types, and one - module per type, each owning its plan syntax, persistence, and progress. + module per type, each owning its plan syntax, persistence, progress, view, + and execution; plus StepContext/StepResult, what a step executes with. - ``model``: the stack, plan steps, and LandingContext, and their (de)serialization. - ``state``: the checkpoint (``--resume``) and the per-branch lock. @@ -29,6 +30,8 @@ - ``display``: rendering a plan's progress. - ``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. - ``engine``: executing a plan. - ``status``: the ``--status`` report. - ``commands``: the subparser, and the fresh / resume / replan flows. diff --git a/src/stack_pr/autoland/commands.py b/src/stack_pr/autoland/commands.py index 9d91ba5..2c2ea7d 100644 --- a/src/stack_pr/autoland/commands.py +++ b/src/stack_pr/autoland/commands.py @@ -27,6 +27,7 @@ parse_plan, plan_from_file, ) +from stack_pr.autoland.services import Services from stack_pr.autoland.state import AutolandCheckpointer, AutolandLock, _state_path from stack_pr.autoland.status import show_status from stack_pr.autoland.worktree import Worktree @@ -428,7 +429,9 @@ def _run_fresh(common: cli.CommonArgs, opts: AutolandOptions) -> None: checkpointer, worktree, opts, - success=engine.execute_plan(ctx, common, opts, checkpointer), + success=engine.execute_plan( + ctx, common, opts, checkpointer, services=Services.production() + ), ) finally: if worktree is not None: @@ -501,7 +504,9 @@ def _run_resume(common: cli.CommonArgs, opts: AutolandOptions) -> None: checkpointer, worktree, opts, - success=engine.execute_plan(ctx, common, opts, checkpointer), + success=engine.execute_plan( + ctx, common, opts, checkpointer, services=Services.production() + ), ) finally: if worktree is not None: @@ -723,7 +728,9 @@ def _replan(common: cli.CommonArgs, opts: AutolandOptions, state_path: Path) -> checkpointer, worktree, opts, - success=engine.execute_plan(ctx, common, opts, checkpointer), + success=engine.execute_plan( + ctx, common, opts, checkpointer, services=Services.production() + ), ) finally: if worktree is not None: diff --git a/src/stack_pr/autoland/display.py b/src/stack_pr/autoland/display.py index 199e9b6..ecff426 100644 --- a/src/stack_pr/autoland/display.py +++ b/src/stack_pr/autoland/display.py @@ -311,13 +311,3 @@ def _describe_step(step: PlanStep, ctx: LandingContext) -> str: differently. """ return step.view(ctx.stack, is_next=False, assumed_completed=False).description - - -def _next_steps_lines( - plan: list[PlanStep], from_index: int, ctx: LandingContext -) -> list[str]: - """Numbered, markup-safe descriptions of the plan steps after `from_index`.""" - return [ - f" {n}. {_escape_markup(_describe_step(s, ctx))}" - for n, s in enumerate(plan[from_index + 1 :], 1) - ] diff --git a/src/stack_pr/autoland/engine.py b/src/stack_pr/autoland/engine.py index 6fcc125..0d75bac 100644 --- a/src/stack_pr/autoland/engine.py +++ b/src/stack_pr/autoland/engine.py @@ -1,15 +1,24 @@ -"""Executing a landing plan step by step.""" +"""Executing a landing plan step by step. + +The engine owns the run: it walks the plan from ``ctx.current_step``, +checkpoints before each step, hands each step a ``StepContext``, records the +run-level facts a step reports in its ``StepResult``, and stops on an abort. +What a step does is up to the step (``AutolandStep.execute``). +""" from __future__ import annotations +import functools + from stack_pr import cli -from stack_pr.autoland import runtime, waits -from stack_pr.autoland.display import _escape_markup, _next_steps_lines -from stack_pr.autoland.model import ConfirmStep, LandingContext, LandStep, WorkflowStep +from stack_pr.autoland.display import _describe_step +from stack_pr.autoland.model import LandingContext from stack_pr.autoland.native_stack import land_as_native_stack, native_stack_run 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.state import AutolandCheckpointer +from stack_pr.autoland.steps import StepContext def execute_plan( @@ -17,169 +26,61 @@ def execute_plan( common: cli.CommonArgs, opts: AutolandOptions, checkpointer: AutolandCheckpointer, + *, + services: Services | None = None, ) -> bool: """Execute the landing plan from ctx.current_step. Returns True on success.""" - # A workflow checkpoint after the already-landed prefix of a plan should - # wait for the code of the last PR that prefix landed. Only that last one - # matters, so don't pay for a fetch + lookup on each of the others. - last_prelanded = landed_prefix_end(ctx.plan) - # The last step of a run whose stack merge fell through; the rest of that - # run lands one PR at a time rather than retrying the stack merge. - unstacked_through = -1 + if services is None: + services = Services.production() + prefix_end = landed_prefix_end(ctx.plan) + native_stack = _NativeStackMerge(ctx, common, opts, checkpointer) + + def upcoming(index: int) -> list[str]: + return [_describe_step(s, ctx) for s in ctx.plan[index + 1 :]] + + def landed_sha() -> str: + if not ctx.last_landed_sha: + _record_landed_sha(ctx, services.repo.landed_sha(common)) + return ctx.last_landed_sha for step_idx in range(ctx.current_step, len(ctx.plan)): step = ctx.plan[step_idx] ctx.current_step = step_idx checkpointer.save(ctx) - if isinstance(step, LandStep) and step.already_landed: - runtime.console.print( - f"\n[green]PR #{step.pr_number} already landed, skipping[/green]" - ) - if step_idx == last_prelanded: - waits._refresh_last_landed_sha(ctx, common, step.pr_number) - - elif isinstance(step, LandStep): - entry = ctx.stack[step.entry_index] - ctx.current_index = step.entry_index - - run_entries = ( - native_stack_run(ctx, step_idx) - if opts.merge_as_stack and step_idx > unstacked_through - else [] - ) - if run_entries: - result = land_as_native_stack( - run_entries, ctx=ctx, common=common, opts=opts - ) - if result.abort_reason: - return _abort(ctx, checkpointer, result.abort_reason) - if not result.landed: - unstacked_through = step_idx + len(run_entries) - 1 - checkpointer.save(ctx) - - if entry.is_merged(): - runtime.console.print( - f"\n[green]PR #{entry.pr_number} already merged, skipping[/green]" - ) - waits._refresh_last_landed_sha(ctx, common, entry.pr_number) - continue - - runtime.console.print( - f"\n{'=' * 60}\n[bold]Step {step_idx + 1}/{len(ctx.plan)}: " - f"Landing PR #{entry.pr_number} — {entry.title}[/bold]\n{'=' * 60}" + abort_reason = native_stack.before_step(step_idx) + if abort_reason: + return _abort(ctx, checkpointer, abort_reason) + + result = step.execute( + StepContext( + index=step_idx, + total=len(ctx.plan), + run=ctx, + opts=opts, + common=common, + services=services, + checkpoint=lambda: checkpointer.save(ctx), + upcoming=functools.partial(upcoming, step_idx), + landed_sha=landed_sha, + ends_landed_prefix=step_idx == prefix_end, ) - - if not waits.wait_for_approval(entry, opts=opts, ctx=ctx): - return _abort( - ctx, - checkpointer, - f"PR #{entry.pr_number} approval wait was aborted", - ) - - if entry.is_merged(): - waits._refresh_last_landed_sha(ctx, common, entry.pr_number) - else: - if not waits.wait_for_checks(entry, opts=opts, ctx=ctx): - return _abort( - ctx, - checkpointer, - f"PR #{entry.pr_number} checks failed after retries", - ) - - if entry.is_merged(): - waits._refresh_last_landed_sha(ctx, common, entry.pr_number) - else: - if not waits.enqueue_and_wait(entry, opts=opts, ctx=ctx): - return _abort( - ctx, - checkpointer, - f"PR #{entry.pr_number} failed to merge", - ) - waits._refresh_last_landed_sha(ctx, common, entry.pr_number) - - # Rebase + resubmit whenever commits remain above the one we just - # landed — not only when more *land steps* follow. On a partial - # land, the PRs we're leaving open still need their bases rebased - # onto the newly-landed commit. - has_commits_above = step.entry_index < len(ctx.stack) - 1 - if has_commits_above: - try: - waits.rebase_and_resubmit(common) - except Exception as e: # noqa: BLE001 - report any resubmit failure - return _abort( - ctx, - checkpointer, - f"Rebase failed after merging #{entry.pr_number}: {e}", - ) - - elif isinstance(step, WorkflowStep): - # "succeeded" before the loop reaches it means --replan carried the - # result over from the run this one replaced. - if step.state in ("skipped", "succeeded"): - continue - runtime.console.print( - f"\n{'=' * 60}\n[bold]Step {step_idx + 1}/{len(ctx.plan)}: " - f"Workflow checkpoint — {step.workflow}[/bold]\n{'=' * 60}" - ) - if not ctx.last_landed_sha: - waits._refresh_last_landed_sha(ctx, common) - if not waits.wait_for_workflow(step, opts=opts, common=common, ctx=ctx): - return _abort( - ctx, - checkpointer, - f"Workflow {step.workflow} failed or timed out", - ) - - elif isinstance(step, ConfirmStep): - if step.confirmed: - continue - question = ( - f'Confirm "{_escape_markup(step.condition)}" is complete — ' - "ready to proceed?" - if step.condition - else "Ready to proceed?" - ) - lines = [ - f"\n{'=' * 60}", - ( - f"[bold yellow]Step {step_idx + 1}/{len(ctx.plan)}: " - "Manual confirmation required[/bold yellow]" - ), - f"{'=' * 60}\n", - f"[bold]{question}[/bold]\n", - ] - next_lines = _next_steps_lines(ctx.plan, step_idx, ctx) - if next_lines: - lines.append("[bold]Next steps:[/bold]") - lines.extend(next_lines) - else: - lines.append("[dim]This is the final step in the plan.[/dim]") - runtime.console.print("\n".join(lines)) - while True: - try: - answer = runtime.console.input( - "[yellow]Type y/Y then Enter to continue " - "(Ctrl+C to abort): [/yellow]" - ).strip() - except EOFError: - return _abort( - ctx, - checkpointer, - "Confirm step received EOF — cannot confirm in " - "non-interactive mode", - ) - if answer in ("y", "Y"): - break - runtime.console.print("[dim]Type 'y' or 'Y' to confirm.[/dim]") - step.confirmed = True - runtime.console.print("[green]Confirmed[/green]") + ) + _record_landed_sha(ctx, result.landed_sha) + if result.aborted: + return _abort(ctx, checkpointer, result.abort_reason) ctx.current_step = len(ctx.plan) checkpointer.save(ctx) return True +def _record_landed_sha(ctx: LandingContext, sha: str) -> None: + # An unknown SHA ("") keeps the last one known. + if sha: + ctx.last_landed_sha = sha + + def _abort( ctx: LandingContext, checkpointer: AutolandCheckpointer, reason: str ) -> bool: @@ -187,3 +88,50 @@ def _abort( ctx.aborted = True checkpointer.save(ctx) return False + + +class _NativeStackMerge: + """TEMPORARY: merge a run of consecutive land steps as one GitHub stack. + + With ``--merge-as-stack``, before each step the engine asks whether a run + of two or more open land steps starts there, and if so lands the run with + one stack merge (``native_stack.land_as_native_stack``). The land steps + then find their PRs merged and only skip. A run whose stack merge fell + through is landed one step at a time instead, without retrying the merge. + + This lives in the engine, not in ``LandStep``, because steps are + independent of each other. It is a stopgap: a plan-optimization pass that + groups land steps into one execution unit replaces it. + """ + + def __init__( + self, + ctx: LandingContext, + common: cli.CommonArgs, + opts: AutolandOptions, + checkpointer: AutolandCheckpointer, + ) -> None: + self._ctx = ctx + self._common = common + self._opts = opts + self._checkpointer = checkpointer + # The last step of a run whose stack merge fell through. + self._unstacked_through = -1 + + def before_step(self, step_idx: int) -> str: + """Stack-merge the run starting at *step_idx*, if any. Returns the + reason the plan must stop, or "".""" + if not self._opts.merge_as_stack or step_idx <= self._unstacked_through: + return "" + run_entries = native_stack_run(self._ctx, step_idx) + if not run_entries: + return "" + result = land_as_native_stack( + run_entries, ctx=self._ctx, common=self._common, opts=self._opts + ) + if result.abort_reason: + return result.abort_reason + if not result.landed: + self._unstacked_through = step_idx + len(run_entries) - 1 + self._checkpointer.save(self._ctx) + return "" diff --git a/src/stack_pr/autoland/services.py b/src/stack_pr/autoland/services.py new file mode 100644 index 0000000..12d3eaa --- /dev/null +++ b/src/stack_pr/autoland/services.py @@ -0,0 +1,182 @@ +"""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. +""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +from typing import TYPE_CHECKING, Protocol, cast + +from stack_pr.autoland import display, gh, runtime, waits +from stack_pr.autoland.model import LandingContext + +if TYPE_CHECKING: + from stack_pr import cli + from stack_pr.autoland.model import StackEntry, WorkflowStep + from stack_pr.autoland.options import AutolandOptions + + +class UI(Protocol): + """Where a step talks to the user.""" + + def print(self, *objects: object) -> None: + """Show console markup (``[bold]...[/bold]``) to the user.""" + + def input(self, prompt: str) -> str: + """Ask the user for a line. Raises EOFError when there is no input.""" + + def escape(self, text: str) -> str: + """*text* made safe to embed in markup, so it shows literally.""" + + +class Waits(Protocol): + """The long-running waits of landing a PR or checking a workflow. + + Each returns whether it succeeded; False also when the run was aborted. + The waits record their progress on the entry or step they are given. + """ + + def wait_for_approval( + self, entry: StackEntry, *, opts: AutolandOptions, ctx: LandingContext + ) -> bool: ... + + def wait_for_checks( + self, entry: StackEntry, *, opts: AutolandOptions, ctx: LandingContext + ) -> bool: ... + + def enqueue_and_wait( + self, entry: StackEntry, *, opts: AutolandOptions, ctx: LandingContext + ) -> bool: ... + + def wait_for_workflow( + self, + step: WorkflowStep, + target_sha: str, + *, + opts: AutolandOptions, + common: cli.CommonArgs, + ctx: LandingContext, + ) -> bool: + """Wait for a successful run of the step's workflow at or after + *target_sha*.""" + + +class Repo(Protocol): + """The local clone and the stack-pr operations on it.""" + + 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.""" + + def rebase_and_resubmit(self, common: cli.CommonArgs) -> None: + """Rebase the local stack onto the target and re-submit it. Raises on + failure.""" + + +@dataclass(frozen=True) +class Services: + """Everything a step may use beyond its ``StepContext``'s run state.""" + + 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] + + @classmethod + def production(cls) -> Services: + """The real console, GitHub client, wait loops, and repo.""" + return cls( + github=cast("gh.GitHub", _LiveGitHub()), + ui=_ConsoleUI(), + waits=_ModuleWaits(), + repo=_GitRepo(), + sleep=_resilient_sleep, + ) + + +# --------------------------------------------------------------------------- +# Production implementations: thin delegates resolved at call time +# --------------------------------------------------------------------------- + + +class _LiveGitHub: + """Forwards every attribute to whatever ``gh.github`` is when it is used.""" + + def __getattr__(self, name: str) -> object: + return getattr(gh.github, name) + + +def _resilient_sleep(seconds: int) -> float: + return runtime.resilient_sleep(seconds) + + +class _ConsoleUI: + def print(self, *objects: object) -> None: + runtime.console.print(*objects) + + def input(self, prompt: str) -> str: + return runtime.console.input(prompt) + + def escape(self, text: str) -> str: + return display._escape_markup(text) + + +class _ModuleWaits: + def wait_for_approval( + self, entry: StackEntry, *, opts: AutolandOptions, ctx: LandingContext + ) -> bool: + return waits.wait_for_approval(entry, opts=opts, ctx=ctx) + + def wait_for_checks( + self, entry: StackEntry, *, opts: AutolandOptions, ctx: LandingContext + ) -> bool: + return waits.wait_for_checks(entry, opts=opts, ctx=ctx) + + def enqueue_and_wait( + self, entry: StackEntry, *, opts: AutolandOptions, ctx: LandingContext + ) -> bool: + return waits.enqueue_and_wait(entry, opts=opts, ctx=ctx) + + def wait_for_workflow( + self, + step: WorkflowStep, + target_sha: str, + *, + opts: AutolandOptions, + common: cli.CommonArgs, + ctx: LandingContext, + ) -> bool: + return waits.wait_for_workflow( + step, opts=opts, common=common, ctx=ctx, target_sha=target_sha + ) + + +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/steps/__init__.py b/src/stack_pr/autoland/steps/__init__.py index 38273a8..296ea2f 100644 --- a/src/stack_pr/autoland/steps/__init__.py +++ b/src/stack_pr/autoland/steps/__init__.py @@ -16,6 +16,7 @@ step_from_dict, ) from stack_pr.autoland.steps.confirm import ConfirmStep +from stack_pr.autoland.steps.execution import StepContext, StepResult from stack_pr.autoland.steps.land import LandStep from stack_pr.autoland.steps.view import StepOutcome, StepTone, StepView from stack_pr.autoland.steps.workflow import WorkflowStep @@ -27,8 +28,10 @@ "LandStep", "PlanParser", "ReplanKey", + "StepContext", "StepOutcome", "StepRegistry", + "StepResult", "StepTone", "StepView", "WorkflowStep", diff --git a/src/stack_pr/autoland/steps/base.py b/src/stack_pr/autoland/steps/base.py index 05f5c9c..fe403fb 100644 --- a/src/stack_pr/autoland/steps/base.py +++ b/src/stack_pr/autoland/steps/base.py @@ -2,7 +2,8 @@ A step type owns everything that differs by kind of step: how it is spelled in a plan file, how it is stored in the state file, how it counts toward a run's -progress, and what it reports about itself for display (a ``StepView``). Code +progress, what it reports about itself for display (a ``StepView``), and how +it executes (given a ``StepContext``, returning a ``StepResult``). Code that walks a plan calls these methods instead of switching on the step's type, so adding a step type means adding a class here rather than a branch in every module that handles plans. @@ -18,6 +19,7 @@ if TYPE_CHECKING: from stack_pr.autoland.model import StackEntry + from stack_pr.autoland.steps.execution import StepContext, StepResult from stack_pr.autoland.steps.view import StepView # A workflow or confirm step's identity for replanning: its keyword, its text, @@ -133,6 +135,16 @@ def view( prefix (see ``plan.is_assumed_completed``). """ + # --- execution ----------------------------------------------------------- + + @abstractmethod + def execute(self, sc: StepContext) -> StepResult: + """Do the step's work, or nothing if it is already done. + + A step is independent of every other: it sees the run only through + *sc*, and uses only ``sc.services`` to reach the outside world. + """ + class StepRegistry: """The step types, looked up by state-file kind or by plan keyword.""" diff --git a/src/stack_pr/autoland/steps/confirm.py b/src/stack_pr/autoland/steps/confirm.py index a25eacd..e9c1aec 100644 --- a/src/stack_pr/autoland/steps/confirm.py +++ b/src/stack_pr/autoland/steps/confirm.py @@ -11,6 +11,7 @@ ReplanKey, register_step, ) +from stack_pr.autoland.steps.execution import StepContext, StepResult from stack_pr.autoland.steps.view import StepOutcome, StepTone, StepView if TYPE_CHECKING: @@ -87,3 +88,41 @@ def view( ), data={"condition": self.condition}, ) + + def execute(self, sc: StepContext) -> StepResult: + """Show what comes next, and wait for the user to type y or Y.""" + if self.confirmed: + return StepResult.done() + ui = sc.services.ui + question = ( + f'Confirm "{ui.escape(self.condition)}" is complete — ready to proceed?' + if self.condition + else "Ready to proceed?" + ) + lines = [ + sc.banner("Manual confirmation required", style="bold yellow") + "\n", + f"[bold]{question}[/bold]\n", + ] + upcoming = sc.upcoming() + if upcoming: + lines.append("[bold]Next steps:[/bold]") + lines.extend(f" {n}. {ui.escape(d)}" for n, d in enumerate(upcoming, 1)) + else: + lines.append("[dim]This is the final step in the plan.[/dim]") + ui.print("\n".join(lines)) + while True: + try: + answer = ui.input( + "[yellow]Type y/Y then Enter to continue (Ctrl+C to abort): " + "[/yellow]" + ).strip() + except EOFError: + return StepResult.abort( + "Confirm step received EOF — cannot confirm in non-interactive mode" + ) + if answer in ("y", "Y"): + break + ui.print("[dim]Type 'y' or 'Y' to confirm.[/dim]") + self.confirmed = True + ui.print("[green]Confirmed[/green]") + return StepResult.done() diff --git a/src/stack_pr/autoland/steps/execution.py b/src/stack_pr/autoland/steps/execution.py new file mode 100644 index 0000000..b4e3630 --- /dev/null +++ b/src/stack_pr/autoland/steps/execution.py @@ -0,0 +1,83 @@ +"""What a step is given when it executes (``StepContext``), and what it reports +back (``StepResult``). + +The engine owns the run: it walks the plan, checkpoints, records run-level +facts, and stops on an abort. A step owns only its own work. It reads the run +through its ``StepContext``, uses the tools in ``StepContext.services``, may +update its own fields (and the stack entry it lands), and reports the outcome +and any run-level facts in its ``StepResult``, which the engine records. +""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from stack_pr import cli + from stack_pr.autoland.model import LandingContext + from stack_pr.autoland.options import AutolandOptions + from stack_pr.autoland.services import Services + + +@dataclass(frozen=True) +class StepResult: + """How a step's execution ended. + + ``abort_reason`` is empty when the step succeeded (or had nothing to do), + and otherwise says why the run has to stop. ``landed_sha`` is the commit of + the code the step landed, or "" if it landed nothing or the commit could + not be found; the engine records it as the run's last landed SHA. + """ + + abort_reason: str = "" + landed_sha: str = "" + + @property + def aborted(self) -> bool: + return bool(self.abort_reason) + + @classmethod + def done(cls, *, landed_sha: str = "") -> StepResult: + return cls(landed_sha=landed_sha) + + @classmethod + def abort(cls, reason: str, *, landed_sha: str = "") -> StepResult: + if not reason: + raise ValueError("an aborted step needs a reason") + return cls(abort_reason=reason, landed_sha=landed_sha) + + +@dataclass(frozen=True) +class StepContext: + """What a step executing at plan position ``index`` may see and use.""" + + # The step's 0-based position in the plan, and the plan's length. + index: int + total: int + # The run's state. A step may update its own fields and the stack entry it + # lands, but reports run-level facts through its StepResult instead. + run: LandingContext + opts: AutolandOptions + common: cli.CommonArgs + services: Services + # Save the run's state now, so a resume picks up from here. + checkpoint: Callable[[], None] + # One-line descriptions of the steps after this one, in plan order. + upcoming: Callable[[], list[str]] + # The commit of the code landed so far: the last SHA a step reported, or, + # if none has been, the head of the target branch (which the engine then + # records as the run's landed SHA). "" if it can't be found. + landed_sha: Callable[[], str] + # Whether this step is the last of the plan's already-landed prefix (see + # ``plan.landed_prefix_end``), the one whose landed code later steps follow. + ends_landed_prefix: bool = False + + def banner(self, title: str, style: str = "bold") -> str: + """The markup heading this step's output: "Step i/n: ".""" + rule = "=" * 60 + return ( + f"\n{rule}\n[{style}]Step {self.index + 1}/{self.total}: " + f"{title}[/{style}]\n{rule}" + ) diff --git a/src/stack_pr/autoland/steps/land.py b/src/stack_pr/autoland/steps/land.py index 10c10d9..13a3c4f 100644 --- a/src/stack_pr/autoland/steps/land.py +++ b/src/stack_pr/autoland/steps/land.py @@ -9,6 +9,7 @@ from stack_pr.autoland import gh from stack_pr.autoland.pr_state import PRState from stack_pr.autoland.steps.base import AutolandStep, PlanParser, register_step +from stack_pr.autoland.steps.execution import StepContext, StepResult from stack_pr.autoland.steps.view import StepOutcome, StepTone, StepView if TYPE_CHECKING: @@ -144,6 +145,60 @@ def view( }, ) + def execute(self, sc: StepContext) -> StepResult: + """Wait for approval and checks, merge through the queue, then rebase + and re-submit whatever is left of the stack above.""" + ui, repo = sc.services.ui, sc.services.repo + if self.already_landed: + ui.print(f"\n[green]PR #{self.pr_number} already landed, skipping[/green]") + # Later steps follow the code of the prefix's last PR; looking up + # the others would only cost a fetch each. + if sc.ends_landed_prefix: + return StepResult.done( + landed_sha=repo.landed_sha(sc.common, self.pr_number) + ) + return StepResult.done() + + entry = sc.run.stack[self.entry_index] + sc.run.current_index = self.entry_index + if entry.is_merged(): + ui.print(f"\n[green]PR #{entry.pr_number} already merged, skipping[/green]") + return StepResult.done( + landed_sha=repo.landed_sha(sc.common, entry.pr_number) + ) + + ui.print(sc.banner(f"Landing PR #{entry.pr_number} — {entry.title}")) + waits, opts, run = sc.services.waits, sc.opts, sc.run + # Each wait may find the PR merged by then (e.g. by hand), which ends + # the landing early. + if not waits.wait_for_approval(entry, opts=opts, ctx=run): + return StepResult.abort(f"PR #{entry.pr_number} approval wait was aborted") + if not entry.is_merged() and not waits.wait_for_checks( + entry, opts=opts, ctx=run + ): + return StepResult.abort( + f"PR #{entry.pr_number} checks failed after retries" + ) + if not entry.is_merged() and not waits.enqueue_and_wait( + entry, opts=opts, ctx=run + ): + return StepResult.abort(f"PR #{entry.pr_number} failed to merge") + landed_sha = repo.landed_sha(sc.common, entry.pr_number) + + # Rebase + resubmit whenever commits remain above the one we just + # landed — not only when more *land steps* follow. On a partial land, + # the PRs we're leaving open still need their bases rebased onto the + # newly-landed commit. + if self.entry_index < len(sc.run.stack) - 1: + try: + repo.rebase_and_resubmit(sc.common) + except Exception as e: # noqa: BLE001 - report any resubmit failure + return StepResult.abort( + f"Rebase failed after merging #{entry.pr_number}: {e}", + landed_sha=landed_sha, + ) + return StepResult.done(landed_sha=landed_sha) + def _current_owner_repo() -> tuple[str, str]: """The repository autoland is landing into, as ``(owner, name)``.""" diff --git a/src/stack_pr/autoland/steps/workflow.py b/src/stack_pr/autoland/steps/workflow.py index 6ec0fc1..143f015 100644 --- a/src/stack_pr/autoland/steps/workflow.py +++ b/src/stack_pr/autoland/steps/workflow.py @@ -11,6 +11,7 @@ ReplanKey, register_step, ) +from stack_pr.autoland.steps.execution import StepContext, StepResult from stack_pr.autoland.steps.view import StepOutcome, StepTone, StepView if TYPE_CHECKING: @@ -83,3 +84,17 @@ def view( detail=(self.error_message,) if self.error_message else (), data={"workflow": self.workflow}, ) + + def execute(self, sc: StepContext) -> StepResult: + """Wait for a successful run of the workflow that includes the code + landed so far.""" + # "succeeded" before the run reaches it means --replan carried the + # result over from the run this one replaced. + if self.state in ("skipped", "succeeded"): + return StepResult.done() + sc.services.ui.print(sc.banner(f"Workflow checkpoint — {self.workflow}")) + if not sc.services.waits.wait_for_workflow( + self, sc.landed_sha(), opts=sc.opts, common=sc.common, ctx=sc.run + ): + return StepResult.abort(f"Workflow {self.workflow} failed or timed out") + return StepResult.done() diff --git a/src/stack_pr/autoland/waits.py b/src/stack_pr/autoland/waits.py index 8c6b8fb..79521a9 100644 --- a/src/stack_pr/autoland/waits.py +++ b/src/stack_pr/autoland/waits.py @@ -168,9 +168,14 @@ def wait_for_workflow( opts: AutolandOptions, common: cli.CommonArgs, ctx: LandingContext, + target_sha: str | None = None, ) -> bool: - """Wait for a workflow to complete with code at or after the landed SHA.""" - target_sha = ctx.last_landed_sha + """Wait for a workflow to complete with code at or after *target_sha*. + + *target_sha* defaults to the run's last landed SHA. + """ + if target_sha is None: + target_sha = ctx.last_landed_sha step.state = "waiting" runtime.console.print( f"\n[bold blue]Waiting for workflow: {step.workflow}[/bold blue]" diff --git a/tests/test_autoland.py b/tests/test_autoland.py index 0e8d3b6..813421b 100644 --- a/tests/test_autoland.py +++ b/tests/test_autoland.py @@ -38,7 +38,6 @@ _POINTER_CELLS, _UNICODE_GLYPHS, _describe_step, - _next_steps_lines, _pick_glyphs, _plan_rows, _StepRow, @@ -1073,25 +1072,6 @@ def test_rich_status_truncates_titles_but_never_the_failure_reason() -> None: assert stack[0].error_message in " ".join(text.split()) -def test_next_steps_lines_numbers_remaining_only() -> None: - stack = [StackEntry(pr_url="u", pr_number=7, branch="b", title="X [1,25] Y")] - plan = [ConfirmStep(), LandStep(entry_index=0), WorkflowStep(workflow="d.yaml")] - ctx = LandingContext(stack=stack, plan=plan) - - lines = _next_steps_lines(plan, 0, ctx) - - assert len(lines) == 2 - assert lines[0].startswith(" 1. Land PR #7:") - assert "1,25" in lines[0] # the title's content survives (escaped or not) - assert lines[1] == " 2. Wait for workflow d.yaml" - - -def test_next_steps_lines_empty_for_final_step() -> None: - plan = [LandStep(entry_index=0), ConfirmStep()] - ctx = LandingContext(stack=_stack(1), plan=plan) - assert _next_steps_lines(plan, 1, ctx) == [] - - # --- executing a partially-landed plan ----------------------------------- diff --git a/tests/test_autoland_step_execution.py b/tests/test_autoland_step_execution.py new file mode 100644 index 0000000..405e18c --- /dev/null +++ b/tests/test_autoland_step_execution.py @@ -0,0 +1,460 @@ +"""Executing each step type in isolation, against fake services. + +Each test builds a ``StepContext`` by hand and injects fakes for the console, +the waits, and the repo, with no module patching: this is the whole contract +between a step and the engine that runs it. +""" + +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.model import ( + ConfirmStep, + LandingContext, + LandStep, + PRState, + 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, +) + + +# --------------------------------------------------------------------------- +# 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, ctx: LandingContext + ) -> bool: + self.calls.append("approval") + return self.approval(entry) + + def wait_for_checks( + self, entry: StackEntry, *, opts: AutolandOptions, ctx: LandingContext + ) -> bool: + self.calls.append("checks") + return self.checks(entry) + + def enqueue_and_wait( + self, entry: StackEntry, *, opts: AutolandOptions, ctx: LandingContext + ) -> 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}") + + +def services( + ui: FakeUI | None = None, + 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()), + ui=ui or FakeUI(), + waits=waits or FakeWaits(), + 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 + ] + + +def run_step( + step: AutolandStep, + svc: Services, + *, + entries: list[StackEntry] | None = None, + upcoming: list[str] | None = None, + landed_sha: str = "", + ends_landed_prefix: bool = False, +) -> StepResult: + """Execute *step* as step 2 of a 3-step plan.""" + run = LandingContext(stack=entries or stack(101, 102), plan=[step]) + return step.execute( + StepContext( + index=1, + total=3, + run=run, + opts=OPTS, + common=common_args(), + services=svc, + checkpoint=lambda: None, + upcoming=lambda: list(upcoming or []), + landed_sha=lambda: landed_sha, + ends_landed_prefix=ends_landed_prefix, + ) + ) + + +# --------------------------------------------------------------------------- +# LandStep +# --------------------------------------------------------------------------- + + +def test_land_step_lands_and_rebases_the_stack_above() -> None: + waits, repo, ui = FakeWaits(), FakeRepo(shas={101: "sha101"}), FakeUI() + entries = stack(101, 102) + + result = run_step( + LandStep(entry_index=0), services(ui, waits, repo), entries=entries + ) + + assert result == StepResult(landed_sha="sha101") + assert waits.calls == ["approval", "checks", "queue"] + assert entries[0].is_merged() + assert repo.lookups == [101] + assert repo.rebases == 1 + assert "Step 2/3: Landing PR #101 — Title 101" in ui.output + + +def test_land_step_at_the_top_of_the_stack_does_not_rebase() -> None: + repo = FakeRepo(shas={102: "sha102"}) + + result = run_step(LandStep(entry_index=1), services(repo=repo)) + + assert result == StepResult(landed_sha="sha102") + assert repo.rebases == 0 + + +def test_land_step_stops_waiting_once_the_pr_has_merged() -> None: + # Merged by hand while waiting for approval: no checks, no queue. + waits = FakeWaits(approval=merges) + + result = run_step(LandStep(entry_index=0), services(waits=waits)) + + assert not result.aborted + assert waits.calls == ["approval"] + + +def test_land_step_skips_a_pr_that_is_already_merged() -> None: + waits, repo, ui = FakeWaits(), FakeRepo(shas={101: "sha101"}), FakeUI() + entries = stack(101, 102) + entries[0].state = PRState.MERGED + + result = run_step( + LandStep(entry_index=0), services(ui, waits, repo), entries=entries + ) + + assert result == StepResult(landed_sha="sha101") + assert waits.calls == [] + assert repo.rebases == 0 + assert "PR #101 already merged, skipping" in ui.output + + +@pytest.mark.parametrize( + ("ends_landed_prefix", "landed_sha", "lookups"), + [(True, "sha99", [99]), (False, "", [])], +) +def test_already_landed_step_reports_its_sha_only_at_the_end_of_the_prefix( + ends_landed_prefix: bool, landed_sha: str, lookups: list[int] +) -> None: + repo, ui = FakeRepo(shas={99: "sha99"}), FakeUI() + + result = run_step( + LandStep(entry_index=-1, pr_number=99), + services(ui, repo=repo), + ends_landed_prefix=ends_landed_prefix, + ) + + assert result == StepResult(landed_sha=landed_sha) + assert repo.lookups == lookups + assert "PR #99 already landed, skipping" in ui.output + + +@pytest.mark.parametrize( + ("waits", "reason"), + [ + (FakeWaits(approval=fails), "PR #101 approval wait was aborted"), + (FakeWaits(checks=fails), "PR #101 checks failed after retries"), + (FakeWaits(queue=fails), "PR #101 failed to merge"), + ], + ids=["approval", "checks", "queue"], +) +def test_land_step_aborts_when_a_wait_fails(waits: FakeWaits, reason: str) -> None: + repo = FakeRepo(shas={101: "sha101"}) + + result = run_step(LandStep(entry_index=0), services(waits=waits, repo=repo)) + + assert result == StepResult(abort_reason=reason) + assert repo.rebases == 0 + + +def test_land_step_aborts_on_a_failed_rebase_but_reports_what_landed() -> None: + repo = FakeRepo(shas={101: "sha101"}, rebase_error=RuntimeError("conflict")) + + result = run_step(LandStep(entry_index=0), services(repo=repo)) + + assert result == StepResult( + abort_reason="Rebase failed after merging #101: conflict", + landed_sha="sha101", + ) + + +# --------------------------------------------------------------------------- +# WorkflowStep +# --------------------------------------------------------------------------- + + +def test_workflow_step_waits_for_the_landed_code() -> None: + waits, ui = FakeWaits(), FakeUI() + step = WorkflowStep(workflow="deploy.yaml") + + result = run_step(step, services(ui, waits), landed_sha="abc123") + + assert result == StepResult() + assert waits.workflow_targets == ["abc123"] + assert step.state == "succeeded" + assert "Step 2/3: Workflow checkpoint — deploy.yaml" in ui.output + + +def test_workflow_step_aborts_when_the_workflow_fails() -> None: + step = WorkflowStep(workflow="deploy.yaml") + + result = run_step(step, services(waits=FakeWaits(workflow_succeeds=False))) + + assert result == StepResult(abort_reason="Workflow deploy.yaml failed or timed out") + assert step.state == "failed" + + +@pytest.mark.parametrize("state", ["skipped", "succeeded"]) +def test_workflow_step_already_done_does_nothing(state: str) -> None: + waits, ui = FakeWaits(), FakeUI() + + result = run_step(WorkflowStep(workflow="d.yaml", state=state), services(ui, waits)) + + assert result == StepResult() + assert waits.workflow_targets == [] + assert ui.lines == [] + + +# --------------------------------------------------------------------------- +# ConfirmStep +# --------------------------------------------------------------------------- + + +def test_confirm_step_waits_for_y() -> None: + ui = FakeUI(answers=["n", "Y"]) + step = ConfirmStep(condition="QA [done]") + + result = run_step(step, services(ui), upcoming=["Land PR #7: X [1,25] Y", "B"]) + + assert result == StepResult() + assert step.confirmed + output = ui.output + assert "Step 2/3: Manual confirmation required" in output + assert r'Confirm "QA \[done]" is complete' in output # escaped + # The steps after this one, numbered from 1, with markup escaped. + assert "Next steps:" in output + assert r" 1. Land PR #7: X \[1,25] Y" in output + assert " 2. B" in output + assert "Type 'y' or 'Y' to confirm." in output # after the "n" + assert ui.lines[-1] == "[green]Confirmed[/green]" + + +def test_confirm_step_as_the_final_step() -> None: + ui = FakeUI(answers=["y"]) + + run_step(ConfirmStep(), services(ui)) + + assert "Ready to proceed?" in ui.output + assert "This is the final step in the plan." in ui.output + assert "Next steps:" not in ui.output + + +def test_confirm_step_aborts_on_eof() -> None: + step = ConfirmStep() + + result = run_step(step, services(FakeUI())) + + assert result == StepResult( + abort_reason=( + "Confirm step received EOF — cannot confirm in non-interactive mode" + ) + ) + assert not step.confirmed + + +def test_confirmed_step_does_not_prompt() -> None: + ui = FakeUI() + + assert run_step(ConfirmStep(confirmed=True), services(ui)) == StepResult() + assert ui.lines == [] + + +# --------------------------------------------------------------------------- +# The engine records what steps report +# --------------------------------------------------------------------------- + + +def test_engine_passes_a_landed_sha_on_to_later_steps(tmp_path: Path) -> None: + waits = FakeWaits() + repo = FakeRepo(shas={101: "sha101", None: "head"}) + plan: list[AutolandStep] = [ + WorkflowStep(workflow="before.yaml"), + LandStep(entry_index=0, pr_number=101), + WorkflowStep(workflow="after.yaml"), + ] + ctx = LandingContext(stack=stack(101, 102), plan=plan) + checkpointer = AutolandCheckpointer( + path=tmp_path / "state.json", branch="feat", base="main" + ) + + assert engine.execute_plan( + ctx, + common_args(), + OPTS, + checkpointer, + services=services(FakeUI(), waits, repo), + ) + + # Before anything lands, a workflow follows the target branch's head; after + # the land step, the commit it reported. + assert waits.workflow_targets == ["head", "sha101"] + assert ctx.last_landed_sha == "sha101" + assert ctx.current_step == len(plan) + + +def test_engine_records_the_reason_a_step_aborted(tmp_path: Path) -> None: + ctx = LandingContext(stack=stack(101), plan=[ConfirmStep()]) + checkpointer = AutolandCheckpointer( + path=tmp_path / "state.json", branch="feat", base="main" + ) + + assert not engine.execute_plan( + ctx, common_args(), OPTS, checkpointer, services=services(FakeUI()) + ) + + assert ctx.aborted + assert ctx.abort_reason.startswith("Confirm step received EOF") + _, saved = AutolandCheckpointer.load(checkpointer.path) + assert saved.abort_reason == ctx.abort_reason diff --git a/tests/test_autoland_steps.py b/tests/test_autoland_steps.py index c1806bf..1233a55 100644 --- a/tests/test_autoland_steps.py +++ b/tests/test_autoland_steps.py @@ -20,8 +20,10 @@ ConfirmStep, LandStep, PlanParser, + StepContext, StepOutcome, StepRegistry, + StepResult, StepTone, StepView, WorkflowStep, @@ -106,6 +108,9 @@ def view( description="Dummy", ) + def execute(self, sc: StepContext) -> StepResult: + return StepResult.done() + @pytest.mark.parametrize( ("kind", "keyword", "message"),