recipe_modules/workflows/api.py

# Copyright 2026 The Pigweed Authors
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not
# use this file except in compliance with the License. You may obtain a copy of
# the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations under
# the License.
"""API for loading workflow configurations."""

from __future__ import annotations

import json
import tomllib
from typing import Any

from google.protobuf import json_format

from recipe_engine import config_types, recipe_api

from PB.pigweed.pw_build.proto import build_driver as build_driver_pb
from PB.pigweed.pw_build.proto import (
    pigweed_build_driver as pigweed_build_driver_pb,
)
from PB.pigweed.pw_build.proto import workflows as workflows_pb

_PROTO_MODULES = (
    build_driver_pb,
    pigweed_build_driver_pb,
    workflows_pb,
)

_CANDIDATES = (
    'workflows.textproto',
    'workflows.json',
    'workflows.yaml',
    'workflows.toml',
)


class NoWorkflowFileFound(recipe_api.StepFailure):
    pass


class TooManyWorkflowFilesFound(recipe_api.StepFailure):
    pass


class WorkflowsApi(recipe_api.RecipeApi):
    """API for loading workflow configurations."""

    NoWorkflowFileFound = NoWorkflowFileFound
    TooManyWorkflowFilesFound = TooManyWorkflowFilesFound

    def load(
        self, directory: config_types.Path | None = None
    ) -> workflows_pb.WorkflowSuite:
        """Loads a WorkflowSuite message from a workflow file in directory."""
        _ = _PROTO_MODULES
        if directory is None:
            directory = self.m.context.cwd or (self.m.path.start_dir / 'co')

        found_files: list[config_types.Path] = []
        for filename in _CANDIDATES:
            path = directory / filename
            if self.m.path.exists(path):
                found_files.append(path)

        if not found_files:
            raise NoWorkflowFileFound(f'no workflow file in {directory}')
        if len(found_files) > 1:
            raise TooManyWorkflowFilesFound(
                f'expected exactly one workflow file in {directory}, '
                f'found {", ".join(str(f) for f in found_files)}'
            )

        target_file = found_files[0]
        filename = target_file.name

        if filename == 'workflows.textproto':
            content: str = self.m.file.read_text(
                f'read {filename}',
                target_file,
                test_data='',
            )
            return self.m.proto.decode(
                content,
                workflows_pb.WorkflowSuite,
                self.m.proto.TEXTPB,
            )

        data: dict[str, Any] = {}
        if filename == 'workflows.yaml':
            data = self.m.yaml.read_file(
                f'read {filename}',
                target_file,
                test_data={},
            )
        else:
            content = self.m.file.read_text(
                f'read {filename}',
                target_file,
                test_data='{}',
            )
            if filename == 'workflows.json':
                data = json.loads(content)
            elif filename == 'workflows.toml':
                data = tomllib.loads(content)
            else:
                raise ValueError(
                    f'unsupported filename: {filename}'
                )  # pragma: no cover

        return json_format.ParseDict(data, workflows_pb.WorkflowSuite())