Merge branch 'dev' into template_water_heater_onoff

tests/components/mipi_rgb/common.yaml

@@ -0,0 +1,52 @@
+display:
+  - platform: mipi_rgb
+    spi_id: spi_bus
+    model: ZX2D10GE01R-V4848
+    update_interval: 1s
+    color_order: BGR
+    draw_rounding: 2
+    pixel_mode: 18bit
+    invert_colors: false
+    use_axis_flips: true
+    pclk_frequency: 15000000.0
+    pclk_inverted: true
+    byte_order: big_endian
+    hsync_pulse_width: 10
+    hsync_back_porch: 10
+    hsync_front_porch: 10
+    vsync_pulse_width: 2
+    vsync_back_porch: 12
+    vsync_front_porch: 14
+    data_pins:
+      red:
+        - number: 10
+        - number: 16
+        - number: 9
+        - number: 15
+        - number: 46
+      green:
+        - number: 8
+        - number: 13
+        - number: 18
+        - number: 12
+        - number: 11
+        - number: 17
+      blue:
+        - number: 47
+        - number: 1
+        - number: 0
+        - number: 42
+        - number: 14
+    de_pin:
+      number: 39
+    pclk_pin:
+      number: 45
+    hsync_pin:
+      number: 38
+    vsync_pin:
+      number: 48
+    data_rate: 1000000.0
+    spi_mode: MODE0
+    cs_pin:
+      number: 21
+    show_test_card: true

tests/components/mipi_rgb/test.esp32-p4-idf.yaml

@@ -0,0 +1,6 @@
+packages:
+  spi: !include ../../test_build_components/common/spi/esp32-p4-idf.yaml
+
+psram:
+
+<<: !include common.yaml

tests/components/mipi_rgb/test.esp32-s3-idf.yaml

@@ -4,58 +4,4 @@ packages:
 psram:
   mode: octal
 
-display:
-  - platform: mipi_rgb
-    spi_id: spi_bus
-    model: ZX2D10GE01R-V4848
-    update_interval: 1s
-    color_order: BGR
-    draw_rounding: 2
-    pixel_mode: 18bit
-    invert_colors: false
-    use_axis_flips: true
-    pclk_frequency: 15000000.0
-    pclk_inverted: true
-    byte_order: big_endian
-    hsync_pulse_width: 10
-    hsync_back_porch: 10
-    hsync_front_porch: 10
-    vsync_pulse_width: 2
-    vsync_back_porch: 12
-    vsync_front_porch: 14
-    data_pins:
-      red:
-        - number: 10
-        - number: 16
-        - number: 9
-        - number: 15
-        - number: 46
-          ignore_strapping_warning: true
-      green:
-        - number: 8
-        - number: 13
-        - number: 18
-        - number: 12
-        - number: 11
-        - number: 17
-      blue:
-        - number: 47
-        - number: 1
-        - number: 0
-          ignore_strapping_warning: true
-        - number: 42
-        - number: 14
-    de_pin:
-      number: 39
-    pclk_pin:
-      number: 45
-      ignore_strapping_warning: true
-    hsync_pin:
-      number: 38
-    vsync_pin:
-      number: 48
-    data_rate: 1000000.0
-    spi_mode: MODE0
-    cs_pin:
-      number: 21
-    show_test_card: true
+<<: !include common.yaml

tests/components/voice_assistant/common-idf.yaml

@@ -68,3 +68,24 @@ voice_assistant:
     - logger.log:
         format: "Voice assistant error - code %s, message: %s"
         args: [code.c_str(), message.c_str()]
+  on_timer_started:
+    - logger.log:
+        format: "Timer started: %s"
+        args: [timer.id.c_str()]
+  on_timer_updated:
+    - logger.log:
+        format: "Timer updated: %s"
+        args: [timer.id.c_str()]
+  on_timer_cancelled:
+    - logger.log:
+        format: "Timer cancelled: %s"
+        args: [timer.id.c_str()]
+  on_timer_finished:
+    - logger.log:
+        format: "Timer finished: %s"
+        args: [timer.id.c_str()]
+  on_timer_tick:
+    - lambda: |-
+        for (auto &timer : timers) {
+          ESP_LOGD("timer", "Timer %s: %" PRIu32 "s left", timer.name.c_str(), timer.seconds_left);
+        }

