core/tests/components/apcupsd/test_sensor.py

"""Test sensors of APCUPSd integration."""

from datetime import timedelta
from unittest.mock import AsyncMock

import pytest
from syrupy.assertion import SnapshotAssertion

from homeassistant.components import automation, script
from homeassistant.components.apcupsd.const import DEPRECATED_SENSORS, DOMAIN
from homeassistant.components.apcupsd.coordinator import REQUEST_REFRESH_COOLDOWN
from homeassistant.const import (
    ATTR_ENTITY_ID,
    STATE_UNAVAILABLE,
    STATE_UNKNOWN,
    Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import (
    device_registry as dr,
    entity_registry as er,
    issue_registry as ir,
)
from homeassistant.setup import async_setup_component
from homeassistant.util import slugify
from homeassistant.util.dt import utcnow

from . import MOCK_MINIMAL_STATUS, MOCK_STATUS

from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform

pytestmark = pytest.mark.usefixtures(
    "entity_registry_enabled_by_default", "init_integration"
)


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


async def test_sensor(
    hass: HomeAssistant,
    entity_registry: er.EntityRegistry,
    device_registry: dr.DeviceRegistry,
    snapshot: SnapshotAssertion,
    mock_config_entry: MockConfigEntry,
    mock_request_status: AsyncMock,
) -> None:
    """Test states of sensor entities."""
    await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)

    # Ensure entities are correctly assigned to device
    device_entry = device_registry.async_get_device(
        identifiers={(DOMAIN, mock_request_status.return_value["SERIALNO"])}
    )
    assert device_entry
    entity_entries = er.async_entries_for_config_entry(
        entity_registry, mock_config_entry.entry_id
    )
    for entity_entry in entity_entries:
        assert entity_entry.device_id == device_entry.id


async def test_state_update(
    hass: HomeAssistant,
    mock_request_status: AsyncMock,
) -> None:
    """Ensure the sensor state changes after updating the data."""
    device_slug = slugify(mock_request_status.return_value["UPSNAME"])
    state = hass.states.get(f"sensor.{device_slug}_load")
    assert state
    assert state.state != STATE_UNAVAILABLE
    assert state.state == "14.0"

    mock_request_status.return_value = MOCK_STATUS | {"LOADPCT": "15.0 Percent"}
    future = utcnow() + timedelta(minutes=2)
    async_fire_time_changed(hass, future)
    await hass.async_block_till_done()

    state = hass.states.get(f"sensor.{device_slug}_load")
    assert state
    assert state.state != STATE_UNAVAILABLE
    assert state.state == "15.0"


async def test_manual_update_entity(
    hass: HomeAssistant,
    mock_request_status: AsyncMock,
) -> None:
    """Test multiple simultaneous manual update entity via service homeassistant/update_entity.

    We should only do network call once for the multiple simultaneous update entity services.
    """
    device_slug = slugify(mock_request_status.return_value["UPSNAME"])
    # Assert the initial state of sensor.ups_load.
    state = hass.states.get(f"sensor.{device_slug}_load")
    assert state
    assert state.state != STATE_UNAVAILABLE
    assert state.state == "14.0"

    # Setup HASS for calling the update_entity service.
    await async_setup_component(hass, "homeassistant", {})

    mock_request_status.return_value = MOCK_STATUS | {
        "LOADPCT": "15.0 Percent",
        "BCHARGE": "99.0 Percent",
    }
    # Now, we fast-forward the time to pass the debouncer cooldown, but put it
    # before the normal update interval to see if the manual update works.
    request_call_count_before = mock_request_status.call_count
    future = utcnow() + timedelta(seconds=REQUEST_REFRESH_COOLDOWN)
    async_fire_time_changed(hass, future)
    await hass.services.async_call(
        "homeassistant",
        "update_entity",
        {
            ATTR_ENTITY_ID: [
                f"sensor.{device_slug}_load",
                f"sensor.{device_slug}_battery",
            ]
        },
        blocking=True,
    )
    # Even if we requested updates for two entities, our integration should smartly
    # group the API calls to just one.
    assert mock_request_status.call_count == request_call_count_before + 1

    # The new state should be effective.
    state = hass.states.get(f"sensor.{device_slug}_load")
    assert state
    assert state.state != STATE_UNAVAILABLE
    assert state.state == "15.0"


