core/tests/components/motioneye/test_camera.py

"""Test the motionEye camera."""
import copy
import logging
from typing import Any
from unittest.mock import AsyncMock, Mock

from aiohttp import web
from aiohttp.web_exceptions import HTTPBadGateway
from motioneye_client.client import (
    MotionEyeClientError,
    MotionEyeClientInvalidAuthError,
)
from motioneye_client.const import (
    KEY_CAMERAS,
    KEY_MOTION_DETECTION,
    KEY_NAME,
    KEY_VIDEO_STREAMING,
)
import pytest

from homeassistant.components.camera import async_get_image, async_get_mjpeg_stream
from homeassistant.components.motioneye import get_motioneye_device_identifier
from homeassistant.components.motioneye.const import (
    CONF_SURVEILLANCE_USERNAME,
    DEFAULT_SCAN_INTERVAL,
    DOMAIN,
    MOTIONEYE_MANUFACTURER,
)
from homeassistant.const import CONF_URL
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.device_registry import async_get_registry
import homeassistant.util.dt as dt_util

from . import (
    TEST_CAMERA_DEVICE_IDENTIFIER,
    TEST_CAMERA_ENTITY_ID,
    TEST_CAMERA_ID,
    TEST_CAMERA_NAME,
    TEST_CAMERAS,
    TEST_CONFIG_ENTRY_ID,
    TEST_SURVEILLANCE_USERNAME,
    create_mock_motioneye_client,
    create_mock_motioneye_config_entry,
    setup_mock_motioneye_config_entry,
)

from tests.common import async_fire_time_changed

_LOGGER = logging.getLogger(__name__)


async def test_setup_camera(hass: HomeAssistant) -> None:
    """Test a basic camera."""
    client = create_mock_motioneye_client()
    await setup_mock_motioneye_config_entry(hass, client=client)

    entity_state = hass.states.get(TEST_CAMERA_ENTITY_ID)
    assert entity_state
    assert entity_state.state == "idle"
    assert entity_state.attributes.get("friendly_name") == TEST_CAMERA_NAME


async def test_setup_camera_auth_fail(hass: HomeAssistant) -> None:
    """Test a successful camera."""
    client = create_mock_motioneye_client()
    client.async_client_login = AsyncMock(side_effect=MotionEyeClientInvalidAuthError)
    await setup_mock_motioneye_config_entry(hass, client=client)
    assert not hass.states.get(TEST_CAMERA_ENTITY_ID)


async def test_setup_camera_client_error(hass: HomeAssistant) -> None:
    """Test a successful camera."""
    client = create_mock_motioneye_client()
    client.async_client_login = AsyncMock(side_effect=MotionEyeClientError)
    await setup_mock_motioneye_config_entry(hass, client=client)
    assert not hass.states.get(TEST_CAMERA_ENTITY_ID)


async def test_setup_camera_empty_data(hass: HomeAssistant) -> None:
    """Test a successful camera."""
    client = create_mock_motioneye_client()
    client.async_get_cameras = AsyncMock(return_value={})
    await setup_mock_motioneye_config_entry(hass, client=client)
    assert not hass.states.get(TEST_CAMERA_ENTITY_ID)


async def test_setup_camera_bad_data(hass: HomeAssistant) -> None:
    """Test bad camera data."""
    client = create_mock_motioneye_client()
    cameras = copy.deepcopy(TEST_CAMERAS)
    del cameras[KEY_CAMERAS][0][KEY_NAME]

    client.async_get_cameras = AsyncMock(return_value=cameras)
    await setup_mock_motioneye_config_entry(hass, client=client)
    assert not hass.states.get(TEST_CAMERA_ENTITY_ID)


async def test_setup_camera_without_streaming(hass: HomeAssistant) -> None:
    """Test a camera without streaming enabled."""
    client = create_mock_motioneye_client()
    cameras = copy.deepcopy(TEST_CAMERAS)
    cameras[KEY_CAMERAS][0][KEY_VIDEO_STREAMING] = False

    client.async_get_cameras = AsyncMock(return_value=cameras)
    await setup_mock_motioneye_config_entry(hass, client=client)
    entity_state = hass.states.get(TEST_CAMERA_ENTITY_ID)
    assert entity_state
    assert entity_state.state == "unavailable"