tests/components/voice_assistant/common.yaml

@@ -58,3 +58,24 @@ voice_assistant:
     - logger.log:
         format: "Voice assistant error - code %s, message: %s"
         args: [code.c_str(), message.c_str()]
+  on_timer_started:
+    - logger.log:
+        format: "Timer started: %s"
+        args: [timer.id.c_str()]
+  on_timer_updated:
+    - logger.log:
+        format: "Timer updated: %s"
+        args: [timer.id.c_str()]
+  on_timer_cancelled:
+    - logger.log:
+        format: "Timer cancelled: %s"
+        args: [timer.id.c_str()]
+  on_timer_finished:
+    - logger.log:
+        format: "Timer finished: %s"
+        args: [timer.id.c_str()]
+  on_timer_tick:
+    - lambda: |-
+        for (auto &timer : timers) {
+          ESP_LOGD("timer", "Timer %s: %" PRIu32 "s left", timer.name.c_str(), timer.seconds_left);
+        }

tests/dashboard/test_settings.py

@@ -1,4 +1,4 @@
-"""Tests for dashboard settings Path-related functionality."""
+"""Tests for DashboardSettings (path resolution and authentication)."""
 
 from __future__ import annotations
 
@@ -10,6 +10,7 @@ import pytest
 
 from esphome.core import CORE
 from esphome.dashboard.settings import DashboardSettings
+from esphome.dashboard.util.password import password_hash
 
 
 @pytest.fixture
@@ -221,3 +222,66 @@ def test_config_path_parent_resolves_to_config_dir(tmp_path: Path) -> None:
     # Verify that CORE.config_path itself uses the sentinel file
     assert CORE.config_path.name == "___DASHBOARD_SENTINEL___.yaml"
     assert not CORE.config_path.exists()  # Sentinel file doesn't actually exist
