ELBv2: describe-target-health: return unavailable target (#6195)

moto/elbv2/models.py

@@ -152,7 +152,17 @@ class FakeTargetGroup(CloudFormationModel):
     def health_for(self, target: Dict[str, Any], ec2_backend: Any) -> FakeHealthStatus:
         t = self.targets.get(target["id"])
         if t is None:
-            raise InvalidTargetError()
+            port = self.port
+            if "port" in target:
+                port = target["port"]
+            return FakeHealthStatus(
+                target["id"],
+                port,
+                self.healthcheck_port,
+                "unavailable",
+                "Target.NotRegistered",
+                "Target is not registered",
+            )
         if t["id"].startswith("i-"):  # EC2 instance ID
             instance = ec2_backend.get_instance_by_id(t["id"])
             if instance.state == "stopped":
@@ -385,7 +395,6 @@ class FakeListenerRule(CloudFormationModel):
         account_id: str,
         region_name: str,
     ) -> "FakeListenerRule":
-
         properties = cloudformation_json["Properties"]
 
         elbv2_backend = elbv2_backends[account_id][region_name]
@@ -765,7 +774,6 @@ class ELBv2Backend(BaseBackend):
     def convert_and_validate_action_properties(
         self, properties: Dict[str, Any]
     ) -> List[Dict[str, Any]]:
-
         # transform Actions to confirm with the rest of the code and XML templates
         default_actions = []
         for i, action in enumerate(properties["Actions"]):
@@ -1094,7 +1102,6 @@ Member must satisfy regular expression pattern: {expression}"
     def convert_and_validate_certificates(
         self, certificates: List[Dict[str, Any]]
     ) -> List[Dict[str, Any]]:
-
         # transform default certificate to conform with the rest of the code and XML templates
         for cert in certificates or []:
             cert["certificate_arn"] = cert["CertificateArn"]
@@ -1104,7 +1111,6 @@ Member must satisfy regular expression pattern: {expression}"
     def convert_and_validate_properties(
         self, properties: Dict[str, Any]
     ) -> List[Dict[str, Any]]:
-
         # transform default actions to confirm with the rest of the code and XML templates
         # Caller: CF create/update for type "AWS::ElasticLoadBalancingV2::Listener"
         default_actions = []

tests/test_elbv2/test_elbv2.py

@@ -501,6 +501,36 @@ def test_register_targets():
     )
     response.get("TargetHealthDescriptions").should.have.length_of(1)
 
+    def get_target_by_instance_id(instance_id):
+        for target in response.get("TargetHealthDescriptions"):
+            if target.get("Target").get("Id") == instance_id:
+                return target
+        return None
+
+    def assert_target_not_registered(target):
+        assert target.get("TargetHealth").get("State") == "unavailable"
+        assert target.get("TargetHealth").get("Reason") == "Target.NotRegistered"
+
+    response = conn.describe_target_health(
+        TargetGroupArn=target_group.get("TargetGroupArn"),
+        Targets=[{"Id": instance_id2}],
+    )
+    response.get("TargetHealthDescriptions").should.have.length_of(1)
+    target_default_port = get_target_by_instance_id(instance_id2)
+    assert target_default_port is not None
+    assert target_default_port.get("Target").get("Port") == 8080
+    assert_target_not_registered(target_default_port)
+
+    response = conn.describe_target_health(
+        TargetGroupArn=target_group.get("TargetGroupArn"),
+        Targets=[{"Id": instance_id2, "Port": 4030}],
+    )
+    response.get("TargetHealthDescriptions").should.have.length_of(1)
+    target_custom_port = get_target_by_instance_id(instance_id2)
+    assert target_custom_port is not None
+    assert target_custom_port.get("Target").get("Port") == 4030
+    assert_target_not_registered(target_custom_port)
+
 
 @mock_ec2
 @mock_elbv2