esphome/tests/unit_tests/components/api/test_api_proto.py

"""Invariant tests for esphome/components/api/api.proto and its generated code.

These guard the DeviceCapabilitiesRequest/DeviceCapabilitiesResponse addition
(API 1.15) against regressions that protoc-based codegen would not catch on
its own, without requiring protoc to be installed at test time:

* script/api_protobuf/api_protobuf.py skips any field marked
  `[deprecated = true]` completely -- it generates no C++ for it at all, so
  the device silently stops sending that value. Six DeviceInfoResponse fields
  were superseded by DeviceCapabilitiesResponse but must keep being sent for
  backward compatibility with clients older than API 1.15. If a future edit
  "tidies up" by marking one of them deprecated, this file breaks that field
  for every existing client with nothing else in CI noticing.
* Field numbers are the wire protocol, not the field names. Renaming a field
  is harmless; renumbering it is a silent breaking change, because an old
  client still decodes by number. This file pins the field number of each of
  the six superseded DeviceInfoResponse fields and of every field on the new
  DeviceCapabilitiesResponse/BluetoothProxyCapabilities/
  VoiceAssistantCapabilities/ZWaveProxyCapabilities sub-messages, so a
  well-intentioned reshuffle of api.proto gets caught here instead of on a
  device in the field.
* Message wire ids must be unique, and the new capabilities RPC must stay
  authenticated-only.

Group A below asserts on the checked-in generated files (api_pb2.h /
api_pb2.cpp), since "the field is present in the generated C++" is exactly
equivalent to "the device still sends it". Group B parses api.proto as plain
text (no protoc). Group C checks the advertised API minor version.
"""

from __future__ import annotations

from pathlib import Path
import re

import esphome

API_DIR = Path(esphome.__file__).parent / "components" / "api"

PROTO_TEXT = (API_DIR / "api.proto").read_text(encoding="utf-8")
HEADER_TEXT = (API_DIR / "api_pb2.h").read_text(encoding="utf-8")
CPP_TEXT = (API_DIR / "api_pb2.cpp").read_text(encoding="utf-8")
API_CONNECTION_TEXT = (API_DIR / "api_connection.cpp").read_text(encoding="utf-8")

# Fields on DeviceInfoResponse that were superseded by DeviceCapabilitiesResponse
# as of API 1.15 but must still be generated (and therefore still sent) for
# backward compatibility with older clients.
SUPERSEDED_FIELDS: dict[str, int] = {
    "bluetooth_proxy_feature_flags": 15,
    "voice_assistant_feature_flags": 17,
    "bluetooth_mac_address": 18,
    "zwave_proxy_feature_flags": 23,
    "zwave_home_id": 24,
    "serial_proxies": 25,
}

# Field numbers on the new capability messages. These are a frozen wire
# contract from the moment they ship: an old client decodes a sub-message
# field purely by number, so renumbering any of these -- even without
# touching a name -- silently corrupts what every already-deployed client
# reads. Keyed by message name so the next capability sub-message is a
# data-only addition here.
NEW_CAPABILITY_FIELDS: dict[str, dict[str, int]] = {
    "DeviceCapabilitiesResponse": {
        "bluetooth_proxy": 1,
        "voice_assistant": 2,
        "zwave_proxy": 3,
        "serial_proxies": 4,
    },
    "BluetoothProxyCapabilities": {
        "feature_flags": 1,
        "mac_address": 2,
    },
    "VoiceAssistantCapabilities": {
        "feature_flags": 1,
    },
    "ZWaveProxyCapabilities": {
        "feature_flags": 1,
        "home_id": 2,
    },
}

# Fields that are genuinely dead and are expected to carry `deprecated=true`.
# Used to prove the deprecated-detection logic below actually detects
# deprecation rather than trivially passing.
GENUINELY_DEPRECATED_FIELDS: tuple[str, ...] = (
    "legacy_bluetooth_proxy_version",
    "legacy_voice_assistant_version",
)