async def test_setup_camera_new_data_same(hass: HomeAssistant) -> None:
    """Test a data refresh with the same data."""
    client = create_mock_motioneye_client()
    await setup_mock_motioneye_config_entry(hass, client=client)
    async_fire_time_changed(hass, dt_util.utcnow() + DEFAULT_SCAN_INTERVAL)
    await hass.async_block_till_done()
    assert hass.states.get(TEST_CAMERA_ENTITY_ID)


async def test_setup_camera_new_data_camera_removed(hass: HomeAssistant) -> None:
    """Test a data refresh with a removed camera."""
    device_registry = await async_get_registry(hass)
    entity_registry = await er.async_get_registry(hass)

    client = create_mock_motioneye_client()
    config_entry = await setup_mock_motioneye_config_entry(hass, client=client)

    # Create some random old devices/entity_ids and ensure they get cleaned up.
    old_device_id = "old-device-id"
    old_entity_unique_id = "old-entity-unique_id"
    old_device = device_registry.async_get_or_create(
        config_entry_id=config_entry.entry_id, identifiers={(DOMAIN, old_device_id)}
    )
    entity_registry.async_get_or_create(
        domain=DOMAIN,
        platform="camera",
        unique_id=old_entity_unique_id,
        config_entry=config_entry,
        device_id=old_device.id,
    )

    await hass.async_block_till_done()
    assert hass.states.get(TEST_CAMERA_ENTITY_ID)
    assert device_registry.async_get_device({TEST_CAMERA_DEVICE_IDENTIFIER})

    client.async_get_cameras = AsyncMock(return_value={KEY_CAMERAS: []})
    async_fire_time_changed(hass, dt_util.utcnow() + DEFAULT_SCAN_INTERVAL)
    await hass.async_block_till_done()
    assert not hass.states.get(TEST_CAMERA_ENTITY_ID)
    assert not device_registry.async_get_device({TEST_CAMERA_DEVICE_IDENTIFIER})
    assert not device_registry.async_get_device({(DOMAIN, old_device_id)})
    assert not entity_registry.async_get_entity_id(
        DOMAIN, "camera", old_entity_unique_id
    )


async def test_setup_camera_new_data_error(hass: HomeAssistant) -> None:
    """Test a data refresh that fails."""
    client = create_mock_motioneye_client()
    await setup_mock_motioneye_config_entry(hass, client=client)
    assert hass.states.get(TEST_CAMERA_ENTITY_ID)
    client.async_get_cameras = AsyncMock(side_effect=MotionEyeClientError)
    async_fire_time_changed(hass, dt_util.utcnow() + DEFAULT_SCAN_INTERVAL)
    await hass.async_block_till_done()
    entity_state = hass.states.get(TEST_CAMERA_ENTITY_ID)
    assert entity_state
    assert entity_state.state == "unavailable"


async def test_setup_camera_new_data_without_streaming(hass: HomeAssistant) -> None:
    """Test a data refresh without streaming."""
    client = create_mock_motioneye_client()
    await setup_mock_motioneye_config_entry(hass, client=client)
    entity_state = hass.states.get(TEST_CAMERA_ENTITY_ID)
    assert entity_state
    assert entity_state.state == "idle"

    cameras = copy.deepcopy(TEST_CAMERAS)
    cameras[KEY_CAMERAS][0][KEY_VIDEO_STREAMING] = False
    client.async_get_cameras = AsyncMock(return_value=cameras)
    async_fire_time_changed(hass, dt_util.utcnow() + DEFAULT_SCAN_INTERVAL)
    await hass.async_block_till_done()
    entity_state = hass.states.get(TEST_CAMERA_ENTITY_ID)
    assert entity_state
    assert entity_state.state == "unavailable"


async def test_unload_camera(hass: HomeAssistant) -> None:
    """Test unloading camera."""
    client = create_mock_motioneye_client()
    entry = await setup_mock_motioneye_config_entry(hass, client=client)
    assert hass.states.get(TEST_CAMERA_ENTITY_ID)
    assert not client.async_client_close.called
    await hass.config_entries.async_unload(entry.entry_id)
    assert client.async_client_close.called


