datastore/ci/jobs/scripts/datastore_version.py

from pathlib import Path

from ci.praktika.info import Info
from ci.praktika.utils import Shell

FILE_WITH_VERSION_PATH = "./cmake/autogenerated_versions.txt"

# Canonical layout of cmake/autogenerated_versions.txt as committed to the repo.
VERSIONS_TEMPLATE = """\
# This variables autochanged by ci/jobs/scripts/create_release.py:

# NOTE: VERSION_REVISION has nothing common with DBMS_TCP_PROTOCOL_VERSION,
# only DBMS_TCP_PROTOCOL_VERSION should be incremented on protocol changes.
SET(VERSION_REVISION {revision})
SET(VERSION_MAJOR {major})
SET(VERSION_MINOR {minor})
SET(VERSION_PATCH {patch})
SET(VERSION_GITHASH {githash})
SET(VERSION_DESCRIBE {describe})
SET(VERSION_STRING {string})
# end of autochange
"""


class VersionType:
    # `new` is the type stamped on master right after a release branch is cut
    # (the `vX.Y.1.1-new` marker); the CreateRelease flow relies on it.
    NEW = "new"
    TESTING = "testing"
    STABLE = "stable"
    LTS = "lts"
    CLOUD = "cloud"
    VALID = {NEW, TESTING, STABLE, LTS, CLOUD}


def _read_versions() -> dict:
    versions = {}
    for line in Path(FILE_WITH_VERSION_PATH).read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line.startswith("SET("):
            continue
        name, value = line[4:-1].split(maxsplit=1)
        name = name.removeprefix("VERSION_").lower()
        try:
            value = int(value)
        except ValueError:
            pass
        versions[name] = value
    return versions


def _write_version_file(version_dict: dict) -> None:
    Path(FILE_WITH_VERSION_PATH).write_text(
        VERSIONS_TEMPLATE.format_map(version_dict), encoding="utf-8"
    )


def _tweak_from_git(githash) -> int:
    """The tweak: the commit count since `githash` on the first-parent history
    (`git rev-list --count`). It does not use tags, so tags need not be fetched.

    Raises on a shallow checkout or any git error instead of silently returning a
    default -- a caller that can tolerate an unknown tweak must opt into that
    explicitly (see `CHVersion.get_current_version`)."""
    return int(
        Shell.get_output(
            f"git rev-list --count --first-parent {githash}..HEAD",
            verbose=True,
        )
    )


def _version_type_from_describe(versions: dict) -> str:
    """Recover the version type from the file's `VERSION_DESCRIBE` suffix, which
    is `v<string>-<version_type>`."""
    describe = str(versions.get("describe", ""))
    return describe.rsplit("-", 1)[1] if "-" in describe else ""


