core/tests/components/scrape/test_sensor.py

"""The tests for the Scrape sensor platform."""
from __future__ import annotations

from datetime import datetime
from unittest.mock import patch

from homeassistant.components.scrape.sensor import SCAN_INTERVAL
from homeassistant.components.sensor import (
    CONF_STATE_CLASS,
    DOMAIN as SENSOR_DOMAIN,
    SensorDeviceClass,
    SensorStateClass,
)
from homeassistant.const import (
    CONF_DEVICE_CLASS,
    CONF_UNIT_OF_MEASUREMENT,
    STATE_UNAVAILABLE,
    STATE_UNKNOWN,
    TEMP_CELSIUS,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component

from . import MockRestData, return_config, return_integration_config

from tests.common import async_fire_time_changed

DOMAIN = "scrape"


async def test_scrape_sensor(hass: HomeAssistant) -> None:
    """Test Scrape sensor minimal."""
    config = {
        DOMAIN: [
            return_integration_config(
                sensors=[{"select": ".current-version h1", "name": "HA version"}]
            )
        ]
    }

    mocker = MockRestData("test_scrape_sensor")
    with patch(
        "homeassistant.components.rest.RestData",
        return_value=mocker,
    ):
        assert await async_setup_component(hass, DOMAIN, config)
        await hass.async_block_till_done()

    state = hass.states.get("sensor.ha_version")
    assert state.state == "Current Version: 2021.12.10"


async def test_scrape_sensor_platform_yaml(hass: HomeAssistant) -> None:
    """Test Scrape sensor load from sensor platform."""
    config = {
        SENSOR_DOMAIN: [
            return_config(
                select=".return",
                name="Auth page",
                username="user@secret.com",
                password="12345678",
                authentication="digest",
            ),
            return_config(
                select=".return",
                name="Auth page2",
                username="user@secret.com",
                password="12345678",
            ),
        ]
    }

    mocker = MockRestData("test_scrape_sensor_authentication")
    with patch(
        "homeassistant.components.rest.RestData",
        return_value=mocker,
    ):
        assert await async_setup_component(hass, SENSOR_DOMAIN, config)
        await hass.async_block_till_done()

    state = hass.states.get("sensor.auth_page")
    assert state.state == "secret text"
    state2 = hass.states.get("sensor.auth_page2")
    assert state2.state == "secret text"


async def test_scrape_sensor_platform_yaml_no_data(hass: HomeAssistant, caplog) -> None:
    """Test Scrape sensor load from sensor platform fetching no data."""
    config = {
        SENSOR_DOMAIN: [
            return_config(
                select=".return",
                name="Auth page",
                username="user@secret.com",
                password="12345678",
                authentication="digest",
            ),
        ]
    }

    mocker = MockRestData("test_scrape_sensor_no_data")
    with patch(
        "homeassistant.components.rest.RestData",
        return_value=mocker,
    ):
        assert await async_setup_component(hass, SENSOR_DOMAIN, config)
        await hass.async_block_till_done()

    state = hass.states.get("sensor.auth_page")
    assert not state
    assert "Platform scrape not ready yet: None; Retrying in background" in caplog.text


async def test_scrape_sensor_value_template(hass: HomeAssistant) -> None:
    """Test Scrape sensor with value template."""
    config = {
        DOMAIN: [
            return_integration_config(
                sensors=[
                    {
                        "select": ".current-version h1",
                        "name": "HA version",
                        "value_template": "{{ value.split(':')[1] }}",
                    }
                ]
            )
        ]
    }

    mocker = MockRestData("test_scrape_sensor")
    with patch(
        "homeassistant.components.rest.RestData",
        return_value=mocker,
    ):
        assert await async_setup_component(hass, DOMAIN, config)
        await hass.async_block_till_done()

    state = hass.states.get("sensor.ha_version")
    assert state.state == "2021.12.10"


async def test_scrape_uom_and_classes(hass: HomeAssistant) -> None:
    """Test Scrape sensor for unit of measurement, device class and state class."""
    config = {
        DOMAIN: [
            return_integration_config(
                sensors=[
                    {
                        "select": ".current-temp h3",
                        "name": "Current Temp",
                        "value_template": "{{ value.split(':')[1] }}",
                        "unit_of_measurement": "°C",
                        "device_class": "temperature",
                        "state_class": "measurement",
                    }
                ]
            )
        ]
    }

    mocker = MockRestData("test_scrape_uom_and_classes")
    with patch(
        "homeassistant.components.rest.RestData",
        return_value=mocker,
    ):
        assert await async_setup_component(hass, DOMAIN, config)
        await hass.async_block_till_done()

    state = hass.states.get("sensor.current_temp")
    assert state.state == "22.1"
    assert state.attributes[CONF_UNIT_OF_MEASUREMENT] == TEMP_CELSIUS
    assert state.attributes[CONF_DEVICE_CLASS] == SensorDeviceClass.TEMPERATURE
    assert state.attributes[CONF_STATE_CLASS] == SensorStateClass.MEASUREMENT


async def test_scrape_unique_id(hass: HomeAssistant) -> None:
    """Test Scrape sensor for unique id."""
    config = {
        DOMAIN: return_integration_config(
            sensors=[
                {
                    "select": ".current-temp h3",
                    "name": "Current Temp",
                    "value_template": "{{ value.split(':')[1] }}",
                    "unique_id": "very_unique_id",
                }
            ]
        )
    }

    mocker = MockRestData("test_scrape_uom_and_classes")
    with patch(
        "homeassistant.components.rest.RestData",
        return_value=mocker,
    ):
        assert await async_setup_component(hass, DOMAIN, config)
        await hass.async_block_till_done()

    state = hass.states.get("sensor.current_temp")
    assert state.state == "22.1"

    registry = er.async_get(hass)
    entry = registry.async_get("sensor.current_temp")
    assert entry
    assert entry.unique_id == "very_unique_id"


async def test_scrape_sensor_authentication(hass: HomeAssistant) -> None:
    """Test Scrape sensor with authentication."""
    config = {
        DOMAIN: [
            return_integration_config(
                authentication="digest",
                username="user@secret.com",
                password="12345678",
                sensors=[
                    {
                        "select": ".return",
                        "name": "Auth page",
                    },
                ],
            ),
            return_integration_config(
                username="user@secret.com",
                password="12345678",
                sensors=[
                    {
                        "select": ".return",
                        "name": "Auth page2",
                    },
                ],
            ),
        ]
    }

    mocker = MockRestData("test_scrape_sensor_authentication")
    with patch(
        "homeassistant.components.rest.RestData",
        return_value=mocker,
    ):
        assert await async_setup_component(hass, DOMAIN, config)
        await hass.async_block_till_done()

    state = hass.states.get("sensor.auth_page")
    assert state.state == "secret text"
    state2 = hass.states.get("sensor.auth_page2")
    assert state2.state == "secret text"


async def test_scrape_sensor_no_data(hass: HomeAssistant) -> None:
    """Test Scrape sensor fails on no data."""
    config = {
        DOMAIN: return_integration_config(
            sensors=[{"select": ".current-version h1", "name": "HA version"}]
        )
    }

    mocker = MockRestData("test_scrape_sensor_no_data")
    with patch(
        "homeassistant.components.rest.RestData",
        return_value=mocker,
    ):
        assert not await async_setup_component(hass, DOMAIN, config)
        await hass.async_block_till_done()

    state = hass.states.get("sensor.ha_version")
    assert state is None


async def test_scrape_sensor_no_data_refresh(hass: HomeAssistant) -> None:
    """Test Scrape sensor no data on refresh."""
    config = {
        DOMAIN: [
            return_integration_config(
                sensors=[{"select": ".current-version h1", "name": "HA version"}]
            )
        ]
    }

    mocker = MockRestData("test_scrape_sensor")
    with patch(
        "homeassistant.components.rest.RestData",
        return_value=mocker,
    ):
        assert await async_setup_component(hass, DOMAIN, config)
        await hass.async_block_till_done()

    state = hass.states.get("sensor.ha_version")
    assert state
    assert state.state == "Current Version: 2021.12.10"

    mocker.payload = "test_scrape_sensor_no_data"
    async_fire_time_changed(hass, datetime.utcnow() + SCAN_INTERVAL)
    await hass.async_block_till_done()

    state = hass.states.get("sensor.ha_version")
    assert state is not None
    assert state.state == STATE_UNAVAILABLE


async def test_scrape_sensor_attribute_and_tag(hass: HomeAssistant) -> None:
    """Test Scrape sensor with attribute and tag."""
    config = {
        DOMAIN: [
            return_integration_config(
                sensors=[
                    {
                        "index": 1,
                        "select": "div",
                        "name": "HA class",
                        "attribute": "class",
                    },
                    {"select": "template", "name": "HA template"},
                ],
            ),
        ]
    }

    mocker = MockRestData("test_scrape_sensor")
    with patch(
        "homeassistant.components.rest.RestData",
        return_value=mocker,
    ):
        assert await async_setup_component(hass, DOMAIN, config)
        await hass.async_block_till_done()

    state = hass.states.get("sensor.ha_class")
    assert state.state == "['links']"
    state2 = hass.states.get("sensor.ha_template")
    assert state2.state == "Trying to get"


async def test_scrape_sensor_errors(hass: HomeAssistant) -> None:
    """Test Scrape sensor handle errors."""
    config = {
        DOMAIN: [
            return_integration_config(
                sensors=[
                    {
                        "index": 5,
                        "select": "div",
                        "name": "HA class",
                        "attribute": "class",
                    },
                    {
                        "select": "div",
                        "name": "HA class2",
                        "attribute": "classes",
                    },
                ],
            ),
        ]
    }

    mocker = MockRestData("test_scrape_sensor")
    with patch(
        "homeassistant.components.rest.RestData",
        return_value=mocker,
    ):
        assert await async_setup_component(hass, DOMAIN, config)
        await hass.async_block_till_done()

    state = hass.states.get("sensor.ha_class")
    assert state.state == STATE_UNKNOWN
    state2 = hass.states.get("sensor.ha_class2")
    assert state2.state == STATE_UNKNOWN


async def test_scrape_sensor_unique_id(hass: HomeAssistant) -> None:
    """Test Scrape sensor with unique_id."""
    config = {
        DOMAIN: [
            return_integration_config(
                sensors=[
                    {
                        "select": ".current-version h1",
                        "name": "HA version",
                        "unique_id": "ha_version_unique_id",
                    }
                ]
            )
        ]
    }

    mocker = MockRestData("test_scrape_sensor")
    with patch(
        "homeassistant.components.rest.RestData",
        return_value=mocker,
    ):
        assert await async_setup_component(hass, DOMAIN, config)
        await hass.async_block_till_done()

    state = hass.states.get("sensor.ha_version")
    assert state.state == "Current Version: 2021.12.10"

    entity_reg = er.async_get(hass)
    entity = entity_reg.async_get("sensor.ha_version")

    assert entity.unique_id == "ha_version_unique_id"


async def test_scrape_sensor_not_configured_sensor(hass: HomeAssistant, caplog) -> None:
    """Test Scrape sensor with missing configured sensors."""
    config = {DOMAIN: [return_integration_config(sensors=None)]}

    mocker = MockRestData("test_scrape_sensor")
    with patch(
        "homeassistant.components.rest.RestData",
        return_value=mocker,
    ):
        assert not await async_setup_component(hass, DOMAIN, config)
        await hass.async_block_till_done()

    assert "No sensors configured" in caplog.text