async def test_get_still_image_from_camera(
    aiohttp_server: Any, hass: HomeAssistant
) -> None:
    """Test getting a still image."""

    image_handler = Mock(return_value="")

    app = web.Application()
    app.add_routes(
        [
            web.get(
                "/foo",
                image_handler,
            )
        ]
    )

    server = await aiohttp_server(app)
    client = create_mock_motioneye_client()
    client.get_camera_snapshot_url = Mock(
        return_value=f"http://localhost:{server.port}/foo"
    )
    config_entry = create_mock_motioneye_config_entry(
        hass,
        data={
            CONF_URL: f"http://localhost:{server.port}",
            CONF_SURVEILLANCE_USERNAME: TEST_SURVEILLANCE_USERNAME,
        },
    )

    await setup_mock_motioneye_config_entry(
        hass, config_entry=config_entry, client=client
    )
    await hass.async_block_till_done()

    # It won't actually get a stream from the dummy handler, so just catch
    # the expected exception, then verify the right handler was called.
    with pytest.raises(HomeAssistantError):
        await async_get_image(hass, TEST_CAMERA_ENTITY_ID, timeout=None)  # type: ignore[no-untyped-call]
    assert image_handler.called


async def test_get_stream_from_camera(aiohttp_server: Any, hass: HomeAssistant) -> None:
    """Test getting a stream."""

    stream_handler = Mock(return_value="")
    app = web.Application()
    app.add_routes([web.get("/", stream_handler)])
    stream_server = await aiohttp_server(app)

    client = create_mock_motioneye_client()
    client.get_camera_stream_url = Mock(
        return_value=f"http://localhost:{stream_server.port}/"
    )
    config_entry = create_mock_motioneye_config_entry(
        hass,
        data={
            CONF_URL: f"http://localhost:{stream_server.port}",
            # The port won't be used as the client is a mock.
            CONF_SURVEILLANCE_USERNAME: TEST_SURVEILLANCE_USERNAME,
        },
    )
    cameras = copy.deepcopy(TEST_CAMERAS)
    client.async_get_cameras = AsyncMock(return_value=cameras)
    await setup_mock_motioneye_config_entry(
        hass, config_entry=config_entry, client=client
    )
    await hass.async_block_till_done()

    # It won't actually get a stream from the dummy handler, so just catch
    # the expected exception, then verify the right handler was called.
    with pytest.raises(HTTPBadGateway):
        await async_get_mjpeg_stream(hass, None, TEST_CAMERA_ENTITY_ID)  # type: ignore[no-untyped-call]
    assert stream_handler.called


async def test_state_attributes(hass: HomeAssistant) -> None:
    """Test state attributes are set correctly."""
    client = create_mock_motioneye_client()
    await setup_mock_motioneye_config_entry(hass, client=client)

    entity_state = hass.states.get(TEST_CAMERA_ENTITY_ID)
    assert entity_state
    assert entity_state.attributes.get("brand") == MOTIONEYE_MANUFACTURER
    assert entity_state.attributes.get("motion_detection")

    cameras = copy.deepcopy(TEST_CAMERAS)
    cameras[KEY_CAMERAS][0][KEY_MOTION_DETECTION] = False
    client.async_get_cameras = AsyncMock(return_value=cameras)
    async_fire_time_changed(hass, dt_util.utcnow() + DEFAULT_SCAN_INTERVAL)
    await hass.async_block_till_done()

    entity_state = hass.states.get(TEST_CAMERA_ENTITY_ID)
    assert entity_state
    assert not entity_state.attributes.get("motion_detection")


async def test_device_info(hass: HomeAssistant) -> None:
    """Verify device information includes expected details."""
    client = create_mock_motioneye_client()
    entry = await setup_mock_motioneye_config_entry(hass, client=client)

    device_identifier = get_motioneye_device_identifier(entry.entry_id, TEST_CAMERA_ID)
    device_registry = dr.async_get(hass)

    device = device_registry.async_get_device({device_identifier})
    assert device
    assert device.config_entries == {TEST_CONFIG_ENTRY_ID}
    assert device.identifiers == {device_identifier}
    assert device.manufacturer == MOTIONEYE_MANUFACTURER
    assert device.model == MOTIONEYE_MANUFACTURER
    assert device.name == TEST_CAMERA_NAME

    entity_registry = await er.async_get_registry(hass)
    entities_from_device = [
        entry.entity_id
        for entry in er.async_entries_for_device(entity_registry, device.id)
    ]
    assert TEST_CAMERA_ENTITY_ID in entities_from_device