core/tests/components/sleepiq/test_init.py

"""Tests for the SleepIQ integration."""

from collections.abc import Callable
from datetime import timedelta
from http import HTTPStatus
from unittest.mock import MagicMock, create_autospec

from asyncsleepiq import (
    Side,
    SleepData,
    SleepIQAPIException,
    SleepIQBed,
    SleepIQFoundation,
    SleepIQLoginException,
    SleepIQSleeper,
    SleepIQTimeoutException,
)
from freezegun.api import FrozenDateTimeFactory
import pytest

from homeassistant.components.sleepiq.const import DOMAIN, IS_IN_BED, SLEEP_NUMBER
from homeassistant.components.sleepiq.coordinator import (
    LONGER_UPDATE_INTERVAL,
    SLEEP_DATA_UPDATE_INTERVAL,
    UPDATE_INTERVAL,
)
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntryState
from homeassistant.const import CONF_USERNAME, PRESSURE
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component
from homeassistant.util.dt import utcnow

from .conftest import (
    BED_ID,
    SLEEPER_L_ID,
    SLEEPER_L_NAME,
    SLEEPER_L_NAME_LOWER,
    SLEEPER_R_ID,
    SLEEPER_R_NAME,
    SLEEPIQ_CONFIG,
    setup_platform,
)

from tests.common import (
    MockConfigEntry,
    RegistryEntryWithDefaults,
    async_fire_time_changed,
    mock_registry,
)

ENTITY_IS_IN_BED = f"sensor.sleepnumber_{BED_ID}_{SLEEPER_L_NAME_LOWER}_{IS_IN_BED}"
ENTITY_PRESSURE = f"sensor.sleepnumber_{BED_ID}_{SLEEPER_L_NAME_LOWER}_{PRESSURE}"
ENTITY_SLEEP_NUMBER = (
    f"sensor.sleepnumber_{BED_ID}_{SLEEPER_L_NAME_LOWER}_{SLEEP_NUMBER}"
)


async def test_unload_entry(hass: HomeAssistant, mock_asyncsleepiq) -> None:
    """Test unloading the SleepIQ entry."""
    entry = await setup_platform(hass, "sensor")
    assert await hass.config_entries.async_unload(entry.entry_id)
    await hass.async_block_till_done()

    assert entry.state is ConfigEntryState.NOT_LOADED
    assert not hass.data.get(DOMAIN)


async def test_entry_setup_login_error(hass: HomeAssistant, mock_asyncsleepiq) -> None:
    """Test when sleepiq client is unable to login."""
    mock_asyncsleepiq.login.side_effect = SleepIQLoginException
    entry = await setup_platform(hass, None)
    assert not await hass.config_entries.async_setup(entry.entry_id)


async def test_entry_setup_timeout_error(
    hass: HomeAssistant, mock_asyncsleepiq
) -> None:
    """Test when sleepiq client timeout."""
    mock_asyncsleepiq.login.side_effect = SleepIQTimeoutException
    entry = await setup_platform(hass, None)
    assert not await hass.config_entries.async_setup(entry.entry_id)


async def test_update_interval(hass: HomeAssistant, mock_asyncsleepiq) -> None:
    """Test update interval."""
    await setup_platform(hass, "sensor")
    assert mock_asyncsleepiq.fetch_bed_statuses.call_count == 1

    async_fire_time_changed(hass, utcnow() + UPDATE_INTERVAL)
    await hass.async_block_till_done()

    assert mock_asyncsleepiq.fetch_bed_statuses.call_count == 2


async def test_api_error(hass: HomeAssistant, mock_asyncsleepiq) -> None:
    """Test when sleepiq client is unable to login."""
    mock_asyncsleepiq.init_beds.side_effect = SleepIQAPIException
    entry = await setup_platform(hass, None)
    assert not await hass.config_entries.async_setup(entry.entry_id)


async def test_api_timeout(hass: HomeAssistant, mock_asyncsleepiq) -> None:
    """Test when sleepiq client timeout."""
    mock_asyncsleepiq.init_beds.side_effect = SleepIQTimeoutException
    entry = await setup_platform(hass, None)
    assert not await hass.config_entries.async_setup(entry.entry_id)


@pytest.mark.parametrize(
    ("get_fetch_mock", "platform", "interval"),
    [
        pytest.param(
            lambda client, bed: client.fetch_bed_statuses,
            "sensor",
            UPDATE_INTERVAL,
            id="bed_status",
        ),
        pytest.param(
            lambda client, bed: bed.fetch_pause_mode,
            "switch",
            LONGER_UPDATE_INTERVAL,
            id="pause_mode",
        ),
        pytest.param(
            lambda client, bed: bed.sleepers[0].fetch_sleep_data,
            "sensor",
            SLEEP_DATA_UPDATE_INTERVAL,
            id="sleep_data",
        ),
    ],
)
async def test_update_auth_error_starts_reauth_flow(
    hass: HomeAssistant,
    freezer: FrozenDateTimeFactory,
    mock_asyncsleepiq: MagicMock,
    mock_bed: MagicMock,
    get_fetch_mock: Callable[[MagicMock, MagicMock], MagicMock],
    platform: str,
    interval: timedelta,
) -> None:
    """Test an authentication failure during an update starts the reauth flow."""
    entry = await setup_platform(hass, platform)

    get_fetch_mock(mock_asyncsleepiq, mock_bed).side_effect = SleepIQAPIException(
        HTTPStatus.UNAUTHORIZED, "unauthorized"
    )
    freezer.tick(interval)
    async_fire_time_changed(hass)
    await hass.async_block_till_done(wait_background_tasks=True)

    flows = hass.config_entries.flow.async_progress_by_handler(DOMAIN)
    assert len(flows) == 1
    assert flows[0]["context"]["source"] == SOURCE_REAUTH
    assert flows[0]["context"]["entry_id"] == entry.entry_id


async def test_update_api_error_does_not_start_reauth_flow(
    hass: HomeAssistant,
    freezer: FrozenDateTimeFactory,
    mock_asyncsleepiq: MagicMock,
) -> None:
    """Test a non-authentication API error during an update does not start reauth."""
    await setup_platform(hass, "sensor")

    mock_asyncsleepiq.fetch_bed_statuses.side_effect = SleepIQAPIException(
        HTTPStatus.INTERNAL_SERVER_ERROR, "server error"
    )
    freezer.tick(UPDATE_INTERVAL)
    async_fire_time_changed(hass)
    await hass.async_block_till_done()

    assert mock_asyncsleepiq.fetch_bed_statuses.call_count == 2
    assert not hass.config_entries.flow.async_progress_by_handler(DOMAIN)


async def test_unique_id_migration(hass: HomeAssistant, mock_asyncsleepiq) -> None:
    """Test migration of sensor unique IDs."""

    mock_entry = MockConfigEntry(
        domain=DOMAIN,
        data=SLEEPIQ_CONFIG,
        unique_id=SLEEPIQ_CONFIG[CONF_USERNAME].lower(),
    )

    mock_entry.add_to_hass(hass)

    mock_registry(
        hass,
        {
            ENTITY_IS_IN_BED: RegistryEntryWithDefaults(
                entity_id=ENTITY_IS_IN_BED,
                unique_id=f"{BED_ID}_{SLEEPER_L_NAME}_{IS_IN_BED}",
                platform=DOMAIN,
                config_entry_id=mock_entry.entry_id,
            ),
            ENTITY_PRESSURE: RegistryEntryWithDefaults(
                entity_id=ENTITY_PRESSURE,
                unique_id=f"{BED_ID}_{SLEEPER_L_NAME}_{PRESSURE}",
                platform=DOMAIN,
                config_entry_id=mock_entry.entry_id,
            ),
            ENTITY_SLEEP_NUMBER: RegistryEntryWithDefaults(
                entity_id=ENTITY_SLEEP_NUMBER,
                unique_id=f"{BED_ID}_{SLEEPER_L_NAME}_{SLEEP_NUMBER}",
                platform=DOMAIN,
                config_entry_id=mock_entry.entry_id,
            ),
        },
    )
    assert await async_setup_component(hass, DOMAIN, {})
    await hass.async_block_till_done()

    ent_reg = er.async_get(hass)  # pylint: disable=home-assistant-tests-registry-fixtures

    sensor_is_in_bed = ent_reg.async_get(ENTITY_IS_IN_BED)
    assert sensor_is_in_bed.unique_id == f"{SLEEPER_L_ID}_{IS_IN_BED}"

    sensor_pressure = ent_reg.async_get(ENTITY_PRESSURE)
    assert sensor_pressure.unique_id == f"{SLEEPER_L_ID}_{PRESSURE}"

    sensor_sleep_number = ent_reg.async_get(ENTITY_SLEEP_NUMBER)
    assert sensor_sleep_number.unique_id == f"{SLEEPER_L_ID}_{SLEEP_NUMBER}"


def _make_controller() -> MagicMock:
    """Build a bare controller bed with no foundation features."""
    controller = create_autospec(SleepIQBed)
    controller.name = "Firmness Control"
    controller.id = "ctrl_001"
    controller.mac_addr = "AA:BB:CC:DD:EE:01"
    controller.model = "Firmness Control, 360, Dual,Boxed"
    controller.paused = False

    ctrl_sleeper_l = create_autospec(SleepIQSleeper)
    ctrl_sleeper_l.side = Side.LEFT
    ctrl_sleeper_l.name = SLEEPER_L_NAME
    ctrl_sleeper_l.sleeper_id = SLEEPER_L_ID
    ctrl_sleeper_l.in_bed = True
    ctrl_sleeper_l.sleep_number = 40
    ctrl_sleeper_l.pressure = 1000
    ctrl_sleeper_l.sleep_data = SleepData(
        duration=28800,
        sleep_score=85,
        heart_rate=60,
        respiratory_rate=14,
        hrv=68,
    )

    ctrl_sleeper_r = create_autospec(SleepIQSleeper)
    ctrl_sleeper_r.side = Side.RIGHT
    ctrl_sleeper_r.name = SLEEPER_R_NAME
    ctrl_sleeper_r.sleeper_id = SLEEPER_R_ID
    ctrl_sleeper_r.in_bed = False
    ctrl_sleeper_r.sleep_number = 80
    ctrl_sleeper_r.pressure = 1400
    ctrl_sleeper_r.sleep_data = SleepData(
        duration=25200,
        sleep_score=78,
        heart_rate=65,
        respiratory_rate=15,
        hrv=72,
    )

    controller.sleepers = [ctrl_sleeper_l, ctrl_sleeper_r]
    controller.foundation = create_autospec(SleepIQFoundation)
    controller.foundation.lights = []
    controller.foundation.actuators = []
    controller.foundation.presets = []
    controller.foundation.foot_warmers = []
    controller.foundation.core_climates = []
    return controller


async def test_duplicate_beds_filtered(
    hass: HomeAssistant,
    entity_registry: er.EntityRegistry,
    mock_asyncsleepiq: MagicMock,
) -> None:
    """Test that duplicate bed objects sharing sleeper IDs are filtered."""
    mock_asyncsleepiq.beds["ctrl_001"] = _make_controller()

    entry = await setup_platform(hass, "sensor")
    assert entry.state is ConfigEntryState.LOADED

    sleeper_l_entities = [
        e
        for e in er.async_entries_for_config_entry(entity_registry, entry.entry_id)
        if SLEEPER_L_ID in e.unique_id
    ]
    sleeper_r_entities = [
        e
        for e in er.async_entries_for_config_entry(entity_registry, entry.entry_id)
        if SLEEPER_R_ID in e.unique_id
    ]

    assert len(sleeper_l_entities) > 0
    assert len(sleeper_r_entities) > 0

    bed_ids = list(mock_asyncsleepiq.beds)
    assert "ctrl_001" not in bed_ids
    assert BED_ID in bed_ids


async def test_duplicate_beds_controller_first(
    hass: HomeAssistant,
    entity_registry: er.EntityRegistry,
    mock_asyncsleepiq: MagicMock,
) -> None:
    """Test that the real bed survives even when the controller appears first."""
    real_bed = mock_asyncsleepiq.beds.pop(BED_ID)
    mock_asyncsleepiq.beds["ctrl_001"] = _make_controller()
    mock_asyncsleepiq.beds[BED_ID] = real_bed

    entry = await setup_platform(hass, "sensor")
    assert entry.state is ConfigEntryState.LOADED

    bed_ids = list(mock_asyncsleepiq.beds)
    assert BED_ID in bed_ids
    assert "ctrl_001" not in bed_ids


async def test_duplicate_beds_none_sleeper_ids_not_filtered(
    hass: HomeAssistant,
    mock_asyncsleepiq: MagicMock,
) -> None:
    """Test that beds with None sleeper IDs are not falsely treated as duplicates."""
    ghost_bed = create_autospec(SleepIQBed)
    ghost_bed.name = "Guest Bed"
    ghost_bed.id = "ghost_001"
    ghost_bed.mac_addr = "AA:BB:CC:DD:EE:02"
    ghost_bed.model = "Guest"
    ghost_bed.paused = False

    ghost_sleeper = create_autospec(SleepIQSleeper)
    ghost_sleeper.side = Side.LEFT
    ghost_sleeper.name = "Guest"
    ghost_sleeper.sleeper_id = None
    ghost_sleeper.in_bed = False
    ghost_sleeper.sleep_number = 50
    ghost_sleeper.pressure = 1200
    ghost_sleeper.sleep_data = SleepData(
        duration=0, sleep_score=0, heart_rate=0, respiratory_rate=0, hrv=0
    )

    ghost_bed.sleepers = [ghost_sleeper]
    ghost_bed.foundation = create_autospec(SleepIQFoundation)
    ghost_bed.foundation.lights = []
    ghost_bed.foundation.actuators = []
    ghost_bed.foundation.presets = []
    ghost_bed.foundation.foot_warmers = []
    ghost_bed.foundation.core_climates = []

    mock_asyncsleepiq.beds["ghost_001"] = ghost_bed

    entry = await setup_platform(hass, "sensor")
    assert entry.state is ConfigEntryState.LOADED
    assert "ghost_001" in mock_asyncsleepiq.beds