DEPRECATED_FIELD_TRAP = (
    "script/api_protobuf/api_protobuf.py skips fields marked `[deprecated = "
    "true]` completely, generating no C++ for them at all. Marking this field "
    "deprecated would silently stop the device from ever sending it, breaking "
    "every existing client that still reads it from DeviceInfoResponse."
)


def _extract_braced_region(text: str, anchor_pattern: str) -> str:
    """Return the region of `text` starting at the first match of
    `anchor_pattern` up to the matching closing brace (inclusive), using
    brace-depth counting so nested braces (e.g. a `for (...) { ... }` loop
    inside a function body) don't cause a premature stop.
    """
    anchor_match = re.search(anchor_pattern, text)
    if anchor_match is None:
        raise AssertionError(f"could not find a match for {anchor_pattern!r}")
    start = anchor_match.start()
    open_brace = text.index("{", start)
    depth = 0
    for i in range(open_brace, len(text)):
        if text[i] == "{":
            depth += 1
        elif text[i] == "}":
            depth -= 1
            if depth == 0:
                return text[start : i + 1]
    raise AssertionError(f"unbalanced braces while scanning after {anchor_pattern!r}")


def _extract_class_body(header_text: str, class_name: str) -> str:
    """Return the body of a generated C++ class, scoped so a field name that
    also happens to exist on some other class cannot satisfy the assertion.
    """
    return _extract_braced_region(header_text, rf"class {re.escape(class_name)}\b")


def _extract_function_body(cpp_text: str, qualified_name: str) -> str:
    """Return the body of a generated `Class::method(...)` definition."""
    return _extract_braced_region(cpp_text, rf"{re.escape(qualified_name)}\(")


def _extract_proto_message(proto_text: str, message_name: str) -> str:
    """Return the body of a top-level `message Name { ... }` block from the
    .proto source. Proto message bodies here contain no nested `{`/`}` of
    their own (options use parens, not braces), so a non-greedy match up to
    the first line that is just `}` is sufficient and keeps the parsing
    simple.
    """
    match = re.search(
        rf"^message {re.escape(message_name)}\s*\{{(.*?)^\}}",
        proto_text,
        re.MULTILINE | re.DOTALL,
    )
    if match is None:
        raise AssertionError(f"could not find `message {message_name}` in api.proto")
    return match.group(1)


def _extract_rpc_body(proto_text: str, rpc_name: str) -> str:
    """Return the option body of an `rpc name (...) returns (...) { ... }`
    declaration from the APIConnection service, robust to it being written
    on one line (`{}`) or spread across several with options inside.
    """
    match = re.search(
        rf"rpc\s+{re.escape(rpc_name)}\s*\([^)]*\)\s*returns\s*\([^)]*\)\s*\{{(.*?)\}}",
        proto_text,
        re.DOTALL,
    )
    if match is None:
        raise AssertionError(f"could not find `rpc {rpc_name}` in api.proto")
    return match.group(1)


def _field_declaration_line(message_body: str, field_name: str) -> str:
    """Return the single source line declaring `field_name` inside a proto
    message body (all fields here are declared on one line).
    """
    for line in message_body.splitlines():
        if re.search(rf"\b{re.escape(field_name)}\s*=\s*\d+", line):
            return line
    raise AssertionError(
        f"could not find a field declaration for {field_name!r} in the given message body"
    )


# ==================== Group A: generated files ====================


def test_superseded_device_info_fields_still_declared_in_header() -> None:
    """Each superseded field must still be a real member of DeviceInfoResponse
    in api_pb2.h -- not merely present somewhere in the file. Several of these
    names (e.g. serial_proxies) also exist on DeviceCapabilitiesResponse, so an
    unscoped substring search over the whole header would pass even if the
    field were removed from DeviceInfoResponse.
    """
    class_body = _extract_class_body(HEADER_TEXT, "DeviceInfoResponse")
    for field_name in SUPERSEDED_FIELDS:
        assert re.search(rf"\b{field_name}\b", class_body), (
            f"{field_name} is missing from the DeviceInfoResponse class body in "
            f"api_pb2.h. {DEPRECATED_FIELD_TRAP}"
        )


