core/tests/components/picnic/test_sensor.py

"""The tests for the Picnic sensor platform."""
import copy
from datetime import timedelta
from typing import Dict
import unittest
from unittest.mock import patch

import pytest
import requests

from homeassistant import config_entries
from homeassistant.components.picnic import const
from homeassistant.components.picnic.const import CONF_COUNTRY_CODE, SENSOR_TYPES
from homeassistant.components.sensor import SensorDeviceClass
from homeassistant.const import (
    CONF_ACCESS_TOKEN,
    CURRENCY_EURO,
    STATE_UNAVAILABLE,
    STATE_UNKNOWN,
)
from homeassistant.helpers.device_registry import DeviceEntryType
from homeassistant.util import dt

from tests.common import (
    MockConfigEntry,
    async_fire_time_changed,
    async_test_home_assistant,
)

DEFAULT_USER_RESPONSE = {
    "user_id": "295-6y3-1nf4",
    "firstname": "User",
    "lastname": "Name",
    "address": {
        "house_number": 123,
        "house_number_ext": "a",
        "postcode": "4321 AB",
        "street": "Commonstreet",
        "city": "Somewhere",
    },
    "total_deliveries": 123,
    "completed_deliveries": 112,
}
DEFAULT_CART_RESPONSE = {
    "items": [],
    "delivery_slots": [
        {
            "slot_id": "611a3b074872b23576bef456a",
            "window_start": "2021-03-03T14:45:00.000+01:00",
            "window_end": "2021-03-03T15:45:00.000+01:00",
            "cut_off_time": "2021-03-02T22:00:00.000+01:00",
            "minimum_order_value": 3500,
        },
    ],
    "selected_slot": {"slot_id": "611a3b074872b23576bef456a", "state": "EXPLICIT"},
    "total_count": 10,
    "total_price": 2535,
}
DEFAULT_DELIVERY_RESPONSE = {
    "delivery_id": "z28fjso23e",
    "creation_time": "2021-02-24T21:48:46.395+01:00",
    "slot": {
        "slot_id": "602473859a40dc24c6b65879",
        "hub_id": "AMS",
        "window_start": "2021-02-26T20:15:00.000+01:00",
        "window_end": "2021-02-26T21:15:00.000+01:00",
        "cut_off_time": "2021-02-25T22:00:00.000+01:00",
        "minimum_order_value": 3500,
    },
    "eta2": {
        "start": "2021-02-26T20:54:00.000+01:00",
        "end": "2021-02-26T21:14:00.000+01:00",
    },
    "status": "COMPLETED",
    "delivery_time": {
        "start": "2021-02-26T20:54:05.221+01:00",
        "end": "2021-02-26T20:58:31.802+01:00",
    },
    "orders": [
        {
            "creation_time": "2021-02-24T21:48:46.418+01:00",
            "total_price": 3597,
        },
        {
            "creation_time": "2021-02-25T17:10:26.816+01:00",
            "total_price": 536,
        },
    ],
}

SENSOR_KEYS = [desc.key for desc in SENSOR_TYPES]


