"""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, OAUTH2_TOKEN 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 homeassistant.util.unit_system import ( METRIC_SYSTEM, US_CUSTOMARY_SYSTEM, UnitSystem, ) from .conftest import ( DEVICES_API_URL, PROFILE_USER_ID, SERVER_ACCESS_TOKEN, TIMESERIES_API_URL_FORMAT, timeseries_response, ) from tests.common import MockConfigEntry 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.fixture(autouse=True) def mock_token_refresh(requests_mock: Mocker) -> None: """Test that platform configuration is imported successfully.""" requests_mock.register_uri( "POST", OAUTH2_TOKEN, status_code=HTTPStatus.OK, json=SERVER_ACCESS_TOKEN, ) @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", "unit_system"), [(["nutrition"], METRIC_SYSTEM), (["nutrition"], US_CUSTOMARY_SYSTEM)], ) async def test_nutrition_scope_config_entry( hass: HomeAssistant, setup_credentials: None, integration_setup: Callable[[], Awaitable[bool]], register_timeseries: Callable[[str, dict[str, Any]], None], unit_system: UnitSystem, snapshot: SnapshotAssertion, ) -> None: """Test nutrition sensors are enabled.""" hass.config.units = unit_system register_timeseries( "foods/log/water", timeseries_response("foods-log-water", "99"), ) register_timeseries( "foods/log/caloriesIn", timeseries_response("foods-log-caloriesIn", "1600"), ) assert await integration_setup() state = hass.states.get("sensor.water") assert state assert (state.state, state.attributes) == snapshot state = hass.states.get("sensor.calories_in") assert state assert (state.state, state.attributes) == snapshot @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 device sensors are enabled.""" 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" # Verify the config entry is in a normal state (no reauth required) flows = hass.config_entries.flow.async_progress() assert not flows @pytest.mark.parametrize( ("scopes"), [(["heartrate"])], ) async def test_sensor_update_failed_requires_reauth( hass: HomeAssistant, setup_credentials: None, integration_setup: Callable[[], Awaitable[bool]], requests_mock: Mocker, ) -> None: """Test a sensor update request requires reauth.""" requests_mock.register_uri( "GET", TIMESERIES_API_URL_FORMAT.format(resource="activities/heart"), status_code=HTTPStatus.UNAUTHORIZED, json={ "errors": [{"errorType": "invalid_grant"}], }, ) assert await integration_setup() state = hass.states.get("sensor.resting_heart_rate") assert state assert state.state == "unavailable" # Verify that reauth is required flows = hass.config_entries.flow.async_progress() assert len(flows) == 1 assert flows[0]["step_id"] == "reauth_confirm" @pytest.mark.parametrize( ("scopes"), [(["heartrate"])], ) async def test_sensor_update_success( 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", TIMESERIES_API_URL_FORMAT.format(resource="activities/heart"), [ { "status_code": HTTPStatus.OK, "json": timeseries_response( "activities-heart", {"restingHeartRate": "60"} ), }, { "status_code": HTTPStatus.OK, "json": timeseries_response( "activities-heart", {"restingHeartRate": "70"} ), }, ], ) assert await integration_setup() state = hass.states.get("sensor.resting_heart_rate") assert state assert state.state == "60" await async_update_entity(hass, "sensor.resting_heart_rate") await hass.async_block_till_done() state = hass.states.get("sensor.resting_heart_rate") assert state assert state.state == "70" @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], }, # 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") await hass.async_block_till_done() state = hass.states.get("sensor.charge_2_battery") assert state assert state.state == "unavailable" # Verify the config entry is in a normal state (no reauth required) flows = hass.config_entries.flow.async_progress() assert not flows @pytest.mark.parametrize( ("scopes", "mock_devices"), [(["settings"], None)], ) async def test_device_battery_level_reauth_required( hass: HomeAssistant, setup_credentials: None, integration_setup: Callable[[], Awaitable[bool]], config_entry: MockConfigEntry, requests_mock: Mocker, ) -> None: """Test API failure requires reauth.""" requests_mock.register_uri( "GET", DEVICES_API_URL, [ { "status_code": HTTPStatus.OK, "json": [DEVICE_RESPONSE_CHARGE_2], }, # Fail when requesting an update { "status_code": HTTPStatus.UNAUTHORIZED, "json": { "errors": [{"errorType": "invalid_grant"}], }, }, ], ) 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") await hass.async_block_till_done() state = hass.states.get("sensor.charge_2_battery") assert state assert state.state == "unavailable" # Verify that reauth is required flows = hass.config_entries.flow.async_progress() assert len(flows) == 1 assert flows[0]["step_id"] == "reauth_confirm"