[tests] Fix flaky log assertion race in oversized payload tests (#14414)

tests/integration/test_oversized_payloads.py

@@ -17,10 +17,10 @@ async def test_oversized_payload_plaintext(
 ) -> None:
     """Test that oversized payloads (>32768 bytes) from client cause disconnection without crashing."""
     process_exited = False
-    helper_log_found = False
+    helper_log_event = asyncio.Event()
 
     def check_logs(line: str) -> None:
-        nonlocal process_exited, helper_log_found
+        nonlocal process_exited
         # Check for signs that the process exited/crashed
         if "Segmentation fault" in line or "core dumped" in line:
             process_exited = True
@@ -30,7 +30,7 @@ async def test_oversized_payload_plaintext(
             and "Bad packet: message size" in line
             and "exceeds maximum" in line
         ):
-            helper_log_found = True
+            helper_log_event.set()
 
     async with run_compiled(yaml_config, line_callback=check_logs):
         async with api_client_connected_with_disconnect() as (client, disconnect_event):
@@ -54,10 +54,13 @@ async def test_oversized_payload_plaintext(
 
         # After disconnection, verify process didn't crash
         assert not process_exited, "ESPHome process should not crash"
-        # Verify we saw the expected HELPER_LOG message
-        assert helper_log_found, (
-            "Expected to see HELPER_LOG about message size exceeding maximum"
-        )
+        # Wait for the expected log message (may arrive after disconnect event)
+        try:
+            await asyncio.wait_for(helper_log_event.wait(), timeout=2.0)
+        except TimeoutError:
+            pytest.fail(
+                "Expected to see HELPER_LOG about message size exceeding maximum"
+            )
 
         # Try to reconnect to verify the process is still running
         async with api_client_connected_with_disconnect() as (client2, _):
@@ -77,10 +80,10 @@ async def test_oversized_protobuf_message_id_plaintext(
     This tests the message type limit - message IDs must fit in a uint16_t (0-65535).
     """
     process_exited = False
-    helper_log_found = False
+    helper_log_event = asyncio.Event()
 
     def check_logs(line: str) -> None:
-        nonlocal process_exited, helper_log_found
+        nonlocal process_exited
         # Check for signs that the process exited/crashed
         if "Segmentation fault" in line or "core dumped" in line:
             process_exited = True
@@ -90,7 +93,7 @@ async def test_oversized_protobuf_message_id_plaintext(
             and "Bad packet: message type" in line
             and "exceeds maximum" in line
         ):
-            helper_log_found = True
+            helper_log_event.set()
 
     async with run_compiled(yaml_config, line_callback=check_logs):
         async with api_client_connected_with_disconnect() as (client, disconnect_event):
@@ -114,10 +117,13 @@ async def test_oversized_protobuf_message_id_plaintext(
 
         # After disconnection, verify process didn't crash
         assert not process_exited, "ESPHome process should not crash"
-        # Verify we saw the expected HELPER_LOG message
-        assert helper_log_found, (
-            "Expected to see HELPER_LOG about message type exceeding maximum"
-        )
+        # Wait for the expected log message (may arrive after disconnect event)
+        try:
+            await asyncio.wait_for(helper_log_event.wait(), timeout=2.0)
+        except TimeoutError:
+            pytest.fail(
+                "Expected to see HELPER_LOG about message type exceeding maximum"
+            )
 
         # Try to reconnect to verify the process is still running
         async with api_client_connected_with_disconnect() as (client2, _):
@@ -135,10 +141,10 @@ async def test_oversized_payload_noise(
     """Test that oversized payloads from client cause disconnection without crashing with noise encryption."""
     noise_key = "N4Yle5YirwZhPiHHsdZLdOA73ndj/84veVaLhTvxCuU="
     process_exited = False
-    helper_log_found = False
+    helper_log_event = asyncio.Event()
 
     def check_logs(line: str) -> None:
-        nonlocal process_exited, helper_log_found
+        nonlocal process_exited
         # Check for signs that the process exited/crashed
         if "Segmentation fault" in line or "core dumped" in line:
             process_exited = True
@@ -149,7 +155,7 @@ async def test_oversized_payload_noise(
             and "Bad packet: message size" in line
             and "exceeds maximum" in line
         ):
-            helper_log_found = True
+            helper_log_event.set()
 
     async with run_compiled(yaml_config, line_callback=check_logs):
         async with api_client_connected_with_disconnect(noise_psk=noise_key) as (
@@ -177,10 +183,13 @@ async def test_oversized_payload_noise(
 
         # After disconnection, verify process didn't crash
         assert not process_exited, "ESPHome process should not crash"
-        # Verify we saw the expected HELPER_LOG message
-        assert helper_log_found, (
-            "Expected to see HELPER_LOG about message size exceeding maximum"
-        )
+        # Wait for the expected log message (may arrive after disconnect event)
+        try:
+            await asyncio.wait_for(helper_log_event.wait(), timeout=2.0)
+        except TimeoutError:
+            pytest.fail(
+                "Expected to see HELPER_LOG about message size exceeding maximum"
+            )
 
         # Try to reconnect to verify the process is still running
         async with api_client_connected_with_disconnect(noise_psk=noise_key) as (
@@ -274,10 +283,10 @@ async def test_noise_corrupt_encrypted_frame(
     """
     noise_key = "N4Yle5YirwZhPiHHsdZLdOA73ndj/84veVaLhTvxCuU="
     process_exited = False
-    cipherstate_failed = False
+    cipherstate_event = asyncio.Event()
 
     def check_logs(line: str) -> None:
-        nonlocal process_exited, cipherstate_failed
+        nonlocal process_exited
         # Check for signs that the process exited/crashed
         if "Segmentation fault" in line or "core dumped" in line:
             process_exited = True
@@ -290,7 +299,7 @@ async def test_noise_corrupt_encrypted_frame(
             "[W][api.connection" in line
             and "Reading failed CIPHERSTATE_DECRYPT_FAILED" in line
         ):
-            cipherstate_failed = True
+            cipherstate_event.set()
 
     async with run_compiled(yaml_config, line_callback=check_logs):
         async with api_client_connected_with_disconnect(noise_psk=noise_key) as (
@@ -326,10 +335,14 @@ async def test_noise_corrupt_encrypted_frame(
         assert not process_exited, (
             "ESPHome process should not crash on corrupt encrypted frames"
         )
-        # Verify we saw the expected log message about decryption failure
-        assert cipherstate_failed, (
-            "Expected to see log about noise_cipherstate_decrypt failure or CIPHERSTATE_DECRYPT_FAILED"
-        )
+        # Wait for the expected log message (may arrive after disconnect event)
+        try:
+            await asyncio.wait_for(cipherstate_event.wait(), timeout=2.0)
+        except TimeoutError:
+            pytest.fail(
+                "Expected to see log about noise_cipherstate_decrypt failure"
+                " or CIPHERSTATE_DECRYPT_FAILED"
+            )
 
         # Verify we can still reconnect after handling the corrupt frame
         async with api_client_connected_with_disconnect(noise_psk=noise_key) as (