@pytest.mark.usefixtures("hass_storage")
class TestPicnicSensor(unittest.IsolatedAsyncioTestCase):
    """Test the Picnic sensor."""

    async def asyncSetUp(self):
        """Set up things to be run when tests are started."""
        self.hass = await async_test_home_assistant(None)
        self.entity_registry = (
            await self.hass.helpers.entity_registry.async_get_registry()
        )

        # Patch the api client
        self.picnic_patcher = patch("homeassistant.components.picnic.PicnicAPI")
        self.picnic_mock = self.picnic_patcher.start()
        self.picnic_mock().session.auth_token = "3q29fpwhulzes"

        # Add a config entry and setup the integration
        config_data = {
            CONF_ACCESS_TOKEN: "x-original-picnic-auth-token",
            CONF_COUNTRY_CODE: "NL",
        }
        self.config_entry = MockConfigEntry(
            domain=const.DOMAIN,
            data=config_data,
            unique_id="295-6y3-1nf4",
        )
        self.config_entry.add_to_hass(self.hass)

    async def asyncTearDown(self):
        """Tear down the test setup, stop hass/patchers."""
        await self.hass.async_stop(force=True)
        self.picnic_patcher.stop()

    @property
    def _coordinator(self):
        return self.hass.data[const.DOMAIN][self.config_entry.entry_id][
            const.CONF_COORDINATOR
        ]

    def _assert_sensor(self, name, state=None, cls=None, unit=None, disabled=False):
        sensor = self.hass.states.get(name)
        if disabled:
            assert sensor is None
            return

        assert sensor.state == state
        if cls:
            assert sensor.attributes["device_class"] == cls
        if unit:
            assert sensor.attributes["unit_of_measurement"] == unit

        assert sensor.attributes["attribution"] == "Data provided by Picnic"

    async def _setup_platform(
        self, use_default_responses=False, enable_all_sensors=True
    ):
        """Set up the Picnic sensor platform."""
        if use_default_responses:
            self.picnic_mock().get_user.return_value = copy.deepcopy(
                DEFAULT_USER_RESPONSE
            )
            self.picnic_mock().get_cart.return_value = copy.deepcopy(
                DEFAULT_CART_RESPONSE
            )
            self.picnic_mock().get_deliveries.return_value = [
                copy.deepcopy(DEFAULT_DELIVERY_RESPONSE)
            ]
            self.picnic_mock().get_delivery_position.return_value = {}

        await self.hass.config_entries.async_setup(self.config_entry.entry_id)
        await self.hass.async_block_till_done()

        if enable_all_sensors:
            await self._enable_all_sensors()

    async def _enable_all_sensors(self):
        """Enable all sensors of the Picnic integration."""
        # Enable the sensors
        for sensor_type in SENSOR_KEYS:
            updated_entry = self.entity_registry.async_update_entity(
                f"sensor.picnic_{sensor_type}", disabled_by=None
            )
            assert updated_entry.disabled is False
        await self.hass.async_block_till_done()

        # Trigger a reload of the data
        async_fire_time_changed(
            self.hass,
            dt.utcnow()
            + timedelta(seconds=config_entries.RELOAD_AFTER_UPDATE_DELAY + 1),
        )
        await self.hass.async_block_till_done()

    async def test_sensor_setup_platform_not_available(self):
        """Test the set-up of the sensor platform if API is not available."""
        # Configure mock requests to yield exceptions
        self.picnic_mock().get_user.side_effect = requests.exceptions.ConnectionError
        self.picnic_mock().get_cart.side_effect = requests.exceptions.ConnectionError
        self.picnic_mock().get_deliveries.side_effect = (
            requests.exceptions.ConnectionError
        )
        self.picnic_mock().get_delivery_position.side_effect = (
            requests.exceptions.ConnectionError
        )
        await self._setup_platform(enable_all_sensors=False)

        # Assert that sensors are not set up
        assert (
            self.hass.states.get("sensor.picnic_selected_slot_max_order_time") is None
        )
        assert self.hass.states.get("sensor.picnic_last_order_status") is None
        assert self.hass.states.get("sensor.picnic_last_order_total_price") is None

    async def test_sensors_setup(self):
        """Test the default sensor setup behaviour."""
        await self._setup_platform(use_default_responses=True)

        self._assert_sensor("sensor.picnic_cart_items_count", "10")
        self._assert_sensor(
            "sensor.picnic_cart_total_price", "25.35", unit=CURRENCY_EURO
        )
        self._assert_sensor(
            "sensor.picnic_selected_slot_start",
            "2021-03-03T13:45:00+00:00",
            cls=SensorDeviceClass.TIMESTAMP,
        )
        self._assert_sensor(
            "sensor.picnic_selected_slot_end",
            "2021-03-03T14:45:00+00:00",
            cls=SensorDeviceClass.TIMESTAMP,
        )
        self._assert_sensor(
            "sensor.picnic_selected_slot_max_order_time",
            "2021-03-02T21:00:00+00:00",
            cls=SensorDeviceClass.TIMESTAMP,
        )
        self._assert_sensor("sensor.picnic_selected_slot_min_order_value", "35.0")
        self._assert_sensor(
            "sensor.picnic_last_order_slot_start",
            "2021-02-26T19:15:00+00:00",
            cls=SensorDeviceClass.TIMESTAMP,
        )
        self._assert_sensor(
            "sensor.picnic_last_order_slot_end",
            "2021-02-26T20:15:00+00:00",
            cls=SensorDeviceClass.TIMESTAMP,
        )
        self._assert_sensor("sensor.picnic_last_order_status", "COMPLETED")
        self._assert_sensor(
            "sensor.picnic_last_order_eta_start",
            "2021-02-26T19:54:00+00:00",
            cls=SensorDeviceClass.TIMESTAMP,
        )
        self._assert_sensor(
            "sensor.picnic_last_order_eta_end",
            "2021-02-26T20:14:00+00:00",
            cls=SensorDeviceClass.TIMESTAMP,
        )
        self._assert_sensor(
            "sensor.picnic_last_order_max_order_time",
            "2021-02-25T21:00:00+00:00",
            cls=SensorDeviceClass.TIMESTAMP,
        )
        self._assert_sensor(
            "sensor.picnic_last_order_delivery_time",
            "2021-02-26T19:54:05+00:00",
            cls=SensorDeviceClass.TIMESTAMP,
        )
        self._assert_sensor(
            "sensor.picnic_last_order_total_price", "41.33", unit=CURRENCY_EURO
        )

    async def test_sensors_setup_disabled_by_default(self):
        """Test that some sensors are disabled by default."""
        await self._setup_platform(use_default_responses=True, enable_all_sensors=False)

        self._assert_sensor("sensor.picnic_cart_items_count", disabled=True)
        self._assert_sensor("sensor.picnic_last_order_slot_start", disabled=True)
        self._assert_sensor("sensor.picnic_last_order_slot_end", disabled=True)
        self._assert_sensor("sensor.picnic_last_order_status", disabled=True)
        self._assert_sensor("sensor.picnic_last_order_total_price", disabled=True)

    async def test_sensors_no_selected_time_slot(self):
        """Test sensor states with no explicit selected time slot."""
        # Adjust cart response
        cart_response = copy.deepcopy(DEFAULT_CART_RESPONSE)
        cart_response["selected_slot"]["state"] = "IMPLICIT"

        # Set mock responses
        self.picnic_mock().get_user.return_value = copy.deepcopy(DEFAULT_USER_RESPONSE)
        self.picnic_mock().get_cart.return_value = cart_response
        self.picnic_mock().get_deliveries.return_value = [
            copy.deepcopy(DEFAULT_DELIVERY_RESPONSE)
        ]
        self.picnic_mock().get_delivery_position.return_value = {}
        await self._setup_platform()

        # Assert sensors are unknown
        self._assert_sensor("sensor.picnic_selected_slot_start", STATE_UNKNOWN)
        self._assert_sensor("sensor.picnic_selected_slot_end", STATE_UNKNOWN)
        self._assert_sensor("sensor.picnic_selected_slot_max_order_time", STATE_UNKNOWN)
        self._assert_sensor(
            "sensor.picnic_selected_slot_min_order_value", STATE_UNKNOWN
        )

    async def test_sensors_last_order_in_future(self):
        """Test sensor states when last order is not yet delivered."""
        # Adjust default delivery response
        delivery_response = copy.deepcopy(DEFAULT_DELIVERY_RESPONSE)
        del delivery_response["delivery_time"]

        # Set mock responses
        self.picnic_mock().get_user.return_value = copy.deepcopy(DEFAULT_USER_RESPONSE)
        self.picnic_mock().get_cart.return_value = copy.deepcopy(DEFAULT_CART_RESPONSE)
        self.picnic_mock().get_deliveries.return_value = [delivery_response]
        self.picnic_mock().get_delivery_position.return_value = {}
        await self._setup_platform()

        # Assert delivery time is not available, but eta is
        self._assert_sensor("sensor.picnic_last_order_delivery_time", STATE_UNKNOWN)
        self._assert_sensor(
            "sensor.picnic_last_order_eta_start", "2021-02-26T19:54:00+00:00"
        )
        self._assert_sensor(
            "sensor.picnic_last_order_eta_end", "2021-02-26T20:14:00+00:00"
        )

    async def test_sensors_eta_date_malformed(self):
        """Test sensor states when last order eta dates are malformed."""
        # Set-up platform with default mock responses
        await self._setup_platform(use_default_responses=True)

        # Set non-datetime strings as eta
        eta_dates: Dict[str, str] = {
            "start": "wrong-time",
            "end": "other-malformed-datetime",
        }
        delivery_response = copy.deepcopy(DEFAULT_DELIVERY_RESPONSE)
        delivery_response["eta2"] = eta_dates
        self.picnic_mock().get_deliveries.return_value = [delivery_response]
        await self._coordinator.async_refresh()

        # Assert eta times are not available due to malformed date strings
        self._assert_sensor("sensor.picnic_last_order_eta_start", STATE_UNKNOWN)
        self._assert_sensor("sensor.picnic_last_order_eta_end", STATE_UNKNOWN)

    async def test_sensors_use_detailed_eta_if_available(self):
        """Test sensor states when last order is not yet delivered."""
        # Set-up platform with default mock responses
        await self._setup_platform(use_default_responses=True)

        # Provide a delivery position response with different ETA and remove delivery time from response
        delivery_response = copy.deepcopy(DEFAULT_DELIVERY_RESPONSE)
        del delivery_response["delivery_time"]
        self.picnic_mock().get_deliveries.return_value = [delivery_response]
        self.picnic_mock().get_delivery_position.return_value = {
            "eta_window": {
                "start": "2021-03-05T10:19:20.452+00:00",
                "end": "2021-03-05T10:39:20.452+00:00",
            }
        }
        await self._coordinator.async_refresh()

        # Assert detailed ETA is used
        self.picnic_mock().get_delivery_position.assert_called_with(
            delivery_response["delivery_id"]
        )
        self._assert_sensor(
            "sensor.picnic_last_order_eta_start", "2021-03-05T10:19:20+00:00"
        )
        self._assert_sensor(
            "sensor.picnic_last_order_eta_end", "2021-03-05T10:39:20+00:00"
        )

    async def test_sensors_no_data(self):
        """Test sensor states when the api only returns empty objects."""
        # Setup platform with default responses
        await self._setup_platform(use_default_responses=True)

        # Change mock responses to empty data and refresh the coordinator
        self.picnic_mock().get_user.return_value = {}
        self.picnic_mock().get_cart.return_value = None
        self.picnic_mock().get_deliveries.return_value = None
        self.picnic_mock().get_delivery_position.side_effect = ValueError
        await self._coordinator.async_refresh()

        # Assert all default-enabled sensors have STATE_UNAVAILABLE because the last update failed
        assert self._coordinator.last_update_success is False
        self._assert_sensor("sensor.picnic_cart_total_price", STATE_UNAVAILABLE)
        self._assert_sensor("sensor.picnic_selected_slot_start", STATE_UNAVAILABLE)
        self._assert_sensor("sensor.picnic_selected_slot_end", STATE_UNAVAILABLE)
        self._assert_sensor(
            "sensor.picnic_selected_slot_max_order_time", STATE_UNAVAILABLE
        )
        self._assert_sensor(
            "sensor.picnic_selected_slot_min_order_value", STATE_UNAVAILABLE
        )
        self._assert_sensor("sensor.picnic_last_order_eta_start", STATE_UNAVAILABLE)
        self._assert_sensor("sensor.picnic_last_order_eta_end", STATE_UNAVAILABLE)
        self._assert_sensor(
            "sensor.picnic_last_order_max_order_time", STATE_UNAVAILABLE
        )
        self._assert_sensor("sensor.picnic_last_order_delivery_time", STATE_UNAVAILABLE)

    async def test_sensors_malformed_delivery_data(self):
        """Test sensor states when the delivery api returns not a list."""
        # Setup platform with default responses
        await self._setup_platform(use_default_responses=True)

        # Change mock responses to empty data and refresh the coordinator
        self.picnic_mock().get_deliveries.return_value = {"error": "message"}
        await self._coordinator.async_refresh()

        # Assert all last-order sensors have STATE_UNAVAILABLE because the delivery info fetch failed
        assert self._coordinator.last_update_success is True
        self._assert_sensor("sensor.picnic_last_order_eta_start", STATE_UNKNOWN)
        self._assert_sensor("sensor.picnic_last_order_eta_end", STATE_UNKNOWN)
        self._assert_sensor("sensor.picnic_last_order_max_order_time", STATE_UNKNOWN)
        self._assert_sensor("sensor.picnic_last_order_delivery_time", STATE_UNKNOWN)

    async def test_sensors_malformed_response(self):
        """Test coordinator update fails when API yields ValueError."""
        # Setup platform with default responses
        await self._setup_platform(use_default_responses=True)

        # Change mock responses to empty data and refresh the coordinator
        self.picnic_mock().get_user.side_effect = ValueError
        self.picnic_mock().get_cart.side_effect = ValueError
        await self._coordinator.async_refresh()

        # Assert coordinator update failed
        assert self._coordinator.last_update_success is False

    async def test_device_registry_entry(self):
        """Test if device registry entry is populated correctly."""
        # Setup platform and default mock responses
        await self._setup_platform(use_default_responses=True)

        device_registry = await self.hass.helpers.device_registry.async_get_registry()
        picnic_service = device_registry.async_get_device(
            identifiers={(const.DOMAIN, DEFAULT_USER_RESPONSE["user_id"])}
        )
        assert picnic_service.model == DEFAULT_USER_RESPONSE["user_id"]
        assert picnic_service.name == "Picnic: Commonstreet 123a"
        assert picnic_service.entry_type is DeviceEntryType.SERVICE

    async def test_auth_token_is_saved_on_update(self):
        """Test that auth-token changes in the session object are reflected by the config entry."""
        # Setup platform and default mock responses
        await self._setup_platform(use_default_responses=True)

        # Set a different auth token in the session mock
        updated_auth_token = "x-updated-picnic-auth-token"
        self.picnic_mock().session.auth_token = updated_auth_token

        # Verify the updated auth token is not set and fetch data using the coordinator
        assert self.config_entry.data.get(CONF_ACCESS_TOKEN) != updated_auth_token
        await self._coordinator.async_refresh()

        # Verify that the updated auth token is saved in the config entry
        assert self.config_entry.data.get(CONF_ACCESS_TOKEN) == updated_auth_token