recipe_modules/checkout/data.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.
"""Data models for the checkout recipe module."""

from __future__ import annotations

import dataclasses
import re
from typing import Any

from recipe_engine import config_types, recipe_api

from PB.recipe_modules.pigweed.checkout.options import Options


def to_dict(obj) -> dict[str, Any]:
    try:
        # Modifications to the dict returned by the built-in vars() function
        # modify the original data structure. Always create a copy for this
        # function to return.
        return __builtins__['vars'](obj).copy()
    except TypeError:  # pragma: no cover
        keys = [x for x in obj.__slots__ if not x.startswith('__')]
        return {k: getattr(obj, k) for k in keys}


@dataclasses.dataclass
class Manifest:
    remotes: dict[str, Remote] = dataclasses.field(default_factory=dict)
    projects: list[Project] = dataclasses.field(default_factory=list)

    def dict(self) -> dict[str, Any]:
        return {
            'remotes': {k: v.dict() for k, v in self.remotes.items()},
            'projects': [x.dict() for x in self.projects],
        }


class Url:
    def __init__(self, url: str, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.url: str = url
        self.https: str | None = None

    def dict(self) -> dict[str, Any]:
        return to_dict(self)


@dataclasses.dataclass
class Remote:
    """Remote config from manifest."""

    name: str
    fetch: Url
    review: str | None = None
    revision: str | None = None
    alias: str | None = None

    def dict(self) -> dict[str, Any]:
        res = to_dict(self)
        res['fetch'] = res['fetch'].dict()
        return res


@dataclasses.dataclass
class Project:
    """Key variables describing a repository/project."""

    name: str
    path: str
    remote: str
    revision: str
    upstream: str
    url: str | None = None

    def path_object(self, root: config_types.Path) -> config_types.Path:
        return root / self.path

    def dict(self) -> dict[str, Any]:
        return to_dict(self)


def _str_or_none(x: Any | None) -> str | None:
    if x is None:
        return x
    return str(x)


def _int_or_none(x: Any | None) -> int | None:
    if x is None:
        return x
    return int(x)


@dataclasses.dataclass
class Change:
    """Data from buildbucket."""

    number: int
    remote: str | None
    ref: str | None
    rebase: bool | None = None
    project: str | None = None
    branch: str | None = None
    gerrit_name: str | None = None
    submitted: bool = False
    patchset: int | None = None
    applied: bool = dataclasses.field(default=False, repr=False)
    path: str | None = None
    base: str | None = None
    base_type: str | None = None
    is_merge: bool = False
    commit_message: str = ''
    topic: str | None = None
    current_revision: str | None = None
    primary: bool = False
    submit_requirements: dict[str, str] = dataclasses.field(
        default_factory=dict,
    )
    uploader: str | None = None

    def __post_init__(self):
        self.number = _int_or_none(self.number)
        self.remote = _str_or_none(self.remote)
        self.ref = _str_or_none(self.ref)
        self.branch = _str_or_none(self.branch)
        self.gerrit_name = _str_or_none(self.gerrit_name)
        self.patchset = _int_or_none(self.patchset)
        self.base = _str_or_none(self.base)
        self.base_type = _str_or_none(self.base_type)

    @property
    def gerrit_host(self) -> str:
        return f'https://{self.gerrit_name}-review.googlesource.com'

    @property
    def gerrit_url(self) -> str:
        if not self.number:
            return self.gitiles_url
        return f'{self.gerrit_host}/c/{self.number}'

    @property
    def gitiles_url(self) -> str:
        return f'{self.remote}/+/{self.ref}'

    @property
    def name(self) -> str:
        return f'{self.gerrit_name}:{self.number}'

    @property
    def name_with_path(self) -> str:
        return f'{self.name} ({self.path})'


@dataclasses.dataclass
class Submodule:
    """Submodule properties."""

    api: recipe_api.RecipeApi = dataclasses.field(repr=False)
    hash: str
    relative_path: str
    path: config_types.Path
    name: str
    describe: str
    remote: str
    initialized: bool
    modified: bool
    conflict: bool
    branch: str
    url: str
    update: str
    ignore: str
    shallow: bool
    fetchRecurseSubmodules: bool

    def __lt__(self, other: Submodule) -> bool:
        return (self.relative_path, self.url) < (
            other.relative_path,
            other.url,
        )


@dataclasses.dataclass
class StatusOfChanges:
    """Changes that were applied or not applied."""

    applied: tuple[Change, ...]
    not_applied: tuple[Change, ...]


@dataclasses.dataclass(slots=False)
class CheckoutContext:
    _api: recipe_api.RecipeApi = dataclasses.field(repr=False)
    options: Options | None = None
    changes: list[Change] | None = None  # List of triggering changes.
    top: config_types.Path = None  # Actual checkout root.
    # Logical checkout root. Usually identical to 'top', but occasionally a
    # subdirectory instead.
    root: config_types.Path = None
    # Which triggering changes were applied or not applied.
    status: StatusOfChanges | None = None
    # Remotes that should be treated identically.
    equivalent_remotes: dict[str, list[str]] | None = dataclasses.field(
        default_factory=dict
    )
    manifest: Manifest | None = None  # Parsed repo manifest.
    # Path to a JSON file containing metadata about the triggering changes.
    changes_json: config_types.Path | None = None
    bazel_overrides: dict[str, config_types.Path] = dataclasses.field(
        default_factory=dict
    )

    # Current revision number.
    def revision(self) -> str:
        if hasattr(self, '_revision'):
            return self._revision

        self._revision = self._api.checkout.get_revision(self.top)
        return self._revision

    @property
    def manifest_path(self) -> config_types.Path:
        return self.top / self.options.manifest_file

    def applied_changes(self) -> list[Change]:
        return [x for x in self.changes if x.applied]

    # Repo manifest with all projects pinned.
    def manifest_snapshot(self):
        if not self.options.use_repo:
            return None

        if hasattr(self, '_manifest_snapshot'):
            return self._manifest_snapshot

        with self._api.context(cwd=self.top):
            self._manifest_snapshot = self._api.repo.manifest_snapshot()
            return self._manifest_snapshot

    # Equivalent of manifest_snapshot() but not as strictly formatted.
    def submodule_snapshot(self):
        if self.options.use_repo:
            return None

        if hasattr(self, '_submodule_snapshot'):
            return self._submodule_snapshot

        with self._api.context(cwd=self.top):
            # To get step_test_data line to pass pylint.
            raw_io_stream_output = self._api.raw_io.test_api.stream_output_text

            self._submodule_snapshot = (
                self._api.git(
                    'submodule-status',
                    'submodule',
                    'status',
                    '--recursive',
                    stdout=self._api.raw_io.output_text(),
                    step_test_data=lambda: raw_io_stream_output(
                        'submodule status filler text',
                    ),
                    ok_ret='any',
                ).stdout.strip()
                or ''
            )
            return self._submodule_snapshot

    def snapshot_to_dir(self, directory: config_types.Path) -> None:
        self._api.file.ensure_directory('mkdir', directory)
        if self.manifest_snapshot():
            self._api.file.write_text(
                'write manifest.xml',
                directory / 'manifest.xml',
                self.manifest_snapshot(),
            )

        if self.submodule_snapshot():
            self._api.file.write_text(
                'write submodule snapshot',
                directory / 'submodules.log',
                self.submodule_snapshot(),
            )

        with self._api.context(cwd=self.top):
            log = self._api.git(
                'log',
                'log',
                '--oneline',
                '-n',
                '10',
                stdout=self._api.raw_io.output_text(),
                ok_ret='any',
            ).stdout
        self._api.file.write_text(
            'write git log',
            directory / 'git.log',
            log,
        )

    def submodules(self, recursive: bool = False) -> list[Submodule]:
        """Return data about all submodules."""

        cmd = [
            'python3',
            self._api.checkout.resource('submodule_status.py'),
            self.top,
            self._api.json.output(),
        ]

        if recursive:
            cmd.append('--recursive')

        submodules = []
        submodule_status = self._api.step(
            'submodule status',
            cmd,
            step_test_data=lambda: self._api.json.test_api.output({}),
        ).json.output
        for sub in submodule_status.values():
            sub['remote'] = self._api.sso.sso_to_https(sub['remote'])
            if sub['remote'].endswith('.git'):
                sub['remote'] = sub['remote'][:-4]
            sub['relative_path'] = sub['path']
            sub['path'] = self.top / sub['path']
            try:
                submodules.append(Submodule(self._api, **sub))
            except Exception as exc:  # pragma: no cover
                self._api.step.empty(
                    'exception',
                    status='FAILURE',
                    raise_on_failure=False,
                ).presentation.step_summary_text = repr(exc)
                raise ValueError(f'submodule {sub["path"]}: {sub}')

        return submodules

    _REMOTE_REGEX = re.compile(r'^https://(?P<host>[^/]+)/(?P<project>.+)$')

    def gerrit_host(self) -> str | None:
        match = self._REMOTE_REGEX.match(self.options.remote)
        if not match:
            return  # pragma: no cover

        gerrit_review_host = f'{match.group("host")}'
        if '-review' not in gerrit_review_host:
            gerrit_review_host = gerrit_review_host.replace('.', '-review.', 1)
        return gerrit_review_host

    def gerrit_project(self) -> str | None:
        match = self._REMOTE_REGEX.match(self.options.remote)
        if not match:
            return  # pragma: no cover

        return match.group('project')

    def remotes_equivalent(
        self,
        remote1: str | None,
        remote2: str | None,
    ) -> bool:
        return self._api.checkout.remotes_equivalent(
            self.equivalent_remotes,
            remote1,
            remote2,
        )