diff --git a/src/stack_pr/autoland/display.py b/src/stack_pr/autoland/display.py index a56dac9..199e9b6 100644 --- a/src/stack_pr/autoland/display.py +++ b/src/stack_pr/autoland/display.py @@ -4,50 +4,22 @@ import sys from dataclasses import astuple, dataclass -from enum import Enum from typing import Any from stack_pr.autoland import runtime -from stack_pr.autoland.model import ( - ConfirmStep, - LandingContext, - LandStep, - PlanStep, - PRState, - StackEntry, - WorkflowStep, -) +from stack_pr.autoland.model import LandingContext, PlanStep from stack_pr.autoland.plan import ( generate_default_plan, is_assumed_completed, landed_prefix_end, ) from stack_pr.autoland.runtime import Group, Padding, Text +from stack_pr.autoland.steps import StepOutcome, StepTone, StepView # --------------------------------------------------------------------------- # Status display # --------------------------------------------------------------------------- -STATE_LABELS = { - PRState.PENDING: "Pending", - PRState.WAITING_FOR_APPROVAL: "Waiting for approval", - PRState.WAITING_FOR_CHECKS: "Waiting for checks", - PRState.IN_MERGE_QUEUE: "In merge queue", - PRState.WAITING_FOR_WORKFLOW: "Waiting for workflow", - PRState.MERGED: "Merged", - PRState.FAILED: "Failed", -} - -STATE_STYLES = { - PRState.PENDING: "dim", - PRState.WAITING_FOR_APPROVAL: "magenta", - PRState.WAITING_FOR_CHECKS: "yellow", - PRState.IN_MERGE_QUEUE: "cyan", - PRState.WAITING_FOR_WORKFLOW: "blue", - PRState.MERGED: "green", - PRState.FAILED: "red bold", -} - _REVIEW_DECISION_DISPLAY = { "APPROVED": ("Approved", "green"), "REVIEW_REQUIRED": ("Review required", "magenta"), @@ -55,15 +27,6 @@ } -class _Outcome(Enum): - """The broad state a step is in, independent of how it is drawn.""" - - DONE = "done" - ACTIVE = "active" - PENDING = "pending" - FAILED = "failed" - - # Every plan line starts with an icon, so a plan scans as a column of glyphs # before any of the words are read. Column alignment depends on the widths # below holding for whichever set is in use, so both sets are fixed-width: @@ -126,7 +89,7 @@ def _pick_glyphs() -> _Glyphs: GLYPHS = _pick_glyphs() -def _icon(outcome: _Outcome) -> str: +def _icon(outcome: StepOutcome) -> str: """The current glyph set's icon for *outcome*. Looked up per render rather than baked into the tables below, so a test can @@ -135,55 +98,57 @@ def _icon(outcome: _Outcome) -> str: return str(getattr(GLYPHS, outcome.value)) -_PR_STATE_OUTCOMES = { - PRState.PENDING: _Outcome.PENDING, - PRState.WAITING_FOR_APPROVAL: _Outcome.ACTIVE, - PRState.WAITING_FOR_CHECKS: _Outcome.ACTIVE, - PRState.IN_MERGE_QUEUE: _Outcome.ACTIVE, - PRState.WAITING_FOR_WORKFLOW: _Outcome.ACTIVE, - PRState.MERGED: _Outcome.DONE, - PRState.FAILED: _Outcome.FAILED, +# The color each kind of status is drawn in. +_TONE_STYLES = { + StepTone.PLAIN: "", + StepTone.MUTED: "dim", + StepTone.SUCCESS: "green", + StepTone.FAILURE: "red bold", + StepTone.AWAITING_REVIEW: "magenta", + StepTone.AWAITING_CHECKS: "yellow", + StepTone.AWAITING_USER: "yellow", + StepTone.IN_QUEUE: "cyan", + StepTone.AWAITING_WORKFLOW: "blue", } -# "skipped" is set only by WorkflowStep.assume_completed, i.e. for a workflow -# that sits inside a plan's already-landed prefix — it must have run before -# those PRs could merge, so the plan assumes it did rather than claiming it was -# deliberately skipped. -_WORKFLOW_STEP_STATUS = { - "pending": (_Outcome.PENDING, "Pending", "dim"), - "waiting": (_Outcome.ACTIVE, "Waiting for workflow", "blue"), - "succeeded": (_Outcome.DONE, "Workflow complete", "green"), - "failed": (_Outcome.FAILED, "Failed", "red bold"), - "skipped": (_Outcome.DONE, "Assumed completed", "green"), -} - -_STEP_KINDS = {LandStep: "Land", WorkflowStep: "Workflow", ConfirmStep: "Confirm"} - - -def _land_entry(ctx: LandingContext, step: LandStep) -> StackEntry | None: - """The stack entry a land step targets, or None if its PR already landed.""" - if step.already_landed: - return None - return ctx.stack[step.entry_index] - @dataclass(frozen=True) class _StepRow: - """One plan step, flattened into the fields the display lays out.""" + """One plan step: its view, plus where it sits in the plan.""" is_next: bool # the first step with work left to do - outcome: _Outcome number: str # "1.", "2." ... - status: str - status_style: str kind: str # Land / Workflow / Confirm - key: str # "#103", "deploy.yaml", the confirm condition - note: str # PR title, or "" when there is nothing to add - detail: str # second line: error message, retry counts, ... + view: StepView + + @property + def outcome(self) -> StepOutcome: + return self.view.outcome + + @property + def status(self) -> str: + return self.view.status + + @property + def status_style(self) -> str: + return _TONE_STYLES[self.view.tone] + + @property + def key(self) -> str: + return self.view.key + + @property + def note(self) -> str: + return self.view.note + + @property + def detail(self) -> str: + """The second line: error message, retry counts, ... ("" for none).""" + return GLYPHS.sep.join(self.view.detail) @property def done(self) -> bool: - return self.outcome is _Outcome.DONE + return self.outcome is StepOutcome.DONE def _pointer_index(ctx: LandingContext, plan: list[PlanStep]) -> int: @@ -198,94 +163,22 @@ def _pointer_index(ctx: LandingContext, plan: list[PlanStep]) -> int: return max(ctx.current_step, landed_prefix_end(plan) + 1) -def _step_status( - step: PlanStep, - ctx: LandingContext, - *, - index: int, - plan: list[PlanStep], - pointer: int, -) -> tuple[_Outcome, str, str]: - """The (outcome, label, style) a step shows in the status column.""" - if isinstance(step, LandStep): - if step.already_landed: - return _Outcome.DONE, "Landed", "green" - entry = ctx.stack[step.entry_index] - return ( - _PR_STATE_OUTCOMES.get(entry.state, _Outcome.PENDING), - STATE_LABELS.get(entry.state, str(entry.state)), - STATE_STYLES.get(entry.state, ""), - ) - if isinstance(step, WorkflowStep): - return _WORKFLOW_STEP_STATUS.get( - step.state, (_Outcome.PENDING, "Pending", "dim") - ) - # ConfirmStep. A confirmation inside the landed prefix was never actually - # given in this run — it is inferred from the fact that later PRs merged — - # so say so rather than claiming the user confirmed it. - assert isinstance(step, ConfirmStep) # the only step type left - if step.confirmed: - if is_assumed_completed(plan, index): - return _Outcome.DONE, "Assumed completed", "green" - return _Outcome.DONE, "Confirmed", "green" - if index == pointer: - return _Outcome.ACTIVE, "Awaiting confirmation", "yellow" - return _Outcome.PENDING, "Pending", "dim" - - -def _step_target(step: PlanStep, ctx: LandingContext) -> tuple[str, str]: - """The step's subject as a (key, note) pair — e.g. ("#103", "Fix the thing").""" - if isinstance(step, LandStep): - entry = _land_entry(ctx, step) - if entry is None: - return f"#{step.pr_number}", "" - return f"#{entry.pr_number}", entry.title or "(untitled)" - if isinstance(step, WorkflowStep): - return step.workflow, "" - assert isinstance(step, ConfirmStep) # the only step type left - return step.condition or "(no condition)", "" - - -def _step_detail(step: PlanStep, ctx: LandingContext) -> str: - """The step's second line, or "" when the header says everything.""" - parts: list[str] = [] - if isinstance(step, LandStep): - entry = _land_entry(ctx, step) - if entry is None: - return "" - if entry.error_message: - parts.append(entry.error_message) - # Retry counts only matter once something has been retried. - if entry.check_retries or entry.queue_retries: - parts.append(f"retries CI {entry.check_retries} / MQ {entry.queue_retries}") - elif isinstance(step, WorkflowStep) and step.error_message: - parts.append(step.error_message) - return GLYPHS.sep.join(parts) - - def _plan_rows(ctx: LandingContext) -> list[_StepRow]: plan = ctx.plan or generate_default_plan(ctx.stack) pointer = _pointer_index(ctx, plan) - rows = [] - for index, step in enumerate(plan): - outcome, status, style = _step_status( - step, ctx, index=index, plan=plan, pointer=pointer - ) - key, note = _step_target(step, ctx) - rows.append( - _StepRow( + return [ + _StepRow( + is_next=index == pointer, + number=f"{index + 1}.", + kind=step.label, + view=step.view( + ctx.stack, is_next=index == pointer, - outcome=outcome, - number=f"{index + 1}.", - status=status, - status_style=style, - kind=_STEP_KINDS[type(step)], - key=key, - note=note, - detail=_step_detail(step, ctx), - ) + assumed_completed=is_assumed_completed(plan, index), + ), ) - return rows + for index, step in enumerate(plan) + ] def _plan_headline(rows: list[_StepRow]) -> str: @@ -413,22 +306,11 @@ def _escape_markup(text: str) -> str: def _describe_step(step: PlanStep, ctx: LandingContext) -> str: """A one-line, human-readable description of a plan step. - Prose for the confirm prompt's "Next steps" list, built from the same - (key, note) the status panel shows, so the two cannot name a step + Prose for the confirm prompt's "Next steps" list. It comes from the same + ``StepView`` the status panel shows, so the two cannot name a step differently. """ - key, note = _step_target(step, ctx) - if isinstance(step, LandStep): - if step.already_landed: - return f"Land PR {key} (already landed)" - return f"Land PR {key}: {note}" - if isinstance(step, WorkflowStep): - return f"Wait for workflow {key}" - # ConfirmStep - assert isinstance(step, ConfirmStep) # the only step type left - if step.condition: - return f"Manual confirmation: {key}" - return "Manual confirmation" + return step.view(ctx.stack, is_next=False, assumed_completed=False).description def _next_steps_lines( diff --git a/src/stack_pr/autoland/model.py b/src/stack_pr/autoland/model.py index 8175cd5..44add89 100644 --- a/src/stack_pr/autoland/model.py +++ b/src/stack_pr/autoland/model.py @@ -3,8 +3,8 @@ from __future__ import annotations from dataclasses import dataclass, field -from enum import Enum +from stack_pr.autoland.pr_state import PRState from stack_pr.autoland.steps import ( AutolandStep, ConfirmStep, @@ -29,16 +29,6 @@ # --------------------------------------------------------------------------- -class PRState(str, Enum): - PENDING = "pending" - WAITING_FOR_APPROVAL = "waiting_for_approval" - WAITING_FOR_CHECKS = "waiting_for_checks" - IN_MERGE_QUEUE = "in_merge_queue" - WAITING_FOR_WORKFLOW = "waiting_for_workflow" - MERGED = "merged" - FAILED = "failed" - - @dataclass class StackEntry: """One PR in the stack (bottom = index 0).""" diff --git a/src/stack_pr/autoland/pr_state.py b/src/stack_pr/autoland/pr_state.py new file mode 100644 index 0000000..c6f0725 --- /dev/null +++ b/src/stack_pr/autoland/pr_state.py @@ -0,0 +1,18 @@ +"""Where a PR in the stack is on its way to landing. + +Its own module so the step types, which ``model`` imports, can use it too. +""" + +from __future__ import annotations + +from enum import Enum + + +class PRState(str, Enum): + PENDING = "pending" + WAITING_FOR_APPROVAL = "waiting_for_approval" + WAITING_FOR_CHECKS = "waiting_for_checks" + IN_MERGE_QUEUE = "in_merge_queue" + WAITING_FOR_WORKFLOW = "waiting_for_workflow" + MERGED = "merged" + FAILED = "failed" diff --git a/src/stack_pr/autoland/status.py b/src/stack_pr/autoland/status.py index b2f3625..b571e4c 100644 --- a/src/stack_pr/autoland/status.py +++ b/src/stack_pr/autoland/status.py @@ -14,18 +14,11 @@ from stack_pr.autoland import runtime from stack_pr.autoland.display import ( _escape_markup, - _land_entry, _plan_rows, _StepRow, print_status, ) -from stack_pr.autoland.model import ( - ConfirmStep, - LandingContext, - LandStep, - PlanStep, - WorkflowStep, -) +from stack_pr.autoland.model import LandingContext, PlanStep from stack_pr.autoland.options import AutolandOptions from stack_pr.autoland.state import AutolandCheckpointer, AutolandLock, _state_path @@ -207,25 +200,16 @@ def _print_status_text(report: _StatusReport) -> None: ) -def _step_json(step: PlanStep, row: _StepRow, ctx: LandingContext) -> dict: +def _step_json(step: PlanStep, row: _StepRow) -> dict: data: dict[str, Any] = { "number": int(row.number.rstrip(".")), "type": step.kind, "outcome": row.outcome.value, "status": row.status, "is_next": row.is_next, + **row.view.data, + "detail": row.detail or None, } - if isinstance(step, LandStep): - entry = _land_entry(ctx, step) - data["pr_number"] = entry.pr_number if entry else step.pr_number - data["pr_url"] = entry.pr_url if entry else None - data["title"] = entry.title if entry else None - elif isinstance(step, WorkflowStep): - data["workflow"] = step.workflow - else: - assert isinstance(step, ConfirmStep) # the only step type left - data["condition"] = step.condition - data["detail"] = row.detail or None return data @@ -240,7 +224,7 @@ def _status_json(report: _StatusReport) -> dict: "total": len(rows), "done": done, "remaining": len(rows) - done, - "steps": [_step_json(s, r, ctx) for s, r in zip(ctx.plan, rows)], + "steps": [_step_json(s, r) for s, r in zip(ctx.plan, rows)], } return { "status": report.status, diff --git a/src/stack_pr/autoland/steps/__init__.py b/src/stack_pr/autoland/steps/__init__.py index f738530..38273a8 100644 --- a/src/stack_pr/autoland/steps/__init__.py +++ b/src/stack_pr/autoland/steps/__init__.py @@ -17,6 +17,7 @@ ) from stack_pr.autoland.steps.confirm import ConfirmStep 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 __all__ = [ @@ -26,7 +27,10 @@ "LandStep", "PlanParser", "ReplanKey", + "StepOutcome", "StepRegistry", + "StepTone", + "StepView", "WorkflowStep", "register_step", "step_from_dict", diff --git a/src/stack_pr/autoland/steps/base.py b/src/stack_pr/autoland/steps/base.py index f5500a6..05f5c9c 100644 --- a/src/stack_pr/autoland/steps/base.py +++ b/src/stack_pr/autoland/steps/base.py @@ -1,10 +1,11 @@ """The base class every plan step derives from, and the registry of step types. 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, and how it counts toward a -run's progress. 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. +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 +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. """ from __future__ import annotations @@ -17,6 +18,7 @@ if TYPE_CHECKING: from stack_pr.autoland.model import StackEntry + from stack_pr.autoland.steps.view import StepView # A workflow or confirm step's identity for replanning: its keyword, its text, # and the PRs landed before it. A step means "once these PRs have landed, this @@ -118,6 +120,19 @@ def credit_from(self, old: AutolandStep) -> None: """Record on this step the result *old*, its match in the previous run's plan, reached.""" + # --- display ------------------------------------------------------------- + + @abstractmethod + def view( + self, stack: list[StackEntry], *, is_next: bool, assumed_completed: bool + ) -> StepView: + """What the step reports about itself: status, subject, and detail. + + ``is_next`` is whether this is the next step with work left to do, and + ``assumed_completed`` whether it sits inside the plan's already-landed + prefix (see ``plan.is_assumed_completed``). + """ + 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 d758218..a25eacd 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.view import StepOutcome, StepTone, StepView if TYPE_CHECKING: from stack_pr.autoland.model import StackEntry @@ -56,3 +57,33 @@ def replan_key(self, landed_prs: frozenset[int]) -> ReplanKey | None: def credit_from(self, old: AutolandStep) -> None: self.confirmed = True + + def view( + self, stack: list[StackEntry], *, is_next: bool, assumed_completed: bool + ) -> StepView: + if self.confirmed: + # A confirmation inside the landed prefix was never actually given + # in this run — it is inferred from the fact that later PRs merged + # — so say so rather than claiming the user confirmed it. + status = "Assumed completed" if assumed_completed else "Confirmed" + outcome, tone = StepOutcome.DONE, StepTone.SUCCESS + elif is_next: + outcome, status, tone = ( + StepOutcome.ACTIVE, + "Awaiting confirmation", + StepTone.AWAITING_USER, + ) + else: + outcome, status, tone = StepOutcome.PENDING, "Pending", StepTone.MUTED + return StepView( + outcome=outcome, + status=status, + tone=tone, + key=self.condition or "(no condition)", + description=( + f"Manual confirmation: {self.condition}" + if self.condition + else "Manual confirmation" + ), + data={"condition": self.condition}, + ) diff --git a/src/stack_pr/autoland/steps/land.py b/src/stack_pr/autoland/steps/land.py index dbedbbc..10c10d9 100644 --- a/src/stack_pr/autoland/steps/land.py +++ b/src/stack_pr/autoland/steps/land.py @@ -7,7 +7,9 @@ from typing import TYPE_CHECKING 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.view import StepOutcome, StepTone, StepView if TYPE_CHECKING: from stack_pr.autoland.model import StackEntry @@ -18,6 +20,28 @@ # 'l #123' is indistinguishable from a bare 'l' with a comment after it. RE_PR_URL = re.compile(r"^https?://[^/]+/([^/]+)/([^/]+)/pull/(\d+)/?$") +_PR_STATE_STATUS = { + PRState.PENDING: (StepOutcome.PENDING, "Pending", StepTone.MUTED), + PRState.WAITING_FOR_APPROVAL: ( + StepOutcome.ACTIVE, + "Waiting for approval", + StepTone.AWAITING_REVIEW, + ), + PRState.WAITING_FOR_CHECKS: ( + StepOutcome.ACTIVE, + "Waiting for checks", + StepTone.AWAITING_CHECKS, + ), + PRState.IN_MERGE_QUEUE: (StepOutcome.ACTIVE, "In merge queue", StepTone.IN_QUEUE), + PRState.WAITING_FOR_WORKFLOW: ( + StepOutcome.ACTIVE, + "Waiting for workflow", + StepTone.AWAITING_WORKFLOW, + ), + PRState.MERGED: (StepOutcome.DONE, "Merged", StepTone.SUCCESS), + PRState.FAILED: (StepOutcome.FAILED, "Failed", StepTone.FAILURE), +} + @register_step @dataclass @@ -76,6 +100,50 @@ def landed_pr(self, stack: list[StackEntry]) -> int | None: return self.pr_number return None if self.already_landed else stack[self.entry_index].pr_number + def view( + self, stack: list[StackEntry], *, is_next: bool, assumed_completed: bool + ) -> StepView: + if self.already_landed: + # The PR is no longer in the stack, so there is no entry to read a + # title or progress from; the pinned number is all there is. + key = f"#{self.pr_number}" + return StepView( + outcome=StepOutcome.DONE, + status="Landed", + tone=StepTone.SUCCESS, + key=key, + description=f"Land PR {key} (already landed)", + data={"pr_number": self.pr_number, "pr_url": None, "title": None}, + ) + entry = stack[self.entry_index] + outcome, status, tone = _PR_STATE_STATUS.get( + entry.state, (StepOutcome.PENDING, str(entry.state), StepTone.PLAIN) + ) + detail = [] + if entry.error_message: + detail.append(entry.error_message) + # Retry counts only matter once something has been retried. + if entry.check_retries or entry.queue_retries: + detail.append( + f"retries CI {entry.check_retries} / MQ {entry.queue_retries}" + ) + key = f"#{entry.pr_number}" + title = entry.title or "(untitled)" + return StepView( + outcome=outcome, + status=status, + tone=tone, + key=key, + note=title, + description=f"Land PR {key}: {title}", + detail=tuple(detail), + data={ + "pr_number": entry.pr_number, + "pr_url": entry.pr_url, + "title": entry.title, + }, + ) + def _current_owner_repo() -> tuple[str, str]: """The repository autoland is landing into, as ``(owner, name)``.""" diff --git a/src/stack_pr/autoland/steps/view.py b/src/stack_pr/autoland/steps/view.py new file mode 100644 index 0000000..96aff81 --- /dev/null +++ b/src/stack_pr/autoland/steps/view.py @@ -0,0 +1,66 @@ +"""What a step reports about itself for display: ``StepView``. + +A step decides what it has to say — its outcome, status, subject, and any +detail — and ``display`` decides how that looks: icons, colors, and layout. +Nothing here names a glyph or a style, so a step type never has to know how +it is drawn, and the display never has to know which type it is drawing. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass, field +from enum import Enum +from typing import Any + + +class StepOutcome(Enum): + """The broad state a step is in, independent of how it is drawn. + + The values appear in ``autoland --status -o json``. + """ + + DONE = "done" + ACTIVE = "active" + PENDING = "pending" + FAILED = "failed" + + +class StepTone(Enum): + """What kind of status a step is reporting, for the display to color. + + Finer than ``StepOutcome``: the active tones say what the step is waiting + on, so the display can tell a PR waiting for review from one waiting for + CI at a glance. + """ + + PLAIN = "plain" # no emphasis either way + MUTED = "muted" # nothing has happened yet + SUCCESS = "success" + FAILURE = "failure" + AWAITING_REVIEW = "awaiting_review" + AWAITING_CHECKS = "awaiting_checks" + AWAITING_USER = "awaiting_user" + IN_QUEUE = "in_queue" + AWAITING_WORKFLOW = "awaiting_workflow" + + +@dataclass(frozen=True) +class StepView: + """One step's state as content: everything the display shows about it.""" + + outcome: StepOutcome + # The status column, e.g. "Waiting for checks" or "Assumed completed". + status: str + tone: StepTone + # The step's subject: "#103", "deploy.yaml", the confirm condition. + key: str + # One line of prose naming the step, for lists such as "Next steps". + description: str + # Says more about the subject (a PR's title); "" when there is nothing. + note: str = "" + # Fragments of the step's second line (an error, retry counts); the + # display decides how to join them. Empty when the header says it all. + detail: tuple[str, ...] = () + # Type-specific fields for ``autoland --status -o json``. + data: Mapping[str, Any] = field(default_factory=dict) diff --git a/src/stack_pr/autoland/steps/workflow.py b/src/stack_pr/autoland/steps/workflow.py index 3b14664..6ec0fc1 100644 --- a/src/stack_pr/autoland/steps/workflow.py +++ b/src/stack_pr/autoland/steps/workflow.py @@ -11,11 +11,25 @@ ReplanKey, register_step, ) +from stack_pr.autoland.steps.view import StepOutcome, StepTone, StepView if TYPE_CHECKING: from stack_pr.autoland.model import StackEntry +# "skipped" is set only by WorkflowStep.assume_completed, i.e. for a workflow +# that sits inside a plan's already-landed prefix — it must have run before +# those PRs could merge, so the plan assumes it did rather than claiming it was +# deliberately skipped. +_STATE_STATUS = { + "pending": (StepOutcome.PENDING, "Pending", StepTone.MUTED), + "waiting": (StepOutcome.ACTIVE, "Waiting for workflow", StepTone.AWAITING_WORKFLOW), + "succeeded": (StepOutcome.DONE, "Workflow complete", StepTone.SUCCESS), + "failed": (StepOutcome.FAILED, "Failed", StepTone.FAILURE), + "skipped": (StepOutcome.DONE, "Assumed completed", StepTone.SUCCESS), +} + + @register_step @dataclass class WorkflowStep(AutolandStep): @@ -53,3 +67,19 @@ def credit_from(self, old: AutolandStep) -> None: if isinstance(old, WorkflowStep): self.state = old.state self.error_message = "" + + def view( + self, stack: list[StackEntry], *, is_next: bool, assumed_completed: bool + ) -> StepView: + outcome, status, tone = _STATE_STATUS.get( + self.state, (StepOutcome.PENDING, "Pending", StepTone.MUTED) + ) + return StepView( + outcome=outcome, + status=status, + tone=tone, + key=self.workflow, + description=f"Wait for workflow {self.workflow}", + detail=(self.error_message,) if self.error_message else (), + data={"workflow": self.workflow}, + ) diff --git a/tests/test_autoland_steps.py b/tests/test_autoland_steps.py index 7d74359..c1806bf 100644 --- a/tests/test_autoland_steps.py +++ b/tests/test_autoland_steps.py @@ -1,14 +1,17 @@ -"""The plan step registry and the per-step-type plan syntax and persistence.""" +"""The plan step registry, and each step type's plan syntax, persistence, and +view.""" from __future__ import annotations +import dataclasses +import inspect import json from dataclasses import dataclass from pathlib import Path import pytest -from stack_pr.autoland.model import LandingContext, StackEntry +from stack_pr.autoland.model import LandingContext, PRState, StackEntry from stack_pr.autoland.plan import format_plan_for_editor, parse_plan from stack_pr.autoland.state import AutolandCheckpointer from stack_pr.autoland.steps import ( @@ -17,7 +20,10 @@ ConfirmStep, LandStep, PlanParser, + StepOutcome, StepRegistry, + StepTone, + StepView, WorkflowStep, ) @@ -39,6 +45,19 @@ def test_every_registered_step_type_has_a_sample() -> None: assert set(REGISTRY.types()) == set(SAMPLES) +@pytest.mark.parametrize("step_type", REGISTRY.types(), ids=lambda t: t.__name__) +def test_every_registered_step_type_has_a_view( + step_type: type[AutolandStep], +) -> None: + # A step type with no view of its own would leave view() abstract. + assert not inspect.isabstract(step_type) + view = SAMPLES[step_type].view(STACK, is_next=False, assumed_completed=False) + assert isinstance(view, StepView) + assert view.status + assert view.key + assert view.description + + @pytest.mark.parametrize( ("kind", "keyword", "step_type"), [ @@ -76,6 +95,17 @@ def parse(cls, args: str, parser: PlanParser) -> _Dummy: def plan_line(self, stack: list[StackEntry]) -> tuple[str, str]: return "d", "" + def view( + self, stack: list[StackEntry], *, is_next: bool, assumed_completed: bool + ) -> StepView: + return StepView( + outcome=StepOutcome.PENDING, + status="Pending", + tone=StepTone.MUTED, + key="d", + description="Dummy", + ) + @pytest.mark.parametrize( ("kind", "keyword", "message"), @@ -128,3 +158,195 @@ def test_unknown_step_type_in_state_file_is_an_error(tmp_path: Path) -> None: with pytest.raises(ValueError, match="Unknown plan step type: 'bogus'"): AutolandCheckpointer.load(sf) + + +# --- views ------------------------------------------------------------------- +# +# A step's view is everything the status panel, the "Next steps" list, and +# `--status -o json` say about it. These pin the contract per step type. + + +def _entry(**changes: object) -> StackEntry: + return dataclasses.replace(STACK[0], **{"title": "Fix the thing", **changes}) + + +def _view( + step: AutolandStep, + stack: list[StackEntry] = STACK, + *, + is_next: bool = False, + assumed_completed: bool = False, +) -> StepView: + return step.view(stack, is_next=is_next, assumed_completed=assumed_completed) + + +def test_land_view_of_a_pending_pr() -> None: + assert _view(LandStep(entry_index=0, pr_number=101), [_entry()]) == StepView( + outcome=StepOutcome.PENDING, + status="Pending", + tone=StepTone.MUTED, + key="#101", + note="Fix the thing", + description="Land PR #101: Fix the thing", + data={ + "pr_number": 101, + "pr_url": "https://github.com/o/r/pull/101", + "title": "Fix the thing", + }, + ) + + +@pytest.mark.parametrize( + ("state", "outcome", "status", "tone"), + [ + ( + PRState.WAITING_FOR_APPROVAL, + StepOutcome.ACTIVE, + "Waiting for approval", + StepTone.AWAITING_REVIEW, + ), + ( + PRState.WAITING_FOR_CHECKS, + StepOutcome.ACTIVE, + "Waiting for checks", + StepTone.AWAITING_CHECKS, + ), + ( + PRState.IN_MERGE_QUEUE, + StepOutcome.ACTIVE, + "In merge queue", + StepTone.IN_QUEUE, + ), + ( + PRState.WAITING_FOR_WORKFLOW, + StepOutcome.ACTIVE, + "Waiting for workflow", + StepTone.AWAITING_WORKFLOW, + ), + (PRState.MERGED, StepOutcome.DONE, "Merged", StepTone.SUCCESS), + (PRState.FAILED, StepOutcome.FAILED, "Failed", StepTone.FAILURE), + ], +) +def test_land_view_follows_the_pr_state( + state: PRState, outcome: StepOutcome, status: str, tone: StepTone +) -> None: + view = _view(LandStep(entry_index=0), [_entry(state=state)]) + assert (view.outcome, view.status, view.tone) == (outcome, status, tone) + + +def test_land_view_detail_has_the_error_and_retries_once_retried() -> None: + step = LandStep(entry_index=0) + assert _view(step, [_entry(error_message="Checks failed")]).detail == ( + "Checks failed", + ) + entry = _entry(error_message="Checks failed", check_retries=2, queue_retries=1) + assert _view(step, [entry]).detail == ("Checks failed", "retries CI 2 / MQ 1") + + +def test_land_view_of_an_untitled_pr() -> None: + view = _view(LandStep(entry_index=0), [_entry(title="")]) + assert view.note == "(untitled)" + assert view.description == "Land PR #101: (untitled)" + # JSON reports the title as GitHub has it, not the placeholder. + assert view.data["title"] == "" + + +def test_land_view_of_an_already_landed_pr() -> None: + assert _view(LandStep(entry_index=-1, pr_number=99)) == StepView( + outcome=StepOutcome.DONE, + status="Landed", + tone=StepTone.SUCCESS, + key="#99", + description="Land PR #99 (already landed)", + data={"pr_number": 99, "pr_url": None, "title": None}, + ) + + +@pytest.mark.parametrize( + ("state", "outcome", "status", "tone"), + [ + ("pending", StepOutcome.PENDING, "Pending", StepTone.MUTED), + ( + "waiting", + StepOutcome.ACTIVE, + "Waiting for workflow", + StepTone.AWAITING_WORKFLOW, + ), + ("succeeded", StepOutcome.DONE, "Workflow complete", StepTone.SUCCESS), + ("failed", StepOutcome.FAILED, "Failed", StepTone.FAILURE), + # Only a workflow inside the landed prefix is "skipped". + ("skipped", StepOutcome.DONE, "Assumed completed", StepTone.SUCCESS), + ], +) +def test_workflow_view_follows_its_state( + state: str, outcome: StepOutcome, status: str, tone: StepTone +) -> None: + assert _view(WorkflowStep(workflow="deploy.yaml", state=state)) == StepView( + outcome=outcome, + status=status, + tone=tone, + key="deploy.yaml", + description="Wait for workflow deploy.yaml", + data={"workflow": "deploy.yaml"}, + ) + + +def test_workflow_view_detail_is_its_error() -> None: + step = WorkflowStep(workflow="deploy.yaml", state="failed", error_message="boom") + assert _view(step).detail == ("boom",) + + +@pytest.mark.parametrize( + ("step", "is_next", "assumed_completed", "expected"), + [ + ( + ConfirmStep(condition="QA"), + False, + False, + (StepOutcome.PENDING, "Pending", StepTone.MUTED), + ), + ( + ConfirmStep(condition="QA"), + True, + False, + (StepOutcome.ACTIVE, "Awaiting confirmation", StepTone.AWAITING_USER), + ), + ( + ConfirmStep(condition="QA", confirmed=True), + True, + False, + (StepOutcome.DONE, "Confirmed", StepTone.SUCCESS), + ), + # Inside the landed prefix nobody confirmed it in this run. + ( + ConfirmStep(condition="QA", confirmed=True), + False, + True, + (StepOutcome.DONE, "Assumed completed", StepTone.SUCCESS), + ), + ], +) +def test_confirm_view_states( + step: ConfirmStep, + is_next: bool, + assumed_completed: bool, + expected: tuple[StepOutcome, str, StepTone], +) -> None: + outcome, status, tone = expected + assert _view(step, is_next=is_next, assumed_completed=assumed_completed) == ( + StepView( + outcome=outcome, + status=status, + tone=tone, + key="QA", + description="Manual confirmation: QA", + data={"condition": "QA"}, + ) + ) + + +def test_confirm_view_without_a_condition() -> None: + view = _view(ConfirmStep()) + assert view.key == "(no condition)" + assert view.description == "Manual confirmation" + assert view.data == {"condition": ""}