[ble_device_base] Hub provider registry and shared consumer helpers (#18081)

Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Edvard Filistovič
2026-08-06 12:13:30 -05:00
committed by GitHub
co-authored by J. Nick Koston
parent abc13a7c0a
commit 3221ed2bad
10 changed files with 553 additions and 28 deletions
@@ -44,6 +44,8 @@ DEPENDENCIES = ["bk72xx"]
AUTO_LOAD = ["ble_device_base", "bk72xx_ble"]
CODEOWNERS = ["@Bl00d-B0b"]
ble_device_base.register_hub_provider("bk72xx_ble_tracker")
bk72xx_ble_tracker_ns = cg.esphome_ns.namespace("bk72xx_ble_tracker")
BK72xxBLETracker = bk72xx_ble_tracker_ns.class_(
"BK72xxBLETracker", ble_device_base.BLEHub, cg.Component
+124 -15
View File
@@ -8,21 +8,33 @@ ESPBLEiBeacon / ESPBTDeviceListener, in ble_device.h) and the tracker contract
BLE consumers (sensor components, bluetooth_proxy) bind to whichever tracker the
configuration declares via `cv.use_id(BLEHub)` — ESPHome resolves any declared
subclass, so there is no platform table here and no dependency in either
direction. A sensor appends inject_ble_hub to its CONFIG_SCHEMA (via cv.All) and
calls register_ble_device() in to_code; a tracker component subclasses BLEHub
(C++ and codegen class). Adding a new BLE chip requires only a new tracker
component.
direction. A sensor extends BLE_DEVICE_SCHEMA in its CONFIG_SCHEMA (so an
explicit ble_hub_id: is a declared key even on strict schemas) and calls
register_ble_device() in to_code; a tracker component subclasses BLEHub (C++
and codegen class) and MUST call register_hub_provider() at import time —
without it _require_hub rejects configs that bind through the generated id
(an explicit ble_hub_id: bypasses the registry). Adding a new BLE chip
requires only a new in-tree tracker component; out-of-tree BLE hubs are
not supported.
AES-CCM decryption for encrypted advertisements is provided portably in
ble_aes_ccm.h.
"""
from collections.abc import Callable
import re
import esphome.codegen as cg
from esphome.components.const import CONF_WINDOW
import esphome.config_validation as cv
from esphome.const import CONF_ACTIVE, CONF_CONTINUOUS, CONF_DURATION, CONF_INTERVAL
from esphome.const import (
CONF_ACTIVE,
CONF_CONTINUOUS,
CONF_DURATION,
CONF_INTERVAL,
KEY_TARGET_PLATFORM,
)
from esphome.core import CORE, ID, KEY_CORE
from esphome.types import ConfigType
CODEOWNERS = ["@Bl00d-B0b"]
@@ -44,17 +56,92 @@ BLEHub = ble_device_base_ns.class_("BLEHub")
ESPBTDeviceListener = ble_device_base_ns.class_("ESPBTDeviceListener")
def inject_ble_hub(config: ConfigType) -> ConfigType:
"""Validator: auto-resolve the configured BLE tracker into the config.
# Config keys that provide a BLEHub, registered by each tracker component at
# import time (a tracker's module is imported iff it can end up in the build).
# Used only to phrase an actionable error when a BLE consumer is configured
# without any tracker — the binding itself resolves any BLEHub subclass and
# needs no platform table. Out-of-tree BLE hubs are not supported; the
# registry and the messages below deal in in-tree trackers only.
_HUB_PROVIDERS: set[str] = set()
Append via cv.All to a BLE consumer's CONFIG_SCHEMA. Uses cv.GenerateID +
cv.use_id(BLEHub): an omitted id resolves to the single declared tracker on
any platform; multiple trackers can be disambiguated with an explicit
ble_hub_id.
"""
return cv.Schema(
{cv.GenerateID(CONF_BLE_HUB_ID): cv.use_id(BLEHub)}, extra=cv.ALLOW_EXTRA
)(config)
# The in-tree trackers per target platform, so the missing-tracker error names
# them even in a fresh process where no tracker module has been imported yet (a
# consumer imports only ble_device_base, so the registry is empty exactly in
# the most common failure: the tracker was simply forgotten). Filtered by the
# current platform so an esp32 config is not told to add a Beken tracker; an
# unknown/absent platform falls back to every in-tree name.
_IN_TREE_HUB_PROVIDERS: dict[str, str] = {
"esp32": "esp32_ble_tracker",
"bk72xx": "bk72xx_ble_tracker",
"rp2": "rp2_ble_tracker",
"ln882x": "ln882h_ble_tracker",
}
def register_hub_provider(component: str) -> None:
"""Called at import time by every component whose config key declares a BLEHub."""
_HUB_PROVIDERS.add(component)
def _require_hub(value: ID) -> ID:
# Without this check a missing tracker surfaces at ID resolution as
# "Couldn't find any component that can be used for 'ble_device_base::BLEHub'"
# — a C++ class name the user never types. Component final validation cannot
# phrase it better: the ID pass runs first and its error skips all later
# steps. All explicitly configured components are loaded before any schema
# validates, so a registered provider in loaded_integrations is exact here.
if value.id is not None:
# Explicit ble_hub_id: — the user is pointing at a specific hub (the
# multi-hub disambiguation case). Let the ID pass judge it; its error
# names the missing id, which is accurate.
return value
if not _HUB_PROVIDERS & CORE.loaded_integrations:
# Defensive lookup rather than CORE.target_platform: the property
# raises when no platform is registered, and this message must never
# be the thing that crashes. In a real run the platform is always set
# (LoadTargetPlatformValidationStep runs before any other domain), so
# the unfiltered all-platforms fallback is reachable only from tests.
platform = CORE.data.get(KEY_CORE, {}).get(KEY_TARGET_PLATFORM)
if platform is not None and platform not in _IN_TREE_HUB_PROVIDERS:
# Known platform with no in-tree hub (esp8266, host, rtl87xx, …):
# listing the other platforms' trackers would misdirect, and
# out-of-tree BLE hubs are not supported.
raise cv.Invalid(
f"No BLE tracker exists for {platform}; BLE components are "
"not supported on this platform"
)
in_tree = (
{tracker}
if (tracker := _IN_TREE_HUB_PROVIDERS.get(platform))
else set(_IN_TREE_HUB_PROVIDERS.values())
)
# in_tree only: _HUB_PROVIDERS is import-time state that outlives
# CORE.reset() in a long-lived process (dashboard), so a tracker from
# an earlier build of another platform must not leak into the message.
# The gate above is immune — loaded_integrations resets per run.
names = ", ".join(sorted(in_tree))
raise cv.Invalid(f"No BLE tracker configured — add one of: {names}")
return value
# Schema fragment binding a consumer to the configured BLE tracker: extend a
# consumer's CONFIG_SCHEMA with this so ble_hub_id: is a declared key — a
# trailing validator after a PREVENT_EXTRA schema would reject the explicit
# form before ever running. An omitted id resolves to the single declared
# tracker on any platform; multiple trackers are disambiguated with an
# explicit ble_hub_id.
BLE_DEVICE_SCHEMA = cv.Schema(
{cv.GenerateID(CONF_BLE_HUB_ID): cv.All(cv.use_id(BLEHub), _require_hub)}
)
def rename_legacy_hub_id(component: str) -> Callable[[ConfigType], ConfigType]:
"""Transitional alias for the pre-migration binding key: esp32_ble_id ->
ble_hub_id. Warns and auto-migrates until removal; every migrated platform
prepends this to its CONFIG_SCHEMA so existing configs keep validating."""
return cv.rename_key(
"esp32_ble_id", CONF_BLE_HUB_ID, removed_in="2027.2.0", component=component
)
def request_irk_support() -> None:
@@ -217,3 +304,25 @@ def as_hex_array(value: str) -> cg.RawExpression:
def as_reversed_hex_array(value: str) -> cg.RawExpression:
return _hex_array_expression(value, reverse=True)
def add_service_uuid(var: cg.MockObj, service_uuid: str) -> None:
"""Emit the width-matched service-UUID setter for a consumer.
16-/32-bit UUIDs go out as plain hex literals, 128-bit as a reversed byte
array (BLE wire order). Shared here so every sensor platform dispatches the
same way instead of carrying its own if/elif copy.
"""
if len(service_uuid) == len(BT_UUID16_FORMAT):
cg.add(var.set_service_uuid16(as_hex(service_uuid)))
elif len(service_uuid) == len(BT_UUID32_FORMAT):
cg.add(var.set_service_uuid32(as_hex(service_uuid)))
elif len(service_uuid) == len(BT_UUID128_FORMAT):
cg.add(var.set_service_uuid128(as_reversed_hex_array(service_uuid)))
else:
# bt_uuid restricts lengths to exactly these three formats; if that
# ever loosens, fail the build instead of emitting no setter (a
# sensor whose match_by_ is unset silently never matches). ValueError,
# not cv.Invalid: this runs from to_code, after validation, where
# voluptuous errors surface as raw tracebacks.
raise ValueError(f"Unsupported UUID format: {service_uuid}")
@@ -8,6 +8,7 @@
#include "ble_aes_ccm.h"
#include "esphome/core/defines.h"
#include "esphome/core/hal.h"
#include "esphome/core/helpers.h"
#include "esphome/core/log.h"
@@ -265,16 +266,21 @@ bool ESPBTUUID::operator==(const ESPBTUUID &other) const {
ESPBLEiBeacon::ESPBLEiBeacon(const uint8_t *data) { memcpy(&this->beacon_data_, data, sizeof(this->beacon_data_)); }
optional<ESPBLEiBeacon> ESPBLEiBeacon::from_manufacturer_data(const ServiceData &data) {
optional<ESPBLEiBeacon> ESPBLEiBeacon::from_manufacturer_data(const ServiceData &data, bool *prefix_rejected) {
// iBeacon manufacturer specific data (after company-ID bytes have been stripped):
// [0x02][0x15][16-byte UUID][2-byte major][2-byte minor][1-byte power] = exactly 23 bytes
// Parity with esp32_ble_tracker: gate on the Apple company ID and length only.
// (Checking the 0x02/0x15 sub-type prefix would be stricter, but is a behavior
// change; it belongs to a follow-up, not this refactor.)
if (!data.uuid.contains(0x4C, 0x00)) // Apple company ID 0x004C
return {};
if (data.data.size() != 23)
return {};
// Require the iBeacon sub-type/length prefix — stricter than the legacy
// esp32 parser, which accepted any 23-byte Apple payload and surfaced
// non-iBeacon frames as garbage beacons.
if (data.data[0] != 0x02 || data.data[1] != 0x15) {
if (prefix_rejected != nullptr)
*prefix_rejected = true;
return {};
}
return ESPBLEiBeacon(data.data.data());
}
@@ -282,6 +288,44 @@ optional<ESPBLEiBeacon> ESPBLEiBeacon::from_manufacturer_data(const ServiceData
// ESPBTDevice
// ---------------------------------------------------------------------------
optional<ESPBLEiBeacon> ESPBTDevice::get_ibeacon() const {
bool prefix_rejected = false;
uint8_t rejected_sub_type = 0;
uint8_t rejected_len = 0;
for (const auto &it : this->manufacturer_datas_) {
bool rejected = false;
auto res = ESPBLEiBeacon::from_manufacturer_data(it, &rejected);
if (res.has_value())
return res;
if (rejected && !prefix_rejected) {
prefix_rejected = true;
rejected_sub_type = it.data[0];
rejected_len = it.data[1];
}
}
if (prefix_rejected) {
// Only when no beacon was found at all: these frames were accepted before
// the prefix check, so their disappearance must be observable at the
// default log level. Throttled so a chatty non-iBeacon Apple advertiser
// cannot flood the log; a different address may bypass the shared window
// so that advertiser cannot mask the device that actually regressed — but
// with a 1 s floor, or two alternating advertisers log every frame.
static uint32_t last_log = 0;
static uint64_t last_addr = 0;
const uint32_t now = millis();
const uint64_t addr = this->address_uint64();
const uint32_t since = now - last_log;
if (last_log == 0 || since > 60000 || (addr != last_addr && since > 1000)) {
last_log = now;
last_addr = addr;
char addr_buf[MAC_ADDRESS_PRETTY_BUFFER_SIZE];
ESP_LOGD(TAG, "%s: 23-byte Apple frame without iBeacon prefix ignored (sub-type 0x%02X len 0x%02X)",
this->address_str_to(addr_buf), rejected_sub_type, rejected_len);
}
}
return {};
}
const char *ESPBTDevice::address_type_str() const {
switch (this->address_type_) {
case BLE_ADDR_TYPE_PUBLIC:
@@ -129,7 +129,12 @@ class ESPBLEiBeacon {
public:
ESPBLEiBeacon() { memset(&this->beacon_data_, 0, sizeof(this->beacon_data_)); }
explicit ESPBLEiBeacon(const uint8_t *data);
static optional<ESPBLEiBeacon> from_manufacturer_data(const ServiceData &data);
/// prefix_rejected: caller must initialise to false; set to true ONLY when a
/// 23-byte Apple frame was refused for lacking the 0x02/0x15 iBeacon prefix —
/// the case the legacy esp32 parser accepted. Never written on accept or on
/// the non-Apple/wrong-size rejects. The caller with the device address does
/// the logging (see ESPBTDevice::get_ibeacon()).
static optional<ESPBLEiBeacon> from_manufacturer_data(const ServiceData &data, bool *prefix_rejected = nullptr);
uint16_t get_major() const { return byteswap(this->beacon_data_.major); }
uint16_t get_minor() const { return byteswap(this->beacon_data_.minor); }
@@ -193,6 +198,8 @@ class ESPBTDevice {
// Historical esp32 signature: consumers assign the result to esp_ble_addr_type_t.
esp_ble_addr_type_t get_address_type() const { return static_cast<esp_ble_addr_type_t>(this->address_type_); }
/// Historical esp32 ingest (esp32 builds only): parse an ESP-IDF scan result.
/// Prefer ESPBTDevice::from_scan_result(); deprecation is a follow-up pending
/// consumer feedback on the raw scan-result fields.
void parse_scan_rst(const esp32_ble::BLEScanResult &scan_result);
// Exposed through a function for use in lambdas
const esp32_ble::BLEScanResult &get_scan_result() const { return *scan_result_; }
@@ -218,14 +225,7 @@ class ESPBTDevice {
/// decryptor; compiled only when a sensor configures irk: (request_irk_support).
bool resolve_irk(const uint8_t *irk) const;
optional<ESPBLEiBeacon> get_ibeacon() const {
for (const auto &it : this->manufacturer_datas_) {
auto res = ESPBLEiBeacon::from_manufacturer_data(it);
if (res.has_value())
return res;
}
return {};
}
optional<ESPBLEiBeacon> get_ibeacon() const;
protected:
void parse_adv_(const uint8_t *payload, uint16_t len);
@@ -43,6 +43,8 @@ AUTO_LOAD = ["ble_device_base", "esp32_ble"]
DEPENDENCIES = ["esp32"]
CODEOWNERS = ["@bdraco"]
ble_device_base.register_hub_provider("esp32_ble_tracker")
CONF_ESP32_BLE_ID = "esp32_ble_id"
CONF_SOFTWARE_COEXISTENCE = "software_coexistence"
@@ -501,6 +501,8 @@ void ESP32BLETracker::process_scan_result_(const BLEScanResult &scan_result) {
if (this->parse_advertisements_) {
#ifdef USE_ESP32_BLE_DEVICE
ESPBTDevice device;
// The historical ingest keeps the raw scan-result fields populated for
// external components.
device.parse_scan_rst(scan_result);
bool found = false;
@@ -29,6 +29,8 @@ DEPENDENCIES = ["ln882x"]
AUTO_LOAD = ["ble_device_base", "ln882h_ble"]
CODEOWNERS = ["@Bl00d-B0b"]
ble_device_base.register_hub_provider("ln882h_ble_tracker")
ln882h_ble_tracker_ns = cg.esphome_ns.namespace("ln882h_ble_tracker")
LN882HBLETracker = ln882h_ble_tracker_ns.class_(
"LN882HBLETracker", ble_device_base.BLEHub, cg.Component
@@ -28,6 +28,8 @@ DEPENDENCIES = ["rp2"]
AUTO_LOAD = ["ble_device_base", "rp2040_ble"]
CODEOWNERS = ["@bdraco"]
ble_device_base.register_hub_provider("rp2_ble_tracker")
rp2_ble_tracker_ns = cg.esphome_ns.namespace("rp2_ble_tracker")
RP2BLETracker = rp2_ble_tracker_ns.class_(
"RP2BLETracker", ble_device_base.BLEHub, cg.Component
@@ -0,0 +1,204 @@
"""Tests for the BLE hub provider registry and the missing-hub diagnostics."""
from collections.abc import Generator
from importlib import import_module
from pathlib import Path
import pytest
import esphome.codegen as cg
from esphome.components import ble_device_base
import esphome.config_validation as cv
from esphome.const import KEY_TARGET_PLATFORM, Platform
from esphome.core import CORE, ID, KEY_CORE
from esphome.cpp_generator import MockObjClass
COMPONENTS_DIR = Path(ble_device_base.__file__).parent.parent
@pytest.fixture
def hub_registry() -> Generator[set[str]]:
"""Save/restore _HUB_PROVIDERS — a module global with no reset hook.
CORE state needs no bookkeeping here: conftest's autouse reset_core
fixture reassigns it after every test.
"""
saved = set(ble_device_base._HUB_PROVIDERS)
yield ble_device_base._HUB_PROVIDERS
ble_device_base._HUB_PROVIDERS.clear()
ble_device_base._HUB_PROVIDERS.update(saved)
def _generated_id() -> ID:
"""An ID as cv.GenerateID leaves it before the ID-assignment pass."""
return ID(None, is_declaration=False, type="ble_device_base::BLEHub")
def _set_platform(platform: str | None) -> None:
core_data = CORE.data.setdefault(KEY_CORE, {})
if platform is None:
core_data.pop(KEY_TARGET_PLATFORM, None)
else:
core_data[KEY_TARGET_PLATFORM] = platform
# The missing-hub diagnostics: one test per path so a regression in one
# scenario cannot mask the others. The hub binding must fail with a
# tracker-naming message, not use_id's C++-class error, regardless of
# config-step ordering internals.
def test_empty_registry_names_every_in_tree_tracker(hub_registry: set[str]) -> None:
# The common failure: a fresh CLI process where the tracker was simply
# forgotten, so no tracker module was ever imported and the registry is
# empty. The error must still name the in-tree trackers.
hub_registry.clear()
CORE.loaded_integrations.clear()
_set_platform(None)
with pytest.raises(
cv.Invalid,
match="add one of: bk72xx_ble_tracker, esp32_ble_tracker, ln882h_ble_tracker, rp2_ble_tracker",
):
ble_device_base._require_hub(_generated_id())
def test_platform_filters_the_suggested_trackers(hub_registry: set[str]) -> None:
hub_registry.clear()
CORE.loaded_integrations.clear()
_set_platform("esp32")
with pytest.raises(cv.Invalid, match="add one of: esp32_ble_tracker$"):
ble_device_base._require_hub(_generated_id())
def test_ble_less_platform_is_not_misdirected(hub_registry: set[str]) -> None:
# A known platform with no in-tree hub must not be pointed at other
# platforms' trackers; out-of-tree BLE hubs are not supported.
hub_registry.clear()
CORE.loaded_integrations.clear()
_set_platform("esp8266")
with pytest.raises(
cv.Invalid,
match="No BLE tracker exists for esp8266; BLE components are not supported",
):
ble_device_base._require_hub(_generated_id())
def test_explicit_id_bypasses_the_registry(hub_registry: set[str]) -> None:
# Explicit ble_hub_id: is the multi-hub disambiguation case; the ID pass
# owns that diagnosis and its error names the missing id.
hub_registry.clear()
CORE.loaded_integrations.clear()
explicit = ID("my_hub", is_declaration=False, type="ble_device_base::BLEHub")
assert ble_device_base._require_hub(explicit) is explicit
def test_registered_and_loaded_provider_passes(hub_registry: set[str]) -> None:
hub_registry.add("esp32_ble_tracker")
CORE.loaded_integrations.add("esp32_ble_tracker")
generated = _generated_id()
assert ble_device_base._require_hub(generated) is generated
def _module_name(path: Path) -> str:
"""Dotted module name for a file under esphome/components."""
rel = path.relative_to(COMPONENTS_DIR.parent)
parts = rel.with_suffix("").parts
if parts[-1] == "__init__":
parts = parts[:-1]
return "esphome." + ".".join(parts)
def _hub_component_modules() -> list[str]:
"""Components whose codegen class inherits ble_device_base.BLEHub.
The source-text pass only selects import candidates (importing all ~900
component packages is too slow); membership is decided by the class
hierarchy via MockObjClass.inherits_from on every module whose source
matched — nested declaring modules included — so a comment mentioning
BLEHub in a consumer cannot produce a false positive.
"""
hub_modules = []
for pkg in sorted(COMPONENTS_DIR.iterdir()):
if pkg.name == "ble_device_base" or not (pkg / "__init__.py").is_file():
continue
matched = [
path
for path in pkg.rglob("*.py")
if "BLEHub" in path.read_text(encoding="utf-8")
]
if not matched:
continue
for path in matched:
mod = import_module(_module_name(path))
if any(
isinstance(attr, MockObjClass)
and attr is not ble_device_base.BLEHub
and attr.inherits_from(ble_device_base.BLEHub)
for attr in vars(mod).values()
):
hub_modules.append(pkg.name)
break
return hub_modules
def test_every_in_tree_hub_registers_as_provider() -> None:
"""A BLEHub subclass that forgets register_hub_provider() makes _require_hub
reject valid configs for that platform — fail CI instead of the user."""
hub_modules = _hub_component_modules()
assert hub_modules, "hub discovery found no BLEHub subclasses — scan stale?"
for name in hub_modules:
assert name in ble_device_base._HUB_PROVIDERS, (
f"{name} subclasses ble_device_base.BLEHub but never calls "
"register_hub_provider(); a valid config using it would be rejected"
)
# The per-platform error table must know every in-tree hub, keyed by real
# platform names — a typo'd key would silently route that platform into
# the no-in-tree-tracker branch.
assert set(ble_device_base._IN_TREE_HUB_PROVIDERS.values()) == set(hub_modules)
platforms = {platform.value for platform in Platform}
assert set(ble_device_base._IN_TREE_HUB_PROVIDERS) <= platforms
def test_ble_device_schema_declares_the_binding_key(hub_registry: set[str]) -> None:
"""Extending BLE_DEVICE_SCHEMA keeps ble_hub_id a declared key on a strict
schema, for both the generated and the explicit form, and the missing-hub
rejection surfaces through the schema itself."""
schema = cv.Schema({}).extend(ble_device_base.BLE_DEVICE_SCHEMA)
hub_registry.clear()
CORE.loaded_integrations.discard("esp32_ble_tracker")
with pytest.raises(cv.Invalid, match="No BLE tracker configured"):
schema({})
hub_registry.add("esp32_ble_tracker")
CORE.loaded_integrations.add("esp32_ble_tracker")
generated = schema({})[ble_device_base.CONF_BLE_HUB_ID]
assert isinstance(generated, ID) and generated.id is None
explicit = schema({"ble_hub_id": "my_hub"})[ble_device_base.CONF_BLE_HUB_ID]
assert explicit.id == "my_hub"
def test_rename_legacy_hub_id_migrates_the_old_key() -> None:
validator = ble_device_base.rename_legacy_hub_id("my_sensor")
migrated = validator({"esp32_ble_id": "tracker1"})
assert migrated == {ble_device_base.CONF_BLE_HUB_ID: "tracker1"}
untouched = validator({"name": "x"})
assert untouched == {"name": "x"}
def test_add_service_uuid_dispatches_by_width(monkeypatch: pytest.MonkeyPatch) -> None:
emitted: list[str] = []
monkeypatch.setattr(
"esphome.components.ble_device_base.cg.add", lambda e: emitted.append(str(e))
)
var = cg.MockObj("trig")
ble_device_base.add_service_uuid(var, "11AA")
ble_device_base.add_service_uuid(var, "11223344")
ble_device_base.add_service_uuid(var, "11223344-5566-7788-99aa-bbccddeeff00")
assert "set_service_uuid16" in emitted[0]
assert "set_service_uuid32" in emitted[1]
assert "set_service_uuid128" in emitted[2]
# BLE wire order: the 128-bit array must be byte-reversed — as_hex_array
# in its place would still emit the right setter name and silently never
# match on-air.
assert "0x00,0xff,0xee,0xdd" in emitted[2]
with pytest.raises(ValueError, match="Unsupported UUID format"):
ble_device_base.add_service_uuid(var, "123")
@@ -0,0 +1,158 @@
#include <gtest/gtest.h>
#include <cstdint>
#include <vector>
#include "esphome/components/ble_device_base/ble_device.h"
namespace esphome::ble_device_base::testing {
// from_manufacturer_data() accepts exactly the iBeacon frame: Apple company ID,
// 23 payload bytes, and the 0x02/0x15 sub-type/length prefix. The prefix check
// is stricter than the legacy esp32 parser (which surfaced any 23-byte Apple
// payload as a beacon) — a declared behavior change; these tests pin the
// accept/reject boundary.
namespace {
ServiceData make_apple_payload(uint8_t sub_type, uint8_t length, size_t size = 23) {
ServiceData data;
data.uuid = ESPBTUUID::from_uint16(0x004C); // Apple company ID
data.data.assign(size, 0);
if (size >= 2) {
data.data[0] = sub_type;
data.data[1] = length;
}
// BeaconData layout: sub_type[0], length[1], proximity_uuid[2..17],
// major[18..19], minor[20..21], signal_power[22] — all wire values big-endian.
if (size >= 23) {
data.data[18] = 0x12; // major 0x1234
data.data[19] = 0x34;
data.data[20] = 0x56; // minor 0x5678
data.data[21] = 0x78;
data.data[22] = 0xC5; // signal power -59 dBm
}
return data;
}
} // namespace
TEST(BleIBeacon, AcceptsWellFormedFrame) {
auto beacon = ESPBLEiBeacon::from_manufacturer_data(make_apple_payload(0x02, 0x15));
ASSERT_TRUE(beacon.has_value());
// Explicit guard: clang-tidy's unchecked-optional-access models neither
// gtest's ASSERT_TRUE nor value() as a check.
if (beacon.has_value()) {
// Pins every scalar accessor's offset and the on-wire big-endian order.
EXPECT_EQ(beacon->get_major(), 0x1234);
EXPECT_EQ(beacon->get_minor(), 0x5678);
EXPECT_EQ(beacon->get_signal_power(), -59);
}
}
TEST(BleIBeacon, RejectsWrongSubType) {
// Apple "nearby" and other frames of coincidental length must not parse.
EXPECT_FALSE(ESPBLEiBeacon::from_manufacturer_data(make_apple_payload(0x10, 0x15)).has_value());
}
TEST(BleIBeacon, RejectsWrongLengthByte) {
EXPECT_FALSE(ESPBLEiBeacon::from_manufacturer_data(make_apple_payload(0x02, 0x14)).has_value());
}
TEST(BleIBeacon, RejectsWrongPayloadSize) {
EXPECT_FALSE(ESPBLEiBeacon::from_manufacturer_data(make_apple_payload(0x02, 0x15, 22)).has_value());
EXPECT_FALSE(ESPBLEiBeacon::from_manufacturer_data(make_apple_payload(0x02, 0x15, 24)).has_value());
}
TEST(BleIBeacon, RejectsNonAppleCompany) {
auto data = make_apple_payload(0x02, 0x15);
data.uuid = ESPBTUUID::from_uint16(0x0059); // Nordic
EXPECT_FALSE(ESPBLEiBeacon::from_manufacturer_data(data).has_value());
}
TEST(BleIBeacon, PrefixRejectedFlagsOnlyTheSubTypeCase) {
// The out-param drives the get_ibeacon() diagnostic for frames the legacy
// parser accepted: exactly the 23-byte Apple payload with a wrong prefix.
// Wrong size and non-Apple frames were never accepted and must stay silent.
bool flagged = false;
EXPECT_FALSE(ESPBLEiBeacon::from_manufacturer_data(make_apple_payload(0x10, 0x15), &flagged).has_value());
EXPECT_TRUE(flagged);
flagged = false;
ESPBLEiBeacon::from_manufacturer_data(make_apple_payload(0x02, 0x15), &flagged);
EXPECT_FALSE(flagged);
flagged = false;
ESPBLEiBeacon::from_manufacturer_data(make_apple_payload(0x10, 0x15, 22), &flagged);
EXPECT_FALSE(flagged);
flagged = false;
auto nordic = make_apple_payload(0x10, 0x15);
nordic.uuid = ESPBTUUID::from_uint16(0x0059);
ESPBLEiBeacon::from_manufacturer_data(nordic, &flagged);
EXPECT_FALSE(flagged);
}
namespace {
// One AD manufacturer-data record: [len][0xFF][company LE][payload...].
void append_mfr_record(std::vector<uint8_t> &adv, uint16_t company, const std::vector<uint8_t> &payload) {
adv.push_back(static_cast<uint8_t>(1 + 2 + payload.size()));
adv.push_back(0xFF);
adv.push_back(static_cast<uint8_t>(company & 0xFF));
adv.push_back(static_cast<uint8_t>(company >> 8));
adv.insert(adv.end(), payload.begin(), payload.end());
}
std::vector<uint8_t> beacon_payload(uint8_t sub_type, uint8_t length) {
std::vector<uint8_t> p(23, 0);
p[0] = sub_type;
p[1] = length;
p[18] = 0x12;
p[19] = 0x34;
p[20] = 0x56;
p[21] = 0x78;
p[22] = 0xC5;
return p;
}
ESPBTDevice device_from(const std::vector<uint8_t> &adv) {
const uint8_t mac[6] = {0x11, 0x22, 0x33, 0x44, 0x55, 0x66};
ESPBTDevice device;
device.from_scan_result(mac, -59, 0, adv.data(), static_cast<uint16_t>(adv.size()));
return device;
}
} // namespace
// get_ibeacon() wraps the parser with first-rejection capture and the log
// gate; pin its short circuits so a regression there needs a code change, not
// a review, to surface.
TEST(BleIBeacon, GetIbeaconReturnsBeaconDespitePrecedingRejectedFrame) {
std::vector<uint8_t> adv;
append_mfr_record(adv, 0x004C, beacon_payload(0x10, 0x15)); // rejected prefix
append_mfr_record(adv, 0x004C, beacon_payload(0x02, 0x15)); // real iBeacon
auto device = device_from(adv);
auto beacon = device.get_ibeacon();
ASSERT_TRUE(beacon.has_value());
if (beacon.has_value()) {
EXPECT_EQ(beacon->get_major(), 0x1234);
}
}
TEST(BleIBeacon, GetIbeaconEmptyWhenOnlyRejectedFrames) {
std::vector<uint8_t> adv;
append_mfr_record(adv, 0x004C, beacon_payload(0x10, 0x15));
auto device = device_from(adv);
EXPECT_FALSE(device.get_ibeacon().has_value());
}
TEST(BleIBeacon, GetIbeaconEmptyWithoutManufacturerData) {
std::vector<uint8_t> adv;
adv.push_back(0x02); // flags record only
adv.push_back(0x01);
adv.push_back(0x06);
auto device = device_from(adv);
EXPECT_FALSE(device.get_ibeacon().has_value());
}
} // namespace esphome::ble_device_base::testing