blob: eac77f0d35dae548f4f56f8142d2e948571ad541 [file] [log] [blame]
# Copyright 2018 The LUCI Authors. All rights reserved.
# Use of this source code is governed under the Apache License, Version 2.0
# that can be found in the LICENSE file.
from recipe_engine import recipe_api
class LedApi(recipe_api.RecipeApi):
"""Interface to the led tool.
"led" stands for LUCI editor. It allows users to debug and modify LUCI jobs.
It can be used to modify many aspects of a LUCI build, most commonly including
the recipes used.
The main interface this module provides is a direct call to the led binary:
led_result = api.led(
'get-builder', ['luci.chromium.try:chromium_presubmit'])
final_data = led_result.then('edit-recipe-bundle').result
See the led binary for full documentation of commands.
class LedResult(object):
"""Holds the result of a led operation. Can be chained using |then|."""
def __init__(self, result, module):
self._result = result
self._module = module
def result(self):
"""The mutable result data of the previous led call as decoded JSON."""
return self._result
def then(self, *cmd):
"""Invoke led, passing it the current `result` data as input.
Returns another LedResult object with the output of the command.
return self.__class__(
self._module._run_command(self._result, *cmd).stdout, self._module)
def __init__(self, **kwargs):
super(LedApi, self).__init__(**kwargs)
self._led_path = None
def _led_binary_path(self):
"""The path to the led binary on disk."""
return self._led_path.join('led')
def __call__(self, *cmd):
"""Runs led with the given arguments. Wraps result in a `LedResult`."""
return self.LedResult(self._run_command(None, *cmd).stdout, self)
def _run_command(self, previous, *cmd):
"""Runs led with a given command and arguments.
cmd: The led command to run, e.g. 'get-builder', 'edit', along with any
previous: The previous led step's json result, if any. This can be
used to chain led commands together. See the tests for an example of
Ensures that led is checked out on disk before trying to execute the
kwargs = {
'stdout': self.m.json.output(),
if previous:
kwargs['stdin'] = self.m.json.input(data=previous)
result = self.m.step(
'led %s' % cmd[0], [self._led_binary_path] + list(cmd), **kwargs)
# If we launched a task, add a link to the swarming task.
if cmd[0] == 'launch':
result.presentation.links['Swarming task'] = 'https://%s/task?id=%s' % (
return result
def _ensure_led(self):
"""Ensures that led is checked out on disk.
Sets _led_path as a side-effect. This will always use `[CACHE]/led` as the
location of the unpacked binaries.
if self._led_path:
ensure_file = self.m.cipd.EnsureFile().add_package(
'infra/tools/luci/led/${platform}', 'latest')
led_path = self.m.path['cache'].join('led')
self.m.cipd.ensure(led_path, ensure_file)
self._led_path = led_path