Add Voicemail Status Entity for Real-time Voicemail Monitoring

custom_components/asterisk/sensor.py

@@ -1,6 +1,6 @@
 import logging
 
-from asterisk.ami import Event
+from asterisk.ami import Event, SimpleAction
 from homeassistant.components.sensor import SensorDeviceClass, SensorEntity
 from homeassistant.const import CONF_DEVICES
 from homeassistant.util.dt import now
@@ -22,6 +22,7 @@ async def async_setup_entry(hass, entry, async_add_entities):
         entities.append(ConnectedLineSensor(hass, entry, device))
         entities.append(DTMFSentSensor(hass, entry, device))
         entities.append(DTMFReceivedSensor(hass, entry, device))
+        entities.append(VoicemailStatusSensor(hass, entry, device))
 
     async_add_entities(entities, False)
 
@@ -251,3 +252,89 @@ def device_class(self) -> SensorDeviceClass:
     def extra_state_attributes(self):
         """Return the state attributes."""
         return self._extra_attributes
+
+
+class VoicemailStatusSensor(AsteriskDeviceEntity, SensorEntity):
+    """Sensor entity for voicemail status."""
+
+    def __init__(self, hass, entry, device):
+        """Initialize the sensor."""
+        super().__init__(hass, entry, device)
+        self._unique_id = f"{self._unique_id_prefix}_voicemail_status"
+        self._name = f"{device['extension']} Voicemail Status"
+        self._state = 0
+        self._extra_attributes = {}
+        self._mailbox = f"{device['extension']}@default"
+
+        # Listen for MWI (Message Waiting Indicator) events
+        self._ami_client.add_event_listener(
+            self.handle_mwi,
+            white_list=["MWI"],
+            Mailbox=self._mailbox,
+        )
+
+        # Get initial voicemail status
+        self._get_initial_status()
+
+    def _get_initial_status(self):
+        """Get initial voicemail status using MailboxStatus action."""
+        try:
+            future = self._ami_client.send_action(
+                SimpleAction("MailboxStatus", Mailbox=self._mailbox)
+            )
+            if future.response and not future.response.is_error():
+                self._update_from_response(future.response.keys)
+        except Exception as e:
+            _LOGGER.warning(
+                f"Failed to get initial voicemail status for {self._mailbox}: {e}"
+            )
+
+    def _update_from_response(self, response_data):
+        """Update sensor state from AMI response data."""
+        try:
+            new_messages = int(response_data.get("NewMessages", 0))
+            old_messages = int(response_data.get("OldMessages", 0))
+
+            self._state = new_messages
+            self._extra_attributes = {
+                "new_messages": new_messages,
+                "old_messages": old_messages,
+                "total_messages": new_messages + old_messages,
+                "mailbox": self._mailbox,
+            }
+        except (ValueError, TypeError) as e:
+            _LOGGER.warning(
+                f"Failed to parse voicemail status data for {self._mailbox}: {e}"
+            )
+
+    def handle_mwi(self, event: Event, **kwargs):
+        """Handle MWI (Message Waiting Indicator) event."""
+        try:
+            new_messages = int(event.get("New", 0))
+            old_messages = int(event.get("Old", 0))
+
+            self._state = new_messages
+            self._extra_attributes = {
+                "new_messages": new_messages,
+                "old_messages": old_messages,
+                "total_messages": new_messages + old_messages,
+                "mailbox": event.get("Mailbox", self._mailbox),
+            }
+            self.schedule_update_ha_state()
+        except (ValueError, TypeError) as e:
+            _LOGGER.warning(f"Failed to parse MWI event for {self._mailbox}: {e}")
+
+    @property
+    def state(self) -> int:
+        """Return the number of new voicemail messages."""
+        return self._state
+
+    @property
+    def extra_state_attributes(self):
+        """Return the state attributes."""
+        return self._extra_attributes
+
+    @property
+    def icon(self) -> str:
+        """Return the icon of the sensor."""
+        return "mdi:voicemail" if self._state > 0 else "mdi:email-outline"

tests/mock_ami_client.py

@@ -14,4 +14,20 @@ def trigger_event(self, event):
             handler(event)
 
     def send_action(self, action):
-        pass
+        # Mock implementation for MailboxStatus action
+        if action.name == "MailboxStatus":
+            return MockFuture({"NewMessages": "2", "OldMessages": "3", "Waiting": "2"})
+        return MockFuture({})
+
+
+class MockFuture:
+    def __init__(self, response_keys):
+        self.response = MockResponse(response_keys)
+
+
+class MockResponse:
+    def __init__(self, keys):
+        self.keys = keys
+
+    def is_error(self):
+        return False

tests/test_sensors.py

@@ -7,6 +7,74 @@
 from .mock_ami_client import MockAMIClient
 
 
+from homeassistant.const import CONF_DEVICES
+from homeassistant.core import HomeAssistant
+from pytest_homeassistant_custom_component.common import MockConfigEntry
+
+from custom_components.asterisk.const import CLIENT, DOMAIN
+
+from .mock_ami_client import MockAMIClient
+
+
+async def test_voicemail_status_sensor(
+    hass: HomeAssistant, config_entry: MockConfigEntry
+):
+    """Test VoicemailStatusSensor."""
+    from custom_components.asterisk.sensor import VoicemailStatusSensor
+
+    client = MockAMIClient()
+    # Setup required hass data
+    hass.data.setdefault(DOMAIN, {})[config_entry.entry_id] = {
+        CLIENT: client,
+    }
+
+    device = {
+        "tech": "PJSIP",
+        "extension": "100",
+        "status": "IN_USE",
+    }
+
+    # Create sensor directly
+    sensor = VoicemailStatusSensor(hass, config_entry, device)
+
+    # Check initial state (should be 2 from mock response)
+    assert sensor.state == 2
+    assert sensor.extra_state_attributes["new_messages"] == 2
+    assert sensor.extra_state_attributes["old_messages"] == 3
+    assert sensor.extra_state_attributes["total_messages"] == 5
+    assert sensor.extra_state_attributes["mailbox"] == "100@default"
+    assert sensor.icon == "mdi:voicemail"
+
+    # Test MWI event handling by calling _update_from_response directly to avoid schedule_update_ha_state
+    sensor._update_from_response(
+        {
+            "NewMessages": "1",
+            "OldMessages": "4",
+        }
+    )
+
+    # Check updated state
+    assert sensor.state == 1
+    assert sensor.extra_state_attributes["new_messages"] == 1
+    assert sensor.extra_state_attributes["old_messages"] == 4
+    assert sensor.extra_state_attributes["total_messages"] == 5
+
+    # Test with no voicemails
+    sensor._update_from_response(
+        {
+            "NewMessages": "0",
+            "OldMessages": "2",
+        }
+    )
+
+    assert sensor.state == 0
+    assert sensor.icon == "mdi:email-outline"
+
+    # Test sensor properties
+    assert sensor.name == "100 Voicemail Status"
+    assert sensor.unique_id == f"{config_entry.entry_id}_100_voicemail_status"
+
+
 # async def test_device_state_sensor(hass: HomeAssistant, config_entry: MockConfigEntry):
 #     """Test DeviceStateSensor."""
 #     client = MockAMIClient()