feat: parse nested load balancer label_selector targets (hetznercloud#633)

When a load balancer has a label_selector type target, the API returns a
nested "targets" array containing the resolved individual server targets
with their health statuses. This data was previously discarded.

- Add `targets` parameter to `LoadBalancerTarget.__init__` in domain.py
- Parse nested targets in `BoundLoadBalancer.__init__` for
label_selector targets, creating `LoadBalancerTarget` objects with
server, health_status, type, and use_private_ip fields
- Add test fixture and test case for nested target parsing

---------

Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: jo <ljonas@riseup.net>

Author: 3 people

hcloud/load_balancers/client.py

@@ -93,36 +93,51 @@ def __init__(
             ]
             data["private_net"] = private_nets
 
-        targets = data.get("targets")
-        if targets:
-            tmp_targets = []
-            for target in targets:
-                tmp_target = LoadBalancerTarget(type=target["type"])
-                if target["type"] == "server":
-                    tmp_target.server = BoundServer(
-                        client._parent.servers, data=target["server"], complete=False
-                    )
-                    tmp_target.use_private_ip = target["use_private_ip"]
-                elif target["type"] == "label_selector":
-                    tmp_target.label_selector = LoadBalancerTargetLabelSelector(
-                        selector=target["label_selector"]["selector"]
+        def _load_balancer_targets(
+            raw_targets: list[dict[str, Any]],
+        ) -> list[LoadBalancerTarget]:
+            return [_load_balancer_target(raw_target) for raw_target in raw_targets]
+
+        def _load_balancer_target(
+            raw_target: dict[str, Any],
+        ) -> LoadBalancerTarget:
+            result = LoadBalancerTarget(type=raw_target["type"])
+
+            if raw_target["type"] == "ip":
+                result.ip = LoadBalancerTargetIP(
+                    ip=raw_target["ip"]["ip"],
+                )
+
+            elif raw_target["type"] == "server":
+                result.server = BoundServer(
+                    client._parent.servers,  # pylint: disable=protected-access
+                    data=raw_target["server"],
+                    complete=False,
+                )
+                result.use_private_ip = raw_target["use_private_ip"]
+
+            elif raw_target["type"] == "label_selector":
+                result.label_selector = LoadBalancerTargetLabelSelector(
+                    selector=raw_target["label_selector"]["selector"]
+                )
+                result.use_private_ip = raw_target["use_private_ip"]
+
+                if (raw_nested_targets := raw_target.get("targets")) is not None:
+                    result.targets = _load_balancer_targets(raw_nested_targets)
+
+            if (raw_health_status := raw_target.get("health_status")) is not None:
+                result.health_status = [
+                    LoadBalancerTargetHealthStatus(
+                        listen_port=item["listen_port"],
+                        status=item["status"],
                     )
-                    tmp_target.use_private_ip = target["use_private_ip"]
-                elif target["type"] == "ip":
-                    tmp_target.ip = LoadBalancerTargetIP(ip=target["ip"]["ip"])
-
-                target_health_status = target.get("health_status")
-                if target_health_status is not None:
-                    tmp_target.health_status = [
-                        LoadBalancerTargetHealthStatus(
-                            listen_port=target_health_status_item["listen_port"],
-                            status=target_health_status_item["status"],
-                        )
-                        for target_health_status_item in target_health_status
-                    ]
+                    for item in raw_health_status
+                ]
+
+            return result
 
-                tmp_targets.append(tmp_target)
-            data["targets"] = tmp_targets
+        if (raw_targets := data.get("targets")) is not None:
+            data["targets"] = _load_balancer_targets(raw_targets)
 
         services = data.get("services")
         if services:

hcloud/load_balancers/domain.py

@@ -411,6 +411,8 @@ class LoadBalancerTarget(BaseDomain):
             use the private IP instead of primary public IP
     :param health_status: list
             List of health statuses of the services on this target. Only present for target types "server" and "ip".
+    :param targets: list
+            List of resolved label selector targets. Only present for target types "label_selector".
     """
 
     __api_properties__ = (
@@ -420,6 +422,7 @@ class LoadBalancerTarget(BaseDomain):
         "ip",
         "use_private_ip",
         "health_status",
+        "targets",
     )
     __slots__ = __api_properties__
 
@@ -431,13 +434,15 @@ def __init__(
         ip: LoadBalancerTargetIP | None = None,
         use_private_ip: bool | None = None,
         health_status: list[LoadBalancerTargetHealthStatus] | None = None,
+        targets: list[LoadBalancerTarget] | None = None,
     ):
         self.type = type
         self.server = server
         self.label_selector = label_selector
         self.ip = ip
         self.use_private_ip = use_private_ip
         self.health_status = health_status
+        self.targets = targets
 
     def to_payload(self) -> dict[str, Any]:
         """

tests/unit/load_balancers/conftest.py

@@ -86,7 +86,23 @@ def response_load_balancer():
                     "health_status": [{"listen_port": 443, "status": "healthy"}],
                     "label_selector": None,
                     "use_private_ip": False,
-                }
+                },
+                {
+                    "type": "label_selector",
+                    "label_selector": {"selector": "env=prod"},
+                    "use_private_ip": True,
+                    "targets": [
+                        {
+                            "type": "server",
+                            "server": {"id": 105054278},
+                            "use_private_ip": True,
+                            "health_status": [
+                                {"listen_port": 443, "status": "healthy"},
+                                {"listen_port": 3000, "status": "healthy"},
+                            ],
+                        }
+                    ],
+                },
             ],
             "algorithm": {"type": "round_robin"},
         }

tests/unit/load_balancers/test_client.py

@@ -19,7 +19,7 @@
 )
 from hcloud.locations import Location
 from hcloud.networks import Network
-from hcloud.servers import Server
+from hcloud.servers import BoundServer, Server
 
 from ..conftest import BoundModelTestCase
 
@@ -60,6 +60,30 @@ def test_init(self, response_load_balancer):
         assert bound_load_balancer.id == 4711
         assert bound_load_balancer.name == "Web Frontend"
 
+    def test_init_label_selector_nested_targets(self, response_load_balancer):
+        bound_load_balancer = BoundLoadBalancer(
+            client=mock.MagicMock(), data=response_load_balancer["load_balancer"]
+        )
+
+        label_selector_target = bound_load_balancer.targets[1]
+        assert label_selector_target.type == "label_selector"
+        assert label_selector_target.label_selector.selector == "env=prod"
+        assert label_selector_target.use_private_ip is True
+        assert label_selector_target.targets is not None
+        assert len(label_selector_target.targets) == 1
+
+        nested = label_selector_target.targets[0]
+        assert nested.type == "server"
+        assert isinstance(nested.server, BoundServer)
+        assert nested.server.id == 105054278
+        assert nested.use_private_ip is True
+        assert nested.health_status is not None
+        assert len(nested.health_status) == 2
+        assert nested.health_status[0].listen_port == 443
+        assert nested.health_status[0].status == "healthy"
+        assert nested.health_status[1].listen_port == 3000
+        assert nested.health_status[1].status == "healthy"
+
 
 class TestLoadBalancerslient:
     @pytest.fixture()