Add options flow for force_poll setting in upnp (#120843)

homeassistant/components/upnp/__init__.py

@@ -15,6 +15,7 @@ from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.helpers import config_validation as cv, device_registry as dr
 
 from .const import (
+    CONFIG_ENTRY_FORCE_POLL,
     CONFIG_ENTRY_HOST,
     CONFIG_ENTRY_MAC_ADDRESS,
     CONFIG_ENTRY_ORIGINAL_UDN,
@@ -82,7 +83,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: UpnpConfigEntry) -> bool
     assert discovery_info is not None
     assert discovery_info.ssdp_udn
     assert discovery_info.ssdp_all_locations
-    force_poll = False
+    force_poll = entry.options.get(CONFIG_ENTRY_FORCE_POLL, False)
     location = get_preferred_location(discovery_info.ssdp_all_locations)
     try:
         device = await async_create_device(hass, location, force_poll)
@@ -98,6 +99,14 @@ async def async_setup_entry(hass: HomeAssistant, entry: UpnpConfigEntry) -> bool
     # Unsubscribe services on unload.
     entry.async_on_unload(device.async_unsubscribe_services)
 
+    # Update force_poll on options update.
+    async def update_listener(hass: HomeAssistant, entry: UpnpConfigEntry):
+        """Handle options update."""
+        force_poll = entry.options.get(CONFIG_ENTRY_FORCE_POLL, False)
+        await device.async_set_force_poll(force_poll)
+
+    entry.async_on_unload(entry.add_update_listener(update_listener))
+
     # Track the original UDN such that existing sensors do not change their unique_id.
     if CONFIG_ENTRY_ORIGINAL_UDN not in entry.data:
         hass.config_entries.async_update_entry(

homeassistant/components/upnp/config_flow.py

@@ -10,16 +10,25 @@ import voluptuous as vol
 
 from homeassistant.components import ssdp
 from homeassistant.components.ssdp import SsdpServiceInfo
-from homeassistant.config_entries import SOURCE_IGNORE, ConfigFlow, ConfigFlowResult
-from homeassistant.core import HomeAssistant
+from homeassistant.config_entries import (
+    SOURCE_IGNORE,
+    ConfigEntry,
+    ConfigFlow,
+    ConfigFlowResult,
+    OptionsFlow,
+    OptionsFlowWithConfigEntry,
+)
+from homeassistant.core import HomeAssistant, callback
 
 from .const import (
+    CONFIG_ENTRY_FORCE_POLL,
     CONFIG_ENTRY_HOST,
     CONFIG_ENTRY_LOCATION,
     CONFIG_ENTRY_MAC_ADDRESS,
     CONFIG_ENTRY_ORIGINAL_UDN,
     CONFIG_ENTRY_ST,
     CONFIG_ENTRY_UDN,
+    DEFAULT_CONFIG_ENTRY_FORCE_POLL,
     DOMAIN,
     DOMAIN_DISCOVERIES,
     LOGGER,
@@ -83,6 +92,12 @@ class UpnpFlowHandler(ConfigFlow, domain=DOMAIN):
     # 1: ssdp(discovery_info) --> ssdp_confirm(None) --> ssdp_confirm({}) --> create_entry()
     # 2: user(None): scan --> user({...}) --> create_entry()
 
+    @staticmethod
+    @callback
+    def async_get_options_flow(config_entry: ConfigEntry) -> OptionsFlow:
+        """Get the options flow for this handler."""
+        return UpnpOptionsFlowHandler(config_entry)
+
     @property
     def _discoveries(self) -> dict[str, SsdpServiceInfo]:
         """Get current discoveries."""
@@ -249,9 +264,14 @@ class UpnpFlowHandler(ConfigFlow, domain=DOMAIN):
             CONFIG_ENTRY_HOST: discovery.ssdp_headers["_host"],
             CONFIG_ENTRY_LOCATION: get_preferred_location(discovery.ssdp_all_locations),
         }
+        options = {
+            CONFIG_ENTRY_FORCE_POLL: False,
+        }
 
         await self.async_set_unique_id(user_input["unique_id"], raise_on_progress=False)
-        return self.async_create_entry(title=user_input["title"], data=data)
+        return self.async_create_entry(
+            title=user_input["title"], data=data, options=options
+        )
 
     async def _async_create_entry_from_discovery(
         self,
@@ -273,4 +293,30 @@ class UpnpFlowHandler(ConfigFlow, domain=DOMAIN):
             CONFIG_ENTRY_MAC_ADDRESS: mac_address,
             CONFIG_ENTRY_HOST: discovery.ssdp_headers["_host"],
         }
-        return self.async_create_entry(title=title, data=data)
+        options = {
+            CONFIG_ENTRY_FORCE_POLL: False,
+        }
+        return self.async_create_entry(title=title, data=data, options=options)
+
+
+class UpnpOptionsFlowHandler(OptionsFlowWithConfigEntry):
+    """Handle an options flow."""
+
+    async def async_step_init(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle options flow."""
+        if user_input is not None:
+            return self.async_create_entry(title="", data=user_input)
+
+        data_schema = vol.Schema(
+            {
+                vol.Optional(
+                    CONFIG_ENTRY_FORCE_POLL,
+                    default=self.options.get(
+                        CONFIG_ENTRY_FORCE_POLL, DEFAULT_CONFIG_ENTRY_FORCE_POLL
+                    ),
+                ): bool,
+            }
+        )
+        return self.async_show_form(step_id="init", data_schema=data_schema)

homeassistant/components/upnp/const.py

@@ -24,6 +24,7 @@ WAN_STATUS = "wan_status"
 PORT_MAPPING_NUMBER_OF_ENTRIES_IPV4 = "port_mapping_number_of_entries"
 ROUTER_IP = "ip"
 ROUTER_UPTIME = "uptime"
+CONFIG_ENTRY_FORCE_POLL = "force_poll"
 CONFIG_ENTRY_ST = "st"
 CONFIG_ENTRY_UDN = "udn"
 CONFIG_ENTRY_ORIGINAL_UDN = "original_udn"
@@ -33,5 +34,6 @@ CONFIG_ENTRY_HOST = "host"
 IDENTIFIER_HOST = "upnp_host"
 IDENTIFIER_SERIAL_NUMBER = "upnp_serial_number"
 DEFAULT_SCAN_INTERVAL = timedelta(seconds=30).total_seconds()
+DEFAULT_CONFIG_ENTRY_FORCE_POLL = False
 ST_IGD_V1 = "urn:schemas-upnp-org:device:InternetGatewayDevice:1"
 ST_IGD_V2 = "urn:schemas-upnp-org:device:InternetGatewayDevice:2"

homeassistant/components/upnp/strings.json

@@ -21,7 +21,8 @@
     "step": {
       "init": {
         "data": {
-          "scan_interval": "Update interval (seconds, minimal 30)"
+          "scan_interval": "Update interval (seconds, minimal 30)",
+          "force_poll": "Force polling of all data"
         }
       }
     }

tests/components/upnp/conftest.py

@@ -248,7 +248,7 @@ async def mock_config_entry(
     ssdp_instant_discovery,
     mock_igd_device: IgdDevice,
     mock_mac_address_from_host,
-):
+) -> MockConfigEntry:
     """Create an initialized integration."""
     entry = MockConfigEntry(
         domain=DOMAIN,

tests/components/upnp/test_config_flow.py

@@ -9,6 +9,7 @@ import pytest
 from homeassistant import config_entries
 from homeassistant.components import ssdp
 from homeassistant.components.upnp.const import (
+    CONFIG_ENTRY_FORCE_POLL,
     CONFIG_ENTRY_HOST,
     CONFIG_ENTRY_LOCATION,
     CONFIG_ENTRY_MAC_ADDRESS,
@@ -473,3 +474,28 @@ async def test_flow_ssdp_with_mismatched_udn(hass: HomeAssistant) -> None:
         CONFIG_ENTRY_MAC_ADDRESS: TEST_MAC_ADDRESS,
         CONFIG_ENTRY_HOST: TEST_HOST,
     }
+
+
+async def test_options_flow(
+    hass: HomeAssistant, mock_config_entry: MockConfigEntry
+) -> None:
+    """Test that the options flow works."""
+    result = await hass.config_entries.options.async_init(mock_config_entry.entry_id)
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "init"
+
+    user_input = {
+        CONFIG_ENTRY_FORCE_POLL: True,
+    }
+    result = await hass.config_entries.options.async_configure(
+        result["flow_id"],
+        user_input,
+    )
+    await hass.async_block_till_done()
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+    assert result["data"] == {
+        CONFIG_ENTRY_FORCE_POLL: True,
+    }
+    assert mock_config_entry.options == {
+        CONFIG_ENTRY_FORCE_POLL: True,
+    }

tests/components/upnp/test_init.py

@@ -10,6 +10,7 @@ import pytest
 
 from homeassistant.components import ssdp
 from homeassistant.components.upnp.const import (
+    CONFIG_ENTRY_FORCE_POLL,
     CONFIG_ENTRY_LOCATION,
     CONFIG_ENTRY_MAC_ADDRESS,
     CONFIG_ENTRY_ORIGINAL_UDN,
@@ -46,6 +47,9 @@ async def test_async_setup_entry_default(
             CONFIG_ENTRY_LOCATION: TEST_LOCATION,
             CONFIG_ENTRY_MAC_ADDRESS: TEST_MAC_ADDRESS,
         },
+        options={
+            CONFIG_ENTRY_FORCE_POLL: False,
+        },
     )
 
     # Load config_entry.
@@ -68,6 +72,9 @@ async def test_async_setup_entry_default_no_mac_address(hass: HomeAssistant) ->
             CONFIG_ENTRY_LOCATION: TEST_LOCATION,
             CONFIG_ENTRY_MAC_ADDRESS: None,
         },
+        options={
+            CONFIG_ENTRY_FORCE_POLL: False,
+        },
     )
 
     # Load config_entry.
@@ -96,6 +103,9 @@ async def test_async_setup_entry_multi_location(
             CONFIG_ENTRY_LOCATION: TEST_LOCATION,
             CONFIG_ENTRY_MAC_ADDRESS: TEST_MAC_ADDRESS,
         },
+        options={
+            CONFIG_ENTRY_FORCE_POLL: False,
+        },
     )
 
     # Load config_entry.
@@ -124,6 +134,9 @@ async def test_async_setup_udn_mismatch(
             CONFIG_ENTRY_LOCATION: TEST_LOCATION,
             CONFIG_ENTRY_MAC_ADDRESS: TEST_MAC_ADDRESS,
         },
+        options={
+            CONFIG_ENTRY_FORCE_POLL: False,
+        },
     )
 
     # Set up device discovery callback.
@@ -148,3 +161,34 @@ async def test_async_setup_udn_mismatch(
 
     # Ensure that the IPv4 location is used.
     mock_async_create_device.assert_called_once_with(TEST_LOCATION)
+
+
+@pytest.mark.usefixtures(
+    "ssdp_instant_discovery",
+    "mock_get_source_ip",
+    "mock_mac_address_from_host",
+)
+async def test_async_setup_entry_force_poll(
+    hass: HomeAssistant, mock_igd_device: IgdDevice
+) -> None:
+    """Test async_setup_entry."""
+    entry = MockConfigEntry(
+        domain=DOMAIN,
+        unique_id=TEST_USN,
+        data={
+            CONFIG_ENTRY_ST: TEST_ST,
+            CONFIG_ENTRY_UDN: TEST_UDN,
+            CONFIG_ENTRY_ORIGINAL_UDN: TEST_UDN,
+            CONFIG_ENTRY_LOCATION: TEST_LOCATION,
+            CONFIG_ENTRY_MAC_ADDRESS: TEST_MAC_ADDRESS,
+        },
+        options={
+            CONFIG_ENTRY_FORCE_POLL: True,
+        },
+    )
+
+    # Load config_entry.
+    entry.add_to_hass(hass)
+    assert await hass.config_entries.async_setup(entry.entry_id) is True
+
+    mock_igd_device.async_subscribe_services.assert_not_called()