def test_superseded_device_info_fields_still_encoded_and_sized() -> None:
    """Each superseded field must still be touched by DeviceInfoResponse's
    generated encode() and calculate_size(), i.e. it is still put on the wire.
    """
    encode_body = _extract_function_body(CPP_TEXT, "DeviceInfoResponse::encode")
    size_body = _extract_function_body(CPP_TEXT, "DeviceInfoResponse::calculate_size")
    for field_name in SUPERSEDED_FIELDS:
        assert f"this->{field_name}" in encode_body, (
            f"DeviceInfoResponse::encode() no longer references {field_name}. "
            f"{DEPRECATED_FIELD_TRAP}"
        )
        assert f"this->{field_name}" in size_body, (
            f"DeviceInfoResponse::calculate_size() no longer references "
            f"{field_name}. {DEPRECATED_FIELD_TRAP}"
        )


def test_new_capability_classes_present_in_header() -> None:
    """The new response message and its capability sub-messages must exist as
    generated classes.
    """
    for class_name in (
        "DeviceCapabilitiesResponse",
        "BluetoothProxyCapabilities",
        "VoiceAssistantCapabilities",
        "ZWaveProxyCapabilities",
    ):
        assert re.search(rf"class {re.escape(class_name)}\b", HEADER_TEXT), (
            f"expected a generated class named {class_name} in api_pb2.h"
        )


# ==================== Group B: api.proto source text ====================


def test_all_message_ids_are_unique() -> None:
    """Every `option (id) = N;` in api.proto must be unique. Two messages
    sharing a wire id would make the client and server misinterpret each
    other's messages -- nothing else currently checks this.
    """
    ids = [int(value) for value in re.findall(r"option \(id\) = (\d+);", PROTO_TEXT)]
    assert ids, "did not find any `option (id) = N;` declarations in api.proto"
    duplicates = sorted({value for value in ids if ids.count(value) > 1})
    assert not duplicates, (
        f"Duplicate `option (id)` values found in api.proto: {duplicates}. Each "
        "message must have a unique wire id."
    )


def test_device_capabilities_request_has_id_149() -> None:
    body = _extract_proto_message(PROTO_TEXT, "DeviceCapabilitiesRequest")
    match = re.search(r"option \(id\) = (\d+);", body)
    assert match is not None, "DeviceCapabilitiesRequest is missing `option (id)`"
    assert int(match.group(1)) == 149, (
        f"DeviceCapabilitiesRequest has id {match.group(1)}, expected 149. "
        "Message ids are part of the wire protocol and must not change once "
        "assigned."
    )


def test_device_capabilities_response_has_id_150() -> None:
    body = _extract_proto_message(PROTO_TEXT, "DeviceCapabilitiesResponse")
    match = re.search(r"option \(id\) = (\d+);", body)
    assert match is not None, "DeviceCapabilitiesResponse is missing `option (id)`"
    assert int(match.group(1)) == 150, (
        f"DeviceCapabilitiesResponse has id {match.group(1)}, expected 150. "
        "Message ids are part of the wire protocol and must not change once "
        "assigned."
    )


def test_superseded_fields_are_not_marked_deprecated_in_proto() -> None:
    """The six superseded fields must not carry `[deprecated = true]` in
    api.proto, or the generator drops them and old clients stop receiving
    them (see module docstring). The second half of this test proves the
    deprecated-detection itself works: two genuinely dead fields
    (legacy_bluetooth_proxy_version, legacy_voice_assistant_version) must
    still be detected as deprecated, so the first half isn't vacuously true.
    """
    body = _extract_proto_message(PROTO_TEXT, "DeviceInfoResponse")

    for field_name in SUPERSEDED_FIELDS:
        line = _field_declaration_line(body, field_name)
        assert "deprecated" not in line, (
            f"{field_name} in DeviceInfoResponse is marked deprecated in "
            f"api.proto ({line.strip()!r}). {DEPRECATED_FIELD_TRAP}"
        )

    for field_name in GENUINELY_DEPRECATED_FIELDS:
        line = _field_declaration_line(body, field_name)
        assert "deprecated" in line, (
            f"expected {field_name} to still carry `deprecated=true` in "
            f"api.proto ({line.strip()!r}). If this fails, the deprecated "
            "detection used above is broken, and the sibling assertion that "
            "the superseded fields are NOT deprecated is not testing anything."
        )


