core/tests/components/scrape/test_sensor.py

"""The tests for the Scrape sensor platform."""

from __future__ import annotations

from datetime import timedelta
from unittest.mock import patch

from freezegun.api import FrozenDateTimeFactory
import pytest

from homeassistant.components.scrape.const import (
    CONF_ENCODING,
    CONF_INDEX,
    CONF_SELECT,
    DEFAULT_ENCODING,
    DEFAULT_SCAN_INTERVAL,
    DEFAULT_VERIFY_SSL,
)
from homeassistant.components.sensor import (
    CONF_STATE_CLASS,
    DOMAIN as SENSOR_DOMAIN,
    SensorDeviceClass,
    SensorStateClass,
)
from homeassistant.const import (
    CONF_DEVICE_CLASS,
    CONF_ICON,
    CONF_METHOD,
    CONF_NAME,
    CONF_RESOURCE,
    CONF_TIMEOUT,
    CONF_UNIQUE_ID,
    CONF_UNIT_OF_MEASUREMENT,
    CONF_VALUE_TEMPLATE,
    CONF_VERIFY_SSL,
    STATE_UNAVAILABLE,
    STATE_UNKNOWN,
    UnitOfTemperature,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.trigger_template_entity import (
    CONF_AVAILABILITY,
    CONF_PICTURE,
)
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util

from . import MockRestData, return_integration_config

from tests.common import MockConfigEntry, 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_xml_content_type(
    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
    """Test Scrape sensor with XML Content-Type header uses XML parser."""
    config = {
        DOMAIN: [
            return_integration_config(
                sensors=[
                    {"select": "title", "name": "RSS Title"},
                    # Test <link> tag - HTML parser treats this as self-closing,
                    # but XML parser correctly parses the content
                    {"select": "item link", "name": "RSS Link"},
                ]
            )
        ]
    }

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

    # Verify XML Content-Type header is set
    assert mocker.headers.get("Content-Type") == "application/rss+xml"

    state = hass.states.get("sensor.rss_title")
    assert state.state == "Test RSS Feed"

    # Verify <link> content is correctly parsed with XML parser
    link_state = hass.states.get("sensor.rss_link")
    assert link_state.state == "https://example.com/item"

    assert "XMLParsedAsHTMLWarning" not in caplog.text


async def test_scrape_xml_declaration(
    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
    """Test Scrape sensor with XML declaration (no XML Content-Type) uses XML parser."""
    config = {
        DOMAIN: [
            return_integration_config(
                sensors=[{"select": "title", "name": "RSS Title"}]
            )
        ]
    }

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

    # Verify non-XML Content-Type but XML parser used due to <?xml declaration
    assert mocker.headers.get("Content-Type") == "text/html"

    state = hass.states.get("sensor.rss_title")
    assert state.state == "Test RSS Feed"
    assert "XMLParsedAsHTMLWarning" not in caplog.text


async def test_scrape_html5_with_xml_declaration(
    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
    """Test HTML5 with XML declaration strips XML prefix and uses HTML parser.

    This test verifies backward compatibility by testing:
    - No Content-Type header (relies on content detection)
    - Uppercase HTML tags with lowercase selectors (case-insensitive matching)
    - Class selectors work correctly
    - No XMLParsedAsHTMLWarning is logged
    """
    config = {
        DOMAIN: [
            return_integration_config(
                sensors=[
                    # Lowercase selector matches uppercase <H1> tag
                    {"select": ".current-version h1", "name": "HA version"},
                    # Lowercase selector matches uppercase <TITLE> tag
                    {"select": "title", "name": "Page Title"},
                ]
            )
        ]
    }

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

    # Verify no Content-Type header is set (tests content-based detection)
    assert "Content-Type" not in mocker.headers

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

    title_state = hass.states.get("sensor.page_title")
    assert title_state.state == "Test Page"

    assert "XMLParsedAsHTMLWarning" not 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] == UnitOfTemperature.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, entity_registry: er.EntityRegistry
) -> 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"

    entry = entity_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, caplog: pytest.LogCaptureFixture
) -> 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 await async_setup_component(hass, DOMAIN, config)
        await hass.async_block_till_done()

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

    assert "Platform scrape not ready yet" in caplog.text


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, dt_util.utcnow() + DEFAULT_SCAN_INTERVAL)
        await hass.async_block_till_done(wait_background_tasks=True)

    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_device_date(hass: HomeAssistant) -> None:
    """Test Scrape sensor with a device of type DATE."""
    config = {
        DOMAIN: [
            return_integration_config(
                sensors=[
                    {
                        "select": ".release-date",
                        "name": "HA Date",
                        "device_class": "date",
                        "value_template": "{{ strptime(value, '%B %d, %Y').strftime('%Y-%m-%d') }}",
                    }
                ],
            ),
        ]
    }

    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_date")
    assert state.state == "2022-01-17"


