Revert poetry update

glocaltokens/client.py

@@ -16,6 +16,7 @@
     ACCESS_TOKEN_DURATION,
     ACCESS_TOKEN_SERVICE,
     ANDROID_ID_LENGTH,
+    DEFAULT_DISCOVERY_PORT,
     DISCOVERY_TIMEOUT,
     GOOGLE_HOME_FOYER_API,
     HOMEGRAPH_DURATION,
@@ -26,6 +27,7 @@
 from .types import DeviceDict
 from .utils import network as net_utils, token as token_utils
 from .utils.logs import censor
+from .utils.network import is_valid_ipv4_address
 
 logging.basicConfig(level=logging.ERROR)
 LOGGER = logging.getLogger(__name__)
@@ -334,7 +336,7 @@ def get_homegraph(self, auth_attempts: int = 3) -> GetHomeGraphResponse | None:
     def get_google_devices(
         self,
         models_list: list[str] | None = None,
-        disable_discovery: bool = False,
+        discovery_address_dict: dict[str, str] | None = None,
         zeroconf_instance: Zeroconf | None = None,
         force_homegraph_reload: bool = False,
         discovery_timeout: int = DISCOVERY_TIMEOUT,
@@ -344,8 +346,9 @@ def get_google_devices(
         and IP and ports if set in models_list.
 
         models_list: The list of accepted model names.
-        disable_discovery: Whether or not the device's IP and port should
-          be searched for in the network.
+        discovery_address_dict: Dict of network devices from the local network
+          ({"name": "ip_address"}). If set to `None` will try to automatically
+          discover network devices. Disable discovery by setting to `{}`.
         zeroconf_instance: If you already have an initialized zeroconf instance,
           use it here.
         force_homegraph_reload: If the stored homegraph should be generated again.
@@ -365,13 +368,31 @@ def get_google_devices(
 
         devices: list[Device] = []
 
+        def is_dict_with_valid_ipv4_addresses(data: dict[str, str]) -> bool:
+            # Test if the data structure is correct and if each entry contains a
+            # valid IPv4 address.
+            return isinstance(data, dict) and all(
+                isinstance(x, str) and is_valid_ipv4_address(x) for x in data.values()
+            )
+
+        if discovery_address_dict and not is_dict_with_valid_ipv4_addresses(
+            discovery_address_dict
+        ):
+            # We need to disable flake8-use-fstring, because of the brackets,
+            # it causes a false positive.
+            LOGGER.error(
+                "Invalid dictionary structure for discovery_address_dict "
+                "argument. Correct structure is {'device_name': 'ipaddress'}"  # noqa
+            )
+            return devices
+
         if homegraph is None:
             LOGGER.debug("Failed to fetch homegraph")
             return devices
 
         network_devices: list[NetworkDevice] = []
-        if disable_discovery is False:
-            LOGGER.debug("Getting network devices...")
+        if discovery_address_dict is None:
+            LOGGER.debug("Automatically discovering network devices...")
             network_devices = discover_devices(
                 models_list,
                 timeout=discovery_timeout,
@@ -385,6 +406,8 @@ def find_device(unique_id: str) -> NetworkDevice | None:
                     return device
             return None
 
+        address_dict = discovery_address_dict if discovery_address_dict else {}
+
         LOGGER.debug("Iterating in %d homegraph devices", len(homegraph.home.devices))
         for item in homegraph.home.devices:
             if item.local_auth_token != "":
@@ -405,6 +428,14 @@ def find_device(unique_id: str) -> NetworkDevice | None:
                         unique_id,
                     )
                     network_device = find_device(unique_id)
+                elif item.device_name in address_dict:
+                    network_device = NetworkDevice(
+                        name=item.device_name,
+                        ip_address=address_dict[item.device_name],
+                        port=DEFAULT_DISCOVERY_PORT,
+                        model=item.hardware.model,
+                        unique_id=item.device_info.device_id,
+                    )
 
                 device = Device(
                     device_id=item.device_info.device_id,
@@ -437,7 +468,7 @@ def get_google_devices_json(
         self,
         models_list: list[str] | None = None,
         indent: int = 2,
-        disable_discovery: bool = False,
+        discovery_address_dict: dict[str, str] | None = None,
         zeroconf_instance: Zeroconf | None = None,
         force_homegraph_reload: bool = False,
     ) -> str:
@@ -447,16 +478,17 @@ def get_google_devices_json(
 
         models_list: The list of accepted model names.
         indent: The indentation for the json formatting.
-        disable_discovery: Whether or not the device's IP and port should
-          be searched for in the network.
+        discovery_address_dict: Dict of network devices from the local network
+          ({"name": "ip_address"}). If set to `None` will try to automatically
+          discover network devices. Disable discovery by setting to `{}`.
         zeroconf_instance: If you already have an initialized zeroconf instance,
           use it here.
         force_homegraph_reload: If the stored homegraph should be generated again.
         """
 
         google_devices = self.get_google_devices(
             models_list=models_list,
-            disable_discovery=disable_discovery,
+            discovery_address_dict=discovery_address_dict,
             zeroconf_instance=zeroconf_instance,
             force_homegraph_reload=force_homegraph_reload,
         )

glocaltokens/const.py

@@ -18,6 +18,7 @@
 HOMEGRAPH_DURATION: Final = 24 * 60 * 60
 
 DISCOVERY_TIMEOUT: Final = 2
+DEFAULT_DISCOVERY_PORT: Final = 0
 
 GOOGLE_HOME_MODELS: Final = [
     "Google Home",

tests/factory/providers.py

@@ -43,14 +43,18 @@ def local_auth_token(self) -> str:
 class HomegraphProvider(TokenProvider):
     """Homegraph provider"""
 
-    def homegraph_device(self) -> GetHomeGraphResponse.Home.Device:
+    def homegraph_device(
+        self, device_name: str | None = None
+    ) -> GetHomeGraphResponse.Home.Device:
         """Using the content from test requests as reference"""
+        device_name = device_name if device_name else faker.word()
+
         return GetHomeGraphResponse.Home.Device(
             local_auth_token=self.local_auth_token(),
             device_info=GetHomeGraphResponse.Home.Device.DeviceInfo(
                 device_id=str(faker.uuid4)
             ),
-            device_name=faker.word(),
+            device_name=device_name,
             hardware=GetHomeGraphResponse.Home.Device.Hardware(model=faker.word()),
         )
 

tests/test_client.py

@@ -322,19 +322,26 @@ def test_get_homegraph_retries(
     def test_get_google_devices(self, m_get_homegraph: NonCallableMock) -> None:
         """Test getting google devices"""
         # With just one device returned from homegraph
-        homegraph_device = faker.homegraph_device()
+        fake_device_name = faker.word()
+        fake_ip_address = faker.ipv4()
+        homegraph_device = faker.homegraph_device(device_name=fake_device_name)
         m_get_homegraph.return_value.home.devices = [homegraph_device]
 
         # With no discover_devices, with no model_list
-        google_devices = self.client.get_google_devices(disable_discovery=True)
+        google_devices = self.client.get_google_devices(
+            discovery_address_dict={fake_device_name: fake_ip_address}
+        )
         self.assertEqual(len(google_devices), 1)
 
         google_device = google_devices[0]
         self.assertDevice(google_device, homegraph_device)
+        self.assertIsNotNone(google_device.network_device)
+        if google_device.network_device is not None:
+            self.assertEqual(google_device.network_device.ip_address, fake_ip_address)
 
         # With two devices returned from homegraph
         # but one device having the invalid token
-        homegraph_device_valid = faker.homegraph_device()
+        homegraph_device_valid = faker.homegraph_device(device_name=fake_device_name)
         homegraph_device_invalid = faker.homegraph_device()
         homegraph_device_invalid.local_auth_token = (
             faker.word()
@@ -345,9 +352,16 @@ def test_get_google_devices(self, m_get_homegraph: NonCallableMock) -> None:
             homegraph_device_invalid,
             homegraph_device_valid,
         ]
-        google_devices = self.client.get_google_devices(disable_discovery=True)
+        google_devices = self.client.get_google_devices(
+            discovery_address_dict={fake_device_name: fake_ip_address}
+        )
         self.assertEqual(len(google_devices), 1)
         self.assertDevice(google_devices[0], homegraph_device_valid)
+        self.assertIsNotNone(google_devices[0].network_device)
+        if google_devices[0].network_device is not None:
+            self.assertEqual(
+                google_devices[0].network_device.ip_address, fake_ip_address
+            )
 
     @patch("glocaltokens.client.GLocalAuthenticationTokens.get_google_devices")
     def test_get_google_devices_json(
@@ -372,7 +386,7 @@ def test_get_google_devices_json(
         )
         m_get_google_devices.return_value = [google_device]
 
-        json_string = self.client.get_google_devices_json(disable_discovery=True)
+        json_string = self.client.get_google_devices_json(discovery_address_dict={})
         self.assertEqual(m_get_google_devices.call_count, 1)
         received_json = json.loads(json_string)
         received_device = received_json[0]