Add collection meta code

changelogs/fragments/168-collection-meta.yml

@@ -0,0 +1,2 @@
+minor_changes:
+  - "Add schema and validation helper for ansible-build-data's collection meta (https://github.com/ansible-community/ansible-build-data/pull/450, https://github.com/ansible-community/antsibull-core/pull/168)."

src/antsibull_core/collection_meta.py

@@ -0,0 +1,161 @@
+# Author: Felix Fontein <[email protected]>
+# Author: Toshio Kuratomi <[email protected]>
+# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or
+# https://www.gnu.org/licenses/gpl-3.0.txt)
+# SPDX-License-Identifier: GPL-3.0-or-later
+# SPDX-FileCopyrightText: Ansible Project, 2020
+
+"""
+Classes to encapsulate collection metadata from collection-meta.yaml
+"""
+
+from __future__ import annotations
+
+import os
+import typing as t
+
+import pydantic as p
+from antsibull_fileutils.yaml import load_yaml_file
+
+from .schemas.collection_meta import (
+    CollectionMetadata,
+    CollectionsMetadata,
+    RemovalInformation,
+)
+
+if t.TYPE_CHECKING:
+    from _typeshed import StrPath
+
+
+class _Validator:
+    """
+    Validate a CollectionsMetadata object.
+
+    Validation error messages are added to the ``errors`` attribute.
+    """
+
+    errors: list[str]
+
+    def __init__(self, *, all_collections: list[str], major_release: int):
+        self.errors = []
+        self.all_collections = all_collections
+        self.major_release = major_release
+
+    def _validate_removal(
+        self, collection: str, removal: RemovalInformation, prefix: str
+    ) -> None:
+        if (
+            removal.major_version != "TBD"
+            and removal.major_version <= self.major_release  # pyre-ignore[58]
+        ):
+            self.errors.append(
+                f"{prefix} major_version: Removal major version {removal.major_version} must"
+                f" be larger than current major version {self.major_release}"
+            )
+
+        if (
+            removal.announce_version is not None
+            and removal.announce_version.major > self.major_release
+        ):
+            self.errors.append(
+                f"{prefix} announce_version: Major version of {removal.announce_version}"
+                f" must not be larger than the current major version {self.major_release}"
+            )
+
+        if removal.redirect_replacement_major_version is not None:
+            if removal.redirect_replacement_major_version <= self.major_release:
+                self.errors.append(
+                    f"{prefix} redirect_replacement_major_version: Redirect removal version"
+                    f" {removal.redirect_replacement_major_version} must be larger than"
+                    f" current major version {self.major_release}"
+                )
+            if (
+                removal.major_version != "TBD"
+                and removal.redirect_replacement_major_version  # pyre-ignore[58]
+                >= removal.major_version
+            ):
+                self.errors.append(
+                    f"{prefix} redirect_replacement_major_version: Redirect removal major version"
+                    f" {removal.redirect_replacement_major_version} must be smaller than"
+                    f" the removal major version {removal.major_version}"
+                )
+
+        if removal.reason == "renamed" and removal.new_name == collection:
+            self.errors.append(f"{prefix} new_name: Must not be the collection's name")
+
+    def _validate_collection(
+        self, collection: str, meta: CollectionMetadata, prefix: str
+    ) -> None:
+        if meta.repository is None:
+            self.errors.append(f"{prefix} repository: Required field not provided")
+
+        if meta.removal:
+            self._validate_removal(collection, meta.removal, f"{prefix} removal ->")
+
+    def validate(self, data: CollectionsMetadata) -> None:
+        # Check order
+        sorted_list = sorted(data.collections)
+        raw_list = list(data.collections)
+        if raw_list != sorted_list:
+            for raw_entry, sorted_entry in zip(raw_list, sorted_list):
+                if raw_entry != sorted_entry:
+                    self.errors.append(
+                        "The collection list must be sorted; "
+                        f"{sorted_entry!r} must come before {raw_entry}"
+                    )
+                    break
+
+        # Validate collection data
+        remaining_collections = set(self.all_collections)
+        for collection, meta in data.collections.items():
+            if collection not in remaining_collections:
+                self.errors.append(
+                    f"collections -> {collection}: Collection not in ansible.in"
+                )
+            else:
+                remaining_collections.remove(collection)
+            self._validate_collection(
+                collection, meta, f"collections -> {collection} ->"
+            )
+
+        # Complain about remaining collections
+        for collection in sorted(remaining_collections):
+            self.errors.append(f"collections: No metadata present for {collection}")
+
+
+def lint_collection_meta(
+    *, collection_meta_path: StrPath, major_release: int, all_collections: list[str]
+) -> list[str]:
+    """Lint collection-meta.yaml."""
+    if not os.path.exists(collection_meta_path):
+        return [f"Cannot find {collection_meta_path}"]
+
+    try:
+        data = load_yaml_file(collection_meta_path)
+    except Exception as exc:  # pylint: disable=broad-exception-caught
+        return [f"Error while parsing YAML file: {exc}"]
+
+    validator = _Validator(
+        all_collections=all_collections,
+        major_release=major_release,
+    )
+
+    for cls in (
+        # NOTE: The order is important here! The most deeply nested classes must come first,
+        #       otherwise extra=forbid might not be used for something deeper in the hierarchy.
+        RemovalInformation,
+        CollectionMetadata,
+        CollectionsMetadata,
+    ):
+        cls.model_config["extra"] = "forbid"
+        cls.model_rebuild(force=True)
+
+    try:
+        parsed_data = CollectionsMetadata.model_validate(data)
+        validator.validate(parsed_data)
+    except p.ValidationError as exc:
+        for error in exc.errors():
+            location = " -> ".join(str(loc) for loc in error["loc"])
+            validator.errors.append(f'{location}: {error["msg"]}')
+
+    return sorted(validator.errors)

