Test that a key provisioned over the api survives a restart

tests/integration/fixtures/api_zero_psk_provisioning_persists_across_restart.yaml

@@ -0,0 +1,6 @@
+esphome:
+  name: zero-psk-provision-test
+host:
+api:
+  encryption:
+logger:

tests/integration/test_api_zero_psk_provisioning.py

@@ -10,11 +10,19 @@ from __future__ import annotations
 
 import asyncio
 import base64
+import socket
 
 from aioesphomeapi import InvalidEncryptionKeyAPIError, RequiresEncryptionAPIError
 import pytest
 
-from .types import APIClientConnectedFactory, RunCompiledFunction
+from .conftest import run_binary_and_wait_for_port
+from .const import LOCALHOST
+from .types import (
+    APIClientConnectedFactory,
+    CompileFunction,
+    ConfigWriter,
+    RunCompiledFunction,
+)
 
 # The well-known provisioning PSK: base64 of 32 zero bytes
 ZERO_PSK = base64.b64encode(bytes(32)).decode()
@@ -125,3 +133,40 @@ async def test_api_zero_psk_provisioning_plaintext(
         with pytest.raises(InvalidEncryptionKeyAPIError):
             async with api_client_connected(noise_psk=ZERO_PSK, timeout=5) as client:
                 await client.device_info()
+
+
+@pytest.mark.asyncio
+async def test_api_zero_psk_provisioning_persists_across_restart(
+    yaml_config: str,
+    write_yaml_config: ConfigWriter,
+    compile_esphome: CompileFunction,
+    reserved_tcp_port: tuple[int, socket.socket],
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """A key provisioned over the api is loaded from preferences on the next
+    boot, so the device comes back requiring that key."""
+    port, port_socket = reserved_tcp_port
+    config_path = await write_yaml_config(yaml_config)
+    binary_path = await compile_esphome(config_path)
+    port_socket.close()
+
+    async with (
+        run_binary_and_wait_for_port(binary_path, LOCALHOST, port),
+        api_client_connected(noise_psk=ZERO_PSK) as client,
+    ):
+        # The key is saved and synced before the response is sent
+        assert await client.noise_encryption_set_key(NEW_KEY) is True
+
+    lines: list[str] = []
+    async with run_binary_and_wait_for_port(
+        binary_path, LOCALHOST, port, line_callback=lines.append
+    ):
+        async with api_client_connected(noise_psk=NEW_KEY.decode()) as client:
+            device_info = await client.device_info()
+            assert device_info.api_encryption_provisionable is False
+
+        with pytest.raises(InvalidEncryptionKeyAPIError):
+            async with api_client_connected(noise_psk=ZERO_PSK, timeout=5) as client:
+                await client.device_info()
+
+    assert any("Loaded saved Noise PSK" in line for line in lines)