async def test_scrape_sensor_device_date_errors(hass: HomeAssistant) -> None:
    """Test Scrape sensor with a device of type DATE."""
    config = {
        DOMAIN: [
            return_integration_config(
                sensors=[
                    {
                        "select": ".current-version h1",
                        "name": "HA Date",
                        "device_class": "date",
                    }
                ],
            ),
        ]
    }

    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_date")
    assert state.state == STATE_UNKNOWN


async def test_scrape_sensor_device_timestamp(hass: HomeAssistant) -> None:
    """Test Scrape sensor with a device of type TIMESTAMP."""
    config = {
        DOMAIN: [
            return_integration_config(
                sensors=[
                    {
                        "select": ".utc-time",
                        "name": "HA Timestamp",
                        "device_class": "timestamp",
                    }
                ],
            ),
        ]
    }

    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_timestamp")
    assert state.state == "2022-12-22T13:15:30+00:00"


async def test_scrape_sensor_device_timestamp_error(hass: HomeAssistant) -> None:
    """Test Scrape sensor with a device of type TIMESTAMP."""
    config = {
        DOMAIN: [
            return_integration_config(
                sensors=[
                    {
                        "select": ".current-time",
                        "name": "HA Timestamp",
                        "device_class": "timestamp",
                    }
                ],
            ),
        ]
    }

    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_timestamp")
    assert state.state == STATE_UNKNOWN


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_UNAVAILABLE
    state2 = hass.states.get("sensor.ha_class2")
    assert state2.state == STATE_UNAVAILABLE


async def test_scrape_sensor_unique_id(
    hass: HomeAssistant, entity_registry: er.EntityRegistry
) -> 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 = entity_registry.async_get("sensor.ha_version")

    assert entity.unique_id == "ha_version_unique_id"


async def test_setup_config_entry(
    hass: HomeAssistant,
    entity_registry: er.EntityRegistry,
    loaded_entry: MockConfigEntry,
) -> None:
    """Test setup from config entry."""

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

    entity = entity_registry.async_get("sensor.current_version")

    assert entity.unique_id == "3699ef88-69e6-11ed-a1eb-0242ac120002"


async def test_templates_with_yaml(hass: HomeAssistant) -> None:
    """Test the Scrape sensor from yaml config with templates."""

    hass.states.async_set("sensor.input1", "on")
    hass.states.async_set("sensor.input2", "on")
    await hass.async_block_till_done()

    config = {
        DOMAIN: [
            return_integration_config(
                sensors=[
                    {
                        CONF_NAME: "Get values with template",
                        CONF_SELECT: ".current-version h1",
                        CONF_INDEX: 0,
                        CONF_UNIQUE_ID: "3699ef88-69e6-11ed-a1eb-0242ac120002",
                        CONF_ICON: '{% if states("sensor.input1")=="on" %} mdi:on {% else %} mdi:off {% endif %}',
                        CONF_PICTURE: '{% if states("sensor.input1")=="on" %} /local/picture1.jpg {% else %} /local/picture2.jpg {% endif %}',
                        CONF_AVAILABILITY: '{{ states("sensor.input2")=="on" }}',
                    }
                ]
            )
        ]
    }

    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.get_values_with_template")
    assert state.state == "Current Version: 2021.12.10"
    assert state.attributes[CONF_ICON] == "mdi:on"
    assert state.attributes["entity_picture"] == "/local/picture1.jpg"

    hass.states.async_set("sensor.input1", "off")
    await hass.async_block_till_done()

    async_fire_time_changed(
        hass,
        dt_util.utcnow() + timedelta(minutes=10),
    )
    await hass.async_block_till_done(wait_background_tasks=True)

    state = hass.states.get("sensor.get_values_with_template")
    assert state.state == "Current Version: 2021.12.10"
    assert state.attributes[CONF_ICON] == "mdi:off"
    assert state.attributes["entity_picture"] == "/local/picture2.jpg"

    hass.states.async_set("sensor.input2", "off")
    await hass.async_block_till_done()

    async_fire_time_changed(
        hass,
        dt_util.utcnow() + timedelta(minutes=20),
    )
    await hass.async_block_till_done(wait_background_tasks=True)

    state = hass.states.get("sensor.get_values_with_template")
    assert state.state == STATE_UNAVAILABLE

    hass.states.async_set("sensor.input1", "on")
    hass.states.async_set("sensor.input2", "on")
    await hass.async_block_till_done()

    async_fire_time_changed(
        hass,
        dt_util.utcnow() + timedelta(minutes=30),
    )
    await hass.async_block_till_done(wait_background_tasks=True)

    state = hass.states.get("sensor.get_values_with_template")
    assert state.state == "Current Version: 2021.12.10"
    assert state.attributes[CONF_ICON] == "mdi:on"
    assert state.attributes["entity_picture"] == "/local/picture1.jpg"