src/antsibull_core/schemas/collection_meta.py

@@ -0,0 +1,156 @@
+# Author: Felix Fontein <[email protected]>
+# Author: Toshio Kuratomi <[email protected]>
+# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or
+# https://www.gnu.org/licenses/gpl-3.0.txt)
+# SPDX-License-Identifier: GPL-3.0-or-later
+# SPDX-FileCopyrightText: Ansible Project, 2020
+
+"""
+Classes to encapsulate collection metadata from collection-meta.yaml
+"""
+
+from __future__ import annotations
+
+import os
+import typing as t
+
+import pydantic as p
+from antsibull_fileutils.yaml import load_yaml_file
+from packaging.version import Version as PypiVer
+from pydantic.functional_validators import BeforeValidator
+from typing_extensions import Annotated, Self
+
+if t.TYPE_CHECKING:
+    from _typeshed import StrPath
+
+
+def _convert_pypi_version(v: t.Any) -> t.Any:
+    if isinstance(v, str):
+        if not v:
+            raise ValueError(f"must be a non-trivial string, got {v!r}")
+        version = PypiVer(v)
+    elif isinstance(v, PypiVer):
+        version = v
+    else:
+        raise ValueError(f"must be a string or PypiVer object, got {v!r}")
+
+    if len(version.release) != 3:
+        raise ValueError(
+            f"must be a version with three release numbers (e.g. 1.2.3, 2.3.4a1), got {v!r}"
+        )
+    return version
+
+
+PydanticPypiVersion = Annotated[PypiVer, BeforeValidator(_convert_pypi_version)]
+
+
+class RemovalInformation(p.BaseModel):
+    """
+    Stores metadata on when and why a collection will get removed.
+    """
+
+    model_config = p.ConfigDict(extra="ignore", arbitrary_types_allowed=True)
+
+    major_version: t.Union[int, t.Literal["TBD"]]
+    reason: t.Literal[
+        "deprecated",
+        "considered-unmaintained",
+        "renamed",
+        "guidelines-violation",
+        "other",
+    ]
+    reason_text: t.Optional[str] = None
+    announce_version: t.Optional[PydanticPypiVersion] = None
+    new_name: t.Optional[str] = None
+    discussion: t.Optional[p.HttpUrl] = None
+    redirect_replacement_major_version: t.Optional[int] = None
+
+    @p.model_validator(mode="after")  # pyre-ignore[56]
+    def _check_reason_text(self) -> Self:
+        reasons_with_text = ("other", "guidelines-violation")
+        if self.reason in reasons_with_text:
+            if self.reason_text is None:
+                reasons = ", ".join(f"'{reason}'" for reason in reasons_with_text)
+                raise ValueError(f"reason_text must be provided if reason is {reasons}")
+        else:
+            if self.reason_text is not None:
+                reasons = ", ".join(f"'{reason}'" for reason in reasons_with_text)
+                raise ValueError(
+                    f"reason_text must not be provided if reason is not {reasons}"
+                )
+        return self
+
+    @p.model_validator(mode="after")  # pyre-ignore[56]
+    def _check_reason_is_renamed(self) -> Self:
+        if self.reason != "renamed":
+            return self
+        if self.new_name is None:
+            raise ValueError("new_name must be provided if reason is 'renamed'")
+        if (
+            self.redirect_replacement_major_version is not None
+            and self.major_version != "TBD"
+            and self.redirect_replacement_major_version
+            >= self.major_version  # pyre-ignore[58]
+        ):
+            raise ValueError(
+                "redirect_replacement_major_version must be smaller than major_version"
+            )
+        return self
+
+    @p.model_validator(mode="after")  # pyre-ignore[56]
+    def _check_reason_is_not_renamed(self) -> Self:
+        if self.reason == "renamed":
+            return self
+        if self.new_name is not None:
+            raise ValueError("new_name must not be provided if reason is not 'renamed'")
+        if self.redirect_replacement_major_version is not None:
+            raise ValueError(
+                "redirect_replacement_major_version must not be provided if reason is not 'renamed'"
+            )
+        if self.major_version == "TBD":
+            raise ValueError("major_version must not be TBD if reason is not 'renamed'")
+        return self
+
+
+class CollectionMetadata(p.BaseModel):
+    """
+    Stores metadata about one collection.
+    """
+
+    model_config = p.ConfigDict(extra="ignore")
+
+    changelog_url: t.Optional[str] = p.Field(alias="changelog-url", default=None)
+    collection_directory: t.Optional[str] = p.Field(
+        alias="collection-directory", default=None
+    )
+    repository: t.Optional[str] = None
+    tag_version_regex: t.Optional[str] = None
+    maintainers: list[str] = []
+    removal: t.Optional[RemovalInformation] = None
+
+
+class CollectionsMetadata(p.BaseModel):
+    """
+    Stores metadata about a set of collections.
+    """
+
+    model_config = p.ConfigDict(extra="ignore")
+
+    collections: dict[str, CollectionMetadata]
+
+    @staticmethod
+    def load_from(deps_dir: StrPath | None) -> CollectionsMetadata:
+        if deps_dir is None:
+            return CollectionsMetadata(collections={})
+        collection_meta_path = os.path.join(deps_dir, "collection-meta.yaml")
+        if not os.path.exists(collection_meta_path):
+            return CollectionsMetadata(collections={})
+        data = load_yaml_file(collection_meta_path)
+        return CollectionsMetadata.model_validate(data)
+
+    def get_meta(self, collection_name: str) -> CollectionMetadata:
+        result = self.collections.get(collection_name)
+        if result is None:
+            result = CollectionMetadata()
+            self.collections[collection_name] = result
+        return result