1
0
mirror of https://github.com/home-assistant/core.git synced 2026-05-08 09:38:58 +01:00

Add helper utility for patching Pydantic model methods in UniFi Protect tests (#159346)

Co-authored-by: RaHehl <rahehl@users.noreply.github.com>
This commit is contained in:
Raphael Hehl
2025-12-23 18:54:56 +01:00
committed by GitHub
parent bcc5985c8b
commit 5107b7012d
9 changed files with 445 additions and 419 deletions
+27 -1
View File
@@ -1,9 +1,12 @@
"""Tests for the UniFi Protect integration."""
from collections.abc import Generator
from contextlib import contextmanager
from unittest.mock import AsyncMock, MagicMock, patch
from typing import Any
from unittest.mock import AsyncMock, MagicMock, Mock, patch
import pytest
from uiprotect.data.base import ProtectModel
from unifi_discovery import AIOUnifiScanner, UnifiDevice, UnifiService
DEVICE_HOSTNAME = "unvr"
@@ -46,3 +49,26 @@ def _patch_discovery(device=None, no_device=False):
yield
return _patcher()
@contextmanager
def patch_ufp_method(
obj: ProtectModel, method: str, *args: Any, **kwargs: Any
) -> Generator[MagicMock]:
"""Patch a method on a UniFi Protect pydantic model.
Pydantic models have frozen fields that cannot be directly patched.
This context manager temporarily modifies the field descriptor to allow
patching.
Note: The field modification is intentionally not restored, as test fixtures
create fresh model instances for each test.
Usage:
with patch_ufp_method(doorbell, "set_lcd_text", new_callable=AsyncMock) as mock:
await hass.services.async_call(...)
mock.assert_called_once_with(...)
"""
obj.__pydantic_fields__[method] = Mock(final=False, frozen=False)
with patch.object(obj, method, *args, **kwargs) as mock_method:
yield mock_method
+11 -10
View File
@@ -44,6 +44,7 @@ from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.setup import async_setup_component
from . import patch_ufp_method
from .utils import (
Camera,
MockUFPFixture,
@@ -615,14 +616,14 @@ async def test_camera_motion_detection(
assert_entity_counts(hass, Platform.CAMERA, 2, 1)
entity_id = "camera.test_camera_high_resolution_channel"
camera.__pydantic_fields__["set_motion_detection"] = Mock(final=False, frozen=False)
camera.set_motion_detection = AsyncMock()
with patch_ufp_method(
camera, "set_motion_detection", new_callable=AsyncMock
) as mock_method:
await hass.services.async_call(
"camera",
service,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
await hass.services.async_call(
"camera",
service,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
camera.set_motion_detection.assert_called_once_with(expected_value)
mock_method.assert_called_once_with(expected_value)
+19 -20
View File
@@ -17,6 +17,7 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import patch_ufp_method
from .utils import (
MockUFPFixture,
adopt_devices,
@@ -210,17 +211,17 @@ async def test_lock_do_lock(
await init_entry(hass, ufp, [doorlock, unadopted_doorlock])
assert_entity_counts(hass, Platform.LOCK, 1, 1)
doorlock.__pydantic_fields__["close_lock"] = Mock(final=False, frozen=False)
doorlock.close_lock = AsyncMock()
with patch_ufp_method(
doorlock, "close_lock", new_callable=AsyncMock
) as mock_method:
await hass.services.async_call(
"lock",
"lock",
{ATTR_ENTITY_ID: "lock.test_lock_lock"},
blocking=True,
)
await hass.services.async_call(
"lock",
"lock",
{ATTR_ENTITY_ID: "lock.test_lock_lock"},
blocking=True,
)
doorlock.close_lock.assert_called_once()
mock_method.assert_called_once()
async def test_lock_do_unlock(
@@ -245,14 +246,12 @@ async def test_lock_do_unlock(
ufp.ws_msg(mock_msg)
await hass.async_block_till_done()
doorlock.__pydantic_fields__["open_lock"] = Mock(final=False, frozen=False)
new_lock.open_lock = AsyncMock()
with patch_ufp_method(new_lock, "open_lock", new_callable=AsyncMock) as mock_method:
await hass.services.async_call(
"lock",
"unlock",
{ATTR_ENTITY_ID: "lock.test_lock_lock"},
blocking=True,
)
await hass.services.async_call(
"lock",
"unlock",
{ATTR_ENTITY_ID: "lock.test_lock_lock"},
blocking=True,
)
new_lock.open_lock.assert_called_once()
mock_method.assert_called_once()
@@ -25,6 +25,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from . import patch_ufp_method
from .utils import (
MockUFPFixture,
adopt_devices,
@@ -116,17 +117,17 @@ async def test_media_player_set_volume(
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.MEDIA_PLAYER, 1, 1)
doorbell.__pydantic_fields__["set_speaker_volume"] = Mock(final=False, frozen=False)
doorbell.set_speaker_volume = AsyncMock()
with patch_ufp_method(
doorbell, "set_speaker_volume", new_callable=AsyncMock
) as mock_method:
await hass.services.async_call(
"media_player",
"volume_set",
{ATTR_ENTITY_ID: "media_player.test_camera_speaker", "volume_level": 0.5},
blocking=True,
)
await hass.services.async_call(
"media_player",
"volume_set",
{ATTR_ENTITY_ID: "media_player.test_camera_speaker", "volume_level": 0.5},
blocking=True,
)
doorbell.set_speaker_volume.assert_called_once_with(50)
mock_method.assert_called_once_with(50)
async def test_media_player_stop(
@@ -173,30 +174,26 @@ async def test_media_player_play(
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.MEDIA_PLAYER, 1, 1)
doorbell.__pydantic_fields__["stop_audio"] = Mock(final=False, frozen=False)
doorbell.__pydantic_fields__["play_audio"] = Mock(final=False, frozen=False)
doorbell.__pydantic_fields__["wait_until_audio_completes"] = Mock(
final=False, frozen=False
)
doorbell.stop_audio = AsyncMock()
doorbell.play_audio = AsyncMock()
doorbell.wait_until_audio_completes = AsyncMock()
with (
patch_ufp_method(doorbell, "stop_audio", new_callable=AsyncMock),
patch_ufp_method(doorbell, "play_audio", new_callable=AsyncMock) as mock_play,
patch_ufp_method(
doorbell, "wait_until_audio_completes", new_callable=AsyncMock
) as mock_wait,
):
await hass.services.async_call(
"media_player",
"play_media",
{
ATTR_ENTITY_ID: "media_player.test_camera_speaker",
"media_content_id": "http://example.com/test.mp3",
"media_content_type": "music",
},
blocking=True,
)
await hass.services.async_call(
"media_player",
"play_media",
{
ATTR_ENTITY_ID: "media_player.test_camera_speaker",
"media_content_id": "http://example.com/test.mp3",
"media_content_type": "music",
},
blocking=True,
)
doorbell.play_audio.assert_called_once_with(
"http://example.com/test.mp3", blocking=False
)
doorbell.wait_until_audio_completes.assert_called_once()
mock_play.assert_called_once_with("http://example.com/test.mp3", blocking=False)
mock_wait.assert_called_once()
async def test_media_player_play_media_source(
@@ -210,18 +207,16 @@ async def test_media_player_play_media_source(
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.MEDIA_PLAYER, 1, 1)
doorbell.__pydantic_fields__["stop_audio"] = Mock(final=False, frozen=False)
doorbell.__pydantic_fields__["play_audio"] = Mock(final=False, frozen=False)
doorbell.__pydantic_fields__["wait_until_audio_completes"] = Mock(
final=False, frozen=False
)
doorbell.stop_audio = AsyncMock()
doorbell.play_audio = AsyncMock()
doorbell.wait_until_audio_completes = AsyncMock()
with patch(
"homeassistant.components.media_source.async_resolve_media",
return_value=Mock(url="http://example.com/test.mp3"),
with (
patch_ufp_method(doorbell, "stop_audio", new_callable=AsyncMock),
patch_ufp_method(doorbell, "play_audio", new_callable=AsyncMock) as mock_play,
patch_ufp_method(
doorbell, "wait_until_audio_completes", new_callable=AsyncMock
) as mock_wait,
patch(
"homeassistant.components.media_source.async_resolve_media",
return_value=Mock(url="http://example.com/test.mp3"),
),
):
await hass.services.async_call(
"media_player",
@@ -234,10 +229,8 @@ async def test_media_player_play_media_source(
blocking=True,
)
doorbell.play_audio.assert_called_once_with(
"http://example.com/test.mp3", blocking=False
)
doorbell.wait_until_audio_completes.assert_called_once()
mock_play.assert_called_once_with("http://example.com/test.mp3", blocking=False)
mock_wait.assert_called_once()
async def test_media_player_play_invalid(
@@ -251,22 +244,22 @@ async def test_media_player_play_invalid(
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.MEDIA_PLAYER, 1, 1)
doorbell.__pydantic_fields__["play_audio"] = Mock(final=False, frozen=False)
doorbell.play_audio = AsyncMock()
with patch_ufp_method(
doorbell, "play_audio", new_callable=AsyncMock
) as mock_method:
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
"media_player",
"play_media",
{
ATTR_ENTITY_ID: "media_player.test_camera_speaker",
"media_content_id": "/test.png",
"media_content_type": "image",
},
blocking=True,
)
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
"media_player",
"play_media",
{
ATTR_ENTITY_ID: "media_player.test_camera_speaker",
"media_content_id": "/test.png",
"media_content_type": "image",
},
blocking=True,
)
assert not doorbell.play_audio.called
assert not mock_method.called
async def test_media_player_play_error(
@@ -280,24 +273,25 @@ async def test_media_player_play_error(
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.MEDIA_PLAYER, 1, 1)
doorbell.__pydantic_fields__["play_audio"] = Mock(final=False, frozen=False)
doorbell.__pydantic_fields__["wait_until_audio_completes"] = Mock(
final=False, frozen=False
)
doorbell.play_audio = AsyncMock(side_effect=StreamError)
doorbell.wait_until_audio_completes = AsyncMock()
with (
patch_ufp_method(
doorbell, "play_audio", new_callable=AsyncMock, side_effect=StreamError
) as mock_play,
patch_ufp_method(
doorbell, "wait_until_audio_completes", new_callable=AsyncMock
) as mock_wait,
):
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
"media_player",
"play_media",
{
ATTR_ENTITY_ID: "media_player.test_camera_speaker",
"media_content_id": "/test.mp3",
"media_content_type": "music",
},
blocking=True,
)
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
"media_player",
"play_media",
{
ATTR_ENTITY_ID: "media_player.test_camera_speaker",
"media_content_id": "/test.mp3",
"media_content_type": "music",
},
blocking=True,
)
assert doorbell.play_audio.called
assert not doorbell.wait_until_audio_completes.called
assert mock_play.called
assert not mock_wait.called
+33 -29
View File
@@ -3,7 +3,7 @@
from __future__ import annotations
from datetime import timedelta
from unittest.mock import AsyncMock, Mock, patch
from unittest.mock import AsyncMock
import pytest
from uiprotect.data import Camera, Doorlock, IRLEDMode, Light
@@ -19,6 +19,7 @@ from homeassistant.const import ATTR_ATTRIBUTION, ATTR_ENTITY_ID, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import patch_ufp_method
from .utils import (
MockUFPFixture,
adopt_devices,
@@ -166,18 +167,21 @@ async def test_number_light_sensitivity(
description = LIGHT_NUMBERS[0]
assert description.ufp_set_method is not None
light.__pydantic_fields__["set_sensitivity"] = Mock(final=False, frozen=False)
light.set_sensitivity = AsyncMock()
_, entity_id = await ids_from_device_description(
hass, Platform.NUMBER, light, description
)
await hass.services.async_call(
"number", "set_value", {ATTR_ENTITY_ID: entity_id, "value": 15.0}, blocking=True
)
with patch_ufp_method(
light, "set_sensitivity", new_callable=AsyncMock
) as mock_method:
await hass.services.async_call(
"number",
"set_value",
{ATTR_ENTITY_ID: entity_id, "value": 15.0},
blocking=True,
)
light.set_sensitivity.assert_called_once_with(15.0)
mock_method.assert_called_once_with(15.0)
async def test_number_light_duration(
@@ -190,18 +194,19 @@ async def test_number_light_duration(
description = LIGHT_NUMBERS[1]
light.__pydantic_fields__["set_duration"] = Mock(final=False, frozen=False)
light.set_duration = AsyncMock()
_, entity_id = await ids_from_device_description(
hass, Platform.NUMBER, light, description
)
await hass.services.async_call(
"number", "set_value", {ATTR_ENTITY_ID: entity_id, "value": 15.0}, blocking=True
)
with patch_ufp_method(light, "set_duration", new_callable=AsyncMock) as mock_method:
await hass.services.async_call(
"number",
"set_value",
{ATTR_ENTITY_ID: entity_id, "value": 15.0},
blocking=True,
)
light.set_duration.assert_called_once_with(timedelta(seconds=15.0))
mock_method.assert_called_once_with(timedelta(seconds=15.0))
@pytest.mark.parametrize("description", CAMERA_NUMBERS)
@@ -221,11 +226,9 @@ async def test_number_camera_simple(
hass, Platform.NUMBER, camera_all_features, description
)
camera_all_features.__pydantic_fields__[description.ufp_set_method] = Mock(
final=False, frozen=False
)
mock_method = AsyncMock()
with patch.object(camera_all_features, description.ufp_set_method, mock_method):
with patch_ufp_method(
camera_all_features, description.ufp_set_method, new_callable=AsyncMock
) as mock_method:
await hass.services.async_call(
"number",
"set_value",
@@ -246,17 +249,18 @@ async def test_number_lock_auto_close(
description = DOORLOCK_NUMBERS[0]
doorlock.__pydantic_fields__["set_auto_close_time"] = Mock(
final=False, frozen=False
)
doorlock.set_auto_close_time = AsyncMock()
_, entity_id = await ids_from_device_description(
hass, Platform.NUMBER, doorlock, description
)
await hass.services.async_call(
"number", "set_value", {ATTR_ENTITY_ID: entity_id, "value": 15.0}, blocking=True
)
with patch_ufp_method(
doorlock, "set_auto_close_time", new_callable=AsyncMock
) as mock_method:
await hass.services.async_call(
"number",
"set_value",
{ATTR_ENTITY_ID: entity_id, "value": 15.0},
blocking=True,
)
doorlock.set_auto_close_time.assert_called_once_with(timedelta(seconds=15.0))
mock_method.assert_called_once_with(timedelta(seconds=15.0))
+111 -115
View File
@@ -31,6 +31,7 @@ from homeassistant.const import ATTR_ATTRIBUTION, ATTR_ENTITY_ID, ATTR_OPTION, P
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import patch_ufp_method
from .utils import (
MockUFPFixture,
adopt_devices,
@@ -262,8 +263,6 @@ async def test_select_update_doorbell_settings(
expected_length += 1
new_nvr = copy(ufp.api.bootstrap.nvr)
new_nvr.__pydantic_fields__["update_all_messages"] = Mock(final=False, frozen=False)
new_nvr.update_all_messages = Mock()
new_nvr.doorbell_settings.all_messages = [
*new_nvr.doorbell_settings.all_messages,
@@ -277,11 +276,12 @@ async def test_select_update_doorbell_settings(
mock_msg.changed_data = {"doorbell_settings": {}}
mock_msg.new_obj = new_nvr
ufp.api.bootstrap.nvr = new_nvr
ufp.ws_msg(mock_msg)
await hass.async_block_till_done()
with patch_ufp_method(new_nvr, "update_all_messages") as mock_method:
ufp.api.bootstrap.nvr = new_nvr
ufp.ws_msg(mock_msg)
await hass.async_block_till_done()
new_nvr.update_all_messages.assert_called_once()
mock_method.assert_called_once()
state = hass.states.get(entity_id)
assert state
@@ -334,19 +334,17 @@ async def test_select_set_option_light_motion(
hass, Platform.SELECT, light, LIGHT_SELECTS[0]
)
light.__pydantic_fields__["set_light_settings"] = Mock(final=False, frozen=False)
light.set_light_settings = AsyncMock()
with patch_ufp_method(
light, "set_light_settings", new_callable=AsyncMock
) as mock_method:
await hass.services.async_call(
"select",
"select_option",
{ATTR_ENTITY_ID: entity_id, ATTR_OPTION: LIGHT_MODE_OFF},
blocking=True,
)
await hass.services.async_call(
"select",
"select_option",
{ATTR_ENTITY_ID: entity_id, ATTR_OPTION: LIGHT_MODE_OFF},
blocking=True,
)
light.set_light_settings.assert_called_once_with(
LightModeType.MANUAL, enable_at=None
)
mock_method.assert_called_once_with(LightModeType.MANUAL, enable_at=None)
async def test_select_set_option_light_camera(
@@ -361,28 +359,28 @@ async def test_select_set_option_light_camera(
hass, Platform.SELECT, light, LIGHT_SELECTS[1]
)
light.__pydantic_fields__["set_paired_camera"] = Mock(final=False, frozen=False)
light.set_paired_camera = AsyncMock()
camera = list(light.api.bootstrap.cameras.values())[0]
await hass.services.async_call(
"select",
"select_option",
{ATTR_ENTITY_ID: entity_id, ATTR_OPTION: camera.name},
blocking=True,
)
with patch_ufp_method(
light, "set_paired_camera", new_callable=AsyncMock
) as mock_method:
await hass.services.async_call(
"select",
"select_option",
{ATTR_ENTITY_ID: entity_id, ATTR_OPTION: camera.name},
blocking=True,
)
light.set_paired_camera.assert_called_once_with(camera)
mock_method.assert_called_once_with(camera)
await hass.services.async_call(
"select",
"select_option",
{ATTR_ENTITY_ID: entity_id, ATTR_OPTION: "Not Paired"},
blocking=True,
)
await hass.services.async_call(
"select",
"select_option",
{ATTR_ENTITY_ID: entity_id, ATTR_OPTION: "Not Paired"},
blocking=True,
)
light.set_paired_camera.assert_called_with(None)
mock_method.assert_called_with(None)
async def test_select_set_option_camera_recording(
@@ -397,17 +395,17 @@ async def test_select_set_option_camera_recording(
hass, Platform.SELECT, doorbell, CAMERA_SELECTS[0]
)
doorbell.__pydantic_fields__["set_recording_mode"] = Mock(final=False, frozen=False)
doorbell.set_recording_mode = AsyncMock()
with patch_ufp_method(
doorbell, "set_recording_mode", new_callable=AsyncMock
) as mock_method:
await hass.services.async_call(
"select",
"select_option",
{ATTR_ENTITY_ID: entity_id, ATTR_OPTION: "never"},
blocking=True,
)
await hass.services.async_call(
"select",
"select_option",
{ATTR_ENTITY_ID: entity_id, ATTR_OPTION: "never"},
blocking=True,
)
doorbell.set_recording_mode.assert_called_once_with(RecordingMode.NEVER)
mock_method.assert_called_once_with(RecordingMode.NEVER)
async def test_select_set_option_camera_ir(
@@ -422,17 +420,17 @@ async def test_select_set_option_camera_ir(
hass, Platform.SELECT, doorbell, CAMERA_SELECTS[1]
)
doorbell.__pydantic_fields__["set_ir_led_model"] = Mock(final=False, frozen=False)
doorbell.set_ir_led_model = AsyncMock()
with patch_ufp_method(
doorbell, "set_ir_led_model", new_callable=AsyncMock
) as mock_method:
await hass.services.async_call(
"select",
"select_option",
{ATTR_ENTITY_ID: entity_id, ATTR_OPTION: "on"},
blocking=True,
)
await hass.services.async_call(
"select",
"select_option",
{ATTR_ENTITY_ID: entity_id, ATTR_OPTION: "on"},
blocking=True,
)
doorbell.set_ir_led_model.assert_called_once_with(IRLEDMode.ON)
mock_method.assert_called_once_with(IRLEDMode.ON)
async def test_select_set_option_camera_doorbell_custom(
@@ -447,19 +445,19 @@ async def test_select_set_option_camera_doorbell_custom(
hass, Platform.SELECT, doorbell, CAMERA_SELECTS[2]
)
doorbell.__pydantic_fields__["set_lcd_text"] = Mock(final=False, frozen=False)
doorbell.set_lcd_text = AsyncMock()
with patch_ufp_method(
doorbell, "set_lcd_text", new_callable=AsyncMock
) as mock_method:
await hass.services.async_call(
"select",
"select_option",
{ATTR_ENTITY_ID: entity_id, ATTR_OPTION: "Test"},
blocking=True,
)
await hass.services.async_call(
"select",
"select_option",
{ATTR_ENTITY_ID: entity_id, ATTR_OPTION: "Test"},
blocking=True,
)
doorbell.set_lcd_text.assert_called_once_with(
DoorbellMessageType.CUSTOM_MESSAGE, text="Test"
)
mock_method.assert_called_once_with(
DoorbellMessageType.CUSTOM_MESSAGE, text="Test"
)
async def test_select_set_option_camera_doorbell_unifi(
@@ -474,34 +472,32 @@ async def test_select_set_option_camera_doorbell_unifi(
hass, Platform.SELECT, doorbell, CAMERA_SELECTS[2]
)
doorbell.__pydantic_fields__["set_lcd_text"] = Mock(final=False, frozen=False)
doorbell.set_lcd_text = AsyncMock()
with patch_ufp_method(
doorbell, "set_lcd_text", new_callable=AsyncMock
) as mock_method:
await hass.services.async_call(
"select",
"select_option",
{
ATTR_ENTITY_ID: entity_id,
ATTR_OPTION: "LEAVE PACKAGE AT DOOR",
},
blocking=True,
)
await hass.services.async_call(
"select",
"select_option",
{
ATTR_ENTITY_ID: entity_id,
ATTR_OPTION: "LEAVE PACKAGE AT DOOR",
},
blocking=True,
)
mock_method.assert_called_once_with(DoorbellMessageType.LEAVE_PACKAGE_AT_DOOR)
doorbell.set_lcd_text.assert_called_once_with(
DoorbellMessageType.LEAVE_PACKAGE_AT_DOOR
)
await hass.services.async_call(
"select",
"select_option",
{
ATTR_ENTITY_ID: entity_id,
ATTR_OPTION: "Default Message (Welcome)",
},
blocking=True,
)
await hass.services.async_call(
"select",
"select_option",
{
ATTR_ENTITY_ID: entity_id,
ATTR_OPTION: "Default Message (Welcome)",
},
blocking=True,
)
doorbell.set_lcd_text.assert_called_with(None)
mock_method.assert_called_with(None)
async def test_select_set_option_camera_doorbell_default(
@@ -516,20 +512,20 @@ async def test_select_set_option_camera_doorbell_default(
hass, Platform.SELECT, doorbell, CAMERA_SELECTS[2]
)
doorbell.__pydantic_fields__["set_lcd_text"] = Mock(final=False, frozen=False)
doorbell.set_lcd_text = AsyncMock()
with patch_ufp_method(
doorbell, "set_lcd_text", new_callable=AsyncMock
) as mock_method:
await hass.services.async_call(
"select",
"select_option",
{
ATTR_ENTITY_ID: entity_id,
ATTR_OPTION: "Default Message (Welcome)",
},
blocking=True,
)
await hass.services.async_call(
"select",
"select_option",
{
ATTR_ENTITY_ID: entity_id,
ATTR_OPTION: "Default Message (Welcome)",
},
blocking=True,
)
doorbell.set_lcd_text.assert_called_once_with(None)
mock_method.assert_called_once_with(None)
async def test_select_set_option_viewer(
@@ -545,16 +541,16 @@ async def test_select_set_option_viewer(
hass, Platform.SELECT, viewer, VIEWER_SELECTS[0]
)
viewer.__pydantic_fields__["set_liveview"] = Mock(final=False, frozen=False)
viewer.set_liveview = AsyncMock()
liveview = list(viewer.api.bootstrap.liveviews.values())[0]
await hass.services.async_call(
"select",
"select_option",
{ATTR_ENTITY_ID: entity_id, ATTR_OPTION: liveview.name},
blocking=True,
)
with patch_ufp_method(
viewer, "set_liveview", new_callable=AsyncMock
) as mock_method:
await hass.services.async_call(
"select",
"select_option",
{ATTR_ENTITY_ID: entity_id, ATTR_OPTION: liveview.name},
blocking=True,
)
viewer.set_liveview.assert_called_once_with(liveview)
mock_method.assert_called_once_with(liveview)
+57 -58
View File
@@ -32,6 +32,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr, entity_registry as er
from . import patch_ufp_method
from .utils import MockUFPFixture, init_entry
@@ -66,19 +67,18 @@ async def test_global_service_bad_device(
"""Test global service, invalid device ID."""
nvr = ufp.api.bootstrap.nvr
nvr.__pydantic_fields__["add_custom_doorbell_message"] = Mock(
final=False, frozen=False
)
nvr.add_custom_doorbell_message = AsyncMock()
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
DOMAIN,
SERVICE_ADD_DOORBELL_TEXT,
{ATTR_DEVICE_ID: "bad_device_id", ATTR_MESSAGE: "Test Message"},
blocking=True,
)
assert not nvr.add_custom_doorbell_message.called
with patch_ufp_method(
nvr, "add_custom_doorbell_message", new_callable=AsyncMock
) as mock_method:
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
DOMAIN,
SERVICE_ADD_DOORBELL_TEXT,
{ATTR_DEVICE_ID: "bad_device_id", ATTR_MESSAGE: "Test Message"},
blocking=True,
)
assert not mock_method.called
async def test_global_service_exception(
@@ -87,19 +87,21 @@ async def test_global_service_exception(
"""Test global service, unexpected error."""
nvr = ufp.api.bootstrap.nvr
nvr.__pydantic_fields__["add_custom_doorbell_message"] = Mock(
final=False, frozen=False
)
nvr.add_custom_doorbell_message = AsyncMock(side_effect=BadRequest)
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
DOMAIN,
SERVICE_ADD_DOORBELL_TEXT,
{ATTR_DEVICE_ID: device.id, ATTR_MESSAGE: "Test Message"},
blocking=True,
)
assert nvr.add_custom_doorbell_message.called
with patch_ufp_method(
nvr,
"add_custom_doorbell_message",
new_callable=AsyncMock,
side_effect=BadRequest,
) as mock_method:
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
DOMAIN,
SERVICE_ADD_DOORBELL_TEXT,
{ATTR_DEVICE_ID: device.id, ATTR_MESSAGE: "Test Message"},
blocking=True,
)
assert mock_method.called
async def test_add_doorbell_text(
@@ -108,18 +110,17 @@ async def test_add_doorbell_text(
"""Test add_doorbell_text service."""
nvr = ufp.api.bootstrap.nvr
nvr.__pydantic_fields__["add_custom_doorbell_message"] = Mock(
final=False, frozen=False
)
nvr.add_custom_doorbell_message = AsyncMock()
await hass.services.async_call(
DOMAIN,
SERVICE_ADD_DOORBELL_TEXT,
{ATTR_DEVICE_ID: device.id, ATTR_MESSAGE: "Test Message"},
blocking=True,
)
nvr.add_custom_doorbell_message.assert_called_once_with("Test Message")
with patch_ufp_method(
nvr, "add_custom_doorbell_message", new_callable=AsyncMock
) as mock_method:
await hass.services.async_call(
DOMAIN,
SERVICE_ADD_DOORBELL_TEXT,
{ATTR_DEVICE_ID: device.id, ATTR_MESSAGE: "Test Message"},
blocking=True,
)
mock_method.assert_called_once_with("Test Message")
async def test_remove_doorbell_text(
@@ -128,18 +129,17 @@ async def test_remove_doorbell_text(
"""Test remove_doorbell_text service."""
nvr = ufp.api.bootstrap.nvr
nvr.__pydantic_fields__["remove_custom_doorbell_message"] = Mock(
final=False, frozen=False
)
nvr.remove_custom_doorbell_message = AsyncMock()
await hass.services.async_call(
DOMAIN,
SERVICE_REMOVE_DOORBELL_TEXT,
{ATTR_DEVICE_ID: subdevice.id, ATTR_MESSAGE: "Test Message"},
blocking=True,
)
nvr.remove_custom_doorbell_message.assert_called_once_with("Test Message")
with patch_ufp_method(
nvr, "remove_custom_doorbell_message", new_callable=AsyncMock
) as mock_method:
await hass.services.async_call(
DOMAIN,
SERVICE_REMOVE_DOORBELL_TEXT,
{ATTR_DEVICE_ID: subdevice.id, ATTR_MESSAGE: "Test Message"},
blocking=True,
)
mock_method.assert_called_once_with("Test Message")
async def test_add_doorbell_text_disabled_config_entry(
@@ -147,24 +147,23 @@ async def test_add_doorbell_text_disabled_config_entry(
) -> None:
"""Test add_doorbell_text service."""
nvr = ufp.api.bootstrap.nvr
nvr.__pydantic_fields__["add_custom_doorbell_message"] = Mock(
final=False, frozen=False
)
nvr.add_custom_doorbell_message = AsyncMock()
await hass.config_entries.async_set_disabled_by(
ufp.entry.entry_id, ConfigEntryDisabler.USER
)
await hass.async_block_till_done()
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
DOMAIN,
SERVICE_ADD_DOORBELL_TEXT,
{ATTR_DEVICE_ID: device.id, ATTR_MESSAGE: "Test Message"},
blocking=True,
)
assert not nvr.add_custom_doorbell_message.called
with patch_ufp_method(
nvr, "add_custom_doorbell_message", new_callable=AsyncMock
) as mock_method:
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
DOMAIN,
SERVICE_ADD_DOORBELL_TEXT,
{ATTR_DEVICE_ID: device.id, ATTR_MESSAGE: "Test Message"},
blocking=True,
)
assert not mock_method.called
async def test_set_chime_paired_doorbells(
+95 -89
View File
@@ -2,7 +2,7 @@
from __future__ import annotations
from unittest.mock import AsyncMock, Mock, patch
from unittest.mock import AsyncMock, Mock
import pytest
from uiprotect.data import Camera, Light, Permission, RecordingMode, VideoMode
@@ -22,6 +22,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from . import patch_ufp_method
from .utils import (
MockUFPFixture,
adopt_devices,
@@ -90,21 +91,20 @@ async def test_switch_nvr(hass: HomeAssistant, ufp: MockUFPFixture) -> None:
assert_entity_counts(hass, Platform.SWITCH, 2, 2)
nvr = ufp.api.bootstrap.nvr
nvr.__pydantic_fields__["set_insights"] = Mock(final=False, frozen=False)
nvr.set_insights = AsyncMock()
entity_id = "switch.unifiprotect_insights_enabled"
await hass.services.async_call(
"switch", "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
with patch_ufp_method(nvr, "set_insights", new_callable=AsyncMock) as mock_method:
await hass.services.async_call(
"switch", "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
nvr.set_insights.assert_called_once_with(True)
mock_method.assert_called_once_with(True)
await hass.services.async_call(
"switch", "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
await hass.services.async_call(
"switch", "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
nvr.set_insights.assert_called_with(False)
mock_method.assert_called_with(False)
async def test_switch_setup_no_perm(
@@ -267,24 +267,24 @@ async def test_switch_light_status(
description = LIGHT_SWITCHES[1]
light.__pydantic_fields__["set_status_light"] = Mock(final=False, frozen=False)
light.set_status_light = AsyncMock()
_, entity_id = await ids_from_device_description(
hass, Platform.SWITCH, light, description
)
await hass.services.async_call(
"switch", "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
with patch_ufp_method(
light, "set_status_light", new_callable=AsyncMock
) as mock_method:
await hass.services.async_call(
"switch", "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
light.set_status_light.assert_called_once_with(True)
mock_method.assert_called_once_with(True)
await hass.services.async_call(
"switch", "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
await hass.services.async_call(
"switch", "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
light.set_status_light.assert_called_with(False)
mock_method.assert_called_with(False)
async def test_switch_camera_ssh(
@@ -297,25 +297,23 @@ async def test_switch_camera_ssh(
description = CAMERA_SWITCHES[0]
doorbell.__pydantic_fields__["set_ssh"] = Mock(final=False, frozen=False)
doorbell.set_ssh = AsyncMock()
_, entity_id = await ids_from_device_description(
hass, Platform.SWITCH, doorbell, description
)
await enable_entity(hass, ufp.entry.entry_id, entity_id)
await hass.services.async_call(
"switch", "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
with patch_ufp_method(doorbell, "set_ssh", new_callable=AsyncMock) as mock_method:
await hass.services.async_call(
"switch", "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
doorbell.set_ssh.assert_called_once_with(True)
mock_method.assert_called_once_with(True)
await hass.services.async_call(
"switch", "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
await hass.services.async_call(
"switch", "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
doorbell.set_ssh.assert_called_with(False)
mock_method.assert_called_with(False)
@pytest.mark.parametrize("description", CAMERA_SWITCHES_NO_EXTRA)
@@ -332,11 +330,9 @@ async def test_switch_camera_simple(
assert description.ufp_set_method is not None
doorbell.__pydantic_fields__[description.ufp_set_method] = Mock(
final=False, frozen=False
)
mock_method = AsyncMock()
with patch.object(doorbell, description.ufp_set_method, mock_method):
with patch_ufp_method(
doorbell, description.ufp_set_method, new_callable=AsyncMock
) as mock_method:
_, entity_id = await ids_from_device_description(
hass, Platform.SWITCH, doorbell, description
)
@@ -364,24 +360,24 @@ async def test_switch_camera_highfps(
description = CAMERA_SWITCHES[3]
doorbell.__pydantic_fields__["set_video_mode"] = Mock(final=False, frozen=False)
doorbell.set_video_mode = AsyncMock()
_, entity_id = await ids_from_device_description(
hass, Platform.SWITCH, doorbell, description
)
await hass.services.async_call(
"switch", "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
with patch_ufp_method(
doorbell, "set_video_mode", new_callable=AsyncMock
) as mock_method:
await hass.services.async_call(
"switch", "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
doorbell.set_video_mode.assert_called_once_with(VideoMode.HIGH_FPS)
mock_method.assert_called_once_with(VideoMode.HIGH_FPS)
await hass.services.async_call(
"switch", "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
await hass.services.async_call(
"switch", "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
doorbell.set_video_mode.assert_called_with(VideoMode.DEFAULT)
mock_method.assert_called_with(VideoMode.DEFAULT)
async def test_switch_camera_privacy(
@@ -397,9 +393,6 @@ async def test_switch_camera_privacy(
description = PRIVACY_MODE_SWITCH
doorbell.__pydantic_fields__["set_privacy"] = Mock(final=False, frozen=False)
doorbell.set_privacy = AsyncMock()
_, entity_id = await ids_from_device_description(
hass, Platform.SWITCH, doorbell, description
)
@@ -409,35 +402,38 @@ async def test_switch_camera_privacy(
assert ATTR_PREV_MIC not in state.attributes
assert ATTR_PREV_RECORD not in state.attributes
await hass.services.async_call(
"switch", "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
with patch_ufp_method(
doorbell, "set_privacy", new_callable=AsyncMock
) as mock_set_privacy:
await hass.services.async_call(
"switch", "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
doorbell.set_privacy.assert_called_with(True, 0, RecordingMode.NEVER)
mock_set_privacy.assert_called_with(True, 0, RecordingMode.NEVER)
new_doorbell = doorbell.model_copy()
new_doorbell.add_privacy_zone()
new_doorbell.mic_volume = 0
new_doorbell.recording_settings.mode = RecordingMode.NEVER
ufp.api.bootstrap.cameras = {new_doorbell.id: new_doorbell}
new_doorbell = doorbell.model_copy()
new_doorbell.add_privacy_zone()
new_doorbell.mic_volume = 0
new_doorbell.recording_settings.mode = RecordingMode.NEVER
ufp.api.bootstrap.cameras = {new_doorbell.id: new_doorbell}
mock_msg = Mock()
mock_msg.changed_data = {}
mock_msg.new_obj = new_doorbell
ufp.ws_msg(mock_msg)
mock_msg = Mock()
mock_msg.changed_data = {}
mock_msg.new_obj = new_doorbell
ufp.ws_msg(mock_msg)
state = hass.states.get(entity_id)
assert state and state.state == "on"
assert state.attributes[ATTR_PREV_MIC] == previous_mic
assert state.attributes[ATTR_PREV_RECORD] == previous_record.value
state = hass.states.get(entity_id)
assert state and state.state == "on"
assert state.attributes[ATTR_PREV_MIC] == previous_mic
assert state.attributes[ATTR_PREV_RECORD] == previous_record.value
doorbell.set_privacy.reset_mock()
mock_set_privacy.reset_mock()
await hass.services.async_call(
"switch", "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
await hass.services.async_call(
"switch", "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
doorbell.set_privacy.assert_called_with(False, previous_mic, previous_record)
mock_set_privacy.assert_called_with(False, previous_mic, previous_record)
async def test_switch_camera_privacy_already_on(
@@ -451,18 +447,18 @@ async def test_switch_camera_privacy_already_on(
description = PRIVACY_MODE_SWITCH
doorbell.__pydantic_fields__["set_privacy"] = Mock(final=False, frozen=False)
doorbell.set_privacy = AsyncMock()
_, entity_id = await ids_from_device_description(
hass, Platform.SWITCH, doorbell, description
)
await hass.services.async_call(
"switch", "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
with patch_ufp_method(
doorbell, "set_privacy", new_callable=AsyncMock
) as mock_set_privacy:
await hass.services.async_call(
"switch", "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
doorbell.set_privacy.assert_called_once_with(False, 100, RecordingMode.ALWAYS)
mock_set_privacy.assert_called_once_with(False, 100, RecordingMode.ALWAYS)
async def test_switch_turn_on_client_error(
@@ -474,14 +470,19 @@ async def test_switch_turn_on_client_error(
description = LIGHT_SWITCHES[1]
light.__pydantic_fields__["set_status_light"] = Mock(final=False, frozen=False)
light.set_status_light = AsyncMock(side_effect=ClientError("Test error"))
_, entity_id = await ids_from_device_description(
hass, Platform.SWITCH, light, description
)
with pytest.raises(HomeAssistantError):
with (
patch_ufp_method(
light,
"set_status_light",
new_callable=AsyncMock,
side_effect=ClientError("Test error"),
),
pytest.raises(HomeAssistantError),
):
await hass.services.async_call(
"switch", "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
@@ -496,14 +497,19 @@ async def test_switch_turn_on_not_authorized(
description = LIGHT_SWITCHES[1]
light.__pydantic_fields__["set_status_light"] = Mock(final=False, frozen=False)
light.set_status_light = AsyncMock(side_effect=NotAuthorized("Not authorized"))
_, entity_id = await ids_from_device_description(
hass, Platform.SWITCH, light, description
)
with pytest.raises(HomeAssistantError):
with (
patch_ufp_method(
light,
"set_status_light",
new_callable=AsyncMock,
side_effect=NotAuthorized("Not authorized"),
),
pytest.raises(HomeAssistantError),
):
await hass.services.async_call(
"switch", "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
+14 -13
View File
@@ -2,7 +2,7 @@
from __future__ import annotations
from unittest.mock import AsyncMock, Mock
from unittest.mock import AsyncMock
from uiprotect.data import Camera, DoorbellMessageType, LCDMessage
@@ -12,6 +12,7 @@ from homeassistant.const import ATTR_ATTRIBUTION, ATTR_ENTITY_ID, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import patch_ufp_method
from .utils import (
MockUFPFixture,
adopt_devices,
@@ -78,16 +79,16 @@ async def test_text_camera_set(
hass, Platform.TEXT, doorbell, description
)
doorbell.__pydantic_fields__["set_lcd_text"] = Mock(final=False, frozen=False)
doorbell.set_lcd_text = AsyncMock()
with patch_ufp_method(
doorbell, "set_lcd_text", new_callable=AsyncMock
) as mock_method:
await hass.services.async_call(
"text",
"set_value",
{ATTR_ENTITY_ID: entity_id, "value": "Test test"},
blocking=True,
)
await hass.services.async_call(
"text",
"set_value",
{ATTR_ENTITY_ID: entity_id, "value": "Test test"},
blocking=True,
)
doorbell.set_lcd_text.assert_called_once_with(
DoorbellMessageType.CUSTOM_MESSAGE, text="Test test"
)
mock_method.assert_called_once_with(
DoorbellMessageType.CUSTOM_MESSAGE, text="Test test"
)