notify-bridge/packages/server/tests/test_home_assistant_parser.py

"""Unit tests for the Home Assistant event parser.

These tests don't need a database or HA server — the parser is a pure
function from ``ha_event_dict`` to :class:`ServiceEvent`.
"""

from __future__ import annotations

from notify_bridge_core.models.events import EventType
from notify_bridge_core.providers.base import ServiceProviderType
from notify_bridge_core.providers.home_assistant.event_parser import parse_event


def _ha_event_envelope(event_type: str, data: dict) -> dict:
    return {
        "event_type": event_type,
        "data": data,
        "time_fired": "2026-05-13T12:34:56.789Z",
    }


def test_state_changed_basic() -> None:
    payload = _ha_event_envelope(
        "state_changed",
        {
            "entity_id": "binary_sensor.front_door",
            "old_state": {"state": "off", "attributes": {}},
            "new_state": {
                "state": "on",
                "attributes": {
                    "friendly_name": "Front Door",
                    "device_class": "door",
                },
            },
        },
    )
    evt = parse_event(payload, provider_name="HA")
    assert evt is not None
    assert evt.event_type is EventType.HA_STATE_CHANGED
    assert evt.provider_type is ServiceProviderType.HOME_ASSISTANT
    assert evt.collection_id == "binary_sensor.front_door"
    assert evt.collection_name == "Front Door"
    assert evt.extra["old_state"] == "off"
    assert evt.extra["new_state"] == "on"
    assert evt.extra["domain"] == "binary_sensor"
    assert evt.extra["device_class"] == "door"
    # Area was not provided in lookup -> None.
    assert evt.extra["area"] is None


def test_state_changed_with_area_lookup() -> None:
    payload = _ha_event_envelope(
        "state_changed",
        {
            "entity_id": "light.kitchen",
            "old_state": {"state": "off", "attributes": {}},
            "new_state": {
                "state": "on",
                "attributes": {"friendly_name": "Kitchen Light"},
            },
        },
    )
    evt = parse_event(
        payload,
        provider_name="HA",
        area_lookup={"light.kitchen": "Kitchen"},
    )
    assert evt is not None
    assert evt.extra["area"] == "Kitchen"


def test_state_changed_entity_removed() -> None:
    """new_state=None means HA removed the entity. Surface as 'removed' so
    templates can branch on it; collection_name falls back to old_state."""
    payload = _ha_event_envelope(
        "state_changed",
        {
            "entity_id": "sensor.dropped",
            "old_state": {
                "state": "online",
                "attributes": {"friendly_name": "Dropped Sensor"},
            },
            "new_state": None,
        },
    )
    evt = parse_event(payload, provider_name="HA")
    assert evt is not None
    assert evt.extra["new_state"] == "removed"
    assert evt.collection_name == "Dropped Sensor"


def test_automation_triggered() -> None:
    payload = _ha_event_envelope(
        "automation_triggered",
        {
            "name": "Front door notification",
            "entity_id": "automation.front_door_notify",
            "source": "state of binary_sensor.front_door",
        },
    )
    evt = parse_event(payload, provider_name="HA")
    assert evt is not None
    assert evt.event_type is EventType.HA_AUTOMATION_TRIGGERED
    assert evt.collection_name == "Front door notification"
    assert evt.collection_id == "automation.front_door_notify"
    assert evt.extra["automation_name"] == "Front door notification"
    assert evt.extra["trigger_source"] == "state of binary_sensor.front_door"


def test_call_service_with_target() -> None:
    payload = _ha_event_envelope(
        "call_service",
        {
            "domain": "light",
            "service": "turn_on",
            "service_data": {"entity_id": "light.kitchen"},
        },
    )
    evt = parse_event(payload, provider_name="HA")
    assert evt is not None
    assert evt.event_type is EventType.HA_SERVICE_CALLED
    assert evt.collection_id == "light.turn_on"
    assert evt.extra["target_entity"] == "light.kitchen"
    assert evt.extra["service_domain"] == "light"
    assert evt.extra["service_name"] == "turn_on"


def test_call_service_with_multi_target() -> None:
    """When the call hits multiple entities, the parser comma-joins them
    so templates can render ``{{ target_entity }}`` without iterating."""
    payload = _ha_event_envelope(
        "call_service",
        {
            "domain": "light",
            "service": "turn_off",
            "service_data": {
                "entity_id": ["light.kitchen", "light.living_room"],
            },
        },
    )
    evt = parse_event(payload, provider_name="HA")
    assert evt is not None
    assert evt.extra["target_entity"] == "light.kitchen, light.living_room"


def test_generic_event_fallback() -> None:
    """Any event_type not in the known set becomes ha_event_fired with the
    raw event_type stashed in extras so loud catch-all subscriptions work."""
    payload = _ha_event_envelope(
        "custom_event_xyz",
        {"foo": "bar"},
    )
    evt = parse_event(payload, provider_name="HA")
    assert evt is not None
    assert evt.event_type is EventType.HA_EVENT_FIRED
    assert evt.extra["ha_event_type"] == "custom_event_xyz"
    assert evt.extra["event_data"] == {"foo": "bar"}


def test_malformed_payload_returns_none() -> None:
    assert parse_event({}, provider_name="HA") is None
    assert parse_event("not a dict", provider_name="HA") is None  # type: ignore[arg-type]
    # state_changed without entity_id is unrecoverable
    bad = _ha_event_envelope("state_changed", {"new_state": None})
    assert parse_event(bad, provider_name="HA") is None
    # call_service without domain/service is unrecoverable
    bad2 = _ha_event_envelope("call_service", {"service": "turn_on"})
    assert parse_event(bad2, provider_name="HA") is None


def test_time_fired_iso_with_z_suffix_parses() -> None:
    """HA uses ``Z`` suffix; older Python ``fromisoformat`` rejects it.
    The parser must handle both forms or we'd lose the timestamp."""
    from datetime import timezone
    payload = _ha_event_envelope(
        "state_changed",
        {
            "entity_id": "sensor.temp",
            "old_state": {"state": "20", "attributes": {}},
            "new_state": {"state": "21", "attributes": {}},
        },
    )
    payload["time_fired"] = "2026-05-13T12:34:56.789Z"
    evt = parse_event(payload, provider_name="HA")
    assert evt is not None
    assert evt.timestamp.tzinfo is not None
    assert evt.timestamp.utcoffset() == timezone.utc.utcoffset(None)