orchestrata_ultimate.cxp.version_compat

Reference › orchestrata_ultimate › cxp › version_compat

version_compat decides whether two CXP peers can talk, using the types in version. VersionChecker.check_compatibility compares a local and a remote version string and returns a CompatibilityResult: COMPATIBLE for the same major and minor, WARN for the same major but a different minor, and INCOMPATIBLE for a different major. negotiate_version returns the version both sides can honour: the shared major with the lower minor and the lower patch of the two, and it raises ValueError when the majors differ. get_supported_versions lists the versions this implementation supports, currently 1.0.0 only. See cxp.

At a glance

Class diagram

Dependency diagram

API

Classes

CompatibilityResult (enum)

Bases: Enum · Decorators: traces(spec='004', story='US-010', reqs=['REQ-004-US-010-007', 'REQ-004-US-010-009']) · Line 26

Outcome of a version compatibility check.

VersionChecker

Line 34

Checks and negotiates CXP protocol version compatibility.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/cxp/version_compat.py

"""CXP Version Compatibility Checking.

Provides compatibility evaluation and version negotiation between
protocol endpoints.
"""

from __future__ import annotations

import logging
from enum import Enum

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.cxp.version import ProtocolVersion, parse_version

# Spec: 004 - CXP Protocol


logger = logging.getLogger(__name__)


@traces(
    spec="004",
    story="US-010",
    reqs=["REQ-004-US-010-007", "REQ-004-US-010-009"],
)
class CompatibilityResult(Enum):
    """Outcome of a version compatibility check."""

    COMPATIBLE = "compatible"
    WARN = "warn"
    INCOMPATIBLE = "incompatible"


class VersionChecker:
    """Checks and negotiates CXP protocol version compatibility."""

    @traces(
        spec="004",
        story="US-007",
        reqs=[
            "REQ-004-US-007-017",
            "REQ-004-US-010-005",
            "REQ-004-US-010-006",
            "REQ-004-US-010-007",
            "REQ-004-US-010-008",
            "REQ-004-US-010-009",
        ],
    )
    def check_compatibility(self, local: str, remote: str) -> CompatibilityResult:
        """Compare *local* and *remote* version strings.

        Returns
        -------
        CompatibilityResult
            COMPATIBLE  – same major and minor
            WARN        – same major, different minor
            INCOMPATIBLE – different major
        """
        local_v = parse_version(local)
        remote_v = parse_version(remote)

        if local_v.major != remote_v.major:
            return CompatibilityResult.INCOMPATIBLE

        if local_v.minor != remote_v.minor:
            return CompatibilityResult.WARN

        return CompatibilityResult.COMPATIBLE

    @traces(
        spec="004",
        story="US-010",
        reqs=[
            "REQ-004-US-010-010",
            "REQ-004-US-010-012",
            "REQ-004-US-010-013",
        ],
    )
    def negotiate_version(self, local: str, remote: str) -> str:
        """Return the highest mutually compatible version.

        Both sides must share the same major version.  The negotiated
        result uses the *lower* minor and patch of the two so that both
        endpoints support every feature implied by the version number.

        Raises
        ------
        ValueError
            If the versions are incompatible (different major).
        """
        local_v = parse_version(local)
        remote_v = parse_version(remote)

        if local_v.major != remote_v.major:
            raise ValueError(
                f"Cannot negotiate between incompatible versions: "
                f"{local} and {remote} (major version mismatch)"
            )

        negotiated_minor = min(local_v.minor, remote_v.minor)
        negotiated_patch = min(local_v.patch, remote_v.patch)

        return str(
            ProtocolVersion(f"{local_v.major}.{negotiated_minor}.{negotiated_patch}")
        )

    @traces(
        spec="004",
        story="US-010",
        reqs=["REQ-004-US-010-011"],
    )
    def get_supported_versions(self) -> list[str]:
        """Return the list of protocol versions this implementation supports."""
        return ["1.0.0"]