1
0
mirror of https://github.com/home-assistant/core.git synced 2026-04-19 00:10:21 +01:00

Add ZoneMinder integration test suite (#163115)

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
Co-authored-by: Joostlek <joostlek@outlook.com>
This commit is contained in:
nic
2026-02-24 16:27:13 -06:00
committed by GitHub
parent e505ad9003
commit bc324a1a6e
11 changed files with 1665 additions and 0 deletions

1
CODEOWNERS generated
View File

@@ -1966,6 +1966,7 @@ build.json @home-assistant/supervisor
/homeassistant/components/zone/ @home-assistant/core
/tests/components/zone/ @home-assistant/core
/homeassistant/components/zoneminder/ @rohankapoorcom @nabbi
/tests/components/zoneminder/ @rohankapoorcom @nabbi
/homeassistant/components/zwave_js/ @home-assistant/z-wave
/tests/components/zwave_js/ @home-assistant/z-wave
/homeassistant/components/zwave_me/ @lawfulchaos @Z-Wave-Me @PoltoS

View File

@@ -2819,6 +2819,9 @@ zha==0.0.90
# homeassistant.components.zinvolt
zinvolt==0.1.0
# homeassistant.components.zoneminder
zm-py==0.5.4
# homeassistant.components.zwave_js
zwave-js-server-python==0.68.0

View File

@@ -0,0 +1 @@
"""Tests for the ZoneMinder integration."""

View File

@@ -0,0 +1,230 @@
"""Shared fixtures for ZoneMinder integration tests."""
from __future__ import annotations
from collections.abc import Generator
from unittest.mock import MagicMock, PropertyMock, patch
import pytest
from zoneminder.monitor import MonitorState, TimePeriod
from homeassistant.components.zoneminder.const import DOMAIN
from homeassistant.const import (
CONF_HOST,
CONF_PASSWORD,
CONF_PATH,
CONF_SSL,
CONF_USERNAME,
CONF_VERIFY_SSL,
)
CONF_PATH_ZMS = "path_zms"
MOCK_HOST = "zm.example.com"
MOCK_HOST_2 = "zm2.example.com"
@pytest.fixture
def single_server_config() -> dict:
"""Return minimal single ZM server YAML config."""
return {
DOMAIN: [
{
CONF_HOST: MOCK_HOST,
CONF_USERNAME: "admin",
CONF_PASSWORD: "secret",
}
]
}
@pytest.fixture
def multi_server_config() -> dict:
"""Return two ZM servers with different settings."""
return {
DOMAIN: [
{
CONF_HOST: MOCK_HOST,
CONF_USERNAME: "admin",
CONF_PASSWORD: "secret",
},
{
CONF_HOST: MOCK_HOST_2,
CONF_USERNAME: "user2",
CONF_PASSWORD: "pass2",
CONF_SSL: True,
CONF_VERIFY_SSL: False,
CONF_PATH: "/zoneminder/",
CONF_PATH_ZMS: "/zoneminder/cgi-bin/nph-zms",
},
]
}
@pytest.fixture
def no_auth_config() -> dict:
"""Return server config without username/password."""
return {
DOMAIN: [
{
CONF_HOST: MOCK_HOST,
}
]
}
@pytest.fixture
def ssl_config() -> dict:
"""Return server config with SSL enabled, verify_ssl disabled."""
return {
DOMAIN: [
{
CONF_HOST: MOCK_HOST,
CONF_SSL: True,
CONF_VERIFY_SSL: False,
CONF_USERNAME: "admin",
CONF_PASSWORD: "secret",
}
]
}
def create_mock_monitor(
monitor_id: int = 1,
name: str = "Front Door",
function: MonitorState = MonitorState.MODECT,
is_recording: bool = False,
is_available: bool = True,
mjpeg_image_url: str = "http://zm.example.com/mjpeg/1",
still_image_url: str = "http://zm.example.com/still/1",
events: dict[TimePeriod, int | None] | None = None,
) -> MagicMock:
"""Create a mock Monitor instance with configurable properties."""
monitor = MagicMock()
monitor.id = monitor_id
monitor.name = name
# function is both a property and a settable attribute in zm-py
monitor.function = function
monitor.is_recording = is_recording
monitor.is_available = is_available
monitor.mjpeg_image_url = mjpeg_image_url
monitor.still_image_url = still_image_url
if events is None:
events = {
TimePeriod.ALL: 100,
TimePeriod.HOUR: 5,
TimePeriod.DAY: 20,
TimePeriod.WEEK: 50,
TimePeriod.MONTH: 80,
}
def mock_get_events(time_period, include_archived=False):
return events.get(time_period, 0)
monitor.get_events = MagicMock(side_effect=mock_get_events)
return monitor
@pytest.fixture
def mock_monitor():
"""Factory fixture returning a function to create mock Monitor instances."""
return create_mock_monitor
@pytest.fixture
def two_monitors():
"""Pre-built list of 2 monitors."""
return [
create_mock_monitor(
monitor_id=1,
name="Front Door",
function=MonitorState.MODECT,
is_recording=True,
is_available=True,
),
create_mock_monitor(
monitor_id=2,
name="Back Yard",
function=MonitorState.MONITOR,
is_recording=False,
is_available=True,
),
]
def create_mock_zm_client(
is_available: bool = True,
verify_ssl: bool = True,
monitors: list | None = None,
login_success: bool = True,
active_state: str | None = "Running",
) -> MagicMock:
"""Create a mock ZoneMinder client."""
client = MagicMock()
client.login.return_value = login_success
client.get_monitors.return_value = monitors or []
# is_available and verify_ssl are properties in zm-py
type(client).is_available = PropertyMock(return_value=is_available)
type(client).verify_ssl = PropertyMock(return_value=verify_ssl)
client.get_active_state.return_value = active_state
client.set_active_state.return_value = True
return client
@pytest.fixture
def mock_zoneminder_client(two_monitors: list[MagicMock]) -> Generator[MagicMock]:
"""Mock a ZoneMinder client."""
with patch(
"homeassistant.components.zoneminder.ZoneMinder",
autospec=True,
) as mock_cls:
client = mock_cls.return_value
client.login.return_value = True
client.get_monitors.return_value = two_monitors
client.get_active_state.return_value = "Running"
client.set_active_state.return_value = True
# is_available and verify_ssl are properties in zm-py
type(client).is_available = PropertyMock(return_value=True)
type(client).verify_ssl = PropertyMock(return_value=True)
# Expose the class mock so tests can inspect constructor call_args
# without needing their own inline patch block.
client.mock_cls = mock_cls
yield client
@pytest.fixture
def sensor_platform_config(single_server_config) -> dict:
"""Return sensor platform YAML with all monitored_conditions."""
config = dict(single_server_config)
config["sensor"] = [
{
"platform": DOMAIN,
"include_archived": True,
"monitored_conditions": ["all", "hour", "day", "week", "month"],
}
]
return config
@pytest.fixture
def switch_platform_config(single_server_config) -> dict:
"""Return switch platform YAML with command_on=Modect, command_off=Monitor."""
config = dict(single_server_config)
config["switch"] = [
{
"platform": DOMAIN,
"command_on": "Modect",
"command_off": "Monitor",
}
]
return config

View File

@@ -0,0 +1,150 @@
"""Tests for ZoneMinder binary sensor entity states (public API)."""
from __future__ import annotations
from datetime import timedelta
from unittest.mock import MagicMock, PropertyMock
from freezegun.api import FrozenDateTimeFactory
from homeassistant.components.binary_sensor import BinarySensorDeviceClass
from homeassistant.components.zoneminder.const import DOMAIN
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from .conftest import MOCK_HOST, MOCK_HOST_2
from tests.common import async_fire_time_changed
# The entity_id uses the hostname with dots replaced by underscores
ENTITY_ID = f"binary_sensor.{MOCK_HOST.replace('.', '_')}"
ENTITY_ID_2 = f"binary_sensor.{MOCK_HOST_2.replace('.', '_')}"
async def _setup_and_update(
hass: HomeAssistant,
config: dict,
mock_zoneminder_client: MagicMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Set up ZM component and trigger first entity update."""
assert await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done(wait_background_tasks=True)
# Trigger the first update poll while mock is still active
freezer.tick(timedelta(seconds=60))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
async def test_binary_sensor_created_per_server(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test one binary sensor entity is created per ZM server."""
await _setup_and_update(hass, single_server_config, mock_zoneminder_client, freezer)
state = hass.states.get(ENTITY_ID)
assert state is not None
async def test_binary_sensor_name_from_hostname(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test binary sensor entity name matches hostname."""
await _setup_and_update(hass, single_server_config, mock_zoneminder_client, freezer)
state = hass.states.get(ENTITY_ID)
assert state is not None
assert state.name == MOCK_HOST
async def test_binary_sensor_device_class_connectivity(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test binary sensor has connectivity device class."""
await _setup_and_update(hass, single_server_config, mock_zoneminder_client, freezer)
state = hass.states.get(ENTITY_ID)
assert state is not None
assert state.attributes.get("device_class") == BinarySensorDeviceClass.CONNECTIVITY
async def test_binary_sensor_state_on_when_available(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test binary sensor state is ON when server is available."""
type(mock_zoneminder_client).is_available = PropertyMock(return_value=True)
await _setup_and_update(hass, single_server_config, mock_zoneminder_client, freezer)
state = hass.states.get(ENTITY_ID)
assert state is not None
assert state.state == STATE_ON
async def test_binary_sensor_state_off_when_unavailable(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test binary sensor state is OFF when server is unavailable."""
type(mock_zoneminder_client).is_available = PropertyMock(return_value=False)
await _setup_and_update(hass, single_server_config, mock_zoneminder_client, freezer)
state = hass.states.get(ENTITY_ID)
assert state is not None
assert state.state == STATE_OFF
async def test_multi_server_creates_multiple_binary_sensors(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
multi_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test multi-server config creates multiple binary sensor entities."""
assert await async_setup_component(hass, DOMAIN, multi_server_config)
await hass.async_block_till_done(wait_background_tasks=True)
freezer.tick(timedelta(seconds=60))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
assert hass.states.get(ENTITY_ID) is not None
assert hass.states.get(ENTITY_ID_2) is not None
async def test_binary_sensor_state_updates_on_poll(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test binary sensor state updates when polled."""
type(mock_zoneminder_client).is_available = PropertyMock(return_value=True)
await _setup_and_update(hass, single_server_config, mock_zoneminder_client, freezer)
state = hass.states.get(ENTITY_ID)
assert state is not None
assert state.state == STATE_ON
# Change availability and trigger another update
type(mock_zoneminder_client).is_available = PropertyMock(return_value=False)
freezer.tick(timedelta(seconds=60))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
state = hass.states.get(ENTITY_ID)
assert state is not None
assert state.state == STATE_OFF

View File

@@ -0,0 +1,209 @@
"""Tests for ZoneMinder camera entities."""
from __future__ import annotations
from datetime import timedelta
from unittest.mock import MagicMock, PropertyMock, patch
from freezegun.api import FrozenDateTimeFactory
from homeassistant.components.camera import CameraState
from homeassistant.components.zoneminder.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from .conftest import create_mock_monitor, create_mock_zm_client
from tests.common import async_fire_time_changed
async def _setup_zm_with_cameras(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
config: dict,
monitors: list,
freezer: FrozenDateTimeFactory,
verify_ssl: bool = True,
) -> None:
"""Set up ZM component with camera platform and given monitors."""
mock_zoneminder_client.get_monitors.return_value = monitors
type(mock_zoneminder_client).verify_ssl = PropertyMock(return_value=verify_ssl)
assert await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done(wait_background_tasks=True)
# Camera uses setup_platform (sync), add camera platform explicitly
assert await async_setup_component(
hass,
"camera",
{"camera": [{"platform": DOMAIN}]},
)
await hass.async_block_till_done(wait_background_tasks=True)
# Trigger first poll to update entity state
freezer.tick(timedelta(seconds=60))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
async def test_one_camera_per_monitor(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
two_monitors: list,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test one camera entity is created per monitor."""
await _setup_zm_with_cameras(
hass, mock_zoneminder_client, single_server_config, two_monitors, freezer
)
states = hass.states.async_all("camera")
assert len(states) == 2
async def test_camera_entity_name(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test camera entity name matches monitor name."""
monitors = [create_mock_monitor(name="Front Door")]
await _setup_zm_with_cameras(
hass, mock_zoneminder_client, single_server_config, monitors, freezer
)
state = hass.states.get("camera.front_door")
assert state is not None
assert state.name == "Front Door"
async def test_camera_recording_state(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test camera recording state reflects monitor is_recording."""
monitors = [
create_mock_monitor(name="Recording Cam", is_recording=True, is_available=True)
]
await _setup_zm_with_cameras(
hass, mock_zoneminder_client, single_server_config, monitors, freezer
)
state = hass.states.get("camera.recording_cam")
assert state is not None
assert state.state == CameraState.RECORDING
async def test_camera_idle_state(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test camera idle state when not recording."""
monitors = [
create_mock_monitor(name="Idle Cam", is_recording=False, is_available=True)
]
await _setup_zm_with_cameras(
hass, mock_zoneminder_client, single_server_config, monitors, freezer
)
state = hass.states.get("camera.idle_cam")
assert state is not None
assert state.state == CameraState.IDLE
async def test_camera_unavailable_state(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test camera unavailable state tracking."""
monitors = [create_mock_monitor(name="Offline Cam", is_available=False)]
await _setup_zm_with_cameras(
hass, mock_zoneminder_client, single_server_config, monitors, freezer
)
state = hass.states.get("camera.offline_cam")
assert state is not None
assert state.state == "unavailable"
async def test_no_monitors_raises_platform_not_ready(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
) -> None:
"""Test PlatformNotReady raised when no monitors returned."""
mock_zoneminder_client.get_monitors.return_value = []
assert await async_setup_component(hass, DOMAIN, single_server_config)
await hass.async_block_till_done()
await async_setup_component(
hass,
"camera",
{"camera": [{"platform": DOMAIN}]},
)
await hass.async_block_till_done()
# No camera entities should exist
states = hass.states.async_all("camera")
assert len(states) == 0
async def test_multi_server_camera_creation(
hass: HomeAssistant,
multi_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test cameras created from multiple ZM servers."""
monitors1 = [create_mock_monitor(monitor_id=1, name="Front Door")]
monitors2 = [create_mock_monitor(monitor_id=2, name="Back Yard")]
clients = iter(
[
create_mock_zm_client(monitors=monitors1),
create_mock_zm_client(monitors=monitors2),
]
)
with patch(
"homeassistant.components.zoneminder.ZoneMinder",
side_effect=lambda *args, **kwargs: next(clients),
):
assert await async_setup_component(hass, DOMAIN, multi_server_config)
await hass.async_block_till_done(wait_background_tasks=True)
assert await async_setup_component(
hass,
"camera",
{"camera": [{"platform": DOMAIN}]},
)
await hass.async_block_till_done(wait_background_tasks=True)
freezer.tick(timedelta(seconds=60))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
states = hass.states.async_all("camera")
assert len(states) == 2
async def test_filter_urllib3_logging_called(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test filter_urllib3_logging() is called in setup_platform."""
monitors = [create_mock_monitor(name="Front Door")]
with patch(
"homeassistant.components.zoneminder.camera.filter_urllib3_logging"
) as mock_filter:
await _setup_zm_with_cameras(
hass, mock_zoneminder_client, single_server_config, monitors, freezer
)
mock_filter.assert_called_once()

View File

@@ -0,0 +1,26 @@
"""Tests for ZoneMinder YAML configuration validation."""
from __future__ import annotations
from homeassistant.components.zoneminder.const import DOMAIN
from homeassistant.const import CONF_HOST, CONF_SSL
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from .conftest import MOCK_HOST
async def test_invalid_config_missing_host(hass: HomeAssistant) -> None:
"""Test that config without host is rejected."""
config: dict = {DOMAIN: [{}]}
result = await async_setup_component(hass, DOMAIN, config)
assert not result
async def test_invalid_config_bad_ssl_type(hass: HomeAssistant) -> None:
"""Test that non-boolean ssl value is rejected."""
config = {DOMAIN: [{CONF_HOST: MOCK_HOST, CONF_SSL: "not_bool"}]}
result = await async_setup_component(hass, DOMAIN, config)
assert not result

View File

@@ -0,0 +1,182 @@
"""Tests for ZoneMinder __init__.py setup flow internals."""
from __future__ import annotations
from unittest.mock import MagicMock, patch
import pytest
from requests.exceptions import ConnectionError as RequestsConnectionError
from homeassistant.components.zoneminder.const import DOMAIN
from homeassistant.const import CONF_HOST, CONF_PATH, Platform
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from .conftest import MOCK_HOST
CONF_PATH_ZMS = "path_zms"
async def test_constructor_called_with_http_prefix(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
) -> None:
"""Test ZM constructor called with http prefix when ssl=false."""
config = {DOMAIN: [{CONF_HOST: MOCK_HOST}]}
assert await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done()
mock_zoneminder_client.mock_cls.assert_called_once_with(
f"http://{MOCK_HOST}",
None, # username
None, # password
"/zm/", # default path
"/zm/cgi-bin/nph-zms", # default path_zms
True, # default verify_ssl
)
async def test_constructor_called_with_https_prefix(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
ssl_config: dict,
) -> None:
"""Test ZM constructor called with https prefix when ssl=true."""
assert await async_setup_component(hass, DOMAIN, ssl_config)
await hass.async_block_till_done()
call_args = mock_zoneminder_client.mock_cls.call_args
assert call_args[0][0] == f"https://{MOCK_HOST}"
async def test_constructor_called_with_auth(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
) -> None:
"""Test ZM constructor called with correct username/password."""
assert await async_setup_component(hass, DOMAIN, single_server_config)
await hass.async_block_till_done()
call_args = mock_zoneminder_client.mock_cls.call_args
assert call_args[0][1] == "admin"
assert call_args[0][2] == "secret"
async def test_constructor_called_with_paths(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
) -> None:
"""Test ZM constructor called with custom paths."""
config = {
DOMAIN: [
{
CONF_HOST: MOCK_HOST,
CONF_PATH: "/custom/",
CONF_PATH_ZMS: "/custom/zms",
}
]
}
assert await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done()
call_args = mock_zoneminder_client.mock_cls.call_args
assert call_args[0][3] == "/custom/"
assert call_args[0][4] == "/custom/zms"
async def test_login_called_in_executor(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
) -> None:
"""Test login() is called during setup."""
assert await async_setup_component(hass, DOMAIN, single_server_config)
await hass.async_block_till_done()
mock_zoneminder_client.login.assert_called_once()
async def test_login_success_returns_true(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
) -> None:
"""Test async_setup returns True on login success."""
mock_zoneminder_client.login.return_value = True
result = await async_setup_component(hass, DOMAIN, single_server_config)
await hass.async_block_till_done()
assert result is True
async def test_login_failure_returns_false(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
) -> None:
"""Test async_setup returns False on login failure."""
mock_zoneminder_client.login.return_value = False
await async_setup_component(hass, DOMAIN, single_server_config)
await hass.async_block_till_done()
async def test_connection_error_logged(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test RequestsConnectionError is logged but doesn't crash setup.
Regression: The original code (lines 76-82) catches the ConnectionError
and logs it, but does NOT set success=False. This means a connection error
doesn't prevent the component from reporting success.
"""
mock_zoneminder_client.login.side_effect = RequestsConnectionError(
"Connection refused"
)
result = await async_setup_component(hass, DOMAIN, single_server_config)
await hass.async_block_till_done()
assert "ZoneMinder connection failure" in caplog.text
assert "Connection refused" in caplog.text
# The component still reports success (this is the regression behavior)
assert result is True
async def test_async_setup_services_invoked(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
) -> None:
"""Test async_setup_services is called during setup."""
with patch(
"homeassistant.components.zoneminder.async_setup_services"
) as mock_services:
assert await async_setup_component(hass, DOMAIN, single_server_config)
await hass.async_block_till_done()
mock_services.assert_called_once_with(hass)
async def test_binary_sensor_platform_load_triggered(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
) -> None:
"""Test binary sensor platform load is triggered during setup."""
with patch("homeassistant.components.zoneminder.async_load_platform") as mock_load:
assert await async_setup_component(hass, DOMAIN, single_server_config)
await hass.async_block_till_done()
mock_load.assert_called_once()
call_args = mock_load.call_args
# Should load binary_sensor platform
assert call_args[0][1] == Platform.BINARY_SENSOR
assert call_args[0][2] == DOMAIN

View File

@@ -0,0 +1,496 @@
"""Tests for ZoneMinder sensor entities."""
from __future__ import annotations
from datetime import timedelta
from unittest.mock import MagicMock, PropertyMock
from freezegun.api import FrozenDateTimeFactory
import pytest
from zoneminder.monitor import MonitorState, TimePeriod
from homeassistant.components.zoneminder.const import DOMAIN
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from .conftest import create_mock_monitor
from tests.common import async_fire_time_changed
async def _setup_zm_with_sensors(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
zm_config: dict,
monitors: list,
freezer: FrozenDateTimeFactory,
sensor_config: dict | None = None,
is_available: bool = True,
active_state: str | None = "Running",
) -> None:
"""Set up ZM component with sensor platform and trigger first poll."""
mock_zoneminder_client.get_monitors.return_value = monitors
type(mock_zoneminder_client).is_available = PropertyMock(return_value=is_available)
mock_zoneminder_client.get_active_state.return_value = active_state
assert await async_setup_component(hass, DOMAIN, zm_config)
await hass.async_block_till_done(wait_background_tasks=True)
if sensor_config is None:
sensor_config = {
"sensor": [
{
"platform": DOMAIN,
"monitored_conditions": [
"all",
"hour",
"day",
"week",
"month",
],
}
]
}
assert await async_setup_component(hass, "sensor", sensor_config)
await hass.async_block_till_done(wait_background_tasks=True)
# Trigger first poll to update entity state
freezer.tick(timedelta(seconds=60))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
# --- Monitor Status Sensor ---
async def test_monitor_status_sensor_exists(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test monitor status sensor is created."""
monitors = [create_mock_monitor(name="Front Door", function=MonitorState.MODECT)]
await _setup_zm_with_sensors(
hass, mock_zoneminder_client, single_server_config, monitors, freezer
)
state = hass.states.get("sensor.front_door_status")
assert state is not None
async def test_monitor_status_sensor_value(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test monitor status sensor shows MonitorState value."""
monitors = [create_mock_monitor(name="Front Door", function=MonitorState.RECORD)]
await _setup_zm_with_sensors(
hass, mock_zoneminder_client, single_server_config, monitors, freezer
)
state = hass.states.get("sensor.front_door_status")
assert state is not None
assert state.state == "Record"
@pytest.mark.parametrize(
("monitor_state", "expected_value"),
[
(MonitorState.NONE, "None"),
(MonitorState.MONITOR, "Monitor"),
(MonitorState.MODECT, "Modect"),
(MonitorState.RECORD, "Record"),
(MonitorState.MOCORD, "Mocord"),
(MonitorState.NODECT, "Nodect"),
],
)
async def test_monitor_status_sensor_all_states(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
monitor_state: MonitorState,
expected_value: str,
) -> None:
"""Test monitor status sensor with all MonitorState values."""
monitors = [create_mock_monitor(name="Cam", function=monitor_state)]
await _setup_zm_with_sensors(
hass, mock_zoneminder_client, single_server_config, monitors, freezer
)
state = hass.states.get("sensor.cam_status")
assert state is not None
assert state.state == expected_value
async def test_monitor_status_sensor_unavailable(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test monitor status sensor when monitor is unavailable."""
monitors = [
create_mock_monitor(name="Front Door", is_available=False, function=None)
]
await _setup_zm_with_sensors(
hass, mock_zoneminder_client, single_server_config, monitors, freezer
)
state = hass.states.get("sensor.front_door_status")
assert state is not None
assert state.state == STATE_UNAVAILABLE
async def test_monitor_status_sensor_null_function(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test monitor status sensor when function is falsy."""
monitors = [
create_mock_monitor(name="Front Door", function=None, is_available=True)
]
await _setup_zm_with_sensors(
hass, mock_zoneminder_client, single_server_config, monitors, freezer
)
state = hass.states.get("sensor.front_door_status")
assert state is not None
# --- Event Sensors ---
@pytest.mark.parametrize(
("condition", "expected_name_suffix", "expected_value"),
[
("all", "Events", "100"),
("hour", "Events Last Hour", "5"),
("day", "Events Last Day", "20"),
("week", "Events Last Week", "50"),
("month", "Events Last Month", "80"),
],
)
async def test_event_sensor_for_each_time_period(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
condition: str,
expected_name_suffix: str,
expected_value: str,
) -> None:
"""Test event sensors for all 5 time periods."""
monitors = [create_mock_monitor(name="Front Door")]
sensor_config = {
"sensor": [
{
"platform": DOMAIN,
"monitored_conditions": [condition],
}
]
}
await _setup_zm_with_sensors(
hass,
mock_zoneminder_client,
single_server_config,
monitors,
freezer,
sensor_config=sensor_config,
)
entity_id = f"sensor.front_door_{expected_name_suffix.lower().replace(' ', '_')}"
state = hass.states.get(entity_id)
assert state is not None
assert state.state == expected_value
async def test_event_sensor_unit_of_measurement(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test event sensors have 'Events' unit of measurement."""
monitors = [create_mock_monitor(name="Front Door")]
await _setup_zm_with_sensors(
hass, mock_zoneminder_client, single_server_config, monitors, freezer
)
state = hass.states.get("sensor.front_door_events")
assert state is not None
assert state.attributes.get("unit_of_measurement") == "Events"
async def test_event_sensor_name_format(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test event sensor name format is '{monitor_name} {time_period_title}'."""
monitors = [create_mock_monitor(name="Back Yard")]
sensor_config = {
"sensor": [
{
"platform": DOMAIN,
"monitored_conditions": ["hour"],
}
]
}
await _setup_zm_with_sensors(
hass,
mock_zoneminder_client,
single_server_config,
monitors,
freezer,
sensor_config=sensor_config,
)
state = hass.states.get("sensor.back_yard_events_last_hour")
assert state is not None
assert state.name == "Back Yard Events Last Hour"
async def test_event_sensor_none_handling(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test event sensor handles None event count."""
monitors = [
create_mock_monitor(
name="Front Door",
events=dict.fromkeys(TimePeriod),
)
]
await _setup_zm_with_sensors(
hass, mock_zoneminder_client, single_server_config, monitors, freezer
)
state = hass.states.get("sensor.front_door_events")
assert state is not None
# --- Run State Sensor ---
async def test_run_state_sensor_exists(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test run state sensor is created."""
monitors = [create_mock_monitor(name="Cam")]
await _setup_zm_with_sensors(
hass, mock_zoneminder_client, single_server_config, monitors, freezer
)
state = hass.states.get("sensor.run_state")
assert state is not None
async def test_run_state_sensor_value(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test run state sensor shows state name."""
monitors = [create_mock_monitor(name="Cam")]
await _setup_zm_with_sensors(
hass,
mock_zoneminder_client,
single_server_config,
monitors,
freezer,
active_state="Home",
)
state = hass.states.get("sensor.run_state")
assert state is not None
assert state.state == "Home"
async def test_run_state_sensor_unavailable(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test run state sensor when server unavailable."""
monitors = [create_mock_monitor(name="Cam")]
await _setup_zm_with_sensors(
hass,
mock_zoneminder_client,
single_server_config,
monitors,
freezer,
is_available=False,
active_state=None,
)
state = hass.states.get("sensor.run_state")
assert state is not None
assert state.state == STATE_UNAVAILABLE
# --- Platform behavior ---
async def test_platform_not_ready_empty_monitors(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
) -> None:
"""Test PlatformNotReady on empty monitors."""
mock_zoneminder_client.get_monitors.return_value = []
assert await async_setup_component(hass, DOMAIN, single_server_config)
await hass.async_block_till_done()
await async_setup_component(
hass,
"sensor",
{"sensor": [{"platform": DOMAIN}]},
)
await hass.async_block_till_done()
# No sensor entities should exist (PlatformNotReady caught by HA)
states = hass.states.async_all("sensor")
assert len(states) == 0
async def test_subset_condition_filtering(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test only selected monitored_conditions get event sensors."""
monitors = [create_mock_monitor(name="Cam")]
sensor_config = {
"sensor": [
{
"platform": DOMAIN,
"monitored_conditions": ["hour", "day"],
}
]
}
await _setup_zm_with_sensors(
hass,
mock_zoneminder_client,
single_server_config,
monitors,
freezer,
sensor_config=sensor_config,
)
# Should have: 1 status + 2 event + 1 run state = 4 sensors
states = hass.states.async_all("sensor")
assert len(states) == 4
# These should exist
assert hass.states.get("sensor.cam_events_last_hour") is not None
assert hass.states.get("sensor.cam_events_last_day") is not None
# These should NOT exist
assert hass.states.get("sensor.cam_events") is None
assert hass.states.get("sensor.cam_events_last_week") is None
assert hass.states.get("sensor.cam_events_last_month") is None
async def test_default_conditions_only_all(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test default monitored_conditions is only 'all'."""
monitors = [create_mock_monitor(name="Cam")]
sensor_config = {"sensor": [{"platform": DOMAIN}]}
await _setup_zm_with_sensors(
hass,
mock_zoneminder_client,
single_server_config,
monitors,
freezer,
sensor_config=sensor_config,
)
# Should have: 1 status + 1 event (all) + 1 run state = 3 sensors
states = hass.states.async_all("sensor")
assert len(states) == 3
async def test_include_archived_flag(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test include_archived flag is passed correctly to get_events."""
monitors = [create_mock_monitor(name="Cam")]
sensor_config = {
"sensor": [
{
"platform": DOMAIN,
"include_archived": True,
"monitored_conditions": ["all"],
}
]
}
await _setup_zm_with_sensors(
hass,
mock_zoneminder_client,
single_server_config,
monitors,
freezer,
sensor_config=sensor_config,
)
# Verify get_events was called with include_archived=True
monitors[0].get_events.assert_called_with(TimePeriod.ALL, True)
async def test_sensor_count_calculation(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test correct number of sensors created per monitor and client.
For each monitor: 1 status + N event sensors
Plus: 1 run state sensor per client
"""
monitors = [
create_mock_monitor(monitor_id=1, name="Cam1"),
create_mock_monitor(monitor_id=2, name="Cam2"),
]
sensor_config = {
"sensor": [
{
"platform": DOMAIN,
"monitored_conditions": ["all", "hour"],
"include_archived": False,
}
]
}
await _setup_zm_with_sensors(
hass,
mock_zoneminder_client,
single_server_config,
monitors,
freezer,
sensor_config=sensor_config,
)
# 2 monitors * (1 status + 2 events) + 1 run state = 7
assert len(hass.states.async_all("sensor")) == 7

View File

@@ -0,0 +1,124 @@
"""Tests for ZoneMinder service calls."""
from __future__ import annotations
from unittest.mock import MagicMock, patch
import pytest
import voluptuous as vol
from homeassistant.components.zoneminder.const import DOMAIN
from homeassistant.const import ATTR_ID, ATTR_NAME
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from .conftest import MOCK_HOST, MOCK_HOST_2, create_mock_zm_client
async def test_set_run_state_service_registered(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
) -> None:
"""Test set_run_state service is registered after setup."""
assert await async_setup_component(hass, DOMAIN, single_server_config)
await hass.async_block_till_done()
assert hass.services.has_service(DOMAIN, "set_run_state")
async def test_set_run_state_valid_call(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
) -> None:
"""Test valid set_run_state call sets state on correct ZM client."""
assert await async_setup_component(hass, DOMAIN, single_server_config)
await hass.async_block_till_done()
await hass.services.async_call(
DOMAIN,
"set_run_state",
{ATTR_ID: MOCK_HOST, ATTR_NAME: "Away"},
blocking=True,
)
await hass.async_block_till_done()
mock_zoneminder_client.set_active_state.assert_called_once_with("Away")
async def test_set_run_state_multi_server_targets_correct_server(
hass: HomeAssistant, multi_server_config: dict
) -> None:
"""Test set_run_state targets specific server by id."""
clients: dict[str, MagicMock] = {}
def make_client(*args, **kwargs):
client = create_mock_zm_client()
# Extract hostname from the server_origin (first positional arg)
origin = args[0]
hostname = origin.split("://")[1]
clients[hostname] = client
return client
with patch(
"homeassistant.components.zoneminder.ZoneMinder",
side_effect=make_client,
):
assert await async_setup_component(hass, DOMAIN, multi_server_config)
await hass.async_block_till_done()
await hass.services.async_call(
DOMAIN,
"set_run_state",
{ATTR_ID: MOCK_HOST_2, ATTR_NAME: "Home"},
blocking=True,
)
await hass.async_block_till_done()
# Only the second server should have been called
clients[MOCK_HOST_2].set_active_state.assert_called_once_with("Home")
clients[MOCK_HOST].set_active_state.assert_not_called()
async def test_set_run_state_missing_fields_rejected(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
) -> None:
"""Test service call with missing required fields is rejected."""
assert await async_setup_component(hass, DOMAIN, single_server_config)
await hass.async_block_till_done()
with pytest.raises(vol.MultipleInvalid):
await hass.services.async_call(
DOMAIN,
"set_run_state",
{ATTR_ID: MOCK_HOST}, # Missing ATTR_NAME
blocking=True,
)
async def test_set_run_state_invalid_host(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test service call with invalid host logs error.
Regression: services.py logs error but doesn't return early,
so it also raises KeyError when trying to access the invalid host.
"""
assert await async_setup_component(hass, DOMAIN, single_server_config)
await hass.async_block_till_done()
with pytest.raises(KeyError):
await hass.services.async_call(
DOMAIN,
"set_run_state",
{ATTR_ID: "invalid.host", ATTR_NAME: "Away"},
blocking=True,
)
assert "Invalid ZoneMinder host provided" in caplog.text

View File

@@ -0,0 +1,243 @@
"""Tests for ZoneMinder switch entities."""
from __future__ import annotations
from datetime import timedelta
from unittest.mock import MagicMock
from freezegun.api import FrozenDateTimeFactory
import pytest
import voluptuous as vol
from zoneminder.monitor import MonitorState
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.components.zoneminder.const import DOMAIN
from homeassistant.components.zoneminder.switch import PLATFORM_SCHEMA
from homeassistant.const import (
ATTR_ENTITY_ID,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
STATE_OFF,
STATE_ON,
)
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from .conftest import create_mock_monitor
from tests.common import async_fire_time_changed
async def _setup_zm_with_switches(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
zm_config: dict,
monitors: list,
freezer: FrozenDateTimeFactory,
command_on: str = "Modect",
command_off: str = "Monitor",
) -> None:
"""Set up ZM component with switch platform and trigger first poll."""
mock_zoneminder_client.get_monitors.return_value = monitors
assert await async_setup_component(hass, DOMAIN, zm_config)
await hass.async_block_till_done(wait_background_tasks=True)
assert await async_setup_component(
hass,
SWITCH_DOMAIN,
{
SWITCH_DOMAIN: [
{
"platform": DOMAIN,
"command_on": command_on,
"command_off": command_off,
}
]
},
)
await hass.async_block_till_done(wait_background_tasks=True)
# Trigger first poll to update entity state
freezer.tick(timedelta(seconds=60))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
async def test_switch_per_monitor(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
two_monitors: list,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test one switch entity is created per monitor."""
await _setup_zm_with_switches(
hass, mock_zoneminder_client, single_server_config, two_monitors, freezer
)
states = hass.states.async_all(SWITCH_DOMAIN)
assert len(states) == 2
async def test_switch_name_format(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test switch name format is '{name} State'."""
monitors = [create_mock_monitor(name="Front Door")]
await _setup_zm_with_switches(
hass, mock_zoneminder_client, single_server_config, monitors, freezer
)
state = hass.states.get("switch.front_door_state")
assert state is not None
assert state.name == "Front Door State"
async def test_switch_on_when_function_matches_command_on(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test switch is ON when monitor function matches command_on."""
monitors = [create_mock_monitor(name="Front Door", function=MonitorState.MODECT)]
await _setup_zm_with_switches(
hass,
mock_zoneminder_client,
single_server_config,
monitors,
freezer,
command_on="Modect",
)
state = hass.states.get("switch.front_door_state")
assert state is not None
assert state.state == STATE_ON
async def test_switch_off_when_function_differs(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test switch is OFF when monitor function differs from command_on."""
monitors = [create_mock_monitor(name="Front Door", function=MonitorState.MONITOR)]
await _setup_zm_with_switches(
hass,
mock_zoneminder_client,
single_server_config,
monitors,
freezer,
command_on="Modect",
)
state = hass.states.get("switch.front_door_state")
assert state is not None
assert state.state == STATE_OFF
async def test_switch_turn_on_service(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test turn_on service sets monitor function to command_on."""
monitors = [create_mock_monitor(name="Front Door", function=MonitorState.MONITOR)]
await _setup_zm_with_switches(
hass, mock_zoneminder_client, single_server_config, monitors, freezer
)
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "switch.front_door_state"},
blocking=True,
)
await hass.async_block_till_done()
# Verify monitor function was set to MonitorState("Modect")
assert monitors[0].function == MonitorState("Modect")
async def test_switch_turn_off_service(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test turn_off service sets monitor function to command_off."""
monitors = [create_mock_monitor(name="Front Door", function=MonitorState.MODECT)]
await _setup_zm_with_switches(
hass, mock_zoneminder_client, single_server_config, monitors, freezer
)
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "switch.front_door_state"},
blocking=True,
)
await hass.async_block_till_done()
assert monitors[0].function == MonitorState("Monitor")
async def test_switch_icon(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test switch icon is mdi:record-rec."""
monitors = [create_mock_monitor(name="Front Door")]
await _setup_zm_with_switches(
hass, mock_zoneminder_client, single_server_config, monitors, freezer
)
state = hass.states.get("switch.front_door_state")
assert state is not None
assert state.attributes.get("icon") == "mdi:record-rec"
async def test_switch_platform_not_ready_empty_monitors(
hass: HomeAssistant,
mock_zoneminder_client: MagicMock,
single_server_config: dict,
) -> None:
"""Test PlatformNotReady on empty monitors."""
mock_zoneminder_client.get_monitors.return_value = []
assert await async_setup_component(hass, DOMAIN, single_server_config)
await hass.async_block_till_done()
await async_setup_component(
hass,
SWITCH_DOMAIN,
{
SWITCH_DOMAIN: [
{
"platform": DOMAIN,
"command_on": "Modect",
"command_off": "Monitor",
}
]
},
)
await hass.async_block_till_done()
states = hass.states.async_all(SWITCH_DOMAIN)
assert len(states) == 0
def test_platform_schema_requires_command_on_off() -> None:
"""Test platform schema requires command_on and command_off."""
# Missing command_on
with pytest.raises(vol.MultipleInvalid):
PLATFORM_SCHEMA({"platform": "zoneminder", "command_off": "Monitor"})
# Missing command_off
with pytest.raises(vol.MultipleInvalid):
PLATFORM_SCHEMA({"platform": "zoneminder", "command_on": "Modect"})