Bump pysnmp to v7 and brother to v5 (#129761)

Co-authored-by: Maciej Bieniek <bieniu@users.noreply.github.com>

homeassistant/components/brother/manifest.json

@@ -8,7 +8,7 @@
   "integration_type": "device",
   "iot_class": "local_polling",
   "loggers": ["brother", "pyasn1", "pysmi", "pysnmp"],
-  "requirements": ["brother==4.3.1"],
+  "requirements": ["brother==5.0.0"],
   "zeroconf": [
     {
       "type": "_printer._tcp.local.",

homeassistant/components/snmp/device_tracker.py

@@ -7,13 +7,13 @@ import logging
 from typing import TYPE_CHECKING
 
 from pysnmp.error import PySnmpError
-from pysnmp.hlapi.asyncio import (
+from pysnmp.hlapi.v3arch.asyncio import (
     CommunityData,
     Udp6TransportTarget,
     UdpTransportTarget,
     UsmUserData,
-    bulkWalkCmd,
-    isEndOfMib,
+    bulk_walk_cmd,
+    is_end_of_mib,
 )
 import voluptuous as vol
 
@@ -59,7 +59,7 @@ async def async_get_scanner(
     hass: HomeAssistant, config: ConfigType
 ) -> SnmpScanner | None:
     """Validate the configuration and return an SNMP scanner."""
-    scanner = SnmpScanner(config[DEVICE_TRACKER_DOMAIN])
+    scanner = await SnmpScanner.create(config[DEVICE_TRACKER_DOMAIN])
     await scanner.async_init(hass)
 
     return scanner if scanner.success_init else None
@@ -69,8 +69,8 @@ class SnmpScanner(DeviceScanner):
     """Queries any SNMP capable Access Point for connected devices."""
 
     def __init__(self, config):
-        """Initialize the scanner and test the target device."""
-        host = config[CONF_HOST]
+        """Initialize the scanner after testing the target device."""
+
         community = config[CONF_COMMUNITY]
         baseoid = config[CONF_BASEOID]
         authkey = config.get(CONF_AUTH_KEY)
@@ -78,19 +78,6 @@ class SnmpScanner(DeviceScanner):
         privkey = config.get(CONF_PRIV_KEY)
         privproto = DEFAULT_PRIV_PROTOCOL
 
-        try:
-            # Try IPv4 first.
-            target = UdpTransportTarget((host, DEFAULT_PORT), timeout=DEFAULT_TIMEOUT)
-        except PySnmpError:
-            # Then try IPv6.
-            try:
-                target = Udp6TransportTarget(
-                    (host, DEFAULT_PORT), timeout=DEFAULT_TIMEOUT
-                )
-            except PySnmpError as err:
-                _LOGGER.error("Invalid SNMP host: %s", err)
-                return
-
         if authkey is not None or privkey is not None:
             if not authkey:
                 authproto = "none"
@@ -109,16 +96,43 @@ class SnmpScanner(DeviceScanner):
                 community, mpModel=SNMP_VERSIONS[DEFAULT_VERSION]
             )
 
-        self._target = target
+        self._target: UdpTransportTarget | Udp6TransportTarget
         self.request_args: RequestArgsType | None = None
         self.baseoid = baseoid
         self.last_results = []
         self.success_init = False
 
+    @classmethod
+    async def create(cls, config):
+        """Asynchronously test the target device before fully initializing the scanner."""
+        host = config[CONF_HOST]
+
+        try:
+            # Try IPv4 first.
+            target = await UdpTransportTarget.create(
+                (host, DEFAULT_PORT), timeout=DEFAULT_TIMEOUT
+            )
+        except PySnmpError:
+            # Then try IPv6.
+            try:
+                target = Udp6TransportTarget(
+                    (host, DEFAULT_PORT), timeout=DEFAULT_TIMEOUT
+                )
+            except PySnmpError as err:
+                _LOGGER.error("Invalid SNMP host: %s", err)
+                return None
+        instance = cls(config)
+        instance._target = target
+
+        return instance
+
     async def async_init(self, hass: HomeAssistant) -> None:
         """Make a one-off read to check if the target device is reachable and readable."""
         self.request_args = await async_create_request_cmd_args(
-            hass, self._auth_data, self._target, self.baseoid
+            hass,
+            self._auth_data,
+            self._target,
+            self.baseoid,
         )
         data = await self.async_get_snmp_data()
         self.success_init = data is not None
@@ -154,7 +168,7 @@ class SnmpScanner(DeviceScanner):
             assert self.request_args is not None
 
         engine, auth_data, target, context_data, object_type = self.request_args
-        walker = bulkWalkCmd(
+        walker = bulk_walk_cmd(
             engine,
             auth_data,
             target,
@@ -177,7 +191,7 @@ class SnmpScanner(DeviceScanner):
                 return None
 
             for _oid, value in res:
-                if not isEndOfMib(res):
+                if not is_end_of_mib(res):
                     try:
                         mac = binascii.hexlify(value.asOctets()).decode("utf-8")
                     except AttributeError:

homeassistant/components/snmp/manifest.json

@@ -6,5 +6,5 @@
   "iot_class": "local_polling",
   "loggers": ["pyasn1", "pysmi", "pysnmp"],
   "quality_scale": "legacy",
-  "requirements": ["pysnmp==6.2.6"]
+  "requirements": ["pysnmp==7.1.21"]
 }

homeassistant/components/snmp/sensor.py

@@ -8,13 +8,13 @@ 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 (
+import pysnmp.hlapi.v3arch.asyncio as hlapi
+from pysnmp.hlapi.v3arch.asyncio import (
     CommunityData,
     Udp6TransportTarget,
     UdpTransportTarget,
     UsmUserData,
-    getCmd,
+    get_cmd,
 )
 from pysnmp.proto.rfc1902 import Opaque
 from pysnmp.proto.rfc1905 import NoSuchObject
@@ -134,7 +134,7 @@ async def async_setup_platform(
 
     try:
         # Try IPv4 first.
-        target = UdpTransportTarget((host, port), timeout=DEFAULT_TIMEOUT)
+        target = await UdpTransportTarget.create((host, port), timeout=DEFAULT_TIMEOUT)
     except PySnmpError:
         # Then try IPv6.
         try:
@@ -159,7 +159,7 @@ async def async_setup_platform(
         auth_data = CommunityData(community, mpModel=SNMP_VERSIONS[version])
 
     request_args = await async_create_request_cmd_args(hass, auth_data, target, baseoid)
-    get_result = await getCmd(*request_args)
+    get_result = await get_cmd(*request_args)
     errindication, _, _, _ = get_result
 
     if errindication and not accept_errors:
@@ -235,7 +235,7 @@ class SnmpData:
     async def async_update(self):
         """Get the latest data from the remote SNMP capable host."""
 
-        get_result = await getCmd(*self._request_args)
+        get_result = await get_cmd(*self._request_args)
         errindication, errstatus, errindex, restable = get_result
 
         if errindication and not self._accept_errors:

homeassistant/components/snmp/switch.py

@@ -5,15 +5,15 @@ from __future__ import annotations
 import logging
 from typing import Any
 
-import pysnmp.hlapi.asyncio as hlapi
-from pysnmp.hlapi.asyncio import (
+import pysnmp.hlapi.v3arch.asyncio as hlapi
+from pysnmp.hlapi.v3arch.asyncio import (
     CommunityData,
     ObjectIdentity,
     ObjectType,
     UdpTransportTarget,
     UsmUserData,
-    getCmd,
-    setCmd,
+    get_cmd,
+    set_cmd,
 )
 from pysnmp.proto.rfc1902 import (
     Counter32,
@@ -169,7 +169,7 @@ async def async_setup_platform(
     else:
         auth_data = CommunityData(community, mpModel=SNMP_VERSIONS[version])
 
-    transport = UdpTransportTarget((host, port))
+    transport = await UdpTransportTarget.create((host, port))
     request_args = await async_create_request_cmd_args(
         hass, auth_data, transport, baseoid
     )
@@ -228,10 +228,17 @@ class SnmpSwitch(SwitchEntity):
         self._state: bool | None = None
         self._payload_on = payload_on
         self._payload_off = payload_off
-        self._target = UdpTransportTarget((host, port))
+        self._host = host
+        self._port = port
         self._request_args = request_args
         self._command_args = command_args
 
+    async def async_added_to_hass(self) -> None:
+        """Run when this Entity has been added to HA."""
+        # The transport creation is done once this entity is registered with HA
+        # (rather than in the __init__)
+        self._target = await UdpTransportTarget.create((self._host, self._port))  # pylint: disable=attribute-defined-outside-init
+
     async def async_turn_on(self, **kwargs: Any) -> None:
         """Turn on the switch."""
         # If vartype set, use it - https://www.pysnmp.com/pysnmp/docs/api-reference.html#pysnmp.smi.rfc1902.ObjectType
@@ -255,7 +262,7 @@ class SnmpSwitch(SwitchEntity):
 
     async def async_update(self) -> None:
         """Update the state."""
-        get_result = await getCmd(*self._request_args)
+        get_result = await get_cmd(*self._request_args)
         errindication, errstatus, errindex, restable = get_result
 
         if errindication:
@@ -291,6 +298,6 @@ class SnmpSwitch(SwitchEntity):
 
     async def _set(self, value: Any) -> None:
         """Set the state of the switch."""
-        await setCmd(
+        await set_cmd(
             *self._command_args, ObjectType(ObjectIdentity(self._commandoid), value)
         )

homeassistant/components/snmp/util.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 import logging
 
-from pysnmp.hlapi.asyncio import (
+from pysnmp.hlapi.v3arch.asyncio import (
     CommunityData,
     ContextData,
     ObjectIdentity,
@@ -14,8 +14,8 @@ from pysnmp.hlapi.asyncio import (
     UdpTransportTarget,
     UsmUserData,
 )
-from pysnmp.hlapi.asyncio.cmdgen import lcd, vbProcessor
-from pysnmp.smi.builder import MibBuilder
+from pysnmp.hlapi.v3arch.asyncio.cmdgen import LCD
+from pysnmp.smi import view
 
 from homeassistant.const import EVENT_HOMEASSISTANT_STOP
 from homeassistant.core import Event, HomeAssistant, callback
@@ -80,7 +80,7 @@ async def async_get_snmp_engine(hass: HomeAssistant) -> SnmpEngine:
     @callback
     def _async_shutdown_listener(ev: Event) -> None:
         _LOGGER.debug("Unconfiguring SNMP engine")
-        lcd.unconfigure(engine, None)
+        LCD.unconfigure(engine, None)
 
     hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _async_shutdown_listener)
     return engine
@@ -89,10 +89,10 @@ async def async_get_snmp_engine(hass: HomeAssistant) -> SnmpEngine:
 def _get_snmp_engine() -> SnmpEngine:
     """Return a cached instance of SnmpEngine."""
     engine = SnmpEngine()
-    mib_controller = vbProcessor.getMibViewController(engine)
-    # Actually load the MIBs from disk so we do
-    # not do it in the event loop
-    builder: MibBuilder = mib_controller.mibBuilder
-    if "PYSNMP-MIB" not in builder.mibSymbols:
-        builder.loadModules()
+    # Actually load the MIBs from disk so we do not do it in the event loop
+    mib_view_controller = view.MibViewController(
+        engine.message_dispatcher.mib_instrum_controller.get_mib_builder()
+    )
+    engine.cache["mibViewController"] = mib_view_controller
+    mib_view_controller.mibBuilder.load_modules()
     return engine

requirements_all.txt

@@ -677,7 +677,7 @@ bring-api==1.1.0
 broadlink==0.19.0
 
 # homeassistant.components.brother
-brother==4.3.1
+brother==5.0.0
 
 # homeassistant.components.brottsplatskartan
 brottsplatskartan==1.0.5
@@ -2363,7 +2363,7 @@ pysml==0.1.5
 pysmlight==0.2.7
 
 # homeassistant.components.snmp
-pysnmp==6.2.6
+pysnmp==7.1.21
 
 # homeassistant.components.snooz
 pysnooz==0.8.6

requirements_test_all.txt

@@ -604,7 +604,7 @@ bring-api==1.1.0
 broadlink==0.19.0
 
 # homeassistant.components.brother
-brother==4.3.1
+brother==5.0.0
 
 # homeassistant.components.brottsplatskartan
 brottsplatskartan==1.0.5
@@ -1966,7 +1966,7 @@ pysml==0.1.5
 pysmlight==0.2.7
 
 # homeassistant.components.snmp
-pysnmp==6.2.6
+pysnmp==7.1.21
 
 # homeassistant.components.snooz
 pysnooz==0.8.6

tests/components/snmp/test_float_sensor.py

@@ -16,7 +16,7 @@ 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",
+        "homeassistant.components.snmp.sensor.get_cmd",
         return_value=(None, None, None, [[mock_data]]),
     ):
         yield

tests/components/snmp/test_init.py

@@ -2,8 +2,8 @@
 
 from unittest.mock import patch
 
-from pysnmp.hlapi.asyncio import SnmpEngine
-from pysnmp.hlapi.asyncio.cmdgen import lcd
+from pysnmp.hlapi.v3arch.asyncio import SnmpEngine
+from pysnmp.hlapi.v3arch.asyncio.cmdgen import LCD
 
 from homeassistant.components import snmp
 from homeassistant.const import EVENT_HOMEASSISTANT_STOP
@@ -16,7 +16,7 @@ async def test_async_get_snmp_engine(hass: HomeAssistant) -> None:
     assert isinstance(engine, SnmpEngine)
     engine2 = await snmp.async_get_snmp_engine(hass)
     assert engine is engine2
-    with patch.object(lcd, "unconfigure") as mock_unconfigure:
+    with patch.object(LCD, "unconfigure") as mock_unconfigure:
         hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
         await hass.async_block_till_done()
     assert mock_unconfigure.called

tests/components/snmp/test_integer_sensor.py

@@ -16,7 +16,7 @@ def hlapi_mock():
     """Mock out 3rd party API."""
     mock_data = Integer32(13)
     with patch(
-        "homeassistant.components.snmp.sensor.getCmd",
+        "homeassistant.components.snmp.sensor.get_cmd",
         return_value=(None, None, None, [[mock_data]]),
     ):
         yield

tests/components/snmp/test_negative_sensor.py

@@ -16,7 +16,7 @@ def hlapi_mock():
     """Mock out 3rd party API."""
     mock_data = Integer32(-13)
     with patch(
-        "homeassistant.components.snmp.sensor.getCmd",
+        "homeassistant.components.snmp.sensor.get_cmd",
         return_value=(None, None, None, [[mock_data]]),
     ):
         yield

tests/components/snmp/test_string_sensor.py

@@ -16,7 +16,7 @@ def hlapi_mock():
     """Mock out 3rd party API."""
     mock_data = OctetString("98F")
     with patch(
-        "homeassistant.components.snmp.sensor.getCmd",
+        "homeassistant.components.snmp.sensor.get_cmd",
         return_value=(None, None, None, [[mock_data]]),
     ):
         yield

tests/components/snmp/test_switch.py

@@ -27,7 +27,7 @@ async def test_snmp_integer_switch_off(hass: HomeAssistant) -> None:
 
     mock_data = Integer32(0)
     with patch(
-        "homeassistant.components.snmp.switch.getCmd",
+        "homeassistant.components.snmp.switch.get_cmd",
         return_value=(None, None, None, [[mock_data]]),
     ):
         assert await async_setup_component(hass, SWITCH_DOMAIN, config)
@@ -41,7 +41,7 @@ async def test_snmp_integer_switch_on(hass: HomeAssistant) -> None:
 
     mock_data = Integer32(1)
     with patch(
-        "homeassistant.components.snmp.switch.getCmd",
+        "homeassistant.components.snmp.switch.get_cmd",
         return_value=(None, None, None, [[mock_data]]),
     ):
         assert await async_setup_component(hass, SWITCH_DOMAIN, config)
@@ -57,7 +57,7 @@ async def test_snmp_integer_switch_unknown(
 
     mock_data = Integer32(3)
     with patch(
-        "homeassistant.components.snmp.switch.getCmd",
+        "homeassistant.components.snmp.switch.get_cmd",
         return_value=(None, None, None, [[mock_data]]),
     ):
         assert await async_setup_component(hass, SWITCH_DOMAIN, config)