backwards_compatibility.py

# Copyright (c) 2022 Project CHIP 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
#
#   http://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.

import dataclasses
import enum
import logging
from typing import Callable, Dict, List, Optional, Protocol, TypeVar

from matter_idl.matter_idl_types import ApiMaturity, Attribute, Bitmap, Cluster, Command, Enum, Event, Field, Idl, Struct


class Compatibility(enum.Enum):
    UNKNOWN = enum.auto()
    COMPATIBLE = enum.auto()
    INCOMPATIBLE = enum.auto()


T = TypeVar('T')


class HasName(Protocol):
    name: str


NAMED = TypeVar('NAMED', bound=HasName)


def group_list(items: List[T], get_id: Callable[[T], str]) -> Dict[str, T]:
    result = {}
    for item in items:
        result[get_id(item)] = item
    return result


def group_list_by_name(items: List[NAMED]) -> Dict[str, NAMED]:
    return group_list(items, lambda x: x.name)


def attribute_name(attribute: Attribute) -> str:
    """Get the name of an attribute."""
    return attribute.definition.name


def not_stable(maturity: ApiMaturity):
    """Determine if the given api maturity allows binary/api changes or not."""
    # TODO: internal and deprecated not currently widely used,
    #       so we enforce stability on them for now.
    return maturity == ApiMaturity.PROVISIONAL