+
+
+@pytest.fixture
+def auth_settings(dashboard_settings: DashboardSettings) -> DashboardSettings:
+    """Create DashboardSettings with auth configured, based on dashboard_settings."""
+    dashboard_settings.username = "admin"
+    dashboard_settings.using_password = True
+    dashboard_settings.password_hash = password_hash("correctpassword")
+    return dashboard_settings
+
+
+def test_check_password_correct_credentials(auth_settings: DashboardSettings) -> None:
+    """Test check_password returns True for correct username and password."""
+    assert auth_settings.check_password("admin", "correctpassword") is True
+
+
+def test_check_password_wrong_password(auth_settings: DashboardSettings) -> None:
+    """Test check_password returns False for wrong password."""
+    assert auth_settings.check_password("admin", "wrongpassword") is False
+
+
+def test_check_password_wrong_username(auth_settings: DashboardSettings) -> None:
+    """Test check_password returns False for wrong username."""
+    assert auth_settings.check_password("notadmin", "correctpassword") is False
+
+
+def test_check_password_both_wrong(auth_settings: DashboardSettings) -> None:
+    """Test check_password returns False when both are wrong."""
+    assert auth_settings.check_password("notadmin", "wrongpassword") is False
+
+
+def test_check_password_no_auth(dashboard_settings: DashboardSettings) -> None:
+    """Test check_password returns True when auth is not configured."""
+    assert dashboard_settings.check_password("anyone", "anything") is True
+
+
+def test_check_password_non_ascii_username(
+    dashboard_settings: DashboardSettings,
+) -> None:
+    """Test check_password handles non-ASCII usernames without TypeError."""
+    dashboard_settings.username = "\u00e9l\u00e8ve"
+    dashboard_settings.using_password = True
+    dashboard_settings.password_hash = password_hash("pass")
+    assert dashboard_settings.check_password("\u00e9l\u00e8ve", "pass") is True
+    assert dashboard_settings.check_password("\u00e9l\u00e8ve", "wrong") is False
+    assert dashboard_settings.check_password("other", "pass") is False
+
+
+def test_check_password_ha_addon_no_password(
+    dashboard_settings: DashboardSettings,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Test check_password doesn't crash in HA add-on mode without a password.
+
+    In HA add-on mode, using_ha_addon_auth can be True while using_password
+    is False, leaving password_hash as b"". This must not raise TypeError
+    in hmac.compare_digest.
+    """
+    monkeypatch.delenv("DISABLE_HA_AUTHENTICATION", raising=False)
+    dashboard_settings.on_ha_addon = True
+    dashboard_settings.using_password = False
+    # password_hash stays as default b""
+    assert dashboard_settings.check_password("anyone", "anything") is False

tests/dashboard/test_web_server.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from argparse import Namespace
 import asyncio
+import base64
 from collections.abc import Generator
 from contextlib import asynccontextmanager
 import gzip
@@ -1741,3 +1742,85 @@ def test_proc_on_exit_skips_when_already_closed() -> None:
 
     handler.write_message.assert_not_called()
     handler.close.assert_not_called()
+
+
+def _make_auth_handler(auth_header: str | None = None) -> Mock:
+    """Create a mock handler with the given Authorization header."""
+    handler = Mock()
+    handler.request = Mock()
+    if auth_header is not None:
+        handler.request.headers = {"Authorization": auth_header}
+    else:
+        handler.request.headers = {}
+    handler.get_secure_cookie = Mock(return_value=None)
+    return handler
+
+
+@pytest.fixture
+def mock_auth_settings(mock_dashboard_settings: MagicMock) -> MagicMock:
+    """Fixture to configure mock dashboard settings with auth enabled."""
+    mock_dashboard_settings.using_auth = True
+    mock_dashboard_settings.on_ha_addon = False
+    return mock_dashboard_settings
+
+
+@pytest.mark.usefixtures("mock_auth_settings")
+def test_is_authenticated_malformed_base64() -> None:
+    """Test that invalid base64 in Authorization header returns False."""
+    handler = _make_auth_handler("Basic !!!not-valid-base64!!!")
+    assert web_server.is_authenticated(handler) is False
+
+
+@pytest.mark.usefixtures("mock_auth_settings")
+def test_is_authenticated_bad_base64_padding() -> None:
+    """Test that incorrect base64 padding (binascii.Error) returns False."""
+    handler = _make_auth_handler("Basic abc")
+    assert web_server.is_authenticated(handler) is False
+
+
+@pytest.mark.usefixtures("mock_auth_settings")
+def test_is_authenticated_invalid_utf8() -> None:
+    """Test that base64 decoding to invalid UTF-8 returns False."""
+    # \xff\xfe is invalid UTF-8
+    bad_payload = base64.b64encode(b"\xff\xfe").decode("ascii")
+    handler = _make_auth_handler(f"Basic {bad_payload}")
+    assert web_server.is_authenticated(handler) is False
+
+
+@pytest.mark.usefixtures("mock_auth_settings")
+def test_is_authenticated_no_colon() -> None:
+    """Test that base64 payload without ':' separator returns False."""
+    no_colon = base64.b64encode(b"nocolonhere").decode("ascii")
+    handler = _make_auth_handler(f"Basic {no_colon}")
+    assert web_server.is_authenticated(handler) is False
+
+
+def test_is_authenticated_valid_credentials(
+    mock_auth_settings: MagicMock,
+) -> None:
+    """Test that valid Basic auth credentials are checked."""
+    creds = base64.b64encode(b"admin:secret").decode("ascii")
+    mock_auth_settings.check_password.return_value = True
+    handler = _make_auth_handler(f"Basic {creds}")
+    assert web_server.is_authenticated(handler) is True
+    mock_auth_settings.check_password.assert_called_once_with("admin", "secret")
+
+
+def test_is_authenticated_wrong_credentials(
+    mock_auth_settings: MagicMock,
+) -> None:
+    """Test that valid Basic auth with wrong credentials returns False."""
+    creds = base64.b64encode(b"admin:wrong").decode("ascii")
+    mock_auth_settings.check_password.return_value = False
+    handler = _make_auth_handler(f"Basic {creds}")
+    assert web_server.is_authenticated(handler) is False
+
+
+def test_is_authenticated_no_auth_configured(
+    mock_dashboard_settings: MagicMock,
+) -> None:
+    """Test that requests pass when auth is not configured."""
+    mock_dashboard_settings.using_auth = False
+    mock_dashboard_settings.on_ha_addon = False
+    handler = _make_auth_handler()
+    assert web_server.is_authenticated(handler) is True

tests/integration/fixtures/scheduler_internal_id_no_collision.yaml

@@ -0,0 +1,109 @@
+esphome:
+  name: scheduler-internal-id-test
+  on_boot:
+    priority: -100
+    then:
+      - logger.log: "Starting scheduler internal ID collision tests"
+
+host:
+api:
+logger:
+  level: VERBOSE
+
+globals:
+  - id: tests_done
+    type: bool
+    initial_value: 'false'
+
+script:
+  - id: test_internal_id_no_collision
+    then:
+      - logger.log: "Testing NUMERIC_ID_INTERNAL vs NUMERIC_ID isolation"
+      - lambda: |-
+          // All tests use the same component and the same uint32_t value (0).
+          // NUMERIC_ID_INTERNAL and NUMERIC_ID are separate NameType values,
+          // so the scheduler must treat them as independent timers.
+          auto *comp = id(test_sensor);
+
+          // ---- Test 1: Both timeout types fire independently ----
+          // Set an internal timeout with ID 0
+          App.scheduler.set_timeout(comp, InternalSchedulerID{0}, 50, []() {
+            ESP_LOGI("test", "Internal timeout 0 fired");
+          });
+          // Set a component numeric timeout with the same ID 0
+          App.scheduler.set_timeout(comp, 0U, 50, []() {
+            ESP_LOGI("test", "Numeric timeout 0 fired");
+          });
+
+          // ---- Test 2: Cancelling numeric ID does NOT cancel internal ID ----
+          // Set an internal timeout with ID 1
+          App.scheduler.set_timeout(comp, InternalSchedulerID{1}, 100, []() {
+            ESP_LOGI("test", "Internal timeout 1 survived cancel");
+          });
+          // Set a numeric timeout with the same ID 1
+          App.scheduler.set_timeout(comp, 1U, 100, []() {
+            ESP_LOGE("test", "ERROR: Numeric timeout 1 should have been cancelled");
+          });
+          // Cancel only the numeric one
+          App.scheduler.cancel_timeout(comp, 1U);
+
+          // ---- Test 3: Cancelling internal ID does NOT cancel numeric ID ----
+          // Set a numeric timeout with ID 2
+          App.scheduler.set_timeout(comp, 2U, 150, []() {
+            ESP_LOGI("test", "Numeric timeout 2 survived cancel");
+          });
+          // Set an internal timeout with the same ID 2
+          App.scheduler.set_timeout(comp, InternalSchedulerID{2}, 150, []() {
+            ESP_LOGE("test", "ERROR: Internal timeout 2 should have been cancelled");
+          });
+          // Cancel only the internal one
+          App.scheduler.cancel_timeout(comp, InternalSchedulerID{2});
+
+          // ---- Test 4: Both interval types fire independently ----
+          static int internal_interval_count = 0;
+          static int numeric_interval_count = 0;
+          App.scheduler.set_interval(comp, InternalSchedulerID{3}, 100, []() {
+            internal_interval_count++;
+            if (internal_interval_count == 2) {
+              ESP_LOGI("test", "Internal interval 3 fired twice");
+              App.scheduler.cancel_interval(id(test_sensor), InternalSchedulerID{3});
+            }
+          });
+          App.scheduler.set_interval(comp, 3U, 100, []() {
+            numeric_interval_count++;
+            if (numeric_interval_count == 2) {
+              ESP_LOGI("test", "Numeric interval 3 fired twice");
+              App.scheduler.cancel_interval(id(test_sensor), 3U);
+            }
+          });
+
+          // ---- Test 5: String name does NOT collide with internal ID ----
+          // Use string name and internal ID 10 on same component
+          App.scheduler.set_timeout(comp, "collision_test", 200, []() {
+            ESP_LOGI("test", "String timeout collision_test fired");
+          });
+          App.scheduler.set_timeout(comp, InternalSchedulerID{10}, 200, []() {
+            ESP_LOGI("test", "Internal timeout 10 fired");
+          });
+
+          // Log completion after all timers should have fired
+          App.scheduler.set_timeout(comp, 9999U, 1500, []() {
+            ESP_LOGI("test", "All collision tests complete");
+          });
+
+sensor:
+  - platform: template
+    name: Test Sensor
+    id: test_sensor
+    lambda: return 1.0;
+    update_interval: never
+
+interval:
+  - interval: 0.1s
+    then:
+      - if:
+          condition:
+            lambda: 'return id(tests_done) == false;'
+          then:
+            - lambda: 'id(tests_done) = true;'
+            - script.execute: test_internal_id_no_collision

tests/integration/test_scheduler_internal_id_no_collision.py

@@ -0,0 +1,124 @@
+"""Test that NUMERIC_ID_INTERNAL and NUMERIC_ID cannot collide.
+
+Verifies that InternalSchedulerID (used by core base classes like
+PollingComponent and DelayAction) and uint32_t numeric IDs (used by
+components) are in completely separate matching namespaces, even when
+the underlying uint32_t values are identical and on the same component.
+"""
+
+import asyncio
+import re
+
+import pytest
+
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_scheduler_internal_id_no_collision(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test that internal and numeric IDs with same value don't collide."""
+    # Test 1: Both types fire independently with same ID
+    internal_timeout_0_fired = asyncio.Event()
+    numeric_timeout_0_fired = asyncio.Event()
+
+    # Test 2: Cancelling numeric doesn't cancel internal
+    internal_timeout_1_survived = asyncio.Event()
+    numeric_timeout_1_error = asyncio.Event()
+
+    # Test 3: Cancelling internal doesn't cancel numeric
+    numeric_timeout_2_survived = asyncio.Event()
+    internal_timeout_2_error = asyncio.Event()
+
+    # Test 4: Both interval types fire independently
+    internal_interval_3_done = asyncio.Event()
+    numeric_interval_3_done = asyncio.Event()
+
+    # Test 5: String name doesn't collide with internal ID
+    string_timeout_fired = asyncio.Event()
+    internal_timeout_10_fired = asyncio.Event()
+
+    # Completion
+    all_tests_complete = asyncio.Event()
+
+    def on_log_line(line: str) -> None:
+        clean_line = re.sub(r"\x1b\[[0-9;]*m", "", line)
+
+        if "Internal timeout 0 fired" in clean_line:
+            internal_timeout_0_fired.set()
+        elif "Numeric timeout 0 fired" in clean_line:
+            numeric_timeout_0_fired.set()
+        elif "Internal timeout 1 survived cancel" in clean_line:
+            internal_timeout_1_survived.set()
+        elif "ERROR: Numeric timeout 1 should have been cancelled" in clean_line:
+            numeric_timeout_1_error.set()
+        elif "Numeric timeout 2 survived cancel" in clean_line:
+            numeric_timeout_2_survived.set()
+        elif "ERROR: Internal timeout 2 should have been cancelled" in clean_line:
+            internal_timeout_2_error.set()
+        elif "Internal interval 3 fired twice" in clean_line:
+            internal_interval_3_done.set()
+        elif "Numeric interval 3 fired twice" in clean_line:
+            numeric_interval_3_done.set()
+        elif "String timeout collision_test fired" in clean_line:
+            string_timeout_fired.set()
+        elif "Internal timeout 10 fired" in clean_line:
+            internal_timeout_10_fired.set()
+        elif "All collision tests complete" in clean_line:
+            all_tests_complete.set()
+
+    async with (
+        run_compiled(yaml_config, line_callback=on_log_line),
+        api_client_connected() as client,
+    ):
+        device_info = await client.device_info()
+        assert device_info is not None
+        assert device_info.name == "scheduler-internal-id-test"
+
+        try:
+            await asyncio.wait_for(all_tests_complete.wait(), timeout=5.0)
+        except TimeoutError:
+            pytest.fail("Not all collision tests completed within 5 seconds")
+
+        # Test 1: Both timeout types with same ID 0 must fire
+        assert internal_timeout_0_fired.is_set(), (
+            "Internal timeout with ID 0 should have fired"
+        )
+        assert numeric_timeout_0_fired.is_set(), (
+            "Numeric timeout with ID 0 should have fired"
+        )
+
+        # Test 2: Cancelling numeric ID must NOT cancel internal ID
+        assert internal_timeout_1_survived.is_set(), (
+            "Internal timeout 1 should survive cancellation of numeric timeout 1"
+        )
+        assert not numeric_timeout_1_error.is_set(), (
+            "Numeric timeout 1 should have been cancelled"
+        )
+
+        # Test 3: Cancelling internal ID must NOT cancel numeric ID
+        assert numeric_timeout_2_survived.is_set(), (
+            "Numeric timeout 2 should survive cancellation of internal timeout 2"
+        )
+        assert not internal_timeout_2_error.is_set(), (
+            "Internal timeout 2 should have been cancelled"
+        )
+
+        # Test 4: Both interval types with same ID must fire independently
+        assert internal_interval_3_done.is_set(), (
+            "Internal interval 3 should have fired at least twice"
+        )
+        assert numeric_interval_3_done.is_set(), (
+            "Numeric interval 3 should have fired at least twice"
+        )
+
+        # Test 5: String name and internal ID don't collide
+        assert string_timeout_fired.is_set(), (
+            "String timeout 'collision_test' should have fired"
+        )
+        assert internal_timeout_10_fired.is_set(), (
+            "Internal timeout 10 should have fired alongside string timeout"
+        )

tests/test_build_components/common/spi/esp32-p4-idf.yaml

@@ -0,0 +1,12 @@
+# Common SPI configuration for ESP32-P4 IDF tests
+
+substitutions:
+  clk_pin: GPIO36
+  mosi_pin: GPIO32
+  miso_pin: GPIO33
+
+spi:
+  - id: spi_bus
+    clk_pin: ${clk_pin}
+    mosi_pin: ${mosi_pin}
+    miso_pin: ${miso_pin}

tests/unit_tests/test_espota2.py

@@ -18,8 +18,8 @@ from esphome import espota2
 from esphome.core import EsphomeError
 
 # Test constants
-MOCK_RANDOM_VALUE = 0.123456
-MOCK_RANDOM_BYTES = b"0.123456"
+MOCK_MD5_CNONCE = "a" * 32  # Mock 32-char hex string from secrets.token_hex(16)
+MOCK_SHA256_CNONCE = "b" * 64  # Mock 64-char hex string from secrets.token_hex(32)
 MOCK_MD5_NONCE = b"12345678901234567890123456789012"  # 32 char nonce for MD5
 MOCK_SHA256_NONCE = b"1234567890123456789012345678901234567890123456789012345678901234"  # 64 char nonce for SHA256
 
@@ -55,10 +55,18 @@ def mock_time() -> Generator[None]:
 
 
 @pytest.fixture
-def mock_random() -> Generator[Mock]:
-    """Mock random for predictable test values."""
-    with patch("random.random", return_value=MOCK_RANDOM_VALUE) as mock_rand:
-        yield mock_rand
+def mock_token_hex() -> Generator[Mock]:
+    """Mock secrets.token_hex for predictable test values."""
+
+    def _token_hex(nbytes: int) -> str:
+        if nbytes == 16:
+            return MOCK_MD5_CNONCE
+        if nbytes == 32:
+            return MOCK_SHA256_CNONCE
+        raise ValueError(f"Unexpected nbytes for token_hex mock: {nbytes}")
+
+    with patch("esphome.espota2.secrets.token_hex", side_effect=_token_hex) as mock:
+        yield mock
 
 
 @pytest.fixture
@@ -236,7 +244,7 @@ def test_send_check_socket_error(mock_socket: Mock) -> None:
 
 @pytest.mark.usefixtures("mock_time")
 def test_perform_ota_successful_md5_auth(
-    mock_socket: Mock, mock_file: io.BytesIO, mock_random: Mock
+    mock_socket: Mock, mock_file: io.BytesIO, mock_token_hex: Mock
 ) -> None:
     """Test successful OTA with MD5 authentication."""
     # Setup socket responses for recv calls
@@ -272,8 +280,11 @@ def test_perform_ota_successful_md5_auth(
         )
     )
 
