[ota] Log prepare, upload, and total OTA timing in espota2 (#18582)

esphome/espota2.py

@@ -460,8 +460,14 @@ def perform_ota(
         (upload_size >> 8) & 0xFF,
         (upload_size >> 0) & 0xFF,
     ]
+    # The device erases flash between receiving the size and acking the
+    # prepare, so this window shows the erase cost (near zero when the
+    # device erases lazily during the upload)
+    prepare_start = time.perf_counter()
     send_check(sock, upload_size_encoded, "binary size")
     receive_exactly(sock, 1, "update prepare result", RESPONSE_UPDATE_PREPARE_OK)
+    prepare_duration = time.perf_counter() - prepare_start
+    _LOGGER.info("Preparing for upload took %.2f seconds", prepare_duration)
 
     upload_md5 = hashlib.md5(upload_contents).hexdigest()
     _LOGGER.debug("MD5 of upload is %s", upload_md5)
@@ -528,11 +534,23 @@ def perform_ota(
     # reboots on its own; the exact commit point is not observable from
     # here, so treat everything past the data phase as non-retryable. A
     # re-upload could flash a device that already updated successfully.
+    commit_start = time.perf_counter()
     try:
         receive_exactly(sock, 1, "update receive result", RESPONSE_RECEIVE_OK)
         receive_exactly(sock, 1, "update end result", RESPONSE_UPDATE_END_OK)
     except OTANetworkError as err:
         raise _committed_error(err) from err
+    commit_duration = time.perf_counter() - commit_start
+
+    # Sum of the named windows so the breakdown is self consistent; connect,
+    # handshake, auth, and the one MD5 round trip are not included
+    _LOGGER.info(
+        "Update took %.2f seconds (prepare %.2f, upload %.2f, commit %.2f)",
+        prepare_duration + duration + commit_duration,
+        prepare_duration,
+        duration,
+        commit_duration,
+    )
 
     try:
         send_check(sock, RESPONSE_OK, "end acknowledgement")

tests/unit_tests/test_espota2.py

@@ -6,6 +6,8 @@ from collections.abc import Generator
 import gzip
 import hashlib
 import io
+import itertools
+import logging
 from pathlib import Path
 import socket
 import struct
@@ -53,8 +55,9 @@ def mock_sleep() -> Generator[Mock]:
 @pytest.fixture
 def mock_time(mock_sleep: Mock) -> Generator[None]:
     """Mock time-related functions for consistent testing."""
-    # Provide enough values for multiple calls (tests may call perform_ota multiple times)
-    with patch("time.perf_counter", side_effect=[0, 1, 0, 1, 0, 1]):
+    # Monotonically increasing, never exhausted regardless of how many timing
+    # windows perform_ota measures or how many times a test calls it
+    with patch("time.perf_counter", side_effect=itertools.count()):
         yield
 
 
@@ -372,7 +375,9 @@ def test_perform_ota_successful_md5_auth(
 
 
 @pytest.mark.usefixtures("mock_time")
-def test_perform_ota_no_auth(mock_socket: Mock, mock_file: io.BytesIO) -> None:
+def test_perform_ota_no_auth(
+    mock_socket: Mock, mock_file: io.BytesIO, caplog: pytest.LogCaptureFixture
+) -> None:
     """Test OTA without authentication."""
     recv_responses = [
         bytes([espota2.RESPONSE_OK]),  # First byte of version response
@@ -387,7 +392,14 @@ def test_perform_ota_no_auth(mock_socket: Mock, mock_file: io.BytesIO) -> None:
 
     mock_socket.recv.side_effect = recv_responses
 
-    espota2.perform_ota(mock_socket, None, mock_file, "test.bin")
+    # Distinct window lengths pin each duration to its label; exactly the 6
+    # expected perf_counter calls, so an unaccounted timing window raises
+    timings = [0.0, 2.0, 10.0, 15.0, 20.0, 27.0]
+    with (
+        patch("time.perf_counter", side_effect=timings),
+        caplog.at_level(logging.INFO),
+    ):
+        espota2.perform_ota(mock_socket, None, mock_file, "test.bin")
 
     # Should not send any auth-related data
     auth_calls = [
@@ -397,6 +409,14 @@ def test_perform_ota_no_auth(mock_socket: Mock, mock_file: io.BytesIO) -> None:
     ]
     assert len(auth_calls) == 0
 
+    # The timing summary is the observable output of the upload; exact strings
+    # pin each duration to its label
+    assert "Preparing for upload took 2.00 seconds" in caplog.text
+    assert (
+        "Update took 14.00 seconds (prepare 2.00, upload 5.00, commit 7.00)"
+        in caplog.text
+    )
+
 
 @pytest.mark.usefixtures("mock_time")
 def test_perform_ota_with_compression(mock_socket: Mock) -> None: