core/tests/components/fitbit/test_sensor.py

"""Tests for the fitbit sensor platform."""


from collections.abc import Awaitable, Callable
from http import HTTPStatus
from typing import Any

import pytest
from requests_mock.mocker import Mocker
from syrupy.assertion import SnapshotAssertion

from homeassistant.components.fitbit.const import DOMAIN
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.entity_component import async_update_entity

from .conftest import (
    DEVICES_API_URL,
    PROFILE_USER_ID,
    TIMESERIES_API_URL_FORMAT,
    timeseries_response,
)

DEVICE_RESPONSE_CHARGE_2 = {
    "battery": "Medium",
    "batteryLevel": 60,
    "deviceVersion": "Charge 2",
    "id": "816713257",
    "lastSyncTime": "2019-11-07T12:00:58.000",
    "mac": "16ADD56D54GD",
    "type": "TRACKER",
}
DEVICE_RESPONSE_ARIA_AIR = {
    "battery": "High",
    "batteryLevel": 95,
    "deviceVersion": "Aria Air",
    "id": "016713257",
    "lastSyncTime": "2019-11-07T12:00:58.000",
    "mac": "06ADD56D54GD",
    "type": "SCALE",
}


@pytest.fixture
def platforms() -> list[str]:
    """Fixture to specify platforms to test."""
    return [Platform.SENSOR]


@pytest.mark.parametrize(
    (
        "monitored_resources",
        "entity_id",
        "api_resource",
        "api_value",
    ),
    [
        (
            ["activities/activityCalories"],
            "sensor.activity_calories",
            "activities/activityCalories",
            "135",
        ),
        (
            ["activities/calories"],
            "sensor.calories",
            "activities/calories",
            "139",
        ),
        (
            ["activities/distance"],
            "sensor.distance",
            "activities/distance",
            "12.7",
        ),
        (
            ["activities/elevation"],
            "sensor.elevation",
            "activities/elevation",
            "7600.24",
        ),
        (
            ["activities/floors"],
            "sensor.floors",
            "activities/floors",
            "8",
        ),
        (
            ["activities/heart"],
            "sensor.resting_heart_rate",
            "activities/heart",
            {"restingHeartRate": 76},
        ),
        (
            ["activities/minutesFairlyActive"],
            "sensor.minutes_fairly_active",
            "activities/minutesFairlyActive",
            35,
        ),
        (
            ["activities/minutesLightlyActive"],
            "sensor.minutes_lightly_active",
            "activities/minutesLightlyActive",
            95,
        ),
        (
            ["activities/minutesSedentary"],
            "sensor.minutes_sedentary",
            "activities/minutesSedentary",
            18,
        ),
        (
            ["activities/minutesVeryActive"],
            "sensor.minutes_very_active",
            "activities/minutesVeryActive",
            20,
        ),
        (
            ["activities/steps"],
            "sensor.steps",
            "activities/steps",
            "5600",
        ),
        (
            ["body/weight"],
            "sensor.weight",
            "body/weight",
            "175",
        ),
        (
            ["body/fat"],
            "sensor.body_fat",
            "body/fat",
            "18",
        ),
        (
            ["body/bmi"],
            "sensor.bmi",
            "body/bmi",
            "23.7",
        ),
        (
            ["sleep/awakeningsCount"],
            "sensor.awakenings_count",
            "sleep/awakeningsCount",
            "7",
        ),
        (
            ["sleep/efficiency"],
            "sensor.sleep_efficiency",
            "sleep/efficiency",
            "80",
        ),
        (
            ["sleep/minutesAfterWakeup"],
            "sensor.minutes_after_wakeup",
            "sleep/minutesAfterWakeup",
            "17",
        ),
        (
            ["sleep/minutesAsleep"],
            "sensor.sleep_minutes_asleep",
            "sleep/minutesAsleep",
            "360",
        ),
        (
            ["sleep/minutesAwake"],
            "sensor.sleep_minutes_awake",
            "sleep/minutesAwake",
            "35",
        ),
        (
            ["sleep/minutesToFallAsleep"],
            "sensor.sleep_minutes_to_fall_asleep",
            "sleep/minutesToFallAsleep",
            "35",
        ),
        (
            ["sleep/startTime"],
            "sensor.sleep_start_time",
            "sleep/startTime",
            "2020-01-27T00:17:30.000",
        ),
        (
            ["sleep/timeInBed"],
            "sensor.sleep_time_in_bed",
            "sleep/timeInBed",
            "462",
        ),
    ],
)
async def test_sensors(
    hass: HomeAssistant,
    fitbit_config_setup: None,
    sensor_platform_setup: Callable[[], Awaitable[bool]],
    register_timeseries: Callable[[str, dict[str, Any]], None],
    entity_registry: er.EntityRegistry,
    entity_id: str,
    api_resource: str,
    api_value: str,
    snapshot: SnapshotAssertion,
) -> None:
    """Test sensors."""

    register_timeseries(
        api_resource, timeseries_response(api_resource.replace("/", "-"), api_value)
    )
    await sensor_platform_setup()
    entries = hass.config_entries.async_entries(DOMAIN)
    assert len(entries) == 1

    state = hass.states.get(entity_id)
    assert state
    entry = entity_registry.async_get(entity_id)
    assert entry
    assert (state.state, state.attributes, entry.unique_id) == snapshot


@pytest.mark.parametrize(
    ("devices_response", "monitored_resources"),
    [([DEVICE_RESPONSE_CHARGE_2, DEVICE_RESPONSE_ARIA_AIR], ["devices/battery"])],
)
async def test_device_battery_level(
    hass: HomeAssistant,
    fitbit_config_setup: None,
    sensor_platform_setup: Callable[[], Awaitable[bool]],
    entity_registry: er.EntityRegistry,
) -> None:
    """Test battery level sensor for devices."""

    assert await sensor_platform_setup()
    entries = hass.config_entries.async_entries(DOMAIN)
    assert len(entries) == 1

    state = hass.states.get("sensor.charge_2_battery")
    assert state
    assert state.state == "Medium"
    assert state.attributes == {
        "attribution": "Data provided by Fitbit.com",
        "friendly_name": "Charge 2 Battery",
        "icon": "mdi:battery-50",
        "model": "Charge 2",
        "type": "tracker",
    }

    entry = entity_registry.async_get("sensor.charge_2_battery")
    assert entry
    assert entry.unique_id == f"{PROFILE_USER_ID}_devices/battery_816713257"

    state = hass.states.get("sensor.aria_air_battery")
    assert state
    assert state.state == "High"
    assert state.attributes == {
        "attribution": "Data provided by Fitbit.com",
        "friendly_name": "Aria Air Battery",
        "icon": "mdi:battery",
        "model": "Aria Air",
        "type": "scale",
    }

    entity_registry = er.async_get(hass)
    entry = entity_registry.async_get("sensor.aria_air_battery")
    assert entry
    assert entry.unique_id == f"{PROFILE_USER_ID}_devices/battery_016713257"


@pytest.mark.parametrize(
    (
        "monitored_resources",
        "profile_locale",
        "configured_unit_system",
        "expected_unit",
    ),
    [
        # Defaults to home assistant unit system unless UK
        (["body/weight"], "en_US", "default", "kg"),
        (["body/weight"], "en_GB", "default", "st"),
        (["body/weight"], "es_ES", "default", "kg"),
        # Use the configured unit system from yaml
        (["body/weight"], "en_US", "en_US", "lb"),
        (["body/weight"], "en_GB", "en_US", "lb"),
        (["body/weight"], "es_ES", "en_US", "lb"),
        (["body/weight"], "en_US", "en_GB", "st"),
        (["body/weight"], "en_GB", "en_GB", "st"),
        (["body/weight"], "es_ES", "en_GB", "st"),
        (["body/weight"], "en_US", "metric", "kg"),
        (["body/weight"], "en_GB", "metric", "kg"),
        (["body/weight"], "es_ES", "metric", "kg"),
    ],
)
async def test_profile_local(
    hass: HomeAssistant,
    fitbit_config_setup: None,
    sensor_platform_setup: Callable[[], Awaitable[bool]],
    register_timeseries: Callable[[str, dict[str, Any]], None],
    expected_unit: str,
) -> None:
    """Test the fitbit profile locale impact on unit of measure."""

    register_timeseries("body/weight", timeseries_response("body-weight", "175"))
    await sensor_platform_setup()
    entries = hass.config_entries.async_entries(DOMAIN)
    assert len(entries) == 1

    state = hass.states.get("sensor.weight")
    assert state
    assert state.attributes.get("unit_of_measurement") == expected_unit


@pytest.mark.parametrize(
    ("sensor_platform_config", "api_response", "expected_state"),
    [
        (
            {"clock_format": "12H", "monitored_resources": ["sleep/startTime"]},
            "17:05",
            "5:05 PM",
        ),
        (
            {"clock_format": "12H", "monitored_resources": ["sleep/startTime"]},
            "5:05",
            "5:05 AM",
        ),
        (
            {"clock_format": "12H", "monitored_resources": ["sleep/startTime"]},
            "00:05",
            "12:05 AM",
        ),
        (
            {"clock_format": "24H", "monitored_resources": ["sleep/startTime"]},
            "17:05",
            "17:05",
        ),
        (
            {"clock_format": "12H", "monitored_resources": ["sleep/startTime"]},
            "",
            "-",
        ),
    ],
)
async def test_sleep_time_clock_format(
    hass: HomeAssistant,
    fitbit_config_setup: None,
    sensor_platform_setup: Callable[[], Awaitable[bool]],
    register_timeseries: Callable[[str, dict[str, Any]], None],
    api_response: str,
    expected_state: str,
) -> None:
    """Test the clock format configuration."""

    register_timeseries(
        "sleep/startTime", timeseries_response("sleep-startTime", api_response)
    )
    await sensor_platform_setup()

    state = hass.states.get("sensor.sleep_start_time")
    assert state
    assert state.state == expected_state


@pytest.mark.parametrize(
    ("scopes"),
    [(["activity"])],
)
async def test_activity_scope_config_entry(
    hass: HomeAssistant,
    setup_credentials: None,
    integration_setup: Callable[[], Awaitable[bool]],
    register_timeseries: Callable[[str, dict[str, Any]], None],
) -> None:
    """Test activity sensors are enabled."""

    for api_resource in (
        "activities/activityCalories",
        "activities/calories",
        "activities/distance",
        "activities/elevation",
        "activities/floors",
        "activities/minutesFairlyActive",
        "activities/minutesLightlyActive",
        "activities/minutesSedentary",
        "activities/minutesVeryActive",
        "activities/steps",
    ):
        register_timeseries(
            api_resource, timeseries_response(api_resource.replace("/", "-"), "0")
        )
    assert await integration_setup()

    states = hass.states.async_all()
    assert {s.entity_id for s in states} == {
        "sensor.activity_calories",
        "sensor.calories",
        "sensor.distance",
        "sensor.elevation",
        "sensor.floors",
        "sensor.minutes_fairly_active",
        "sensor.minutes_lightly_active",
        "sensor.minutes_sedentary",
        "sensor.minutes_very_active",
        "sensor.steps",
    }


@pytest.mark.parametrize(
    ("scopes"),
    [(["heartrate"])],
)
async def test_heartrate_scope_config_entry(
    hass: HomeAssistant,
    setup_credentials: None,
    integration_setup: Callable[[], Awaitable[bool]],
    register_timeseries: Callable[[str, dict[str, Any]], None],
) -> None:
    """Test heartrate sensors are enabled."""

    register_timeseries(
        "activities/heart",
        timeseries_response("activities-heart", {"restingHeartRate": "0"}),
    )
    assert await integration_setup()

    states = hass.states.async_all()
    assert {s.entity_id for s in states} == {
        "sensor.resting_heart_rate",
    }


@pytest.mark.parametrize(
    ("scopes"),
    [(["sleep"])],
)
async def test_sleep_scope_config_entry(
    hass: HomeAssistant,
    setup_credentials: None,
    integration_setup: Callable[[], Awaitable[bool]],
    register_timeseries: Callable[[str, dict[str, Any]], None],
) -> None:
    """Test sleep sensors are enabled."""

    for api_resource in (
        "sleep/startTime",
        "sleep/timeInBed",
        "sleep/minutesToFallAsleep",
        "sleep/minutesAwake",
        "sleep/minutesAsleep",
        "sleep/minutesAfterWakeup",
        "sleep/efficiency",
        "sleep/awakeningsCount",
    ):
        register_timeseries(
            api_resource,
            timeseries_response(api_resource.replace("/", "-"), "0"),
        )
    assert await integration_setup()

    states = hass.states.async_all()
    assert {s.entity_id for s in states} == {
        "sensor.awakenings_count",
        "sensor.sleep_efficiency",
        "sensor.minutes_after_wakeup",
        "sensor.sleep_minutes_asleep",
        "sensor.sleep_minutes_awake",
        "sensor.sleep_minutes_to_fall_asleep",
        "sensor.sleep_time_in_bed",
        "sensor.sleep_start_time",
    }


@pytest.mark.parametrize(
    ("scopes"),
    [(["weight"])],
)
async def test_weight_scope_config_entry(
    hass: HomeAssistant,
    setup_credentials: None,
    integration_setup: Callable[[], Awaitable[bool]],
    register_timeseries: Callable[[str, dict[str, Any]], None],
) -> None:
    """Test sleep sensors are enabled."""

    register_timeseries("body/weight", timeseries_response("body-weight", "0"))
    assert await integration_setup()

    states = hass.states.async_all()
    assert [s.entity_id for s in states] == [
        "sensor.weight",
    ]


@pytest.mark.parametrize(
    ("scopes", "devices_response"),
    [(["settings"], [DEVICE_RESPONSE_CHARGE_2])],
)
async def test_settings_scope_config_entry(
    hass: HomeAssistant,
    setup_credentials: None,
    integration_setup: Callable[[], Awaitable[bool]],
    register_timeseries: Callable[[str, dict[str, Any]], None],
) -> None:
    """Test heartrate sensors are enabled."""

    for api_resource in ("activities/heart",):
        register_timeseries(
            api_resource,
            timeseries_response(
                api_resource.replace("/", "-"), {"restingHeartRate": "0"}
            ),
        )
    assert await integration_setup()

    states = hass.states.async_all()
    assert [s.entity_id for s in states] == [
        "sensor.charge_2_battery",
    ]


@pytest.mark.parametrize(
    ("scopes"),
    [(["heartrate"])],
)
async def test_sensor_update_failed(
    hass: HomeAssistant,
    setup_credentials: None,
    integration_setup: Callable[[], Awaitable[bool]],
    requests_mock: Mocker,
) -> None:
    """Test a failed sensor update when talking to the API."""

    requests_mock.register_uri(
        "GET",
        TIMESERIES_API_URL_FORMAT.format(resource="activities/heart"),
        status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
    )

    assert await integration_setup()

    state = hass.states.get("sensor.resting_heart_rate")
    assert state
    assert state.state == "unavailable"


@pytest.mark.parametrize(
    ("scopes", "mock_devices"),
    [(["settings"], None)],
)
async def test_device_battery_level_update_failed(
    hass: HomeAssistant,
    setup_credentials: None,
    integration_setup: Callable[[], Awaitable[bool]],
    requests_mock: Mocker,
) -> None:
    """Test API failure for a battery level sensor for devices."""

    requests_mock.register_uri(
        "GET",
        DEVICES_API_URL,
        [
            {
                "status_code": HTTPStatus.OK,
                "json": [DEVICE_RESPONSE_CHARGE_2],
            },
            # A second spurious update request on startup
            {
                "status_code": HTTPStatus.OK,
                "json": [DEVICE_RESPONSE_CHARGE_2],
            },
            # Fail when requesting an update
            {
                "status_code": HTTPStatus.INTERNAL_SERVER_ERROR,
                "json": {
                    "errors": [
                        {
                            "errorType": "request",
                            "message": "An error occurred",
                        }
                    ]
                },
            },
        ],
    )

    assert await integration_setup()

    state = hass.states.get("sensor.charge_2_battery")
    assert state
    assert state.state == "Medium"

    # Request an update for the entity which will fail
    await async_update_entity(hass, "sensor.charge_2_battery")

    state = hass.states.get("sensor.charge_2_battery")
    assert state
    assert state.state == "unavailable"