@pytest.mark.parametrize(
    ("mock_request_status", "entity_id", "known_status"),
    [
        pytest.param(
            # Even though the "LASTSTEST" field is not available, we should still create the entity.
            MOCK_MINIMAL_STATUS,
            "sensor.apc_ups_last_self_test",
            MOCK_MINIMAL_STATUS | {"LASTSTEST": "1970-01-01 00:00:00 +0000"},
            id="last_self_test_missing",
        ),
        pytest.param(
            MOCK_MINIMAL_STATUS | {"XOFFBATT": "N/A"},
            "sensor.apc_ups_transfer_from_battery",
            MOCK_MINIMAL_STATUS | {"XOFFBATT": "1970-01-01 00:00:00 +0000"},
            id="xoffbatt_na",
        ),
        pytest.param(
            MOCK_MINIMAL_STATUS | {"XOFFBATT": "invalid-time-string"},
            "sensor.apc_ups_transfer_from_battery",
            MOCK_MINIMAL_STATUS | {"XOFFBATT": "1970-01-01 00:00:00 +0000"},
            id="xoffbatt_invalid_time_string",
        ),
    ],
    indirect=["mock_request_status"],
)
async def test_sensor_unknown(
    hass: HomeAssistant,
    mock_request_status: AsyncMock,
    entity_id: str,
    known_status: dict[str, str],
) -> None:
    """Test if our integration can properly mark certain sensors as known/unknown when it becomes so."""
    base_status = mock_request_status.return_value

    # The state should be unknown initially.
    state = hass.states.get(entity_id)
    assert state
    assert state.state == STATE_UNKNOWN

    # Update to a payload that should make the entity known.
    mock_request_status.return_value = known_status
    future = utcnow() + timedelta(minutes=2)
    async_fire_time_changed(hass, future)
    await hass.async_block_till_done()

    state = hass.states.get(entity_id)
    assert state
    assert state.state != STATE_UNKNOWN

    # Revert back to the initial status, and the state should now be unknown again.
    mock_request_status.return_value = base_status
    future = utcnow() + timedelta(minutes=2)
    async_fire_time_changed(hass, future)
    await hass.async_block_till_done()

    state = hass.states.get(entity_id)
    assert state
    assert state.state == STATE_UNKNOWN


@pytest.mark.parametrize(("entity_key", "issue_key"), DEPRECATED_SENSORS.items())
async def test_deprecated_sensor_issue(
    hass: HomeAssistant,
    mock_config_entry: MockConfigEntry,
    mock_request_status: AsyncMock,
    entity_registry: er.EntityRegistry,
    snapshot: SnapshotAssertion,
    entity_key: str,
    issue_key: str,
) -> None:
    """Ensure the issue lists automations and scripts referencing a deprecated sensor."""
    issue_registry = ir.async_get(hass)
    unique_id = f"{mock_request_status.return_value['SERIALNO']}_{entity_key}"
    entity_id = entity_registry.async_get_entity_id("sensor", DOMAIN, unique_id)
    assert entity_id

    # No issue yet.
    issue_id = f"{issue_key}_{entity_id}"
    assert issue_registry.async_get_issue(DOMAIN, issue_id) is None

    # Add automations and scripts referencing the deprecated sensor.
    entity_slug = slugify(entity_key)
    automation_object_id = f"apcupsd_auto_{entity_slug}"
    assert await async_setup_component(
        hass,
        automation.DOMAIN,
        {
            automation.DOMAIN: {
                "id": automation_object_id,
                "alias": f"APC UPS automation ({entity_key})",
                "trigger": {"platform": "state", "entity_id": entity_id},
                "action": {
                    "action": "automation.turn_on",
                    "target": {"entity_id": f"automation.{automation_object_id}"},
                },
            }
        },
    )

    assert await async_setup_component(
        hass,
        script.DOMAIN,
        {
            script.DOMAIN: {
                f"apcupsd_script_{entity_slug}": {
                    "alias": f"APC UPS script ({entity_key})",
                    "sequence": [
                        {
                            "condition": "state",
                            "entity_id": entity_id,
                            "state": "on",
                        }
                    ],
                }
            }
        },
    )
    await hass.config_entries.async_reload(mock_config_entry.entry_id)
    await hass.async_block_till_done()

    issue = issue_registry.async_get_issue(DOMAIN, issue_id)
    # Redact the device ID in the placeholder for consistency.
    issue.translation_placeholders["device_id"] = "<ANY>"
    assert issue == snapshot

    await hass.config_entries.async_unload(mock_config_entry.entry_id)
    await hass.async_block_till_done()

    # Assert the issue is no longer present.
    assert not issue_registry.async_get_issue(DOMAIN, issue_id)
    assert len(issue_registry.issues) == 0