jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-30 16:57:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use non-autospec mock in Reolink's remaining tests (#149565)

Browse Source 
Co-authored-by: starkillerOG <starkiller.og@gmail.com>
This commit is contained in:
Abílio Costa 2025-07-28 18:38:06 +01:00 committed by GitHub
parent b1dd742a57
commit dda46e7e0b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 102 additions and 176 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

65
tests/components/reolink/conftest.py
View File

@ -1,11 +1,10 @@
"""Setup the Reolink tests.""" """Setup the Reolink tests."""
from collections.abc import Generator from collections.abc import Generator
from unittest.mock import AsyncMock, MagicMock, create_autospec, patch from unittest.mock import AsyncMock, MagicMock, patch
import pytest import pytest
from reolink_aio.api import Chime from reolink_aio.api import Chime
from reolink_aio.baichuan import Baichuan
from reolink_aio.exceptions import ReolinkError from reolink_aio.exceptions import ReolinkError
from homeassistant.components.reolink.config_flow import DEFAULT_PROTOCOL from homeassistant.components.reolink.config_flow import DEFAULT_PROTOCOL
@ -91,6 +90,8 @@ def _init_host_mock(host_mock: MagicMock) -> None:
host_mock.expire_session = AsyncMock() host_mock.expire_session = AsyncMock()
host_mock.set_volume = AsyncMock() host_mock.set_volume = AsyncMock()
host_mock.set_hub_audio = AsyncMock() host_mock.set_hub_audio = AsyncMock()
host_mock.play_quick_reply = AsyncMock()
host_mock.update_firmware = AsyncMock()
host_mock.is_nvr = True host_mock.is_nvr = True
host_mock.is_hub = False host_mock.is_hub = False
host_mock.mac_address = TEST_MAC host_mock.mac_address = TEST_MAC
@ -155,6 +156,7 @@ def _init_host_mock(host_mock: MagicMock) -> None:
host_mock.recording_packing_time = "60 Minutes" host_mock.recording_packing_time = "60 Minutes"
# Baichuan # Baichuan
host_mock.baichuan = MagicMock()
host_mock.baichuan_only = False host_mock.baichuan_only = False
# Disable tcp push by default for tests # Disable tcp push by default for tests
host_mock.baichuan.port = TEST_BC_PORT host_mock.baichuan.port = TEST_BC_PORT
@ -163,6 +165,8 @@ def _init_host_mock(host_mock: MagicMock) -> None:
host_mock.baichuan.unsubscribe_events = AsyncMock() host_mock.baichuan.unsubscribe_events = AsyncMock()
host_mock.baichuan.check_subscribe_events = AsyncMock() host_mock.baichuan.check_subscribe_events = AsyncMock()
host_mock.baichuan.get_privacy_mode = AsyncMock() host_mock.baichuan.get_privacy_mode = AsyncMock()
host_mock.baichuan.set_privacy_mode = AsyncMock()
host_mock.baichuan.set_scene = AsyncMock()
host_mock.baichuan.mac_address.return_value = TEST_MAC_CAM host_mock.baichuan.mac_address.return_value = TEST_MAC_CAM
host_mock.baichuan.privacy_mode.return_value = False host_mock.baichuan.privacy_mode.return_value = False
host_mock.baichuan.day_night_state.return_value = "day" host_mock.baichuan.day_night_state.return_value = "day"
@ -180,38 +184,20 @@ def _init_host_mock(host_mock: MagicMock) -> None:
host_mock.baichuan.smart_ai_name.return_value = "zone1" host_mock.baichuan.smart_ai_name.return_value = "zone1"
@pytest.fixture(scope="module") @pytest.fixture
def reolink_connect_class() -> Generator[MagicMock]: def reolink_host_class() -> Generator[MagicMock]:
"""Mock reolink connection and return both the host_mock and host_mock_class.""" """Mock reolink connection and return both the host_mock and host_mock_class."""
with ( with patch(
patch( "homeassistant.components.reolink.host.Host", autospec=False
"homeassistant.components.reolink.host.Host", autospec=True ) as host_mock_class:
) as host_mock_class, _init_host_mock(host_mock_class.return_value)
):
host_mock = host_mock_class.return_value
host_mock.baichuan = create_autospec(Baichuan)
_init_host_mock(host_mock)
yield host_mock_class yield host_mock_class
@pytest.fixture @pytest.fixture
def reolink_connect( def reolink_host(reolink_host_class: MagicMock) -> Generator[MagicMock]:
reolink_connect_class: MagicMock,
) -> Generator[MagicMock]:
"""Mock reolink connection."""
return reolink_connect_class.return_value
@pytest.fixture
def reolink_host() -> Generator[MagicMock]:
"""Mock reolink Host class.""" """Mock reolink Host class."""
with patch( return reolink_host_class.return_value
"homeassistant.components.reolink.host.Host", autospec=False
) as host_mock_class:
host_mock = host_mock_class.return_value
host_mock.baichuan = MagicMock()
_init_host_mock(host_mock)
yield host_mock
@pytest.fixture @pytest.fixture
@ -246,29 +232,6 @@ def config_entry(hass: HomeAssistant) -> MockConfigEntry:
return config_entry return config_entry
@pytest.fixture
def test_chime(reolink_connect: MagicMock) -> None:
"""Mock a reolink chime."""
TEST_CHIME = Chime(
host=reolink_connect,
dev_id=12345678,
channel=0,
)
TEST_CHIME.name = "Test chime"
TEST_CHIME.volume = 3
TEST_CHIME.connect_state = 2
TEST_CHIME.led_state = True
TEST_CHIME.event_info = {
"md": {"switch": 0, "musicId": 0},
"people": {"switch": 0, "musicId": 1},
"visitor": {"switch": 1, "musicId": 2},
}
reolink_connect.chime_list = [TEST_CHIME]
reolink_connect.chime.return_value = TEST_CHIME
return TEST_CHIME
@pytest.fixture @pytest.fixture
def reolink_chime(reolink_host: MagicMock) -> None: def reolink_chime(reolink_host: MagicMock) -> None:
"""Mock a reolink chime.""" """Mock a reolink chime."""

