recipe_modules/cipd_roll/test_api.py

# Copyright 2024 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.
"""Test API for cipd_roll."""

from __future__ import annotations

from collections.abc import Sequence
from typing import Any

from recipe_engine import recipe_test_api

from PB.recipe_modules.pigweed.cipd_roll.package import Package


class CipdRollTestApi(recipe_test_api.RecipeTestApi):
    """Test API for cipd_roller."""

    def package(
        self,
        cipd_path: str,
        old_version: str,
        platforms: list[str] | None = None,
    ) -> dict[str, Any]:
        result = {
            'path': cipd_path,
            'tags': [old_version],
            '_comment': 'comments should be preserved',
        }

        if platforms is not None:
            result['platforms'] = list(platforms)

        return result

    def _describe(
        self,
        name: str,
        package_path: str,
        tags: Sequence[tuple[str, str]] | None = None,
        tstamp: int | None = None,
        **kwargs,
    ) -> recipe_test_api.StepTestData:
        describe_kwargs = {}
        if tags:
            describe_kwargs['test_data_tags'] = [
                f'{name}:{value}' for name, value in tags
            ]
        if tstamp:
            describe_kwargs['tstamp'] = tstamp

        return self.step_data(
            name,
            self.m.cipd.example_describe(
                package_path,
                **describe_kwargs,
            ),
            **kwargs,
        )

    def describe(
        self,
        prefix: str,
        package_path: str,
        tags: Sequence[tuple[str, str]] | None = None,
        **kwargs,
    ) -> recipe_test_api.StepTestData:
        prefix = f'{prefix.rstrip(".")}.' if prefix else ''
        return self._describe(
            name=f'{prefix}cipd describe {package_path}',
            package_path=package_path,
            tags=tags,
            **kwargs,
        )

    def describe_current(
        self,
        prefix: str,
        package_path: str,
        tags: Sequence[tuple[str, str]] | None = None,
        **kwargs,
    ) -> recipe_test_api.StepTestData:
        return self._describe(
            name=(
                f'{prefix}.checking against old version.{package_path}.'
                f'cipd describe {package_path}'
            ),
            package_path=package_path,
            tags=tags,
            **kwargs,
        )

    def no_ref(
        self,
        prefix: str,
        package_paths: Sequence[str],
    ) -> recipe_test_api.StepTestData:
        res = None
        for package_path in package_paths:
            step = self.describe(prefix, package_path, (), retcode=1)
            res = res + step if res else step
        return res

    def no_common_tags(
        self,
        prefix: str,
        package_paths: Sequence[str],
        tagname: str = 'git_revision',
    ) -> recipe_test_api.StepTestData:
        res = None
        for i, package_path in enumerate(package_paths):
            step = self.describe(prefix, package_path, ((tagname, i),))
            res = res + step if res else step
        return res

    def multiple_common_tags(
        self,
        prefix: str,
        package_paths: Sequence[str],
        tagname: str = 'git_revision',
        versions: Sequence[int] = (1, 2, 3),
    ) -> recipe_test_api.StepTestData:
        res = None
        for package_path in package_paths:
            step = self.describe(
                prefix,
                package_path,
                tuple((tagname, x) for x in versions),
            )
            res = res + step if res else step
        return res

    def relaxing_works(
        self,
        prefix: str,
        package_paths: Sequence[str],
        tagname: str = 'git_revision',
    ) -> recipe_test_api.StepTestData:
        return self.step_data(
            f'{prefix}.find shared tag.cipd describe {package_paths[1]}',
            self.m.cipd.example_describe(
                package_paths[1],
                test_data_tags=[f'{tagname}:{0}'],
            ),
        )

    def relaxing_does_not_work(
        self,
        prefix: str,
        package_paths: Sequence[str],
    ) -> recipe_test_api.StepTestData:
        # No version of package_paths[1] with hoped-for tag.
        return self.step_data(
            f'{prefix}.find shared tag.cipd describe {package_paths[0]}',
            self.m.cipd.example_describe(package_paths[0]),
            retcode=1,
        ) + self.step_data(
            f'{prefix}.find shared tag.cipd describe {package_paths[1]}',
            self.m.cipd.example_describe(package_paths[1]),
            retcode=1,
        )

    def read_file_step_data(
        self,
        prefix: str,
        package_json_name: str,
        *packages: dict[str, Any],
    ) -> recipe_test_api.StepTestData:
        return self.step_data(
            f'{prefix}.read {package_json_name}',
            self.m.file.read_json({'packages': packages}),
        )

    def package_props(self, spec: str, **kwargs) -> Package:
        kwargs.setdefault('spec', spec)
        kwargs.setdefault('ref', 'latest')
        kwargs.setdefault('tag', 'git_revision')
        kwargs.setdefault(
            'json_path',
            'pw_env_setup/py/pw_env_setup/cipd_setup/pigweed.json',
        )
        return Package(**kwargs)