recipe_modules/publish/api.py

# Copyright 2021 The Fuchsia Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

from recipe_engine import recipe_api


class PublishApi(recipe_api.RecipeApi):
    """APIs for publishing releases."""

    def release(
        self,
        step_name,
        product_shortname,
        release_version,
        channel,
        publish_host=None,
        rollout_pct=None,
        urgent_update=False,
        stop_rollout=False,
        switch_channel=None,
        **kwargs,
    ):
        """Publish a release to a channel.

        Args:
            step_name (str): Name of the step.
            product_shortname (str): String identifier for the product.
            release_version (release.ReleaseVersion): Version for the release.
            channel (str): Channel to publish to.
            publish_host (str): API hostname. Default host is used if not
              specified.
            rollout_pct (int): Rollout percentage. Default value is used if not
              specified.
            urgent_update (bool): Whether to request an urgent update.
            stop_rollout (bool): Whether to stop rollout to devices.
            switch_channel (str): Optional channel to redirect devices to.
            kwargs (dict): Passed to self.m.step.
        """
        args = [
            "release",
            "-product-shortname",
            product_shortname,
            "-version",
            str(release_version),
            "-channel",
            channel,
            "-json-output",
            self.m.json.output(),
        ]
        if publish_host:
            args += ["-host", publish_host]
        if rollout_pct:
            args += ["-rollout-pct", rollout_pct]
        if urgent_update:
            args.append("-urgent-update")
        if stop_rollout:
            args.append("-stop-rollout")
        if switch_channel:
            args += ["-switch-channel", switch_channel]
        step = self._run(step_name, args, **kwargs)
        return step

    @property
    def _publish_tool(self):
        return self.m.ensure_tool("publish", self.resource("tool_manifest.json"))

    def _run(self, step_name, args, **kwargs):
        return self.m.step(step_name, [self._publish_tool] + args, **kwargs)