snmp: Better sensor support to resolve previous issues (#113624)

Co-authored-by: Christian Kühnel <christian.kuehnel@gmail.com>
Co-authored-by: jan iversen <jancasacondor@gmail.com>

homeassistant/components/snmp/sensor.py

@@ -4,7 +4,9 @@ from __future__ import annotations
 
 from datetime import timedelta
 import logging
+from struct import unpack
 
+from pyasn1.codec.ber import decoder
 from pysnmp.error import PySnmpError
 import pysnmp.hlapi.asyncio as hlapi
 from pysnmp.hlapi.asyncio import (
@@ -18,6 +20,8 @@ from pysnmp.hlapi.asyncio import (
     UsmUserData,
     getCmd,
 )
+from pysnmp.proto.rfc1902 import Opaque
+from pysnmp.proto.rfc1905 import NoSuchObject
 import voluptuous as vol
 
 from homeassistant.components.sensor import CONF_STATE_CLASS, PLATFORM_SCHEMA
@@ -165,7 +169,10 @@ async def async_setup_platform(
     errindication, _, _, _ = get_result
 
     if errindication and not accept_errors:
-        _LOGGER.error("Please check the details in the configuration file")
+        _LOGGER.error(
+            "Please check the details in the configuration file: %s",
+            errindication,
+        )
         return
 
     name = config.get(CONF_NAME, Template(DEFAULT_NAME, hass))
@@ -248,10 +255,44 @@ class SnmpData:
             _LOGGER.error(
                 "SNMP error: %s at %s",
                 errstatus.prettyPrint(),
-                errindex and restable[-1][int(errindex) - 1] or "?",
+                restable[-1][int(errindex) - 1] if errindex else "?",
             )
         elif (errindication or errstatus) and self._accept_errors:
             self.value = self._default_value
         else:
             for resrow in restable:
-                self.value = resrow[-1].prettyPrint()
+                self.value = self._decode_value(resrow[-1])
+
+    def _decode_value(self, value):
+        """Decode the different results we could get into strings."""
+
+        _LOGGER.debug(
+            "SNMP OID %s received type=%s and data %s",
+            self._baseoid,
+            type(value),
+            bytes(value),
+        )
+        if isinstance(value, NoSuchObject):
+            _LOGGER.error(
+                "SNMP error for OID %s: No Such Object currently exists at this OID",
+                self._baseoid,
+            )
+            return self._default_value
+
+        if isinstance(value, Opaque):
+            # Float data type is not supported by the pyasn1 library,
+            # so we need to decode this type ourselves based on:
+            # https://tools.ietf.org/html/draft-perkins-opaque-01
+            if bytes(value).startswith(b"\x9f\x78"):
+                return str(unpack("!f", bytes(value)[3:])[0])
+            # Otherwise Opaque types should be asn1 encoded
+            try:
+                decoded_value, _ = decoder.decode(bytes(value))
+                return str(decoded_value)
+            # pylint: disable=broad-except
+            except Exception as decode_exception:
+                _LOGGER.error(
+                    "SNMP error in decoding opaque type: %s", decode_exception
+                )
+                return self._default_value
+        return str(value)

tests/components/snmp/test_float_sensor.py

@@ -0,0 +1,79 @@
+"""SNMP sensor tests."""
+
+from unittest.mock import patch
+
+from pysnmp.proto.rfc1902 import Opaque
+import pytest
+
+from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import entity_registry as er
+from homeassistant.setup import async_setup_component
+
+
+@pytest.fixture(autouse=True)
+def hlapi_mock():
+    """Mock out 3rd party API."""
+    mock_data = Opaque(value=b"\x9fx\x04=\xa4\x00\x00")
+    with patch(
+        "homeassistant.components.snmp.sensor.getCmd",
+        return_value=(None, None, None, [[mock_data]]),
+    ):
+        yield
+
+
+async def test_basic_config(hass: HomeAssistant) -> None:
+    """Test basic entity configuration."""
+
+    config = {
+        SENSOR_DOMAIN: {
+            "platform": "snmp",
+            "host": "192.168.1.32",
+            "baseoid": "1.3.6.1.4.1.2021.10.1.3.1",
+        },
+    }
+
+    assert await async_setup_component(hass, SENSOR_DOMAIN, config)
+    await hass.async_block_till_done()
+
+    state = hass.states.get("sensor.snmp")
+    assert state.state == "0.080078125"
+    assert state.attributes == {"friendly_name": "SNMP"}
+
+
+async def test_entity_config(hass: HomeAssistant) -> None:
+    """Test entity configuration."""
+
+    config = {
+        SENSOR_DOMAIN: {
+            # SNMP configuration
+            "platform": "snmp",
+            "host": "192.168.1.32",
+            "baseoid": "1.3.6.1.4.1.2021.10.1.3.1",
+            # Entity configuration
+            "icon": "{{'mdi:one_two_three'}}",
+            "picture": "{{'blabla.png'}}",
+            "device_class": "temperature",
+            "name": "{{'SNMP' + ' ' + 'Sensor'}}",
+            "state_class": "measurement",
+            "unique_id": "very_unique",
+            "unit_of_measurement": "°C",
+        },
+    }
+
+    assert await async_setup_component(hass, SENSOR_DOMAIN, config)
+    await hass.async_block_till_done()
+
+    entity_registry = er.async_get(hass)
+    assert entity_registry.async_get("sensor.snmp_sensor").unique_id == "very_unique"
+
+    state = hass.states.get("sensor.snmp_sensor")
+    assert state.state == "0.080078125"
+    assert state.attributes == {
+        "device_class": "temperature",
+        "entity_picture": "blabla.png",
+        "friendly_name": "SNMP Sensor",
+        "icon": "mdi:one_two_three",
+        "state_class": "measurement",
+        "unit_of_measurement": "°C",
+    }

tests/components/snmp/test_integer_sensor.py

@@ -1,7 +1,8 @@
 """SNMP sensor tests."""
 
-from unittest.mock import MagicMock, Mock, patch
+from unittest.mock import patch
 
+from pysnmp.hlapi import Integer32
 import pytest
 
 from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
@@ -13,8 +14,7 @@ from homeassistant.setup import async_setup_component
 @pytest.fixture(autouse=True)
 def hlapi_mock():
     """Mock out 3rd party API."""
-    mock_data = MagicMock()
-    mock_data.prettyPrint = Mock(return_value="13.5")
+    mock_data = Integer32(13)
     with patch(
         "homeassistant.components.snmp.sensor.getCmd",
         return_value=(None, None, None, [[mock_data]]),
@@ -37,7 +37,7 @@ async def test_basic_config(hass: HomeAssistant) -> None:
     await hass.async_block_till_done()
 
     state = hass.states.get("sensor.snmp")
-    assert state.state == "13.5"
+    assert state.state == "13"
     assert state.attributes == {"friendly_name": "SNMP"}
 
 
@@ -68,7 +68,7 @@ async def test_entity_config(hass: HomeAssistant) -> None:
     assert entity_registry.async_get("sensor.snmp_sensor").unique_id == "very_unique"
 
     state = hass.states.get("sensor.snmp_sensor")
-    assert state.state == "13.5"
+    assert state.state == "13"
     assert state.attributes == {
         "device_class": "temperature",
         "entity_picture": "blabla.png",

tests/components/snmp/test_string_sensor.py

@@ -0,0 +1,73 @@
+"""SNMP sensor tests."""
+
+from unittest.mock import patch
+
+from pysnmp.proto.rfc1902 import OctetString
+import pytest
+
+from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import entity_registry as er
+from homeassistant.setup import async_setup_component
+
+
+@pytest.fixture(autouse=True)
+def hlapi_mock():
+    """Mock out 3rd party API."""
+    mock_data = OctetString("98F")
+    with patch(
+        "homeassistant.components.snmp.sensor.getCmd",
+        return_value=(None, None, None, [[mock_data]]),
+    ):
+        yield
+
+
+async def test_basic_config(hass: HomeAssistant) -> None:
+    """Test basic entity configuration."""
+
+    config = {
+        SENSOR_DOMAIN: {
+            "platform": "snmp",
+            "host": "192.168.1.32",
+            "baseoid": "1.3.6.1.4.1.2021.10.1.3.1",
+        },
+    }
+
+    assert await async_setup_component(hass, SENSOR_DOMAIN, config)
+    await hass.async_block_till_done()
+
+    state = hass.states.get("sensor.snmp")
+    assert state.state == "98F"
+    assert state.attributes == {"friendly_name": "SNMP"}
+
+
+async def test_entity_config(hass: HomeAssistant) -> None:
+    """Test entity configuration."""
+
+    config = {
+        SENSOR_DOMAIN: {
+            # SNMP configuration
+            "platform": "snmp",
+            "host": "192.168.1.32",
+            "baseoid": "1.3.6.1.4.1.2021.10.1.3.1",
+            # Entity configuration
+            "icon": "{{'mdi:one_two_three'}}",
+            "picture": "{{'blabla.png'}}",
+            "name": "{{'SNMP' + ' ' + 'Sensor'}}",
+            "unique_id": "very_unique",
+        },
+    }
+
+    assert await async_setup_component(hass, SENSOR_DOMAIN, config)
+    await hass.async_block_till_done()
+
+    entity_registry = er.async_get(hass)
+    assert entity_registry.async_get("sensor.snmp_sensor").unique_id == "very_unique"
+
+    state = hass.states.get("sensor.snmp_sensor")
+    assert state.state == "98F"
+    assert state.attributes == {
+        "entity_picture": "blabla.png",
+        "friendly_name": "SNMP Sensor",
+        "icon": "mdi:one_two_three",
+    }