114
tests/components/reolink/test_config_flow.py
View File

@ -58,7 +58,7 @@ from .conftest import (
from tests.common import MockConfigEntry, async_fire_time_changed from tests.common import MockConfigEntry, async_fire_time_changed
pytestmark = pytest.mark.usefixtures("reolink_connect") pytestmark = pytest.mark.usefixtures("reolink_host")
async def test_config_flow_manual_success( async def test_config_flow_manual_success(
@ -101,11 +101,11 @@ async def test_config_flow_manual_success(
async def test_config_flow_privacy_success( async def test_config_flow_privacy_success(
hass: HomeAssistant, reolink_connect: MagicMock, mock_setup_entry: MagicMock hass: HomeAssistant, reolink_host: MagicMock, mock_setup_entry: MagicMock
) -> None: ) -> None:
"""Successful flow when privacy mode is turned on.""" """Successful flow when privacy mode is turned on."""
reolink_connect.baichuan.privacy_mode.return_value = True reolink_host.baichuan.privacy_mode.return_value = True
reolink_connect.get_host_data.side_effect = LoginPrivacyModeError("Test error") reolink_host.get_host_data.side_effect = LoginPrivacyModeError("Test error")
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER} DOMAIN, context={"source": config_entries.SOURCE_USER}
@ -128,13 +128,13 @@ async def test_config_flow_privacy_success(
assert result["step_id"] == "privacy" assert result["step_id"] == "privacy"
assert result["errors"] is None assert result["errors"] is None
assert reolink_connect.baichuan.set_privacy_mode.call_count == 0 assert reolink_host.baichuan.set_privacy_mode.call_count == 0
reolink_connect.get_host_data.reset_mock(side_effect=True) reolink_host.get_host_data.reset_mock(side_effect=True)
with patch("homeassistant.components.reolink.config_flow.API_STARTUP_TIME", new=0): with patch("homeassistant.components.reolink.config_flow.API_STARTUP_TIME", new=0):
result = await hass.config_entries.flow.async_configure(result["flow_id"], {}) result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert reolink_connect.baichuan.set_privacy_mode.call_count == 1 assert reolink_host.baichuan.set_privacy_mode.call_count == 1
assert result["type"] is FlowResultType.CREATE_ENTRY assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == TEST_NVR_NAME assert result["title"] == TEST_NVR_NAME
@ -153,14 +153,12 @@ async def test_config_flow_privacy_success(
} }
assert result["result"].unique_id == TEST_MAC assert result["result"].unique_id == TEST_MAC
reolink_connect.baichuan.privacy_mode.return_value = False
async def test_config_flow_baichuan_only( async def test_config_flow_baichuan_only(
hass: HomeAssistant, reolink_connect: MagicMock, mock_setup_entry: MagicMock hass: HomeAssistant, reolink_host: MagicMock, mock_setup_entry: MagicMock
) -> None: ) -> None:
"""Successful flow manually initialized by the user for baichuan only device.""" """Successful flow manually initialized by the user for baichuan only device."""
reolink_connect.baichuan_only = True reolink_host.baichuan_only = True
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER} DOMAIN, context={"source": config_entries.SOURCE_USER}
@ -196,11 +194,9 @@ async def test_config_flow_baichuan_only(
} }
assert result["result"].unique_id == TEST_MAC assert result["result"].unique_id == TEST_MAC
reolink_connect.baichuan_only = False
async def test_config_flow_errors( async def test_config_flow_errors(
hass: HomeAssistant, reolink_connect: MagicMock, mock_setup_entry: MagicMock hass: HomeAssistant, reolink_host: MagicMock, mock_setup_entry: MagicMock
) -> None: ) -> None:
"""Successful flow manually initialized by the user after some errors.""" """Successful flow manually initialized by the user after some errors."""
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
@ -211,10 +207,10 @@ async def test_config_flow_errors(
assert result["step_id"] == "user" assert result["step_id"] == "user"
assert result["errors"] == {} assert result["errors"] == {}
reolink_connect.is_admin = False reolink_host.is_admin = False
reolink_connect.user_level = "guest" reolink_host.user_level = "guest"
reolink_connect.unsubscribe.side_effect = ReolinkError("Test error") reolink_host.unsubscribe.side_effect = ReolinkError("Test error")
reolink_connect.logout.side_effect = ReolinkError("Test error") reolink_host.logout.side_effect = ReolinkError("Test error")
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
{ {
@ -228,9 +224,9 @@ async def test_config_flow_errors(
assert result["step_id"] == "user" assert result["step_id"] == "user"
assert result["errors"] == {CONF_USERNAME: "not_admin"} assert result["errors"] == {CONF_USERNAME: "not_admin"}
reolink_connect.is_admin = True reolink_host.is_admin = True
reolink_connect.user_level = "admin" reolink_host.user_level = "admin"
reolink_connect.get_host_data.side_effect = ReolinkError("Test error") reolink_host.get_host_data.side_effect = ReolinkError("Test error")
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
{ {
@ -244,7 +240,7 @@ async def test_config_flow_errors(
assert result["step_id"] == "user" assert result["step_id"] == "user"
assert result["errors"] == {CONF_HOST: "cannot_connect"} assert result["errors"] == {CONF_HOST: "cannot_connect"}
reolink_connect.get_host_data.side_effect = ReolinkWebhookException("Test error") reolink_host.get_host_data.side_effect = ReolinkWebhookException("Test error")
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
{ {
@ -258,7 +254,7 @@ async def test_config_flow_errors(
assert result["step_id"] == "user" assert result["step_id"] == "user"
assert result["errors"] == {"base": "webhook_exception"} assert result["errors"] == {"base": "webhook_exception"}
reolink_connect.get_host_data.side_effect = json.JSONDecodeError( reolink_host.get_host_data.side_effect = json.JSONDecodeError(
"test_error", "test", 1 "test_error", "test", 1
) )
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
@ -274,7 +270,7 @@ async def test_config_flow_errors(
assert result["step_id"] == "user" assert result["step_id"] == "user"
assert result["errors"] == {CONF_HOST: "unknown"} assert result["errors"] == {CONF_HOST: "unknown"}
reolink_connect.get_host_data.side_effect = CredentialsInvalidError("Test error") reolink_host.get_host_data.side_effect = CredentialsInvalidError("Test error")
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
{ {
@ -288,7 +284,7 @@ async def test_config_flow_errors(
assert result["step_id"] == "user" assert result["step_id"] == "user"
assert result["errors"] == {CONF_PASSWORD: "invalid_auth"} assert result["errors"] == {CONF_PASSWORD: "invalid_auth"}
reolink_connect.get_host_data.side_effect = LoginFirmwareError("Test error") reolink_host.get_host_data.side_effect = LoginFirmwareError("Test error")
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
{ {
@ -302,7 +298,7 @@ async def test_config_flow_errors(
assert result["step_id"] == "user" assert result["step_id"] == "user"
assert result["errors"] == {"base": "update_needed"} assert result["errors"] == {"base": "update_needed"}
reolink_connect.valid_password.return_value = False reolink_host.valid_password.return_value = False
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
{ {
@ -316,8 +312,8 @@ async def test_config_flow_errors(
assert result["step_id"] == "user" assert result["step_id"] == "user"
assert result["errors"] == {CONF_PASSWORD: "password_incompatible"} assert result["errors"] == {CONF_PASSWORD: "password_incompatible"}
reolink_connect.valid_password.return_value = True reolink_host.valid_password.return_value = True
reolink_connect.get_host_data.side_effect = ApiError("Test error") reolink_host.get_host_data.side_effect = ApiError("Test error")
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
{ {
@ -331,7 +327,7 @@ async def test_config_flow_errors(
assert result["step_id"] == "user" assert result["step_id"] == "user"
assert result["errors"] == {CONF_HOST: "api_error"} assert result["errors"] == {CONF_HOST: "api_error"}
reolink_connect.get_host_data.reset_mock(side_effect=True) reolink_host.get_host_data.reset_mock(side_effect=True)
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
{ {
@ -360,9 +356,6 @@ async def test_config_flow_errors(
CONF_PROTOCOL: DEFAULT_PROTOCOL, CONF_PROTOCOL: DEFAULT_PROTOCOL,
} }
reolink_connect.unsubscribe.reset_mock(side_effect=True)
reolink_connect.logout.reset_mock(side_effect=True)
async def test_options_flow(hass: HomeAssistant, mock_setup_entry: MagicMock) -> None: async def test_options_flow(hass: HomeAssistant, mock_setup_entry: MagicMock) -> None:
"""Test specifying non default settings using options flow.""" """Test specifying non default settings using options flow."""
@ -450,7 +443,7 @@ async def test_reauth(hass: HomeAssistant, mock_setup_entry: MagicMock) -> None:
async def test_reauth_abort_unique_id_mismatch( async def test_reauth_abort_unique_id_mismatch(
hass: HomeAssistant, mock_setup_entry: MagicMock, reolink_connect: MagicMock hass: HomeAssistant, mock_setup_entry: MagicMock, reolink_host: MagicMock
) -> None: ) -> None:
"""Test a reauth flow.""" """Test a reauth flow."""
config_entry = MockConfigEntry( config_entry = MockConfigEntry(
@ -475,7 +468,7 @@ async def test_reauth_abort_unique_id_mismatch(
assert await hass.config_entries.async_setup(config_entry.entry_id) assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done() await hass.async_block_till_done()
reolink_connect.mac_address = "aa:aa:aa:aa:aa:aa" reolink_host.mac_address = "aa:aa:aa:aa:aa:aa"
result = await config_entry.start_reauth_flow(hass) result = await config_entry.start_reauth_flow(hass)
@ -497,8 +490,6 @@ async def test_reauth_abort_unique_id_mismatch(
assert config_entry.data[CONF_USERNAME] == TEST_USERNAME assert config_entry.data[CONF_USERNAME] == TEST_USERNAME
assert config_entry.data[CONF_PASSWORD] == TEST_PASSWORD assert config_entry.data[CONF_PASSWORD] == TEST_PASSWORD
reolink_connect.mac_address = TEST_MAC
async def test_dhcp_flow(hass: HomeAssistant, mock_setup_entry: MagicMock) -> None: async def test_dhcp_flow(hass: HomeAssistant, mock_setup_entry: MagicMock) -> None:
"""Successful flow from DHCP discovery.""" """Successful flow from DHCP discovery."""
@ -544,8 +535,8 @@ async def test_dhcp_flow(hass: HomeAssistant, mock_setup_entry: MagicMock) -> No
async def test_dhcp_ip_update_aborted_if_wrong_mac( async def test_dhcp_ip_update_aborted_if_wrong_mac(
hass: HomeAssistant, hass: HomeAssistant,
freezer: FrozenDateTimeFactory, freezer: FrozenDateTimeFactory,
reolink_connect_class: MagicMock, reolink_host_class: MagicMock,
reolink_connect: MagicMock, reolink_host: MagicMock,
) -> None: ) -> None:
"""Test dhcp discovery does not update the IP if the mac address does not match.""" """Test dhcp discovery does not update the IP if the mac address does not match."""
config_entry = MockConfigEntry( config_entry = MockConfigEntry(
@ -572,7 +563,7 @@ async def test_dhcp_ip_update_aborted_if_wrong_mac(
assert config_entry.state is ConfigEntryState.LOADED assert config_entry.state is ConfigEntryState.LOADED
# ensure the last_update_succes is False for the device_coordinator. # ensure the last_update_succes is False for the device_coordinator.
reolink_connect.get_states.side_effect = ReolinkError("Test error") reolink_host.get_states.side_effect = ReolinkError("Test error")
freezer.tick(DEVICE_UPDATE_INTERVAL) freezer.tick(DEVICE_UPDATE_INTERVAL)
async_fire_time_changed(hass) async_fire_time_changed(hass)
await hass.async_block_till_done() await hass.async_block_till_done()
@ -583,7 +574,7 @@ async def test_dhcp_ip_update_aborted_if_wrong_mac(
macaddress=DHCP_FORMATTED_MAC, macaddress=DHCP_FORMATTED_MAC,
) )
reolink_connect.mac_address = "aa:aa:aa:aa:aa:aa" reolink_host.mac_address = "aa:aa:aa:aa:aa:aa"
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_DHCP}, data=dhcp_data DOMAIN, context={"source": config_entries.SOURCE_DHCP}, data=dhcp_data
@ -602,9 +593,9 @@ async def test_dhcp_ip_update_aborted_if_wrong_mac(
bc_port=TEST_BC_PORT, bc_port=TEST_BC_PORT,
bc_only=False, bc_only=False,
) )
assert expected_call in reolink_connect_class.call_args_list assert expected_call in reolink_host_class.call_args_list
for exc_call in reolink_connect_class.call_args_list: for exc_call in reolink_host_class.call_args_list:
assert exc_call[0][0] in [TEST_HOST, TEST_HOST2] assert exc_call[0][0] in [TEST_HOST, TEST_HOST2]
get_session = exc_call[1]["aiohttp_get_session_callback"] get_session = exc_call[1]["aiohttp_get_session_callback"]
assert isinstance(get_session(), ClientSession) assert isinstance(get_session(), ClientSession)
@ -616,10 +607,6 @@ async def test_dhcp_ip_update_aborted_if_wrong_mac(
# Check that IP was not updated # Check that IP was not updated
assert config_entry.data[CONF_HOST] == TEST_HOST assert config_entry.data[CONF_HOST] == TEST_HOST
reolink_connect.get_states.side_effect = None
reolink_connect_class.reset_mock()
reolink_connect.mac_address = TEST_MAC
@pytest.mark.parametrize( @pytest.mark.parametrize(
("attr", "value", "expected", "host_call_list"), ("attr", "value", "expected", "host_call_list"),
@ -641,8 +628,8 @@ async def test_dhcp_ip_update_aborted_if_wrong_mac(
async def test_dhcp_ip_update( async def test_dhcp_ip_update(
hass: HomeAssistant, hass: HomeAssistant,
freezer: FrozenDateTimeFactory, freezer: FrozenDateTimeFactory,
reolink_connect_class: MagicMock, reolink_host_class: MagicMock,
reolink_connect: MagicMock, reolink_host: MagicMock,
attr: str, attr: str,
value: Any, value: Any,
expected: str, expected: str,
@ -673,7 +660,7 @@ async def test_dhcp_ip_update(
assert config_entry.state is ConfigEntryState.LOADED assert config_entry.state is ConfigEntryState.LOADED
# ensure the last_update_succes is False for the device_coordinator. # ensure the last_update_succes is False for the device_coordinator.
reolink_connect.get_states.side_effect = ReolinkError("Test error") reolink_host.get_states.side_effect = ReolinkError("Test error")
freezer.tick(DEVICE_UPDATE_INTERVAL) freezer.tick(DEVICE_UPDATE_INTERVAL)
async_fire_time_changed(hass) async_fire_time_changed(hass)
await hass.async_block_till_done() await hass.async_block_till_done()
@ -685,8 +672,7 @@ async def test_dhcp_ip_update(
) )
if attr is not None: if attr is not None:
original = getattr(reolink_connect, attr) setattr(reolink_host, attr, value)
setattr(reolink_connect, attr, value)
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_DHCP}, data=dhcp_data DOMAIN, context={"source": config_entries.SOURCE_DHCP}, data=dhcp_data
@ -705,9 +691,9 @@ async def test_dhcp_ip_update(
bc_port=TEST_BC_PORT, bc_port=TEST_BC_PORT,
bc_only=False, bc_only=False,
) )
assert expected_call in reolink_connect_class.call_args_list assert expected_call in reolink_host_class.call_args_list
for exc_call in reolink_connect_class.call_args_list: for exc_call in reolink_host_class.call_args_list:
assert exc_call[0][0] in host_call_list assert exc_call[0][0] in host_call_list
get_session = exc_call[1]["aiohttp_get_session_callback"] get_session = exc_call[1]["aiohttp_get_session_callback"]
assert isinstance(get_session(), ClientSession) assert isinstance(get_session(), ClientSession)
@ -718,17 +704,12 @@ async def test_dhcp_ip_update(
await hass.async_block_till_done() await hass.async_block_till_done()
assert config_entry.data[CONF_HOST] == expected assert config_entry.data[CONF_HOST] == expected
reolink_connect.get_states.side_effect = None
reolink_connect_class.reset_mock()
if attr is not None:
setattr(reolink_connect, attr, original)
async def test_dhcp_ip_update_ingnored_if_still_connected( async def test_dhcp_ip_update_ingnored_if_still_connected(
hass: HomeAssistant, hass: HomeAssistant,
freezer: FrozenDateTimeFactory, freezer: FrozenDateTimeFactory,
reolink_connect_class: MagicMock, reolink_host_class: MagicMock,
reolink_connect: MagicMock, reolink_host: MagicMock,
) -> None: ) -> None:
"""Test dhcp discovery is ignored when the camera is still properly connected to HA.""" """Test dhcp discovery is ignored when the camera is still properly connected to HA."""
config_entry = MockConfigEntry( config_entry = MockConfigEntry(
@ -776,9 +757,9 @@ async def test_dhcp_ip_update_ingnored_if_still_connected(
bc_port=TEST_BC_PORT, bc_port=TEST_BC_PORT,
bc_only=False, bc_only=False,
) )
assert expected_call in reolink_connect_class.call_args_list assert expected_call in reolink_host_class.call_args_list
for exc_call in reolink_connect_class.call_args_list: for exc_call in reolink_host_class.call_args_list:
assert exc_call[0][0] == TEST_HOST assert exc_call[0][0] == TEST_HOST
get_session = exc_call[1]["aiohttp_get_session_callback"] get_session = exc_call[1]["aiohttp_get_session_callback"]
assert isinstance(get_session(), ClientSession) assert isinstance(get_session(), ClientSession)
@ -789,9 +770,6 @@ async def test_dhcp_ip_update_ingnored_if_still_connected(
await hass.async_block_till_done() await hass.async_block_till_done()
assert config_entry.data[CONF_HOST] == TEST_HOST assert config_entry.data[CONF_HOST] == TEST_HOST
reolink_connect.get_states.side_effect = None
reolink_connect_class.reset_mock()
async def test_reconfig(hass: HomeAssistant, mock_setup_entry: MagicMock) -> None: async def test_reconfig(hass: HomeAssistant, mock_setup_entry: MagicMock) -> None:
"""Test a reconfiguration flow.""" """Test a reconfiguration flow."""
@ -840,7 +818,7 @@ async def test_reconfig(hass: HomeAssistant, mock_setup_entry: MagicMock) -> Non
async def test_reconfig_abort_unique_id_mismatch( async def test_reconfig_abort_unique_id_mismatch(
hass: HomeAssistant, mock_setup_entry: MagicMock, reolink_connect: MagicMock hass: HomeAssistant, mock_setup_entry: MagicMock, reolink_host: MagicMock
) -> None: ) -> None:
"""Test a reconfiguration flow aborts if the unique id does not match.""" """Test a reconfiguration flow aborts if the unique id does not match."""
config_entry = MockConfigEntry( config_entry = MockConfigEntry(
@ -865,7 +843,7 @@ async def test_reconfig_abort_unique_id_mismatch(
assert await hass.config_entries.async_setup(config_entry.entry_id) assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done() await hass.async_block_till_done()
reolink_connect.mac_address = "aa:aa:aa:aa:aa:aa" reolink_host.mac_address = "aa:aa:aa:aa:aa:aa"
result = await config_entry.start_reconfigure_flow(hass) result = await config_entry.start_reconfigure_flow(hass)
@ -887,5 +865,3 @@ async def test_reconfig_abort_unique_id_mismatch(
assert config_entry.data[CONF_HOST] == TEST_HOST assert config_entry.data[CONF_HOST] == TEST_HOST
assert config_entry.data[CONF_USERNAME] == TEST_USERNAME assert config_entry.data[CONF_USERNAME] == TEST_USERNAME
assert config_entry.data[CONF_PASSWORD] == TEST_PASSWORD assert config_entry.data[CONF_PASSWORD] == TEST_PASSWORD
reolink_connect.mac_address = TEST_MAC

49
tests/components/reolink/test_select.py
View File

@ -29,7 +29,7 @@ async def test_floodlight_mode_select(
hass: HomeAssistant, hass: HomeAssistant,
freezer: FrozenDateTimeFactory, freezer: FrozenDateTimeFactory,
config_entry: MockConfigEntry, config_entry: MockConfigEntry,
reolink_connect: MagicMock, reolink_host: MagicMock,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
) -> None: ) -> None:
"""Test select entity with floodlight_mode.""" """Test select entity with floodlight_mode."""
@ -47,9 +47,9 @@ async def test_floodlight_mode_select(
{ATTR_ENTITY_ID: entity_id, "option": "off"}, {ATTR_ENTITY_ID: entity_id, "option": "off"},
blocking=True, blocking=True,
) )
reolink_connect.set_whiteled.assert_called_once() reolink_host.set_whiteled.assert_called_once()
reolink_connect.set_whiteled.side_effect = ReolinkError("Test error") reolink_host.set_whiteled.side_effect = ReolinkError("Test error")
with pytest.raises(HomeAssistantError): with pytest.raises(HomeAssistantError):
await hass.services.async_call( await hass.services.async_call(
SELECT_DOMAIN, SELECT_DOMAIN,
@ -58,7 +58,7 @@ async def test_floodlight_mode_select(
blocking=True, blocking=True,
) )
reolink_connect.set_whiteled.side_effect = InvalidParameterError("Test error") reolink_host.set_whiteled.side_effect = InvalidParameterError("Test error")
with pytest.raises(ServiceValidationError): with pytest.raises(ServiceValidationError):
await hass.services.async_call( await hass.services.async_call(
SELECT_DOMAIN, SELECT_DOMAIN,
@ -67,24 +67,22 @@ async def test_floodlight_mode_select(
blocking=True, blocking=True,
) )
reolink_connect.whiteled_mode.return_value = -99 # invalid value reolink_host.whiteled_mode.return_value = -99 # invalid value
freezer.tick(DEVICE_UPDATE_INTERVAL) freezer.tick(DEVICE_UPDATE_INTERVAL)
async_fire_time_changed(hass) async_fire_time_changed(hass)
await hass.async_block_till_done() await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_UNKNOWN assert hass.states.get(entity_id).state == STATE_UNKNOWN
reolink_connect.set_whiteled.reset_mock(side_effect=True)
async def test_play_quick_reply_message( async def test_play_quick_reply_message(
hass: HomeAssistant, hass: HomeAssistant,
config_entry: MockConfigEntry, config_entry: MockConfigEntry,
reolink_connect: MagicMock, reolink_host: MagicMock,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
) -> None: ) -> None:
"""Test select play_quick_reply_message entity.""" """Test select play_quick_reply_message entity."""
reolink_connect.quick_reply_dict.return_value = {0: "off", 1: "test message"} reolink_host.quick_reply_dict.return_value = {0: "off", 1: "test message"}
with patch("homeassistant.components.reolink.PLATFORMS", [Platform.SELECT]): with patch("homeassistant.components.reolink.PLATFORMS", [Platform.SELECT]):
assert await hass.config_entries.async_setup(config_entry.entry_id) assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done() await hass.async_block_till_done()
@ -99,16 +97,14 @@ async def test_play_quick_reply_message(
{ATTR_ENTITY_ID: entity_id, "option": "test message"}, {ATTR_ENTITY_ID: entity_id, "option": "test message"},
blocking=True, blocking=True,
) )
reolink_connect.play_quick_reply.assert_called_once() reolink_host.play_quick_reply.assert_called_once()
reolink_connect.quick_reply_dict = MagicMock()
async def test_host_scene_select( async def test_host_scene_select(
hass: HomeAssistant, hass: HomeAssistant,
freezer: FrozenDateTimeFactory, freezer: FrozenDateTimeFactory,
config_entry: MockConfigEntry, config_entry: MockConfigEntry,
reolink_connect: MagicMock, reolink_host: MagicMock,
) -> None: ) -> None:
"""Test host select entity with scene mode.""" """Test host select entity with scene mode."""
with patch("homeassistant.components.reolink.PLATFORMS", [Platform.SELECT]): with patch("homeassistant.components.reolink.PLATFORMS", [Platform.SELECT]):
@ -125,9 +121,9 @@ async def test_host_scene_select(
{ATTR_ENTITY_ID: entity_id, "option": "home"}, {ATTR_ENTITY_ID: entity_id, "option": "home"},
blocking=True, blocking=True,
) )
reolink_connect.baichuan.set_scene.assert_called_once() reolink_host.baichuan.set_scene.assert_called_once()
reolink_connect.baichuan.set_scene.side_effect = ReolinkError("Test error") reolink_host.baichuan.set_scene.side_effect = ReolinkError("Test error")
with pytest.raises(HomeAssistantError): with pytest.raises(HomeAssistantError):
await hass.services.async_call( await hass.services.async_call(
SELECT_DOMAIN, SELECT_DOMAIN,
@ -136,7 +132,7 @@ async def test_host_scene_select(
blocking=True, blocking=True,
) )
reolink_connect.baichuan.set_scene.side_effect = InvalidParameterError("Test error") reolink_host.baichuan.set_scene.side_effect = InvalidParameterError("Test error")
with pytest.raises(ServiceValidationError): with pytest.raises(ServiceValidationError):
await hass.services.async_call( await hass.services.async_call(
SELECT_DOMAIN, SELECT_DOMAIN,
@ -145,23 +141,20 @@ async def test_host_scene_select(
blocking=True, blocking=True,
) )
reolink_connect.baichuan.active_scene = "Invalid value" reolink_host.baichuan.active_scene = "Invalid value"
freezer.tick(DEVICE_UPDATE_INTERVAL) freezer.tick(DEVICE_UPDATE_INTERVAL)
async_fire_time_changed(hass) async_fire_time_changed(hass)
await hass.async_block_till_done() await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_UNKNOWN assert hass.states.get(entity_id).state == STATE_UNKNOWN
reolink_connect.baichuan.set_scene.reset_mock(side_effect=True)
reolink_connect.baichuan.active_scene = "off"
async def test_chime_select( async def test_chime_select(
hass: HomeAssistant, hass: HomeAssistant,
freezer: FrozenDateTimeFactory, freezer: FrozenDateTimeFactory,
config_entry: MockConfigEntry, config_entry: MockConfigEntry,
reolink_connect: MagicMock, reolink_host: MagicMock,
test_chime: Chime, reolink_chime: Chime,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
) -> None: ) -> None:
"""Test chime select entity.""" """Test chime select entity."""
@ -175,16 +168,16 @@ async def test_chime_select(
assert hass.states.get(entity_id).state == "pianokey" assert hass.states.get(entity_id).state == "pianokey"
# Test selecting chime ringtone option # Test selecting chime ringtone option
test_chime.set_tone = AsyncMock() reolink_chime.set_tone = AsyncMock()
await hass.services.async_call( await hass.services.async_call(
SELECT_DOMAIN, SELECT_DOMAIN,
SERVICE_SELECT_OPTION, SERVICE_SELECT_OPTION,
{ATTR_ENTITY_ID: entity_id, "option": "off"}, {ATTR_ENTITY_ID: entity_id, "option": "off"},
blocking=True, blocking=True,
) )
test_chime.set_tone.assert_called_once() reolink_chime.set_tone.assert_called_once()
test_chime.set_tone.side_effect = ReolinkError("Test error") reolink_chime.set_tone.side_effect = ReolinkError("Test error")
with pytest.raises(HomeAssistantError): with pytest.raises(HomeAssistantError):
await hass.services.async_call( await hass.services.async_call(
SELECT_DOMAIN, SELECT_DOMAIN,
@ -193,7 +186,7 @@ async def test_chime_select(
blocking=True, blocking=True,
) )
test_chime.set_tone.side_effect = InvalidParameterError("Test error") reolink_chime.set_tone.side_effect = InvalidParameterError("Test error")
with pytest.raises(ServiceValidationError): with pytest.raises(ServiceValidationError):
await hass.services.async_call( await hass.services.async_call(
SELECT_DOMAIN, SELECT_DOMAIN,
@ -203,11 +196,9 @@ async def test_chime_select(
) )
# Test unavailable # Test unavailable
test_chime.event_info = {} reolink_chime.event_info = {}
freezer.tick(DEVICE_UPDATE_INTERVAL) freezer.tick(DEVICE_UPDATE_INTERVAL)
async_fire_time_changed(hass) async_fire_time_changed(hass)
await hass.async_block_till_done() await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_UNKNOWN assert hass.states.get(entity_id).state == STATE_UNKNOWN
test_chime.set_tone.reset_mock(side_effect=True)

50
tests/components/reolink/test_update.py
View File

@ -30,11 +30,11 @@ TEST_RELEASE_NOTES = "bugfix 1, bugfix 2"
async def test_no_update( async def test_no_update(
hass: HomeAssistant, hass: HomeAssistant,
config_entry: MockConfigEntry, config_entry: MockConfigEntry,
reolink_connect: MagicMock, reolink_host: MagicMock,
entity_name: str, entity_name: str,
) -> None: ) -> None:
"""Test update state when no update available.""" """Test update state when no update available."""
reolink_connect.camera_name.return_value = TEST_CAM_NAME reolink_host.camera_name.return_value = TEST_CAM_NAME
with patch("homeassistant.components.reolink.PLATFORMS", [Platform.UPDATE]): with patch("homeassistant.components.reolink.PLATFORMS", [Platform.UPDATE]):
assert await hass.config_entries.async_setup(config_entry.entry_id) assert await hass.config_entries.async_setup(config_entry.entry_id)
@ -49,12 +49,12 @@ async def test_no_update(
async def test_update_str( async def test_update_str(
hass: HomeAssistant, hass: HomeAssistant,
config_entry: MockConfigEntry, config_entry: MockConfigEntry,
reolink_connect: MagicMock, reolink_host: MagicMock,
entity_name: str, entity_name: str,
) -> None: ) -> None:
"""Test update state when update available with string from API.""" """Test update state when update available with string from API."""
reolink_connect.camera_name.return_value = TEST_CAM_NAME reolink_host.camera_name.return_value = TEST_CAM_NAME
reolink_connect.firmware_update_available.return_value = "New firmware available" reolink_host.firmware_update_available.return_value = "New firmware available"
with patch("homeassistant.components.reolink.PLATFORMS", [Platform.UPDATE]): with patch("homeassistant.components.reolink.PLATFORMS", [Platform.UPDATE]):
assert await hass.config_entries.async_setup(config_entry.entry_id) assert await hass.config_entries.async_setup(config_entry.entry_id)
@ -69,21 +69,21 @@ async def test_update_str(
async def test_update_firm( async def test_update_firm(
hass: HomeAssistant, hass: HomeAssistant,
config_entry: MockConfigEntry, config_entry: MockConfigEntry,
reolink_connect: MagicMock, reolink_host: MagicMock,
hass_ws_client: WebSocketGenerator, hass_ws_client: WebSocketGenerator,
freezer: FrozenDateTimeFactory, freezer: FrozenDateTimeFactory,
entity_name: str, entity_name: str,
) -> None: ) -> None:
"""Test update state when update available with firmware info from reolink.com.""" """Test update state when update available with firmware info from reolink.com."""
reolink_connect.camera_name.return_value = TEST_CAM_NAME reolink_host.camera_name.return_value = TEST_CAM_NAME
reolink_connect.sw_upload_progress.return_value = 100 reolink_host.sw_upload_progress.return_value = 100
reolink_connect.camera_sw_version.return_value = "v1.1.0.0.0.0000" reolink_host.camera_sw_version.return_value = "v1.1.0.0.0.0000"
new_firmware = NewSoftwareVersion( new_firmware = NewSoftwareVersion(
version_string="v3.3.0.226_23031644", version_string="v3.3.0.226_23031644",
download_url=TEST_DOWNLOAD_URL, download_url=TEST_DOWNLOAD_URL,
release_notes=TEST_RELEASE_NOTES, release_notes=TEST_RELEASE_NOTES,
) )
reolink_connect.firmware_update_available.return_value = new_firmware reolink_host.firmware_update_available.return_value = new_firmware
with patch("homeassistant.components.reolink.PLATFORMS", [Platform.UPDATE]): with patch("homeassistant.components.reolink.PLATFORMS", [Platform.UPDATE]):
assert await hass.config_entries.async_setup(config_entry.entry_id) assert await hass.config_entries.async_setup(config_entry.entry_id)
@ -117,9 +117,9 @@ async def test_update_firm(
{ATTR_ENTITY_ID: entity_id}, {ATTR_ENTITY_ID: entity_id},
blocking=True, blocking=True,
) )
reolink_connect.update_firmware.assert_called() reolink_host.update_firmware.assert_called()
reolink_connect.sw_upload_progress.return_value = 50 reolink_host.sw_upload_progress.return_value = 50
freezer.tick(POLL_PROGRESS) freezer.tick(POLL_PROGRESS)
async_fire_time_changed(hass) async_fire_time_changed(hass)
await hass.async_block_till_done() await hass.async_block_till_done()
@ -127,7 +127,7 @@ async def test_update_firm(
assert hass.states.get(entity_id).attributes["in_progress"] assert hass.states.get(entity_id).attributes["in_progress"]
assert hass.states.get(entity_id).attributes["update_percentage"] == 50 assert hass.states.get(entity_id).attributes["update_percentage"] == 50
reolink_connect.sw_upload_progress.return_value = 100 reolink_host.sw_upload_progress.return_value = 100
freezer.tick(POLL_AFTER_INSTALL) freezer.tick(POLL_AFTER_INSTALL)
async_fire_time_changed(hass) async_fire_time_changed(hass)
await hass.async_block_till_done() await hass.async_block_till_done()
@ -135,7 +135,7 @@ async def test_update_firm(
assert not hass.states.get(entity_id).attributes["in_progress"] assert not hass.states.get(entity_id).attributes["in_progress"]
assert hass.states.get(entity_id).attributes["update_percentage"] is None assert hass.states.get(entity_id).attributes["update_percentage"] is None
reolink_connect.update_firmware.side_effect = ReolinkError("Test error") reolink_host.update_firmware.side_effect = ReolinkError("Test error")
with pytest.raises(HomeAssistantError): with pytest.raises(HomeAssistantError):
await hass.services.async_call( await hass.services.async_call(
UPDATE_DOMAIN, UPDATE_DOMAIN,
@ -144,7 +144,7 @@ async def test_update_firm(
blocking=True, blocking=True,
) )
reolink_connect.update_firmware.side_effect = ApiError( reolink_host.update_firmware.side_effect = ApiError(
"Test error", translation_key="firmware_rate_limit" "Test error", translation_key="firmware_rate_limit"
) )
with pytest.raises(HomeAssistantError): with pytest.raises(HomeAssistantError):
@ -156,34 +156,32 @@ async def test_update_firm(
) )
# test _async_update_future # test _async_update_future
reolink_connect.camera_sw_version.return_value = "v3.3.0.226_23031644" reolink_host.camera_sw_version.return_value = "v3.3.0.226_23031644"
reolink_connect.firmware_update_available.return_value = False reolink_host.firmware_update_available.return_value = False
freezer.tick(POLL_AFTER_INSTALL) freezer.tick(POLL_AFTER_INSTALL)
async_fire_time_changed(hass) async_fire_time_changed(hass)
await hass.async_block_till_done() await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_OFF assert hass.states.get(entity_id).state == STATE_OFF
reolink_connect.update_firmware.side_effect = None
@pytest.mark.parametrize("entity_name", [TEST_NVR_NAME, TEST_CAM_NAME]) @pytest.mark.parametrize("entity_name", [TEST_NVR_NAME, TEST_CAM_NAME])
async def test_update_firm_keeps_available( async def test_update_firm_keeps_available(
hass: HomeAssistant, hass: HomeAssistant,
config_entry: MockConfigEntry, config_entry: MockConfigEntry,
reolink_connect: MagicMock, reolink_host: MagicMock,
hass_ws_client: WebSocketGenerator, hass_ws_client: WebSocketGenerator,
entity_name: str, entity_name: str,
) -> None: ) -> None:
"""Test update entity keeps being available during update.""" """Test update entity keeps being available during update."""
reolink_connect.camera_name.return_value = TEST_CAM_NAME reolink_host.camera_name.return_value = TEST_CAM_NAME
reolink_connect.camera_sw_version.return_value = "v1.1.0.0.0.0000" reolink_host.camera_sw_version.return_value = "v1.1.0.0.0.0000"
new_firmware = NewSoftwareVersion( new_firmware = NewSoftwareVersion(
version_string="v3.3.0.226_23031644", version_string="v3.3.0.226_23031644",
download_url=TEST_DOWNLOAD_URL, download_url=TEST_DOWNLOAD_URL,
release_notes=TEST_RELEASE_NOTES, release_notes=TEST_RELEASE_NOTES,
) )
reolink_connect.firmware_update_available.return_value = new_firmware reolink_host.firmware_update_available.return_value = new_firmware
with patch("homeassistant.components.reolink.PLATFORMS", [Platform.UPDATE]): with patch("homeassistant.components.reolink.PLATFORMS", [Platform.UPDATE]):
assert await hass.config_entries.async_setup(config_entry.entry_id) assert await hass.config_entries.async_setup(config_entry.entry_id)
@ -196,7 +194,7 @@ async def test_update_firm_keeps_available(
async def mock_update_firmware(*args, **kwargs) -> None: async def mock_update_firmware(*args, **kwargs) -> None:
await asyncio.sleep(0.000005) await asyncio.sleep(0.000005)
reolink_connect.update_firmware = mock_update_firmware reolink_host.update_firmware = mock_update_firmware
# test install # test install
with patch("homeassistant.components.reolink.update.POLL_PROGRESS", 0.000001): with patch("homeassistant.components.reolink.update.POLL_PROGRESS", 0.000001):
@ -207,11 +205,9 @@ async def test_update_firm_keeps_available(
blocking=True, blocking=True,
) )
reolink_connect.session_active = False reolink_host.session_active = False
async_fire_time_changed(hass, utcnow() + timedelta(seconds=1)) async_fire_time_changed(hass, utcnow() + timedelta(seconds=1))
await hass.async_block_till_done() await hass.async_block_till_done()
# still available # still available
assert hass.states.get(entity_id).state == STATE_ON assert hass.states.get(entity_id).state == STATE_ON
reolink_connect.session_active = True