-    # Verify cnonce was sent (MD5 of random.random())
-    cnonce = hashlib.md5(MOCK_RANDOM_BYTES).hexdigest()
+    # Verify token_hex was called with MD5 digest size
+    mock_token_hex.assert_called_once_with(16)
+
+    # Verify cnonce was sent
+    cnonce = MOCK_MD5_CNONCE
     assert mock_socket.sendall.call_args_list[2] == call(cnonce.encode())
 
     # Verify auth result was computed correctly
@@ -366,7 +377,7 @@ def test_perform_ota_auth_without_password(mock_socket: Mock) -> None:
 
 @pytest.mark.usefixtures("mock_time")
 def test_perform_ota_md5_auth_wrong_password(
-    mock_socket: Mock, mock_file: io.BytesIO, mock_random: Mock
+    mock_socket: Mock, mock_file: io.BytesIO, mock_token_hex: Mock
 ) -> None:
     """Test OTA fails when MD5 authentication is rejected due to wrong password."""
     # Setup socket responses for recv calls
@@ -390,7 +401,7 @@ def test_perform_ota_md5_auth_wrong_password(
 
 @pytest.mark.usefixtures("mock_time")
 def test_perform_ota_sha256_auth_wrong_password(
-    mock_socket: Mock, mock_file: io.BytesIO, mock_random: Mock
+    mock_socket: Mock, mock_file: io.BytesIO, mock_token_hex: Mock
 ) -> None:
     """Test OTA fails when SHA256 authentication is rejected due to wrong password."""
     # Setup socket responses for recv calls
@@ -603,7 +614,7 @@ def test_progress_bar(capsys: CaptureFixture[str]) -> None:
 # Tests for SHA256 authentication
 @pytest.mark.usefixtures("mock_time")
 def test_perform_ota_successful_sha256_auth(
-    mock_socket: Mock, mock_file: io.BytesIO, mock_random: Mock
+    mock_socket: Mock, mock_file: io.BytesIO, mock_token_hex: Mock
 ) -> None:
     """Test successful OTA with SHA256 authentication."""
     # Setup socket responses for recv calls
@@ -639,8 +650,11 @@ def test_perform_ota_successful_sha256_auth(
         )
     )
 