@pytest.mark.parametrize(
    "get_config",
    [
        {
            CONF_RESOURCE: "https://www.home-assistant.io",
            CONF_METHOD: "GET",
            CONF_VERIFY_SSL: DEFAULT_VERIFY_SSL,
            CONF_TIMEOUT: 10,
            CONF_ENCODING: DEFAULT_ENCODING,
            SENSOR_DOMAIN: [
                {
                    CONF_SELECT: ".current-version h1",
                    CONF_NAME: "Current version",
                    CONF_VALUE_TEMPLATE: "{{ value.split(':')[1] }}",
                    CONF_INDEX: 0,
                    CONF_UNIQUE_ID: "3699ef88-69e6-11ed-a1eb-0242ac120002",
                    CONF_AVAILABILITY: '{{ states("sensor.input1")=="on" }}',
                    CONF_ICON: 'mdi:o{{ "n" if states("sensor.input1")=="on" else "ff" }}',
                    CONF_PICTURE: 'o{{ "n" if states("sensor.input1")=="on" else "ff" }}.jpg',
                }
            ],
        }
    ],
)
async def test_availability(
    hass: HomeAssistant,
    loaded_entry: MockConfigEntry,
    freezer: FrozenDateTimeFactory,
) -> None:
    """Test availability when setup from config entry."""

    hass.states.async_set("sensor.input1", "on")
    freezer.tick(timedelta(minutes=10))
    async_fire_time_changed(hass)
    await hass.async_block_till_done(wait_background_tasks=True)

    state = hass.states.get("sensor.current_version")
    assert state.state == "2021.12.10"
    assert state.attributes["icon"] == "mdi:on"
    assert state.attributes["entity_picture"] == "on.jpg"

    hass.states.async_set("sensor.input1", "off")
    await hass.async_block_till_done()

    freezer.tick(timedelta(minutes=10))
    async_fire_time_changed(hass)
    await hass.async_block_till_done(wait_background_tasks=True)

    state = hass.states.get("sensor.current_version")
    assert state.state == STATE_UNAVAILABLE
    assert "icon" not in state.attributes
    assert "entity_picture" not in state.attributes


async def test_template_render_with_availability_syntax_error(
    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
    """Test availability template render with syntax errors."""
    config = {
        DOMAIN: [
            return_integration_config(
                sensors=[
                    {
                        "select": ".current-version h1",
                        "name": "Current version",
                        "unique_id": "ha_version_unique_id",
                        CONF_VALUE_TEMPLATE: "{{ value.split(':')[1] }}",
                        CONF_AVAILABILITY: "{{ what_the_heck == 2 }}",
                    }
                ]
            )
        ]
    }

    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.current_version")
    assert state.state == "2021.12.10"

    assert (
        "Error rendering availability template for sensor.current_version: UndefinedError: 'what_the_heck' is undefined"
        in caplog.text
    )


async def test_availability_blocks_value_template(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test availability blocks value_template from rendering."""
    error = "Error parsing value for sensor.current_version: 'x' is undefined"
    config = {
        DOMAIN: [
            return_integration_config(
                sensors=[
                    {
                        "select": ".current-version h1",
                        "name": "Current version",
                        "unique_id": "ha_version_unique_id",
                        CONF_VALUE_TEMPLATE: "{{ x - 1 }}",
                        CONF_AVAILABILITY: '{{ states("sensor.input1")=="on" }}',
                    }
                ]
            )
        ]
    }

    hass.states.async_set("sensor.input1", "off")
    await hass.async_block_till_done()

    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()

    assert error not in caplog.text

    state = hass.states.get("sensor.current_version")
    assert state
    assert state.state == STATE_UNAVAILABLE

    hass.states.async_set("sensor.input1", "on")
    await hass.async_block_till_done()

    async_fire_time_changed(
        hass,
        dt_util.utcnow() + timedelta(minutes=10),
    )
    await hass.async_block_till_done(wait_background_tasks=True)

    assert error in caplog.text