def test_superseded_fields_keep_their_wire_numbers() -> None:
    """Each superseded field must stay on the field number recorded in
    SUPERSEDED_FIELDS. Old clients decode DeviceInfoResponse purely by field
    number, so renumbering one of these -- even without touching its name --
    would make an old client read a completely different value out of the
    wire, with nothing else in CI noticing.
    """
    body = _extract_proto_message(PROTO_TEXT, "DeviceInfoResponse")

    for field_name, field_number in SUPERSEDED_FIELDS.items():
        line = _field_declaration_line(body, field_name)
        assert re.search(rf"\b{field_name}\s*=\s*{field_number}\b", line), (
            f"{field_name} in DeviceInfoResponse is no longer declared at "
            f"field number {field_number} ({line.strip()!r}). Field numbers "
            "are the wire protocol -- renumbering this field silently breaks "
            "every existing client that still decodes DeviceInfoResponse by "
            "the old numbering."
        )


def test_capability_message_fields_keep_their_wire_numbers() -> None:
    """Every field on DeviceCapabilitiesResponse and its three capability
    sub-messages must stay on the field number recorded in
    NEW_CAPABILITY_FIELDS. These messages are brand new as of API 1.15, but
    the moment a device ships with them, their field numbers are a frozen
    wire contract -- a client decodes a sub-message field purely by number,
    so a later "cleanup" that renumbers one of these would silently corrupt
    what every already-deployed client reads, with nothing else in CI
    noticing.
    """
    for message_name, fields in NEW_CAPABILITY_FIELDS.items():
        body = _extract_proto_message(PROTO_TEXT, message_name)
        for field_name, field_number in fields.items():
            line = _field_declaration_line(body, field_name)
            assert re.search(rf"\b{field_name}\s*=\s*{field_number}\b", line), (
                f"{field_name} on {message_name} is no longer declared at "
                f"field number {field_number} ({line.strip()!r}). Field "
                "numbers are the wire protocol -- renumbering this field "
                "silently breaks every existing client that decodes this "
                "message by the old numbering."
            )


def test_device_capabilities_rpc_requires_authentication() -> None:
    """The `device_capabilities` RPC must not set
    `option (needs_authentication) = false;` (or set it to anything at all).
    Leaving it unset makes it inherit needs_authentication = true, keeping
    capability data behind authentication (and encryption, when configured).
    """
    body = _extract_rpc_body(PROTO_TEXT, "device_capabilities")
    assert "needs_authentication" not in body, (
        "rpc device_capabilities sets a `needs_authentication` option in "
        "api.proto. It must stay unset so it inherits needs_authentication = "
        "true; otherwise device capability data could be requested over an "
        "unauthenticated connection."
    )


# ==================== Group C: advertised API version ====================


def test_api_version_minor_is_at_least_15() -> None:
    """Clients gate sending DeviceCapabilitiesRequest on seeing
    api_version >= 1.15 in HelloResponse. Regressing api_version_minor below
    15 would make every client believe capabilities are unsupported even
    though the RPC exists, so this must never go backwards. Use >= rather
    than == so the next unrelated minor-version bump doesn't need to touch
    this test.
    """
    match = re.search(r"resp\.api_version_minor\s*=\s*(\d+);", API_CONNECTION_TEXT)
    assert match is not None, (
        "could not find `resp.api_version_minor = N;` in api_connection.cpp"
    )
    minor = int(match.group(1))
    assert minor >= 15, (
        f"api_version_minor is {minor}, but device_capabilities requires "
        "clients to see api_version >= 1.15 in HelloResponse before they will "
        "ever request it."
    )