-    # Verify cnonce was sent (SHA256 of random.random())
-    cnonce = hashlib.sha256(MOCK_RANDOM_BYTES).hexdigest()
+    # Verify token_hex was called with SHA256 digest size
+    mock_token_hex.assert_called_once_with(32)
+
+    # Verify cnonce was sent
+    cnonce = MOCK_SHA256_CNONCE
     assert mock_socket.sendall.call_args_list[2] == call(cnonce.encode())
 
     # Verify auth result was computed correctly with SHA256
@@ -654,7 +668,7 @@ def test_perform_ota_successful_sha256_auth(
 
 @pytest.mark.usefixtures("mock_time")
 def test_perform_ota_sha256_fallback_to_md5(
-    mock_socket: Mock, mock_file: io.BytesIO, mock_random: Mock
+    mock_socket: Mock, mock_file: io.BytesIO, mock_token_hex: Mock
 ) -> None:
     """Test SHA256-capable client falls back to MD5 for compatibility."""
     # This test verifies the temporary backward compatibility
@@ -692,7 +706,8 @@ def test_perform_ota_sha256_fallback_to_md5(
     )
 
     # But authentication was done with MD5
-    cnonce = hashlib.md5(MOCK_RANDOM_BYTES).hexdigest()
+    mock_token_hex.assert_called_once_with(16)
+    cnonce = MOCK_MD5_CNONCE
     expected_hash = hashlib.md5()
     expected_hash.update(b"testpass")
     expected_hash.update(MOCK_MD5_NONCE)