class CompatibilityChecker:
    def __init__(self, original: Idl, updated: Idl):
        self._original_idl = original
        self._updated_idl = updated
        self.compatible = Compatibility.UNKNOWN
        self.errors: List[str] = []
        self.logger = logging.getLogger(__name__)

    def _mark_incompatible(self, reason: str):
        self.logger.error(reason)
        self.errors.append(reason)
        self.compatible = Compatibility.INCOMPATIBLE

    def _check_field_lists_are_the_same(self, location: str, original: List[Field], updated: List[Field]):
        """Validates no compatibility changes in a list of fields.

        Specifically no changes are allowed EXCEPT names of fields.
        """

        # Every field MUST be the same except that
        # name does not matter and order does not matter
        #
        # Comparison is done on a dict (so order does not matter)
        # and replacing names with a fixed name based on code.
        a = {}
        for item in original:
            a[item.code] = dataclasses.replace(item, name=f"entry{item.code}")

        b = {}
        for item in updated:
            b[item.code] = dataclasses.replace(item, name=f"entry{item.code}")

        if a != b:
            self._mark_incompatible(f"{location} has field changes")

    def _check_enum_compatible(self, cluster_name: str, original: Enum, updated: Optional[Enum]):
        if not updated:
            self._mark_incompatible(
                f"Enumeration {cluster_name}::{original.name} was deleted")
            return

        if original.base_type != updated.base_type:
            self._mark_incompatible(
                f"Enumeration {cluster_name}::{original.name} switched base type from {original.base_type} to {updated.base_type}")

        # Validate that all old entries exist
        for entry in original.entries:
            # old entry must exist and have identical code
            existing = [
                item for item in updated.entries if item.name == entry.name]
            if len(existing) == 0:
                self._mark_incompatible(
                    f"Enumeration {cluster_name}::{original.name} removed entry {entry.name}")
            elif existing[0].code != entry.code:
                self._mark_incompatible(
                    f"Enumeration {cluster_name}::{original.name} changed code for entry {entry.name} from {entry.code} to {existing[0].code}")

    def _check_bitmap_compatible(self, cluster_name: str, original: Bitmap, updated: Optional[Bitmap]):
        if not updated:
            self._mark_incompatible(
                f"Bitmap {cluster_name}::{original.name} was deleted")
            return

        if original.base_type != updated.base_type:
            self._mark_incompatible(
                f"Bitmap {cluster_name}::{original.name} switched base type from {original.base_type} to {updated.base_type}")

        # Validate that all old entries exist
        for entry in original.entries:
            # old entry must exist and have identical code
            existing = [
                item for item in updated.entries if item.name == entry.name]
            if len(existing) == 0:
                self._mark_incompatible(
                    f"Bitmap {original.name} removed entry {entry.name}")
            elif existing[0].code != entry.code:
                self._mark_incompatible(
                    f"Bitmap {original.name} changed code for entry {entry.name} from {entry.code} to {existing[0].code}")

    def _check_event_compatible(self, cluster_name: str, event: Event, updated_event: Optional[Event]):
        if not updated_event:
            self._mark_incompatible(
                f"Event {cluster_name}::{event.name} was removed")
            return

        if event.code != updated_event.code:
            self._mark_incompatible(
                f"Event {cluster_name}::{event.name} code changed from {event.code} to {updated_event.code}")

        self._check_field_lists_are_the_same(
            f"Event {cluster_name}::{event.name}", event.fields, updated_event.fields)

    def _check_command_compatible(self, cluster_name: str, command: Command, updated_command: Optional[Command]):
        self.logger.debug(f"  Checking command {cluster_name}::{command.name}")
        if not updated_command:
            self._mark_incompatible(
                f"Command {cluster_name}::{command.name} was removed")
            return

        if command.code != updated_command.code:
            self._mark_incompatible(
                f"Command {cluster_name}::{command.name} code changed from {command.code} to {updated_command.code}")

        if command.input_param != updated_command.input_param:
            self._mark_incompatible(
                f"Command {cluster_name}::{command.name} input changed from {command.input_param} to {updated_command.input_param}")

        if command.output_param != updated_command.output_param:
            self._mark_incompatible(
                f"Command {cluster_name}::{command.name} output changed from {command.output_param} to {updated_command.output_param}")

        if command.qualities != updated_command.qualities:
            self._mark_incompatible(
                f"Command {cluster_name}::{command.name} qualities changed from {command.qualities} to {updated_command.qualities}")

    def _check_struct_compatible(self, cluster_name: str, original: Struct, updated: Optional[Struct]):
        self.logger.debug(f"  Checking struct {original.name}")
        if not updated:
            self._mark_incompatible(
                f"Struct {cluster_name}::{original.name} has been deleted.")
            return

        self._check_field_lists_are_the_same(
            f"Struct {cluster_name}::{original.name}", original.fields, updated.fields)

        if original.tag != updated.tag:
            self._mark_incompatible(
                f"Struct {cluster_name}::{original.name} has modified tags")

        if original.code != updated.code:
            self._mark_incompatible(
                f"Struct {cluster_name}::{original.name} has modified code (likely resnopse difference)")

        if original.qualities != updated.qualities:
            self._mark_incompatible(
                f"Struct {cluster_name}::{original.name} has modified qualities")

    def _check_attribute_compatible(self, cluster_name: str, original: Attribute, updated: Optional[Attribute]):
        self.logger.debug(
            f"  Checking attribute {cluster_name}::{original.definition.name}")
        if not updated:
            self._mark_incompatible(
                f"Attribute {cluster_name}::{original.definition.name} has been deleted.")
            return

        if original.definition.code != updated.definition.code:
            self._mark_incompatible(
                f"Attribute {cluster_name}::{original.definition.name} changed its code.")

        if original.definition.data_type != updated.definition.data_type:
            self._mark_incompatible(
                f"Attribute {cluster_name}::{original.definition.name} changed its data type.")

        if original.definition.is_list != updated.definition.is_list:
            self._mark_incompatible(
                f"Attribute {cluster_name}::{original.definition.name} changed its list status.")

        if original.definition.qualities != updated.definition.qualities:
            # optional/nullable
            self._mark_incompatible(
                f"Attribute {cluster_name}::{original.definition.name} changed its data type qualities.")

        if original.qualities != updated.qualities:
            # read/write/subscribe/timed status
            self._mark_incompatible(
                f"Attribute {cluster_name}::{original.definition.name} changed its qualities.")

    def _check_enum_list_compatible(self, cluster_name: str, original: List[Enum], updated: List[Enum]):
        updated_enums = group_list_by_name(updated)

        for original_enum in original:
            updated_enum = updated_enums.get(original_enum.name)
            self._check_enum_compatible(
                cluster_name, original_enum, updated_enum)

    def _check_bitmap_list_compatible(self, cluster_name: str, original: List[Bitmap], updated: List[Bitmap]):
        updated_bitmaps = {}
        for item in updated:
            updated_bitmaps[item.name] = item

        for original_bitmap in original:
            updated_bitmap = updated_bitmaps.get(original_bitmap.name)
            self._check_bitmap_compatible(
                cluster_name, original_bitmap, updated_bitmap)

    def _check_struct_list_compatible(self, cluster_name: str, original: List[Struct], updated: List[Struct]):
        updated_structs = group_list_by_name(updated)

        for struct in original:
            self._check_struct_compatible(
                cluster_name, struct, updated_structs.get(struct.name))

    def _check_command_list_compatible(self, cluster_name: str, original: List[Command], updated: List[Command]):
        updated_commands = group_list_by_name(updated)

        for command in original:
            updated_command = updated_commands.get(command.name)
            self._check_command_compatible(
                cluster_name, command, updated_command)

    def _check_event_list_compatible(self, cluster_name: str, original: List[Event], updated: List[Event]):
        updated_events = group_list_by_name(updated)

        for event in original:
            updated_event = updated_events.get(event.name)
            self._check_event_compatible(cluster_name, event, updated_event)

    def _check_attribute_list_compatible(self, cluster_name: str, original: List[Attribute], updated: List[Attribute]):
        updated_attributes = group_list(updated, attribute_name)

        for attribute in original:
            self._check_attribute_compatible(
                cluster_name, attribute, updated_attributes.get(attribute_name(attribute)))

    def _check_cluster_list_compatible(self, original: List[Cluster], updated: List[Cluster]):
        updated_clusters = group_list(updated, lambda c: c.name)

        for original_cluster in original:
            updated_cluster = updated_clusters.get(original_cluster.name)

            if not_stable(original_cluster.api_maturity):
                continue

            if updated_cluster and not_stable(updated_cluster.api_maturity):
                continue

            self._check_cluster_compatible(original_cluster, updated_cluster)

    def _check_cluster_compatible(self, original_cluster: Cluster, updated_cluster: Optional[Cluster]):
        self.logger.debug(
            f"Checking cluster {original_cluster.name}")
        if not updated_cluster:
            self._mark_incompatible(
                f"Cluster {original_cluster.name} was deleted")
            return

        if original_cluster.code != updated_cluster.code:
            self._mark_incompatible(
                f"Cluster {original_cluster.name} has different codes {original_cluster.code} != {updated_cluster.code}")

        self._check_enum_list_compatible(
            original_cluster.name, original_cluster.enums, updated_cluster.enums)
        self._check_struct_list_compatible(
            original_cluster.name, original_cluster.structs, updated_cluster.structs)
        self._check_bitmap_list_compatible(
            original_cluster.name, original_cluster.bitmaps, updated_cluster.bitmaps)
        self._check_command_list_compatible(
            original_cluster.name, original_cluster.commands, updated_cluster.commands)
        self._check_event_list_compatible(
            original_cluster.name, original_cluster.events, updated_cluster.events)
        self._check_attribute_list_compatible(
            original_cluster.name, original_cluster.attributes, updated_cluster.attributes)

    def check(self):
        # assume ok, and then validate
        self.compatible = Compatibility.COMPATIBLE

        self._check_cluster_list_compatible(
            self._original_idl.clusters, self._updated_idl.clusters)

        return self.compatible


def is_backwards_compatible(original: Idl, updated: Idl):
    """
    Validate that 'updated' IDL contains only
    incremental changes from 'original'
    """
    checker = CompatibilityChecker(original, updated)
    return checker.check() == Compatibility.COMPATIBLE