class CHVersion:
    """The Datastore release version.

    A stateful model (`major.minor.patch.tweak`, plus `revision`, `githash` and
    `version_type`, with the derived `string`/`describe` kept in sync) that the
    release process reads, bumps and writes. `get_current_version` /
    `get_release_version` read it from `cmake/autogenerated_versions.txt` without
    consulting the CI `Info` context, so they are safe inside a checkout of an
    arbitrary release commit; `get_current_version_from_ci_pipeline` reads it back
    from the pipeline kv storage for jobs that run without a full checkout.
    """

    def __init__(
        self,
        major,
        minor,
        patch,
        revision,
        tweak=1,
        githash="",
        version_type="",
    ):
        self.major = int(major)
        self.minor = int(minor)
        self.patch = int(patch)
        self.revision = int(revision)
        self.tweak = int(tweak)
        self.githash = githash
        self.version_type = version_type
        self._refresh()

    def _refresh(self) -> "CHVersion":
        """Recompute the derived `string`/`describe` members from the parts."""
        self.string = f"{self.major}.{self.minor}.{self.patch}.{self.tweak}"
        self.describe = (
            f"v{self.string}-{self.version_type}" if self.version_type else ""
        )
        return self

    # --- reading / writing the version model ---------------------------------

    @classmethod
    def get_current_version(
        cls, with_tweak: bool = True, no_strict: bool = False
    ) -> "CHVersion":
        """The version currently in `cmake/autogenerated_versions.txt`.

        `version_type` is read from the file's `VERSION_DESCRIBE` suffix and the
        githash is the current `HEAD`. With `with_tweak`, the tweak is the commit
        count since the file's githash (git history, no tags); on a shallow
        checkout that count cannot be computed, so this raises unless `no_strict`
        is set, in which case the tweak stays 1. Consults neither the CI `Info`
        context nor `_Environment`, so it is safe inside a checkout of an
        arbitrary release commit."""
        versions = _read_versions()
        tweak = 1
        if with_tweak:
            try:
                tweak = _tweak_from_git(versions["githash"])
            except Exception:
                if not no_strict:
                    raise
        return cls(
            versions["major"],
            versions["minor"],
            versions["patch"],
            versions["revision"],
            tweak=tweak,
            githash=Shell.get_output("git rev-parse HEAD") or "0" * 40,
            version_type=_version_type_from_describe(versions),
        )

    @classmethod
    def get_release_version(cls) -> "CHVersion":
        """The release version exactly as stored in the file -- githash and tweak
        as committed, without recomputing anything from git. For the CreateRelease
        workflow."""
        versions = _read_versions()
        string = str(versions.get("string", ""))
        tweak = int(string.rsplit(".", 1)[1]) if string.count(".") >= 3 else 1
        return cls(
            versions["major"],
            versions["minor"],
            versions["patch"],
            versions["revision"],
            tweak=tweak,
            githash=str(versions.get("githash", "")),
            version_type=_version_type_from_describe(versions),
        )

    @classmethod
    def get_current_version_from_ci_pipeline(cls):
        """The version stored in the CI pipeline kv data during the build (see
        `store_version_data_in_ci_pipeline`), for jobs that run without a full
        checkout. Returns None when nothing was stored yet."""
        data = Info().get_kv_data("version")
        return cls.from_dict(data) if data else None

    @classmethod
    def from_dict(cls, data: dict) -> "CHVersion":
        string = str(data.get("string", ""))
        tweak = data.get("tweak")
        if tweak is None:
            tweak = int(string.rsplit(".", 1)[1]) if string.count(".") >= 3 else 1
        return cls(
            data["major"],
            data["minor"],
            data["patch"],
            data["revision"],
            tweak=tweak,
            githash=str(data.get("githash", "")),
            version_type=_version_type_from_describe(data),
        )

    def to_dict(self) -> dict:
        version_dict = {
            "revision": self.revision,
            "major": self.major,
            "minor": self.minor,
            "patch": self.patch,
            "tweak": self.tweak,
            "githash": self.githash,
            "describe": self.describe,
            "string": self.string,
        }
        missing = [k for k in ("githash", "describe", "string") if not version_dict[k]]
        assert not missing, f"version fields not set: {missing} in {version_dict}"
        return version_dict

    def write(self) -> None:
        """Write this version to `cmake/autogenerated_versions.txt`."""
        _write_version_file(self.to_dict())

    # --- bumping -------------------------------------------------------------

    def bump_release(self) -> "CHVersion":
        """Bump the minor version (or wrap to the next major when minor reaches 12)."""
        if self.minor < 12:
            self.minor += 1
        else:
            self.major += 1
            self.minor = 1
        self.revision += 1
        self.patch = 1
        self.tweak = 1
        return self._refresh()

    def bump_patch(self) -> "CHVersion":
        self.revision += 1
        self.patch += 1
        self.tweak = 1
        return self._refresh()

    # --- comparison / rendering ----------------------------------------------

    def with_description(self, version_type: str) -> "CHVersion":
        if version_type not in VersionType.VALID:
            raise ValueError(f"version type {version_type!r} not in {VersionType.VALID}")
        self.version_type = version_type
        return self._refresh()

    def with_tweak(self, tweak: int) -> "CHVersion":
        self.tweak = int(tweak)
        return self._refresh()

    def get_stable_release_type(self) -> str:
        return VersionType.LTS if self.minor % 5 == 3 else VersionType.STABLE

    def is_older(self, other: "CHVersion") -> bool:
        """Whether this is an older release than `other`, by release triple
        (major, minor, patch). The tweak is a within-patch commit count and does
        not define a newer release."""
        return (self.major, self.minor, self.patch) < (
            other.major,
            other.minor,
            other.patch,
        )

    # --- CI pipeline kv storage ----------------------------------------------

    def store_version_data_in_ci_pipeline(self) -> None:
        version = self.to_dict()
        print(f"Store version in pipeline kv data: [version={version}]")
        Info().store_kv_data("version", version)


if __name__ == "__main__":
    # test;
    print(CHVersion.get_current_version().to_dict())