|
|
|
@@ -12,11 +12,12 @@ arbiter a full-duty scan would starve wifi, so the 30 ms default is kept.
|
|
|
|
|
from __future__ import annotations
|
|
|
|
|
|
|
|
|
|
from collections.abc import Callable
|
|
|
|
|
from pathlib import Path
|
|
|
|
|
|
|
|
|
|
import pytest
|
|
|
|
|
|
|
|
|
|
from esphome import config_validation as cv
|
|
|
|
|
from esphome.components.ble_device_base import to_ble_units
|
|
|
|
|
from esphome.components.ble_device_base import CONF_CONNECTION_SCAN_WINDOW, to_ble_units
|
|
|
|
|
from esphome.components.const import CONF_SCAN_PARAMETERS, CONF_WINDOW
|
|
|
|
|
from esphome.components.esp32 import KEY_IDF_VERSION
|
|
|
|
|
from esphome.components.esp32_ble_tracker import (
|
|
|
|
@@ -120,3 +121,103 @@ def test_short_interval_without_window_still_rejected(
|
|
|
|
|
stage_esp32("5.5.5", wifi=True)
|
|
|
|
|
with pytest.raises(cv.Invalid, match="needs to be smaller than scan interval"):
|
|
|
|
|
_scan_params({"scan_parameters": {"interval": "20ms"}})
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# The connection-time fallback window: while a GATT connection is active the
|
|
|
|
|
# scanner drops from a raised full-duty window back to this value so the
|
|
|
|
|
# connection gets guaranteed airtime.
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_raise_arms_connection_scan_window_default(
|
|
|
|
|
stage_esp32: Callable[..., None],
|
|
|
|
|
) -> None:
|
|
|
|
|
stage_esp32("5.5.5", wifi=True)
|
|
|
|
|
params = _scan_params({})
|
|
|
|
|
assert params[CONF_WINDOW] == params[CONF_INTERVAL]
|
|
|
|
|
assert to_ble_units(params[CONF_CONNECTION_SCAN_WINDOW]) == 48
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_user_connection_scan_window_survives_raise(
|
|
|
|
|
stage_esp32: Callable[..., None],
|
|
|
|
|
) -> None:
|
|
|
|
|
stage_esp32("5.5.5", wifi=True)
|
|
|
|
|
params = _scan_params({"scan_parameters": {"connection_scan_window": "60ms"}})
|
|
|
|
|
assert params[CONF_WINDOW] == params[CONF_INTERVAL]
|
|
|
|
|
assert to_ble_units(params[CONF_CONNECTION_SCAN_WINDOW]) == 96
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_unraised_window_gets_no_connection_scan_window_default(
|
|
|
|
|
stage_esp32: Callable[..., None],
|
|
|
|
|
) -> None:
|
|
|
|
|
stage_esp32("5.5.4", wifi=True)
|
|
|
|
|
assert CONF_CONNECTION_SCAN_WINDOW not in _scan_params({})
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_connection_scan_window_above_interval_rejected(
|
|
|
|
|
stage_esp32: Callable[..., None],
|
|
|
|
|
) -> None:
|
|
|
|
|
stage_esp32("5.5.5", wifi=True)
|
|
|
|
|
with pytest.raises(
|
|
|
|
|
cv.Invalid, match="connection_scan_window .* needs to be smaller"
|
|
|
|
|
):
|
|
|
|
|
_scan_params({"scan_parameters": {"connection_scan_window": "400ms"}})
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_connection_scan_window_above_window_rejected(
|
|
|
|
|
stage_esp32: Callable[..., None],
|
|
|
|
|
) -> None:
|
|
|
|
|
"""A connection window above the (post-raise) window would widen the scan
|
|
|
|
|
during connections; the reject runs after the raise so a fallback below a
|
|
|
|
|
raised window still validates (covered by the survives-raise test)."""
|
|
|
|
|
stage_esp32("5.5.5", wifi=True)
|
|
|
|
|
with pytest.raises(
|
|
|
|
|
cv.Invalid, match="connection_scan_window .* needs to be smaller"
|
|
|
|
|
):
|
|
|
|
|
_scan_params(
|
|
|
|
|
{"scan_parameters": {"window": "30ms", "connection_scan_window": "300ms"}}
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_connection_scan_window_truncation_collapse_rejected(
|
|
|
|
|
stage_esp32: Callable[..., None],
|
|
|
|
|
) -> None:
|
|
|
|
|
"""A connection window that truncates into the interval's 0.625 ms unit
|
|
|
|
|
would silently program a full-duty scan during connections."""
|
|
|
|
|
stage_esp32("5.5.5", wifi=True)
|
|
|
|
|
with pytest.raises(cv.Invalid, match="connection_scan_window .* both truncate"):
|
|
|
|
|
_scan_params(
|
|
|
|
|
{
|
|
|
|
|
"scan_parameters": {
|
|
|
|
|
"interval": "320.5ms",
|
|
|
|
|
"connection_scan_window": "320.2ms",
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
|
|
|
("config_file", "window_call", "connection_call", "warns"),
|
|
|
|
|
[
|
|
|
|
|
# Raised window with GATT clients: the injected fallback is emitted.
|
|
|
|
|
("scan_window_raised.yaml", "set_scan_window(512)", True, False),
|
|
|
|
|
# Explicit window: nothing injected.
|
|
|
|
|
("scan_window_explicit.yaml", "set_scan_window(48)", False, False),
|
|
|
|
|
# Scan-only build compiles the path out: the injected default is
|
|
|
|
|
# dropped silently, a user-set value warns.
|
|
|
|
|
("scan_window_scan_only.yaml", "set_scan_window(512)", False, False),
|
|
|
|
|
("scan_window_user_set_scan_only.yaml", "set_scan_window(512)", False, True),
|
|
|
|
|
],
|
|
|
|
|
)
|
|
|
|
|
def test_connection_scan_window_codegen(
|
|
|
|
|
generate_main: Callable[[str | Path], str],
|
|
|
|
|
component_config_path: Callable[[str], Path],
|
|
|
|
|
caplog: pytest.LogCaptureFixture,
|
|
|
|
|
config_file: str,
|
|
|
|
|
window_call: str,
|
|
|
|
|
connection_call: bool,
|
|
|
|
|
warns: bool,
|
|
|
|
|
) -> None:
|
|
|
|
|
main_cpp = generate_main(component_config_path(config_file))
|
|
|
|
|
assert window_call in main_cpp
|
|
|
|
|
assert ("set_connection_scan_window(48)" in main_cpp) == connection_call
|
|
|
|
|
assert ("'connection_scan_window' has no effect" in caplog.text) == warns
|
|
|
|
|