Discharge7443/brave-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[brockit] Kick out Rust/WASM CI jobs (#36903)

Browse Source 
This PR adds a new command to brockit, `brockit
gen-rust-wasm-toolchain`, which will be used to kick out a new toolchain
run across all platforms.

This is the first use of CI in `brockit`, but we potentially want this
type of thing for all toolchains, so further work will follow to make
this code more generic to other CI tasks.

This is a very experimental implementation.

Bug: https://github.com/brave/brave-browser/issues/55812
This commit is contained in:
cdesouza-chromium
2026-06-01 22:50:42 +01:00
committed by GitHub
parent b5a0618e0f
commit 119a40ec55
4 changed files with 1731 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/cr/brockit.py
+563 -33
View File
@@ -190,6 +190,26 @@ Pass `--culprit=<hash>` to provide a specific culprit for the toolchain update.
If none is provided, `brockit` will trying to determine the culprit by looking
for the last commit updating the toolchain in Chromium.
### `brockit.py gen-rust-toolchain`
This command triggers the Rust/WASM toolchain Jenkins pipelines (one per
platform) for a given Chromium tag. It reads credentials from `~/.jenkins.json`
and kicks off a parameterized build for each pipeline.
```sh
tools/cr/brockit.py gen-rust-toolchain 150.0.7850.1
```
The `tag` argument also accepts the same `@latest-*` labels (e.g. `@latest-tag`,
etc).
Pass `--watch` to show a live-updating table of each pipeline's stage and
status until all finish. Pressing Ctrl+C stops watching, but the builds keep
running.
```sh
tools/cr/brockit.py gen-rust-toolchain @latest-canary --watch
```
### `brockit.py reassign`
This command is used to change the authorship of a given commit in the branch.
It generates an empty commit with the message `reassign! {original_subject}`
@@ -226,10 +246,16 @@ import pickle
import platform
import re
import requests
from rich.box import Box
from rich.live import Live
from rich.markdown import Markdown
from rich.padding import Padding
from rich.spinner import Spinner
from rich.table import Table
from rich.text import Text
import subprocess
import sys
import time
from git_status import GitStatus
from patchfile import Patchfile
@@ -239,7 +265,8 @@ import rebase_v2
from rebase_v2 import REASSIGN_COMMIT_MSG_PREFIX
import repository
from repository import Repository
from terminal import IncendiaryErrorHandler, console, is_verbose, terminal
from terminal import (IncendiaryErrorHandler, Task as BaseTask, console,
is_verbose, terminal)
import versioning
from versioning import Version
from vpython_utils import VPYTHON3_PATH
@@ -279,6 +306,65 @@ CHROMIUM_MAC_SDK_GNI = 'build/config/mac/mac_sdk.gni'
# Google dash link used to check the latest version for a given channel
CHROMIUMDASH_LATEST_RELEASE = 'https://chromiumdash.appspot.com/fetch_releases?channel={channel}&platform={platform}&num=1'
# Configuration file holding the Jenkins credentials used to trigger toolchain
# pipelines. See `GenRustToolchain` for the expected schema.
JENKINS_CONFIG_FILE = Path.home() / '.jenkins.json'
# The Jenkins jobs that build and publish the Rust/WASM toolchain, one per
# platform. Each accepts a single `CHROMIUM_TAG` build parameter.
RUST_TOOLCHAIN_JOBS = (
'brave-browser-rust-toolchain-aux-build-linux-x64',
'brave-browser-rust-toolchain-aux-build-macos-arm64',
'brave-browser-rust-toolchain-aux-build-macos-x64',
'brave-browser-rust-toolchain-aux-build-windows-x64',
)
# How often `gen-rust-toolchain --watch` polls Jenkins for pipeline progress.
WATCH_POLL_INTERVAL_SECONDS = 8
# Maps Pipeline Stage View (`wfapi`) statuses onto the canonical states the
# watch table tracks. Statuses not listed (e.g. NOT_EXECUTED) leave the state
# unchanged.
_WFAPI_STATUS_TO_STATE = {
'IN_PROGRESS': 'RUNNING',
'PAUSED_PENDING_INPUT': 'RUNNING',
'SUCCESS': 'SUCCESS',
'FAILED': 'FAILURE',
'ABORTED': 'ABORTED',
'UNSTABLE': 'UNSTABLE',
}
# States meaning a pipeline has finished and no longer needs polling.
_TERMINAL_WATCH_STATES = frozenset(
{'SUCCESS', 'FAILURE', 'ABORTED', 'UNSTABLE', 'CANCELLED'})
# Minimalist box for the `--watch` table: column dividers plus a single
# header rule, no outer frame (paired with `show_edge=False`). Each of the
# eight lines is four chars -- left edge, cell fill, column divider, right
# edge -- in the order Rich's `Box` expects (top, head, head-rule, mid, row,
# foot-rule, foot, bottom). Edge/foot lines are placeholders never drawn once
# the edge is hidden and the table has no footer.
_WATCH_TABLE_BOX = Box(' \n' # top
' │ \n' # head
' ── \n' # head rule
' │ \n' # mid
' │ \n' # row
' ── \n' # foot rule
' │ \n' # foot
' \n') # bottom
# Icon + rich style for each watch state, used to render the State column.
_WATCH_STATE_STYLE = {
'QUEUED': ('⏳', 'dim'),
'RUNNING': ('🔧', 'cyan'),
'SUCCESS': ('✔️', 'green'),
'FAILURE': ('🚨', 'bold red'),
'ABORTED': ('🛑', 'yellow'),
'CANCELLED': ('🚫', 'yellow'),
'UNSTABLE': ('⚠️', 'yellow'),
'UNKNOWN': ('🤔', 'dim'),
}
# A decorator to be shown for messages that the user should address before
# continuing.
ACTION_NEEDED_DECORATOR = '[bold yellow](action needed)[/]'
@@ -575,42 +661,24 @@ class ActionNeededException(Exception):
console.log(message)
class Task:
""" Base class for all tasks in brockit.
# Banners framing every Brockit task run, shared with `run_watching`.
_BROCKIT_START_BANNER = '[italic]🚀 Brockit!'
_BROCKIT_END_BANNER = '[bold]💥 Done!'
This class provides a common interface for other tasks to build upon. It
provides a run method that will execute the task, and a status_message
method that will return a string to be displayed while the task is running.
class Task(BaseTask):
"""Base class for all Brockit tasks.
Adds the Brockit banners around the generic `terminal.Task` run; the actual
behaviour (status spinner, `execute`/`status_message` contract) lives in the
base class. Subclasses provide `status_message`.
"""
def run(self, **kwargs) -> bool:
"""Runs the task with a status message.
# Abstract base: concrete subclasses implement `status_message`.
# pylint: disable=abstract-method
This function will run the task inside the scope of a status message.
Args:
an open set of argument to be passed along to the derived class's
execute method.
Returns:
Return 1 if the task failed, 0 if the task succeeded. This is used
as the process' exit code.
"""
console.log('[italic]🚀 Brockit!')
with terminal.with_status(self.status_message()):
# Calling `self.execute` triggers the linter, as there's no
# `execute` method in this class. The derived classes are expected
# to provide this method.
# pylint: disable=no-member
self.execute(**kwargs)
console.log('[bold]💥 Done!')
def status_message(self) -> str:
"""Returns a status message for the task.
This function has to be implemented by the derived class.
"""
raise NotImplementedError
start_banner = _BROCKIT_START_BANNER
end_banner = _BROCKIT_END_BANNER
class Versioned(Task):
@@ -2396,6 +2464,441 @@ class UpdateXcodeToolchain(Task):
})
@dataclass
class _WatchedJob:
"""Mutable per-pipeline state tracked while `--watch` polls Jenkins."""
# The Jenkins job name.
job: str
# The transient queue-item URL from the trigger's `Location` header.
queue_url: str
# The build URL, resolved once an executor dequeues the job.
build_url: str | None = None
# Canonical state: QUEUED / RUNNING / SUCCESS / FAILURE / ABORTED /
# UNSTABLE / CANCELLED / UNKNOWN.
state: str = 'QUEUED'
# Human-readable current stage (or the queue reason while QUEUED).
stage: str = ''
# Pre-formatted elapsed build time (e.g. "12m04s"), as of the last poll.
# Used as-is for non-running rows; RUNNING rows tick forward from the two
# fields below instead (see `_ElapsedClock`).
elapsed: str = ''
# The raw server-reported build duration (ms) at the last poll, and the
# monotonic clock read at that same instant. Together they anchor a
# RUNNING row's locally-ticking elapsed counter. Both None until a build
# is resolved.
duration_millis: int | None = None
elapsed_anchor: float | None = None
# The pipeline's configured `display-name`, resolved once when watching
# starts. None until resolved, or when the pipeline sets no display name.
display_name: str | None = None
@property
def bot(self) -> str:
"""Label for the table's Bot column.
The pipeline's configured `display-name` when it has one, otherwise
its Jenkins job name.
"""
return self.display_name or self.job
@property
def is_terminal(self) -> bool:
"""Whether the pipeline has finished and no longer needs polling."""
return self.state in _TERMINAL_WATCH_STATES
def link(self, base_url: str) -> str:
"""The best link available: the build URL, else the job page."""
return self.build_url or f'{base_url}/job/{self.job}/'
class _ElapsedClock:
"""Renderable that advances a RUNNING job's elapsed time between polls.
Rich re-renders the `Live` tree on every refresh -- the same mechanism
that animates the RUNNING spinner -- so recomputing the value here lets
the counter tick smoothly at the Live refresh rate instead of jumping once
per poll. `base_millis` is the duration the server reported at the last
poll and `anchor` is the monotonic clock read at that same instant; the
rendered value is `base + (now - anchor)`. Every poll re-anchors both, so
the local ticking stays corrected by the authoritative server value and
settles on it once the build leaves RUNNING.
"""
def __init__(self, base_millis: int, anchor: float) -> None:
self._base_millis = base_millis
self._anchor = anchor
def __rich__(self) -> Text:
elapsed_millis = (self._base_millis +
(time.monotonic() - self._anchor) * 1000)
return Text(GenRustToolchain._format_duration(elapsed_millis),
justify='right')
class GenRustToolchain(Task):
"""Triggers the Rust/WASM toolchain Jenkins pipelines for a Chromium tag.
There is one pipeline per platform (see `RUST_TOOLCHAIN_JOBS`), and each
takes a single `CHROMIUM_TAG` build parameter. This task resolves the
requested version (accepting the same labels as `lift --to`, e.g.
`@latest-canary`), reads the Jenkins credentials from `~/.jenkins.json`,
and kicks off a parameterized build for every pipeline.
The expected `~/.jenkins.json` schema is:
{
"url": "https://ci.brave.com",
"username": "<ldap-user>",
"token": "<jenkins-api-token>"
}
"""
def status_message(self):
return "Triggering Rust toolchain builds..."
def run_watching(self, tag: str) -> None:
"""Entry point for `--watch` that bypasses `Task.run`.
The watch table is a `rich.live.Live` display, and rich permits only
one live display at a time. `Task.run` wraps `execute` in the shared
status spinner -- itself a live display -- so the watch flow cannot go
through it. This mirrors `run`'s banner framing but drives `execute`
directly, without the spinner.
"""
console.log(self.start_banner)
self.execute(tag=tag, watch=True)
console.log(self.end_banner)
@staticmethod
def _load_jenkins_config() -> tuple[str, str, str]:
"""Reads the Jenkins base URL and credentials from `~/.jenkins.json`.
Returns:
A `(base_url, username, token)` tuple, with any trailing slash
stripped from the base URL.
"""
if not JENKINS_CONFIG_FILE.is_file():
raise InvalidInputException(
f'Jenkins config not found at {JENKINS_CONFIG_FILE}. Create it '
'with [bold cyan]url[/], [bold cyan]username[/], and '
'[bold cyan]token[/] fields.')
try:
config = json.loads(
JENKINS_CONFIG_FILE.read_bytes().decode('utf-8'))
except json.JSONDecodeError as e:
raise InvalidInputException(
f'Failed to parse {JENKINS_CONFIG_FILE}: {e}') from e
missing = [
key for key in ('url', 'username', 'token') if not config.get(key)
]
if missing:
raise InvalidInputException(
f'{JENKINS_CONFIG_FILE} is missing required field(s): '
f'{", ".join(missing)}.')
return config['url'].rstrip('/'), config['username'], config['token']
@staticmethod
def _get_crumb(session: requests.Session, base_url: str) -> dict[str, str]:
"""Fetches a Jenkins CSRF crumb as a ready-to-merge header dict.
Returns an empty dict when the crumb issuer is unavailable. API-token
auth is usually crumb-exempt, so a missing issuer is treated as "no
crumb needed" rather than a hard failure.
"""
try:
response = session.get(f'{base_url}/crumbIssuer/api/json',
timeout=15)
response.raise_for_status()
data = response.json()
return {data['crumbRequestField']: data['crumb']}
except (requests.RequestException, KeyError, ValueError):
return {}
def execute(self, tag: str, watch: bool = False):
# Resolve the version first so a bad tag/label fails before we touch
# any Jenkins state.
version = _fetch_chromium_tag(tag)
base_url, username, token = self._load_jenkins_config()
session = requests.Session()
session.auth = (username, token)
crumb = self._get_crumb(session, base_url)
terminal.log_task(
f'Triggering Rust toolchain pipelines for Chromium {version}:')
watched: list[_WatchedJob] = []
failures: list[str] = []
for job in RUST_TOOLCHAIN_JOBS:
try:
response = session.post(
f'{base_url}/job/{job}/buildWithParameters',
params={'CHROMIUM_TAG': str(version)},
headers=crumb,
timeout=30)
response.raise_for_status()
except requests.RequestException as e:
failures.append(job)
console.log(Padding(f'✘ {job}: {e}', (0, 4)))
continue
# Jenkins' 201 Location header points at the transient queue item,
# not the build: the build number isn't assigned until an executor
# picks the job up, and the queue URL only serves JSON.
queue_url = response.headers.get('Location', '')
watched.append(_WatchedJob(job=job, queue_url=queue_url))
if not watch:
# Link the job page, which always resolves and surfaces the
# queued/running build.
terminal.log_task(f'[bold]✔️ [/]{job} ➜ {base_url}/job/{job}/')
if failures:
raise BadOutcomeException(
'Failed to trigger the following Rust toolchain pipelines:\n%s'
% '\n'.join(f' * {job}' for job in failures))
if watch:
self._watch(session, base_url, version, watched)
@staticmethod
def _get_json(session: requests.Session, url: str) -> dict | None:
"""GETs `url` and returns the parsed JSON, or None on any failure."""
try:
response = session.get(url, timeout=15)
response.raise_for_status()
return response.json()
except (requests.RequestException, ValueError):
return None
def _resolve_display_name(self, session: requests.Session, base_url: str,
job: _WatchedJob) -> None:
"""Records a pipeline's `display-name` for the Bot column, if set.
Jenkins returns the job name as `displayName` when no display name is
configured, so a value equal to the job name is treated as "unset" and
left as None -- `_WatchedJob.bot` then falls back to the job name.
"""
info = self._get_json(
session,
f'{base_url}/job/{job.job}/api/json?tree=displayName,name')
if info is None:
return
display_name = info.get('displayName')
if display_name and display_name != info.get('name'):
job.display_name = display_name
@staticmethod
def _format_duration(millis) -> str:
"""Formats a Jenkins millisecond duration as e.g. "12m04s"."""
if not millis:
return ''
total_seconds = int(millis) // 1000
minutes, seconds = divmod(total_seconds, 60)
if minutes:
return f'{minutes}m{seconds:02d}s'
return f'{seconds}s'
def _poll_job(self, session: requests.Session, job: _WatchedJob) -> None:
"""Refreshes one pipeline's state in place.
Resolves the queue item to a build the first time an executor picks it
up, then tracks the running build's stage and result via the Pipeline
Stage View API (falling back to the plain build API).
"""
if job.is_terminal:
return
if job.build_url is None:
if not job.queue_url:
job.state = 'UNKNOWN'
job.stage = 'no queue item returned'
return
item = self._get_json(session, f'{job.queue_url}api/json')
if item is None:
return # Transient; retry on the next poll.
if item.get('cancelled'):
job.state = 'CANCELLED'
job.stage = 'queue item cancelled'
return
executable = item.get('executable')
if not executable:
# Still queued; surface Jenkins' reason (e.g. "Waiting for
# next available executor").
job.state = 'QUEUED'
job.stage = (item.get('why') or 'waiting').strip()
return
job.build_url = executable.get('url') or ''
job.state = 'RUNNING'
self._refresh_build_state(session, job)
def _record_elapsed(self, job: _WatchedJob, millis) -> None:
"""Anchors a job's elapsed time to the latest server-reported duration.
Stores the raw duration and a monotonic timestamp so a RUNNING row can
tick forward between polls (see `_ElapsedClock`), and keeps the
formatted string used to render every non-running row.
"""
job.duration_millis = int(millis) if millis else None
job.elapsed_anchor = time.monotonic()
job.elapsed = self._format_duration(millis)
def _refresh_build_state(self, session: requests.Session,
job: _WatchedJob) -> None:
"""Updates state/stage/elapsed for a job that already has a build."""
describe = self._get_json(session, f'{job.build_url}wfapi/describe')
if describe is not None:
job.state = _WFAPI_STATUS_TO_STATE.get(describe.get('status', ''),
job.state)
self._record_elapsed(job, describe.get('durationMillis'))
stages = describe.get('stages') or []
running = [s for s in stages if s.get('status') == 'IN_PROGRESS']
if running:
# The deepest reported in-progress stage is the most specific.
job.stage = running[-1].get('name', '')
elif job.is_terminal:
job.stage = '(done)'
elif stages:
job.stage = stages[-1].get('name', '')
return
# Fallback for jobs without the Stage View plugin: the plain build API
# gives building/result but no per-stage detail.
info = self._get_json(session, f'{job.build_url}api/json')
if info is None:
return
self._record_elapsed(job, info.get('duration'))
if info.get('building'):
job.state = 'RUNNING'
job.stage = 'building'
else:
job.state = info.get('result') or 'UNKNOWN'
job.stage = '(done)'
@staticmethod
def _state_cell(state: str) -> str | Spinner:
"""Renders the State column.
RUNNING gets an animated throbber (the `Live` display drives the
animation); every other state is a static icon + label.
"""
if state == 'RUNNING':
return Spinner('dots', text='RUNNING', style='cyan')
icon, style = _WATCH_STATE_STYLE.get(state, ('?', 'dim'))
return f'[{style}]{icon} {state}[/]'
@staticmethod
def _elapsed_cell(job: _WatchedJob) -> str | _ElapsedClock:
"""Renders the Elapsed column.
A RUNNING build with a resolved duration ticks forward between polls
via `_ElapsedClock`; every other state shows the static,
server-accurate value captured at the last poll.
"""
if (job.state == 'RUNNING' and job.duration_millis is not None
and job.elapsed_anchor is not None):
return _ElapsedClock(job.duration_millis, job.elapsed_anchor)
return job.elapsed or '—'
@staticmethod
def _link_cell(url: str) -> Text:
"""Renders a URL as a styled, clickable hyperlink for the Build column.
Table cells skip the `ReprHighlighter` that `console.log`/`print` run
over their output, so a bare URL string renders as plain text. Wrapping
it in a `Text` with a `link` style emits the OSC 8 hyperlink escape
(clickable in capable terminals) and the blue underline mirrors the
look Rich gives auto-detected URLs elsewhere in brockit.
"""
return Text(url, style=f'underline blue link {url}')
@staticmethod
def _dim_cell(renderable: str | Text, dim: bool) -> str | Text:
"""Dims a non-state cell for finished, non-successful rows.
Returns the renderable untouched when `dim` is False. Otherwise wraps
it in Rich's `dim` style (preserving any existing style, such as the
Build column's hyperlink) so the whole row reads as muted -- except the
State cell, which the caller leaves coloured so the outcome stands out.
"""
if not dim:
return renderable
text = renderable if isinstance(renderable, Text) else Text(
str(renderable))
text.stylize('dim')
return text
def _render_table(self, version: Version, base_url: str,
jobs: list[_WatchedJob]) -> Table:
"""Builds the per-pipeline progress table rendered in place by Live."""
table = Table(title=f'Rust toolchain · Chromium {version}',
title_justify='left',
box=_WATCH_TABLE_BOX,
show_edge=False,
expand=True)
table.add_column('Bot', no_wrap=True)
table.add_column('State', no_wrap=True)
table.add_column('Stage', no_wrap=True)
table.add_column('Elapsed', justify='right', no_wrap=True)
table.add_column('Build', overflow='fold')
for job in jobs:
# A job that has finished as anything other than SUCCESS is dimmed
# across the row, except its State cell, which keeps its colour so
# the failure still stands out.
dim = job.is_terminal and job.state != 'SUCCESS'
table.add_row(
self._dim_cell(job.bot, dim), self._state_cell(job.state),
self._dim_cell(job.stage or '—', dim),
self._dim_cell(self._elapsed_cell(job), dim),
self._dim_cell(self._link_cell(job.link(base_url)), dim))
return table
def _watch(self, session: requests.Session, base_url: str,
version: Version, jobs: list[_WatchedJob]) -> None:
"""Polls the triggered pipelines, updating an in-place table until all
finish or the user interrupts with Ctrl+C (which detaches and leaves
the builds running).
"""
terminal.log_task('Watching pipelines — press [bold cyan]Ctrl+C[/] to '
'stop watching (builds keep running).')
# Resolve each pipeline's display name once up front so the Bot column
# shows it from the very first render.
for job in jobs:
self._resolve_display_name(session, base_url, job)
try:
# The throbber on RUNNING rows animates off the Live's own refresh
# (~12.5fps matches the `dots` spinner cadence); the data itself is
# only re-polled every WATCH_POLL_INTERVAL_SECONDS.
with Live(self._render_table(version, base_url, jobs),
console=console,
refresh_per_second=12.5) as live:
while True:
for job in jobs:
self._poll_job(session, job)
live.update(self._render_table(version, base_url, jobs))
if all(job.is_terminal for job in jobs):
break
time.sleep(WATCH_POLL_INTERVAL_SECONDS)
except KeyboardInterrupt:
# Detach cleanly: the builds keep running on CI.
console.log('[yellow]Stopped watching. Builds continue on CI:[/]')
for job in jobs:
console.log(Padding(f'{job.bot}: {job.link(base_url)}',
(0, 4)))
def fetch_chromium_dash_version(channel: str) -> Version:
"""Fetches the highest latest version across all platforms for a channel.
@@ -2682,6 +3185,25 @@ def main():
f'{CHROMIUM_MAC_SDK_GNI}.',
dest='culprit')
gen_rust_parser = subparsers.add_parser(
'gen-rust-toolchain',
parents=[global_parser],
formatter_class=argparse.RawTextHelpFormatter,
help='Triggers the Rust/WASM toolchain Jenkins pipelines for a '
'Chromium tag. Requires credentials in ~/.jenkins.json.')
gen_rust_parser.add_argument(
'tag',
help=('The Chromium version to build the toolchain for (e.g.\n'
'150.0.7850.1), or one of the @latest-* labels accepted by\n'
'`lift --to` (e.g. @latest-canary, @latest-m150,\n'
'@latest-for-branch, @latest-tag).'))
gen_rust_parser.add_argument(
'--watch',
action='store_true',
help='After triggering, show a live-updating table of each '
"pipeline's\nstage and status until all finish. Press Ctrl+C to stop "
'watching;\nthe builds keep running.')
subparsers.add_parser('reference',
help='Detailed documentation for this tool.')
args = parser.parse_args()
@@ -2741,6 +3263,14 @@ def main():
Reassign().run(change=args.change)
if args.command == 'update-xcode-toolchain':
UpdateXcodeToolchain().run(url=args.url, culprit=args.culprit)
if args.command == 'gen-rust-toolchain':
task = GenRustToolchain()
if args.watch:
# `--watch` provdes live updates, therefore it doesn't use the
# spinner provided by `Task.run`.
task.run_watching(tag=args.tag)
else:
task.run(tag=args.tag)
if args.command == 'reference':
console.print(Markdown(__doc__))
if args.command == 'show':
tools/cr/gen_rust_toolchain_integration_test.py
Executable
+446
View File
@@ -0,0 +1,446 @@
#!/usr/bin/env vpython3
# Copyright (c) 2026 The Brave Authors. All rights reserved.
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at https://mozilla.org/MPL/2.0/.
"""Integration tests for `brockit.GenRustToolchain` against a fake Jenkins.
Where `gen_rust_toolchain_test.py` mocks `requests.Session`, these tests stand
up a real in-process HTTP server (`FakeJenkins`) that emulates the slice of the
Jenkins REST API the task talks to:
* the CSRF crumb issuer (`/crumbIssuer/api/json`),
* the parameterised trigger (`/job/<job>/buildWithParameters`, answered with a
201 + `Location` header pointing at a transient queue item),
* the build queue item (`/queue/item/<id>/api/json`), which hands back an
`executable` build URL once an "executor" picks the job up, and
* the Pipeline Stage View (`<build>/wfapi/describe`) with a plain build-API
fallback (`<build>/api/json`).
The task drives genuine `requests` calls over a loopback socket, so basic-auth
and crumb headers, the `CHROMIUM_TAG` parameter, the `Location`-header handoff,
and the full queue -> build -> result lifecycle are all exercised end to end.
The fake advances each pipeline deterministically by counting polls (no wall
clock involved): a build stays QUEUED for `queue_polls_before_start` polls,
RUNNING for `running_polls_before_done` polls, then settles on `final_status`.
"""
from __future__ import annotations
import base64
import json
import tempfile
import threading
import unittest
from collections import defaultdict
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, urlparse
from unittest.mock import patch
import brockit
# A concrete version so `_fetch_chromium_tag` short-circuits to `Version(...)`
# without touching git or the network.
TAG = '150.0.7850.1'
# Maps the canonical `final_status` onto the equivalent Pipeline Stage View
# (`wfapi`) status string the fake reports for a finished build.
_FINAL_TO_WFAPI = {
'SUCCESS': 'SUCCESS',
'FAILURE': 'FAILED',
'UNSTABLE': 'UNSTABLE',
'ABORTED': 'ABORTED',
}
class _NullLive:
"""Stand-in for `rich.live.Live` so the watch loop renders nothing."""
def __init__(self, *args, **kwargs):
pass
def __enter__(self) -> _NullLive:
return self
def __exit__(self, *exc) -> bool:
return False
def update(self, *args, **kwargs) -> None:
pass
def _job_from_build_path(path: str) -> str:
"""Pulls `<job>` out of a `/job/<job>/<build-no>` path."""
parts = [segment for segment in path.split('/') if segment]
# parts == ['job', '<job>', '<build-no>'].
return parts[1]
class _Handler(BaseHTTPRequestHandler):
"""Routes the handful of endpoints the task hits to the owning fake."""
# Silence the default per-request stderr logging.
def log_message(self, *args) -> None:
pass
@property
def fake(self) -> FakeJenkins:
return self.server.fake
def send_json(self, payload: dict, status: int = 200) -> None:
body = json.dumps(payload).encode('utf-8')
self.send_response(status)
self.send_header('Content-Type', 'application/json')
self.send_header('Content-Length', str(len(body)))
self.end_headers()
self.wfile.write(body)
def send_empty(self, status: int, headers: dict | None = None) -> None:
self.send_response(status)
for key, value in (headers or {}).items():
self.send_header(key, value)
self.send_header('Content-Length', '0')
self.end_headers()
def do_GET(self) -> None:
path = urlparse(self.path).path
if path == '/crumbIssuer/api/json':
self.fake.handle_crumb(self)
elif path.startswith('/queue/item/') and path.endswith('/api/json'):
qid = path[len('/queue/item/'):-len('/api/json')].strip('/')
self.fake.handle_queue(self, qid)
elif path.endswith('/wfapi/describe'):
self.fake.handle_describe(
self, _job_from_build_path(path[:-len('/wfapi/describe')]))
elif path.startswith('/job/') and path.endswith('/api/json'):
parts = [segment for segment in path.split('/') if segment]
# ['job', <job>, 'api', 'json'] is the pipeline-level info, whereas
# ['job', <job>, <build-no>, 'api', 'json'] is a build's API.
if len(parts) == 4:
self.fake.handle_job_info(self, parts[1])
else:
self.fake.handle_build_api(self, parts[1])
else:
self.send_empty(404)
def do_POST(self) -> None:
path = urlparse(self.path).path
if path.startswith('/job/') and path.endswith('/buildWithParameters'):
job = path[len('/job/'):-len('/buildWithParameters')].strip('/')
self.fake.handle_trigger(self, job)
else:
self.send_empty(404)
class FakeJenkins:
"""An in-process HTTP server emulating the Jenkins endpoints the task uses.
Use as a context manager; `base_url` is populated on entry. Behaviour is
tuned via the public attributes below before (or during) a run.
"""
def __init__(self,
*,
username: str = 'alice',
token: str = 'secret-token'):
self.username = username
self.token = token
# The crumb to serve as `(field, value)`, or None to 404 the issuer
# (which API-token auth treats as "no crumb needed").
self.crumb: tuple[str, str] | None = ('Jenkins-Crumb', 'deadbeef')
# Per-job HTTP status override for the trigger POST (job -> status).
# Anything >= 400 makes that pipeline's trigger fail.
self.trigger_status: dict[str, int] = {}
# Per-job configured `display-name` (job -> display name). A job absent
# here reports its own name as `displayName`, i.e. no display name set.
self.display_names: dict[str, str] = {}
# Lifecycle knobs, applied uniformly to every pipeline.
self.queue_polls_before_start = 0
self.running_polls_before_done = 1
self.final_status = 'SUCCESS'
self.serve_wfapi = True
# Recorded trigger POSTs, one dict per call (job, params, auth, crumb).
self.triggers: list[dict] = []
self._lock = threading.Lock()
self._next_qid = 1
self._qid_to_job: dict[str, str] = {}
self._queue_polls: dict[str, int] = defaultdict(int)
self._describe_polls: dict[str, int] = defaultdict(int)
self._build_polls: dict[str, int] = defaultdict(int)
# Server state, populated on `__enter__` once the server is bound.
self._httpd: ThreadingHTTPServer | None = None
self._thread: threading.Thread | None = None
self.base_url = ''
def __enter__(self) -> FakeJenkins:
self._httpd = ThreadingHTTPServer(('127.0.0.1', 0), _Handler)
self._httpd.fake = self
self._thread = threading.Thread(target=self._httpd.serve_forever,
daemon=True)
self._thread.start()
host, port = self._httpd.server_address
self.base_url = f'http://{host}:{port}'
return self
def __exit__(self, *exc) -> bool:
self._httpd.shutdown()
self._httpd.server_close()
self._thread.join()
return False
def write_config(self, path: Path) -> None:
"""Writes a `~/.jenkins.json` pointing at this server."""
path.write_text(json.dumps({
'url': self.base_url,
'username': self.username,
'token': self.token,
}),
encoding='utf-8',
newline='')
@staticmethod
def _decode_auth(handler: _Handler) -> tuple[str, str] | None:
header = handler.headers.get('Authorization', '')
if not header.startswith('Basic '):
return None
raw = base64.b64decode(header[len('Basic '):]).decode('utf-8')
user, _, token = raw.partition(':')
return user, token
def handle_crumb(self, handler: _Handler) -> None:
if self.crumb is None:
handler.send_empty(404)
return
field, value = self.crumb
handler.send_json({'crumbRequestField': field, 'crumb': value})
def handle_trigger(self, handler: _Handler, job: str) -> None:
crumb_field = self.crumb[0] if self.crumb else None
self.triggers.append({
'job': job,
'params': parse_qs(urlparse(handler.path).query),
'auth': self._decode_auth(handler),
'crumb_header': handler.headers.get(crumb_field)
if crumb_field else None,
})
status = self.trigger_status.get(job, 201)
if status >= 400:
handler.send_empty(status)
return
with self._lock:
qid = str(self._next_qid)
self._next_qid += 1
self._qid_to_job[qid] = job
handler.send_empty(201,
{'Location': f'{self.base_url}/queue/item/{qid}/'})
def handle_job_info(self, handler: _Handler, job: str) -> None:
# Jenkins reports the job name as `displayName` when none is set.
handler.send_json({
'name': job,
'displayName': self.display_names.get(job, job),
})
def handle_queue(self, handler: _Handler, qid: str) -> None:
job = self._qid_to_job.get(qid)
if job is None:
handler.send_empty(404)
return
with self._lock:
polls = self._queue_polls[qid]
self._queue_polls[qid] += 1
if polls < self.queue_polls_before_start:
handler.send_json({
'cancelled': False,
'executable': None,
'why': 'Waiting for next available executor',
})
return
build_url = f'{self.base_url}/job/{job}/100/'
handler.send_json({
'cancelled': False,
'executable': {
'url': build_url
}
})
def handle_describe(self, handler: _Handler, job: str) -> None:
if not self.serve_wfapi:
# Emulate a controller without the Stage View plugin; the task
# falls back to the plain build API.
handler.send_empty(404)
return
with self._lock:
polls = self._describe_polls[job]
self._describe_polls[job] += 1
if polls < self.running_polls_before_done:
handler.send_json({
'status': 'IN_PROGRESS',
'durationMillis': 1000 * (polls + 1),
'stages': [
{
'name': 'env',
'status': 'SUCCESS'
},
{
'name': 'build',
'status': 'IN_PROGRESS'
},
],
})
return
wfapi_status = _FINAL_TO_WFAPI.get(self.final_status, 'SUCCESS')
handler.send_json({
'status': wfapi_status,
'durationMillis': 99000,
'stages': [{
'name': 'build',
'status': wfapi_status
}],
})
def handle_build_api(self, handler: _Handler, job: str) -> None:
with self._lock:
polls = self._build_polls[job]
self._build_polls[job] += 1
if polls < self.running_polls_before_done:
handler.send_json({
'building': True,
'duration': 0,
'result': None
})
return
handler.send_json({
'building': False,
'duration': 50000,
'result': self.final_status,
})
class GenRustToolchainIntegrationTest(unittest.TestCase):
"""Drives `GenRustToolchain.execute` against `FakeJenkins`."""
def setUp(self):
tmp = tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
self.config_path = Path(tmp.name) / '.jenkins.json'
patcher = patch.object(brockit, 'JENKINS_CONFIG_FILE',
self.config_path)
patcher.start()
self.addCleanup(patcher.stop)
def _execute(self, fake: FakeJenkins, *, watch: bool = False) -> list:
"""Runs `execute` against `fake`, returning the watched-job list.
The job list is captured by intercepting `_render_table`; because the
`_WatchedJob` instances are mutated in place, the captured reference
reflects their final state once `execute` returns.
"""
fake.write_config(self.config_path)
captured: dict[str, list] = {}
def _record(_self, _version, _base_url, jobs):
captured['jobs'] = jobs
if watch:
with patch.object(brockit.GenRustToolchain, '_render_table',
_record), \
patch('brockit.time.sleep'), \
patch('brockit.Live', _NullLive):
brockit.GenRustToolchain().execute(tag=TAG, watch=True)
else:
brockit.GenRustToolchain().execute(tag=TAG)
return captured.get('jobs', [])
def test_triggers_every_pipeline_with_auth_param_and_crumb(self):
with FakeJenkins() as fake:
self._execute(fake)
self.assertEqual({trigger['job']
for trigger in fake.triggers},
set(brockit.RUST_TOOLCHAIN_JOBS))
self.assertEqual(len(fake.triggers), len(brockit.RUST_TOOLCHAIN_JOBS))
for trigger in fake.triggers:
self.assertEqual(trigger['params'], {'CHROMIUM_TAG': [TAG]})
self.assertEqual(trigger['auth'], ('alice', 'secret-token'))
self.assertEqual(trigger['crumb_header'], 'deadbeef')
def test_missing_crumb_issuer_still_triggers_without_header(self):
with FakeJenkins() as fake:
fake.crumb = None
self._execute(fake)
self.assertEqual(len(fake.triggers), len(brockit.RUST_TOOLCHAIN_JOBS))
for trigger in fake.triggers:
self.assertIsNone(trigger['crumb_header'])
def test_trigger_failure_raises_bad_outcome(self):
with FakeJenkins() as fake:
fake.trigger_status = {brockit.RUST_TOOLCHAIN_JOBS[0]: 500}
fake.write_config(self.config_path)
with self.assertRaises(brockit.BadOutcomeException):
brockit.GenRustToolchain().execute(tag=TAG)
def test_watch_runs_full_queue_to_success_lifecycle(self):
with FakeJenkins() as fake:
# Force a QUEUED -> RUNNING -> RUNNING -> SUCCESS progression so the
# queue handoff and the running-state polling are both exercised.
fake.queue_polls_before_start = 1
fake.running_polls_before_done = 2
jobs = self._execute(fake, watch=True)
self.assertEqual(len(jobs), len(brockit.RUST_TOOLCHAIN_JOBS))
for job in jobs:
self.assertEqual(job.state, 'SUCCESS')
self.assertEqual(job.stage, '(done)')
self.assertEqual(job.elapsed, '1m39s') # 99000 ms
# The build URL was resolved from the queue item's executable.
self.assertIn(f'/job/{job.job}/', job.build_url)
def test_watch_uses_pipeline_display_name_for_bot(self):
with FakeJenkins() as fake:
fake.display_names = {
job: f'Display of {job}'
for job in brockit.RUST_TOOLCHAIN_JOBS
}
jobs = self._execute(fake, watch=True)
for job in jobs:
self.assertEqual(job.display_name, f'Display of {job.job}')
self.assertEqual(job.bot, f'Display of {job.job}')
def test_watch_bot_falls_back_to_job_name_without_display_name(self):
with FakeJenkins() as fake:
# No display names configured: Jenkins echoes the job name as
# `displayName`, which must be treated as "unset".
jobs = self._execute(fake, watch=True)
for job in jobs:
self.assertIsNone(job.display_name)
self.assertEqual(job.bot, job.job)
def test_watch_falls_back_to_build_api_and_reports_failure(self):
with FakeJenkins() as fake:
fake.serve_wfapi = False
fake.final_status = 'FAILURE'
jobs = self._execute(fake, watch=True)
for job in jobs:
self.assertEqual(job.state, 'FAILURE')
self.assertEqual(job.stage, '(done)')
self.assertEqual(job.elapsed, '50s') # 50000 ms
if __name__ == '__main__':
unittest.main()
tools/cr/gen_rust_toolchain_test.py
Executable
+677
View File
@@ -0,0 +1,677 @@
#!/usr/bin/env vpython3
# Copyright (c) 2026 The Brave Authors. All rights reserved.
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at https://mozilla.org/MPL/2.0/.
"""Tests for `brockit.GenRustToolchain`.
The file is split in two layers:
* `LoadJenkinsConfigTest` exercises the pure `~/.jenkins.json` reader,
pointing `brockit.JENKINS_CONFIG_FILE` at a temp file. No network involved.
* `GenRustToolchainExecuteTest` drives `GenRustToolchain.execute` with a mocked
`requests.Session`, asserting that every pipeline is triggered with the right
URL, `CHROMIUM_TAG` parameter, auth, and CSRF crumb, and that a per-job
failure surfaces as a `BadOutcomeException`.
"""
from __future__ import annotations
import json
import tempfile
import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch
from rich.console import Console
import brockit
# A valid Jenkins config, written into the temp file by the execute tests.
VALID_CONFIG = {
'url': 'https://ci.brave.com',
'username': 'alice',
'token': 'secret-token',
}
class LoadJenkinsConfigTest(unittest.TestCase):
"""Tests for `GenRustToolchain._load_jenkins_config`."""
def setUp(self):
tmp = tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
self.config_path = Path(tmp.name) / '.jenkins.json'
patcher = patch.object(brockit, 'JENKINS_CONFIG_FILE',
self.config_path)
patcher.start()
self.addCleanup(patcher.stop)
def _write(self, data) -> None:
self.config_path.write_text(json.dumps(data),
encoding='utf-8',
newline='')
def test_reads_all_fields_and_strips_slash(self):
"""All three fields are returned, with any trailing slash dropped from
the base URL so it can be concatenated with `/job/...` paths."""
self._write({
'url': 'https://ci.example.com/',
'username': 'bob',
'token': 'tok',
})
base_url, user, token = (
brockit.GenRustToolchain._load_jenkins_config())
self.assertEqual(base_url, 'https://ci.example.com')
self.assertEqual(user, 'bob')
self.assertEqual(token, 'tok')
def test_missing_file_raises(self):
with self.assertRaises(brockit.InvalidInputException):
brockit.GenRustToolchain._load_jenkins_config()
def test_missing_field_raises(self):
self._write({'url': 'https://ci.brave.com', 'username': 'bob'})
with self.assertRaises(brockit.InvalidInputException):
brockit.GenRustToolchain._load_jenkins_config()
def test_empty_field_raises(self):
"""A present-but-empty field is treated as missing."""
self._write({
'url': 'https://ci.brave.com',
'username': '',
'token': 'tok',
})
with self.assertRaises(brockit.InvalidInputException):
brockit.GenRustToolchain._load_jenkins_config()
def test_invalid_json_raises(self):
self.config_path.write_text('{not valid json',
encoding='utf-8',
newline='')
with self.assertRaises(brockit.InvalidInputException):
brockit.GenRustToolchain._load_jenkins_config()
class GenRustToolchainExecuteTest(unittest.TestCase):
"""End-to-end tests for `GenRustToolchain.execute`."""
def setUp(self):
tmp = tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
self.config_path = Path(tmp.name) / '.jenkins.json'
self.config_path.write_text(json.dumps(VALID_CONFIG),
encoding='utf-8',
newline='')
patcher = patch.object(brockit, 'JENKINS_CONFIG_FILE',
self.config_path)
patcher.start()
self.addCleanup(patcher.stop)
@staticmethod
def _make_session() -> MagicMock:
"""Builds a `requests.Session` mock whose crumb GET and build POST both
succeed."""
session = MagicMock()
crumb_resp = MagicMock()
crumb_resp.json.return_value = {
'crumbRequestField': 'Jenkins-Crumb',
'crumb': 'deadbeef',
}
crumb_resp.raise_for_status.return_value = None
session.get.return_value = crumb_resp
post_resp = MagicMock()
post_resp.headers = {'Location': 'https://ci.brave.com/queue/item/1/'}
post_resp.raise_for_status.return_value = None
session.post.return_value = post_resp
return session
@patch('brockit.requests.Session')
def test_triggers_all_jobs(self, session_cls):
session = self._make_session()
session_cls.return_value = session
brockit.GenRustToolchain().execute(tag='150.0.7850.1')
# Basic-auth credentials come straight from the config.
self.assertEqual(session.auth, ('alice', 'secret-token'))
# One POST per pipeline, each at the right buildWithParameters URL.
self.assertEqual(session.post.call_count,
len(brockit.RUST_TOOLCHAIN_JOBS))
posted_urls = {call.args[0] for call in session.post.call_args_list}
self.assertEqual(
posted_urls, {
f'https://ci.brave.com/job/{job}/buildWithParameters'
for job in brockit.RUST_TOOLCHAIN_JOBS
})
# Every call carries the resolved CHROMIUM_TAG and the crumb header.
for call in session.post.call_args_list:
self.assertEqual(call.kwargs['params'],
{'CHROMIUM_TAG': '150.0.7850.1'})
self.assertEqual(call.kwargs['headers'],
{'Jenkins-Crumb': 'deadbeef'})
@patch('brockit.requests.Session')
@patch('brockit._fetch_chromium_tag')
def test_resolves_label_via_fetch_chromium_tag(self, fetch, session_cls):
"""A label like `@latest-canary` is resolved through
`_fetch_chromium_tag`, and the resolved version is what gets sent as
`CHROMIUM_TAG`."""
fetch.return_value = brockit.Version('151.0.1.2')
session = self._make_session()
session_cls.return_value = session
brockit.GenRustToolchain().execute(tag='@latest-canary')
fetch.assert_called_once_with('@latest-canary')
for call in session.post.call_args_list:
self.assertEqual(call.kwargs['params'],
{'CHROMIUM_TAG': '151.0.1.2'})
@patch('brockit.requests.Session')
def test_missing_crumb_issuer_proceeds_without_header(self, session_cls):
"""When the crumb issuer is unavailable, the builds are still triggered
with no crumb header (API-token auth is crumb-exempt)."""
session = self._make_session()
session.get.side_effect = brockit.requests.RequestException('no crumb')
session_cls.return_value = session
brockit.GenRustToolchain().execute(tag='150.0.7850.1')
self.assertEqual(session.post.call_count,
len(brockit.RUST_TOOLCHAIN_JOBS))
for call in session.post.call_args_list:
self.assertEqual(call.kwargs['headers'], {})
@patch('brockit.requests.Session')
def test_job_failure_raises_bad_outcome(self, session_cls):
"""A non-2xx response from any pipeline trigger surfaces as a
`BadOutcomeException`."""
session = self._make_session()
failing = MagicMock()
failing.raise_for_status.side_effect = brockit.requests.HTTPError(
'403 Forbidden')
session.post.return_value = failing
session_cls.return_value = session
with self.assertRaises(brockit.BadOutcomeException):
brockit.GenRustToolchain().execute(tag='150.0.7850.1')
@patch('brockit.requests.Session')
def test_missing_config_raises_before_any_request(self, session_cls):
"""With no config file present, the task fails before issuing any
Jenkins request."""
self.config_path.unlink()
session = self._make_session()
session_cls.return_value = session
with self.assertRaises(brockit.InvalidInputException):
brockit.GenRustToolchain().execute(tag='150.0.7850.1')
session.post.assert_not_called()
def _json_response(payload):
"""A response mock whose `.json()` yields `payload` and which is 2xx."""
resp = MagicMock()
resp.json.return_value = payload
resp.raise_for_status.return_value = None
return resp
def _dispatching_session(routes):
"""A session mock whose `.get` dispatches by first matching URL substring.
`routes` is an ordered list of `(substring, payload)` pairs; a payload that
is an `Exception` is raised (to simulate a failed/absent endpoint),
otherwise it is returned as a JSON response.
"""
session = MagicMock()
def _get(url, timeout=None): # pylint: disable=unused-argument
for needle, payload in routes:
if needle in url:
if isinstance(payload, Exception):
raise payload
return _json_response(payload)
raise brockit.requests.RequestException(f'unexpected GET {url}')
session.get.side_effect = _get
return session
class RunWatchingTest(unittest.TestCase):
"""`--watch` must bypass `Task.run`'s status spinner (only one live
display at a time) and drive `execute` with `watch=True`."""
def test_drives_execute_with_watch_without_spinner(self):
task = brockit.GenRustToolchain()
with patch.object(task, 'execute') as execute, \
patch('brockit.terminal.with_status') as with_status:
task.run_watching(tag='150.0.7850.1')
execute.assert_called_once_with(tag='150.0.7850.1', watch=True)
with_status.assert_not_called()
class WatchedJobTest(unittest.TestCase):
"""Tests for the `_WatchedJob` helpers."""
def _job(self, **kwargs):
defaults = {
'job': 'brave-browser-rust-toolchain-aux-build-linux-x64',
'queue_url': 'https://ci.brave.com/queue/item/1/',
}
defaults.update(kwargs)
return brockit._WatchedJob(**defaults)
def test_bot_is_pipeline_name(self):
self.assertEqual(self._job().bot,
'brave-browser-rust-toolchain-aux-build-linux-x64')
def test_bot_prefers_display_name(self):
self.assertEqual(self._job(display_name='Linux x64').bot, 'Linux x64')
def test_is_terminal(self):
self.assertTrue(self._job(state='SUCCESS').is_terminal)
self.assertTrue(self._job(state='FAILURE').is_terminal)
self.assertFalse(self._job(state='RUNNING').is_terminal)
self.assertFalse(self._job(state='QUEUED').is_terminal)
def test_link_prefers_build_url(self):
job = self._job(build_url='https://ci.brave.com/job/x/5/')
self.assertEqual(job.link('https://ci.brave.com'),
'https://ci.brave.com/job/x/5/')
def test_link_falls_back_to_job_page(self):
self.assertEqual(
self._job().link('https://ci.brave.com'),
'https://ci.brave.com/job/'
'brave-browser-rust-toolchain-aux-build-linux-x64/')
class FormatDurationTest(unittest.TestCase):
"""Tests for `GenRustToolchain._format_duration`."""
def test_empty_for_missing(self):
self.assertEqual(brockit.GenRustToolchain._format_duration(None), '')
self.assertEqual(brockit.GenRustToolchain._format_duration(0), '')
def test_seconds_only(self):
self.assertEqual(brockit.GenRustToolchain._format_duration(5000), '5s')
def test_minutes_and_seconds_zero_padded(self):
self.assertEqual(brockit.GenRustToolchain._format_duration(62000),
'1m02s')
self.assertEqual(brockit.GenRustToolchain._format_duration(100000),
'1m40s')
class StateCellTest(unittest.TestCase):
"""Tests for `GenRustToolchain._state_cell`."""
def test_running_is_animated_spinner(self):
cell = brockit.GenRustToolchain._state_cell('RUNNING')
self.assertIsInstance(cell, brockit.Spinner)
def test_other_states_are_static_markup(self):
cell = brockit.GenRustToolchain._state_cell('SUCCESS')
self.assertIsInstance(cell, str)
self.assertIn('SUCCESS', cell)
class LinkCellTest(unittest.TestCase):
"""The Build column renders URLs as styled, clickable hyperlinks."""
def test_wraps_url_in_link_style(self):
url = 'https://ci.brave.com/job/x/5/'
cell = brockit.GenRustToolchain._link_cell(url)
self.assertIsInstance(cell, brockit.Text)
# The raw URL is preserved (so terminal URL detection still works)...
self.assertEqual(cell.plain, url)
# ...and an OSC 8 link plus the blue underline are applied.
self.assertIn(f'link {url}', cell.style)
self.assertIn('underline', cell.style)
class DimCellTest(unittest.TestCase):
"""`_dim_cell` mutes finished, non-successful rows except the State cell."""
def test_passes_through_when_not_dim(self):
self.assertEqual(brockit.GenRustToolchain._dim_cell('build', False),
'build')
def test_wraps_string_in_dim(self):
cell = brockit.GenRustToolchain._dim_cell('build', True)
self.assertIsInstance(cell, brockit.Text)
self.assertEqual(cell.plain, 'build')
self.assertIn('dim', [span.style for span in cell.spans])
def test_dims_existing_text_preserving_base_style(self):
link = brockit.GenRustToolchain._link_cell('https://ci.brave.com/x/')
cell = brockit.GenRustToolchain._dim_cell(link, True)
# The hyperlink styling is preserved and dim is layered on top.
self.assertIn('link https://ci.brave.com/x/', cell.style)
self.assertIn('dim', [span.style for span in cell.spans])
class RenderTableDimTest(unittest.TestCase):
"""Finished non-successful rows render dim; successful ones do not."""
DIM = '\x1b[2m' # SGR code Rich emits for the `dim` style.
def _render(self, state):
job = brockit._WatchedJob(
job='brave-browser-rust-toolchain-aux-build-linux-x64',
queue_url='',
build_url='https://ci.brave.com/job/x/5/',
state=state,
stage='(done)',
elapsed='5m00s')
table = brockit.GenRustToolchain()._render_table(
brockit.Version('150.0.7850.1'), 'https://ci.brave.com', [job])
console = Console(force_terminal=True, width=200)
with console.capture() as capture:
console.print(table)
return capture.get()
def test_failure_row_is_dimmed(self):
self.assertIn(self.DIM, self._render('FAILURE'))
def test_aborted_row_is_dimmed(self):
self.assertIn(self.DIM, self._render('ABORTED'))
def test_success_row_is_not_dimmed(self):
self.assertNotIn(self.DIM, self._render('SUCCESS'))
class ElapsedClockTest(unittest.TestCase):
"""`_ElapsedClock` ticks forward from its anchor on each render."""
@patch('brockit.time.monotonic')
def test_ticks_from_anchor(self, monotonic):
clock = brockit._ElapsedClock(base_millis=60000, anchor=100.0)
monotonic.return_value = 130.0 # 30s past the anchor
self.assertEqual(clock.__rich__().plain, '1m30s')
@patch('brockit.time.monotonic')
def test_uses_base_when_no_time_passed(self, monotonic):
clock = brockit._ElapsedClock(base_millis=5000, anchor=100.0)
monotonic.return_value = 100.0
self.assertEqual(clock.__rich__().plain, '5s')
class ElapsedCellTest(unittest.TestCase):
"""The Elapsed cell is a live clock only while RUNNING with a duration."""
def _job(self, **kwargs):
defaults = {'job': 'job-linux', 'queue_url': ''}
defaults.update(kwargs)
return brockit._WatchedJob(**defaults)
def test_running_with_anchor_is_clock(self):
job = self._job(state='RUNNING',
duration_millis=1000,
elapsed_anchor=10.0,
elapsed='1s')
self.assertIsInstance(brockit.GenRustToolchain._elapsed_cell(job),
brockit._ElapsedClock)
def test_running_without_duration_is_static(self):
"""A RUNNING build whose duration hasn't resolved yet shows a dash, not
a clock anchored to nothing."""
self.assertEqual(
brockit.GenRustToolchain._elapsed_cell(self._job(state='RUNNING')),
'—')
def test_terminal_shows_static_server_value(self):
job = self._job(state='SUCCESS',
duration_millis=100000,
elapsed_anchor=10.0,
elapsed='1m40s')
self.assertEqual(brockit.GenRustToolchain._elapsed_cell(job), '1m40s')
class ResolveDisplayNameTest(unittest.TestCase):
"""Tests for `GenRustToolchain._resolve_display_name`."""
BASE = 'https://ci.brave.com'
def _job(self):
return brockit._WatchedJob(job='job-linux', queue_url='')
def test_sets_display_name_when_distinct_from_job_name(self):
job = self._job()
session = _dispatching_session([('/job/job-linux/api/json', {
'name': 'job-linux',
'displayName': 'Linux x64',
})])
brockit.GenRustToolchain()._resolve_display_name(
session, self.BASE, job)
self.assertEqual(job.display_name, 'Linux x64')
self.assertEqual(job.bot, 'Linux x64')
def test_leaves_none_when_display_name_equals_job_name(self):
"""Jenkins echoes the job name as `displayName` when none is set; that
must not be treated as a real display name."""
job = self._job()
session = _dispatching_session([('api/json', {
'name': 'job-linux',
'displayName': 'job-linux',
})])
brockit.GenRustToolchain()._resolve_display_name(
session, self.BASE, job)
self.assertIsNone(job.display_name)
self.assertEqual(job.bot, 'job-linux')
def test_leaves_none_on_request_failure(self):
job = self._job()
session = _dispatching_session([
('api/json', brockit.requests.RequestException('boom'))
])
brockit.GenRustToolchain()._resolve_display_name(
session, self.BASE, job)
self.assertIsNone(job.display_name)
class PollJobTest(unittest.TestCase):
"""Drives the `_poll_job` state machine through its transitions."""
QUEUE_URL = 'https://ci.brave.com/queue/item/1/'
BUILD_URL = 'https://ci.brave.com/job/x/5/'
def _job(self, **kwargs):
defaults = {'job': 'job-linux', 'queue_url': self.QUEUE_URL}
defaults.update(kwargs)
return brockit._WatchedJob(**defaults)
def test_still_queued_records_reason(self):
job = self._job()
session = _dispatching_session([('queue/item', {
'cancelled': False,
'executable': None,
'why': 'Waiting for next available executor',
})])
brockit.GenRustToolchain()._poll_job(session, job)
self.assertEqual(job.state, 'QUEUED')
self.assertEqual(job.stage, 'Waiting for next available executor')
self.assertIsNone(job.build_url)
def test_cancelled_queue_item(self):
job = self._job()
session = _dispatching_session([('queue/item', {'cancelled': True})])
brockit.GenRustToolchain()._poll_job(session, job)
self.assertEqual(job.state, 'CANCELLED')
def test_dequeue_then_running_stage(self):
"""Once an executor picks the job up, the build is resolved and the
running stage is read from wfapi in the same poll."""
job = self._job()
session = _dispatching_session([
('queue/item', {
'executable': {
'url': self.BUILD_URL
}
}),
('wfapi/describe', {
'status': 'IN_PROGRESS',
'durationMillis': 62000,
'stages': [
{
'name': 'env',
'status': 'SUCCESS'
},
{
'name': 'build',
'status': 'IN_PROGRESS'
},
],
}),
])
brockit.GenRustToolchain()._poll_job(session, job)
self.assertEqual(job.build_url, self.BUILD_URL)
self.assertEqual(job.state, 'RUNNING')
self.assertEqual(job.stage, 'build')
self.assertEqual(job.elapsed, '1m02s')
# The raw duration is also anchored so the row can tick between polls.
self.assertEqual(job.duration_millis, 62000)
self.assertIsNotNone(job.elapsed_anchor)
def test_running_to_success(self):
job = self._job(build_url=self.BUILD_URL, state='RUNNING')
session = _dispatching_session([('wfapi/describe', {
'status': 'SUCCESS',
'durationMillis': 100000,
'stages': [{
'name': 's3-upload',
'status': 'SUCCESS'
}],
})])
brockit.GenRustToolchain()._poll_job(session, job)
self.assertEqual(job.state, 'SUCCESS')
self.assertEqual(job.stage, '(done)')
self.assertEqual(job.elapsed, '1m40s')
def test_falls_back_to_build_api_without_stage_view(self):
"""When wfapi is unavailable, state/result come from the plain build
API instead."""
job = self._job(build_url=self.BUILD_URL, state='RUNNING')
session = _dispatching_session([
('wfapi/describe',
brockit.requests.HTTPError('404 no stage view')),
('api/json', {
'building': False,
'result': 'FAILURE',
'duration': 50000
}),
])
brockit.GenRustToolchain()._poll_job(session, job)
self.assertEqual(job.state, 'FAILURE')
self.assertEqual(job.stage, '(done)')
self.assertEqual(job.elapsed, '50s')
def test_terminal_job_is_not_repolled(self):
job = self._job(state='SUCCESS')
session = _dispatching_session([])
brockit.GenRustToolchain()._poll_job(session, job)
session.get.assert_not_called()
def test_missing_queue_url_marks_unknown(self):
job = self._job(queue_url='')
session = _dispatching_session([])
brockit.GenRustToolchain()._poll_job(session, job)
self.assertEqual(job.state, 'UNKNOWN')
session.get.assert_not_called()
class WatchLoopTest(unittest.TestCase):
"""Smoke test for `execute(watch=True)`: triggers all jobs, then polls
until every pipeline reaches a terminal state."""
def setUp(self):
tmp = tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
self.config_path = Path(tmp.name) / '.jenkins.json'
self.config_path.write_text(json.dumps(VALID_CONFIG),
encoding='utf-8',
newline='')
patcher = patch.object(brockit, 'JENKINS_CONFIG_FILE',
self.config_path)
patcher.start()
self.addCleanup(patcher.stop)
@patch('brockit.time.sleep')
@patch('brockit.Live')
@patch('brockit.requests.Session')
def test_watch_polls_until_all_terminal(self, session_cls, live_cls,
sleep):
# Every build is already SUCCESS on the first poll, so the loop should
# break before ever sleeping.
def _get(url, timeout=None): # pylint: disable=unused-argument
if 'crumbIssuer' in url:
return _json_response({
'crumbRequestField': 'Jenkins-Crumb',
'crumb': 'x'
})
if 'queue/item' in url:
return _json_response(
{'executable': {
'url': 'https://ci.brave.com/job/x/5/'
}})
if 'wfapi/describe' in url:
return _json_response({
'status': 'SUCCESS',
'durationMillis': 1000,
'stages': [{
'name': 'build',
'status': 'SUCCESS'
}],
})
raise brockit.requests.RequestException(url)
post_resp = MagicMock()
post_resp.headers = {'Location': 'https://ci.brave.com/queue/item/1/'}
post_resp.raise_for_status.return_value = None
session = MagicMock()
session.post.return_value = post_resp
session.get.side_effect = _get
session_cls.return_value = session
brockit.GenRustToolchain().execute(tag='150.0.7850.1', watch=True)
# All four pipelines were triggered, the live display was entered, and
# because every build was terminal on the first poll, we never slept.
self.assertEqual(session.post.call_count,
len(brockit.RUST_TOOLCHAIN_JOBS))
live_cls.assert_called()
sleep.assert_not_called()
if __name__ == '__main__':
unittest.main()
tools/cr/terminal.py
+45
View File
@@ -335,3 +335,48 @@ class Terminal:
terminal = Terminal()
class Task:
"""Base class for a console task that runs under a status spinner.
A task encapsulates a unit of work and the message shown while it runs.
`run` drives the work inside a live status spinner (see
`Terminal.with_status`), optionally framed by a start and an end banner.
Subclasses implement:
* `execute(**kwargs)` -- the actual work; keyword arguments are
forwarded verbatim from `run`.
* `status_message()` -- the text shown in the spinner while running.
A subclass (or a whole tool) can set `start_banner` / `end_banner` to log
a line immediately before and after the work, e.g. to frame a command-line
run with a recognisable header and footer.
"""
# Optional banner lines logged immediately before and after `execute`.
# Leave as None to run the task without any framing output.
start_banner: str | None = None
end_banner: str | None = None
def run(self, **kwargs) -> None:
"""Runs the task inside a status spinner, framed by the banners.
Keyword arguments are forwarded verbatim to the subclass's `execute`.
"""
if self.start_banner is not None:
console.log(self.start_banner)
with terminal.with_status(self.status_message()):
# `execute` is provided by subclasses; the base class deliberately
# does not define it.
# pylint: disable=no-member
self.execute(**kwargs)
if self.end_banner is not None:
console.log(self.end_banner)
def status_message(self) -> str:
"""Returns the message shown in the status spinner while running.
Must be implemented by the derived class.
"""
raise NotImplementedError