Prevent errors when Reolink privacy mode is turned on (#136506)

tests/components/reolink/test_init.py

@@ -1,13 +1,18 @@
 """Test the Reolink init."""
 
 import asyncio
+from collections.abc import Callable
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock, Mock, patch
 
 from freezegun.api import FrozenDateTimeFactory
 import pytest
 from reolink_aio.api import Chime
-from reolink_aio.exceptions import CredentialsInvalidError, ReolinkError
+from reolink_aio.exceptions import (
+    CredentialsInvalidError,
+    LoginPrivacyModeError,
+    ReolinkError,
+)
 
 from homeassistant.components.reolink import (
     DEVICE_UPDATE_INTERVAL,
@@ -16,7 +21,13 @@ from homeassistant.components.reolink import (
 )
 from homeassistant.components.reolink.const import DOMAIN
 from homeassistant.config_entries import ConfigEntryState
-from homeassistant.const import CONF_PORT, STATE_OFF, STATE_UNAVAILABLE, Platform
+from homeassistant.const import (
+    CONF_PORT,
+    STATE_OFF,
+    STATE_ON,
+    STATE_UNAVAILABLE,
+    Platform,
+)
 from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
 from homeassistant.core_config import async_process_ha_core_config
 from homeassistant.helpers import (
@@ -749,3 +760,102 @@ async def test_port_changed(
     await hass.async_block_till_done()
 
     assert config_entry.data[CONF_PORT] == 4567
+
+
+async def test_privacy_mode_on(
+    hass: HomeAssistant,
+    freezer: FrozenDateTimeFactory,
+    reolink_connect: MagicMock,
+    config_entry: MockConfigEntry,
+) -> None:
+    """Test successful setup even when privacy mode is turned on."""
+    reolink_connect.baichuan.privacy_mode.return_value = True
+    reolink_connect.get_states = AsyncMock(
+        side_effect=LoginPrivacyModeError("Test error")
+    )
+
+    with patch("homeassistant.components.reolink.PLATFORMS", [Platform.SWITCH]):
+        assert await hass.config_entries.async_setup(config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    assert config_entry.state == ConfigEntryState.LOADED
+
+    reolink_connect.baichuan.privacy_mode.return_value = False
+
+
+async def test_LoginPrivacyModeError(
+    hass: HomeAssistant,
+    freezer: FrozenDateTimeFactory,
+    reolink_connect: MagicMock,
+    config_entry: MockConfigEntry,
+) -> None:
+    """Test normal update when get_states returns a LoginPrivacyModeError."""
+    reolink_connect.baichuan.privacy_mode.return_value = False
+    reolink_connect.get_states = AsyncMock(
+        side_effect=LoginPrivacyModeError("Test error")
+    )
+
+    with patch("homeassistant.components.reolink.PLATFORMS", [Platform.SWITCH]):
+        assert await hass.config_entries.async_setup(config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    reolink_connect.baichuan.check_subscribe_events.reset_mock()
+    assert reolink_connect.baichuan.check_subscribe_events.call_count == 0
+
+    freezer.tick(DEVICE_UPDATE_INTERVAL)
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+
+    assert reolink_connect.baichuan.check_subscribe_events.call_count >= 1
+
+
+async def test_privacy_mode_change_callback(
+    hass: HomeAssistant,
+    freezer: FrozenDateTimeFactory,
+    config_entry: MockConfigEntry,
+    reolink_connect: MagicMock,
+) -> None:
+    """Test privacy mode changed callback."""
+
+    class callback_mock_class:
+        callback_func = None
+
+        def register_callback(
+            self, callback_id: str, callback: Callable[[], None], *args, **key_args
+        ) -> None:
+            if callback_id == "privacy_mode_change":
+                self.callback_func = callback
+
+    callback_mock = callback_mock_class()
+
+    reolink_connect.model = TEST_HOST_MODEL
+    reolink_connect.baichuan.events_active = True
+    reolink_connect.baichuan.subscribe_events.reset_mock(side_effect=True)
+    reolink_connect.baichuan.register_callback = callback_mock.register_callback
+    reolink_connect.baichuan.privacy_mode.return_value = True
+    reolink_connect.audio_record.return_value = True
+    reolink_connect.get_states = AsyncMock()
+
+    with patch("homeassistant.components.reolink.PLATFORMS", [Platform.SWITCH]):
+        assert await hass.config_entries.async_setup(config_entry.entry_id)
+    await hass.async_block_till_done()
+    assert config_entry.state is ConfigEntryState.LOADED
+
+    entity_id = f"{Platform.SWITCH}.{TEST_NVR_NAME}_record_audio"
+    assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
+
+    # simulate a TCP push callback signaling a privacy mode change
+    reolink_connect.baichuan.privacy_mode.return_value = False
+    assert callback_mock.callback_func is not None
+    callback_mock.callback_func()
+
+    # check that a coordinator update was scheduled.
+    reolink_connect.get_states.reset_mock()
+    assert reolink_connect.get_states.call_count == 0
+
+    freezer.tick(5)
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+
+    assert reolink_connect.get_states.call_count >= 1
+    assert hass.states.get(entity_id).state == STATE_ON