sarasacw-omrf/packer/src/sarasacw_omrf_packer/metadata.py

import json
import os
import subprocess
from dataclasses import dataclass
from pathlib import Path
from typing import Any
import tomli
from semver import Version


# YYC MARK:
# We use `tomli` with at least 2.4.0 version by design.
# Considering that Cargo has approve the change allowing TOML 1.1 syntax in `Cargo.toml`,
# Python embedded `toml` package, which only support TOML 1.0 syntax until Python 3.15,
# is not suit for parsing `Cargo.toml` in future.
# So we use the upstream of official `toml` package, i.e. `tomli` as our solution.
# And according to the document if `tomli`, the version starting support TOML 1.1 syntax is 2.4.0.


@dataclass(frozen=True)
class MetadataHeader:
    from_path: str
    to_path: str

    def __post_init__(self) -> None:
        if self.from_path == "":
            raise ValueError("header asset 'from' must not be empty")

    @staticmethod
    def from_dict(d: dict[str, Any]) -> "MetadataHeader":
        from_path = d.get("from")
        if from_path is None:
            raise ValueError("header asset is missing required key 'from'")
        if not isinstance(from_path, str):
            raise TypeError("header asset 'from' must be a string")
        to_path = d.get("to", "")
        if not isinstance(to_path, str):
            raise TypeError("header asset 'to' must be a string")
        return MetadataHeader(from_path=from_path, to_path=to_path)


@dataclass(frozen=True)
class MetadataCMake:
    namespace_name: str | None
    target_name: str | None

    @staticmethod
    def from_dict(d: dict[str, Any]) -> "MetadataCMake":
        namespace_name = d.get("namespace_name")
        if namespace_name is not None and not isinstance(namespace_name, str):
            raise TypeError("cmake 'namespace_name' must be a string or None")
        target_name = d.get("target_name")
        if target_name is not None and not isinstance(target_name, str):
            raise TypeError("cmake 'target_name' must be a string or None")
        return MetadataCMake(namespace_name=namespace_name, target_name=target_name)


@dataclass(frozen=True)
class MetadataPkgConfig:
    name: str | None
    description: str | None

    @staticmethod
    def from_dict(d: dict[str, Any]) -> "MetadataPkgConfig":
        name = d.get("name")
        if name is not None and not isinstance(name, str):
            raise TypeError("pkgconfig 'name' must be a string or None")
        description = d.get("description")
        if description is not None and not isinstance(description, str):
            raise TypeError("pkgconfig 'description' must be a string or None")
        return MetadataPkgConfig(name=name, description=description)


@dataclass(frozen=True)
class Metadata:
    min_version: Version | None
    headers: tuple[MetadataHeader, ...]
    cmake: MetadataCMake | None
    pkgconfig: MetadataPkgConfig | None

    @staticmethod
    def from_dict(d: dict[str, Any]) -> "Metadata":
        raw_min_version = d.get("min_version")
        if raw_min_version is not None:
            if not isinstance(raw_min_version, str):
                raise TypeError("min_version must be a string")
            try:
                min_version = Version.parse(raw_min_version)
            except ValueError as e:
                raise ValueError(f"invalid min_version {raw_min_version!r}: {e}") from e
        else:
            min_version = None

        raw_headers = d.get("headers")
        if raw_headers is None:
            raise ValueError(
                "headers is required (use an empty list to distribute no headers)"
            )
        if not isinstance(raw_headers, list):
            raise TypeError("headers must be a list")
        headers_list: list[MetadataHeader] = []
        for i, item in enumerate(raw_headers):
            if not isinstance(item, dict):
                raise TypeError(f"headers[{i}] must be a table")
            headers_list.append(MetadataHeader.from_dict(item))
        headers = tuple(headers_list)

        raw_cmake = d.get("cmake")
        if raw_cmake is not None:
            if not isinstance(raw_cmake, dict):
                raise TypeError("cmake must be a table")
            cmake = MetadataCMake.from_dict(raw_cmake)
        else:
            cmake = None

        raw_pkgconfig = d.get("pkgconfig")
        if raw_pkgconfig is not None:
            if not isinstance(raw_pkgconfig, dict):
                raise TypeError("pkgconfig must be a table")
            pkgconfig = MetadataPkgConfig.from_dict(raw_pkgconfig)
        else:
            pkgconfig = None

        return Metadata(
            min_version=min_version,
            headers=headers,
            cmake=cmake,
            pkgconfig=pkgconfig,
        )


class MetadataExtractor:
    __cargo_toml_path: Path
    __cargo_metadata: dict[str, Any]
    __cargo_toml: dict[str, Any]

    def __init__(self, cargo_toml: Path) -> None:
        self.__cargo_toml_path = cargo_toml
        self.__cargo_toml = MetadataExtractor.__extract_cargo_toml(cargo_toml)
        self.__cargo_metadata = MetadataExtractor.__extract_cargo_metadata(cargo_toml)

    def extract_omrf_metadata(self) -> Metadata:
        omrf = self.__cargo_toml.get("package", {}).get("metadata", {}).get("omrf")
        if not isinstance(omrf, dict):
            raise ValueError("[package.metadata.omrf] is required")
        return Metadata.from_dict(omrf)

    @staticmethod
    def __extract_cargo_toml(cargo_toml: Path) -> dict[str, Any]:
        with open(cargo_toml, "rb") as f:
            return tomli.load(f)

    @staticmethod
    def __extract_cargo_metadata(cargo_toml: Path) -> dict[str, Any]:
        cargo_bin = os.getenv("OMRF_PACKER_CARGO", "cargo")
        cmd = [
            cargo_bin,
            "metadata",
            "--no-deps",
            "--format-version",
            "1",
            "--manifest-path",
            str(cargo_toml),
        ]
        proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        try:
            stdout, stderr = proc.communicate(timeout=10)
        except subprocess.TimeoutExpired:
            proc.kill()
            proc.communicate()
            raise RuntimeError("fail to fetch cargo metadata: timed out")
        if proc.returncode != 0:
            raise RuntimeError(
                "fail to fetch cargo metadata: "
                + stderr.decode("utf-8", errors="ignore")
            )
        else:
            return json.loads(stdout)