ELBv2: Describe/Modify Listener Attributes (#8280)

moto/elbv2/models.py

@@ -315,6 +315,8 @@ def __init__(
             is_default=True,
         )
 
+        self.attributes = {"tcp.idle_timeout.seconds": "350"}
+
     @property
     def physical_resource_id(self) -> str:
         return self.arn
@@ -2000,6 +2002,21 @@ def describe_tags(self, resource_arns: List[str]) -> Dict[str, Dict[str, str]]:
             for arn in resource_arns
         }
 
+    def describe_listener_attributes(self, listener_arn: str) -> Dict[str, str]:
+        listener = self.describe_listeners(
+            load_balancer_arn=None, listener_arns=[listener_arn]
+        )[0]
+        return listener.attributes
+
+    def modify_listener_attributes(
+        self, listener_arn: str, attrs: List[Dict[str, str]]
+    ) -> Dict[str, str]:
+        listener = self.describe_listeners(
+            load_balancer_arn=None, listener_arns=[listener_arn]
+        )[0]
+        listener.attributes.update({a["Key"]: a["Value"] for a in attrs})
+        return listener.attributes
+
     def _get_resource_by_arn(self, arn: str) -> Any:
         if ":targetgroup" in arn:
             resource: Any = self.target_groups.get(arn)

moto/elbv2/responses.py

@@ -822,6 +822,21 @@ def remove_listener_certificates(self) -> str:
         template = self.response_template(REMOVE_LISTENER_CERTIFICATES_TEMPLATE)
         return template.render()
 
+    def describe_listener_attributes(self) -> str:
+        arn = self._get_param("ListenerArn")
+        attrs = self.elbv2_backend.describe_listener_attributes(arn)
+        template = self.response_template(DESCRIBE_LISTENER_ATTRIBUTES)
+        return template.render(attributes=attrs)
+
+    def modify_listener_attributes(self) -> str:
+        arn = self._get_param("ListenerArn")
+        attrs = self._get_params()["Attributes"]
+        updated_attrs = self.elbv2_backend.modify_listener_attributes(
+            listener_arn=arn, attrs=attrs
+        )
+        template = self.response_template(MODIFY_LISTENER_ATTRIBUTES)
+        return template.render(attributes=updated_attrs)
+
 
 ADD_TAGS_TEMPLATE = """<AddTagsResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
   <AddTagsResult/>
@@ -1912,3 +1927,35 @@ def remove_listener_certificates(self) -> str:
     <RequestId>{{ request_id }}</RequestId>
   </ResponseMetadata>
 </RemoveListenerCertificatesResponse>"""
+
+DESCRIBE_LISTENER_ATTRIBUTES = """<DescribeListenerAttributesResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
+  <DescribeListenerAttributesResult>
+    <Attributes>
+      {% for attr, value in attributes.items() %}
+      <ListenerAttribute>
+        <Key>{{ attr }}</Key>
+        <Value>{{ value }}</Value>
+      </ListenerAttribute>
+      {% endfor %}
+    </Attributes>
+  </DescribeListenerAttributesResult>
+  <ResponseMetadata>
+    <RequestId>{{ request_id }}</RequestId>
+  </ResponseMetadata>
+</DescribeListenerAttributesResponse>"""
+
+MODIFY_LISTENER_ATTRIBUTES = """<ModifyListenerAttributesResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
+  <ModifyListenerAttributesResult>
+    <Attributes>
+      {% for attr, value in attributes.items() %}
+      <ListenerAttribute>
+        <Key>{{ attr }}</Key>
+        <Value>{{ value }}</Value>
+      </ListenerAttribute>
+      {% endfor %}
+    </Attributes>
+  </ModifyListenerAttributesResult>
+  <ResponseMetadata>
+    <RequestId>{{ request_id }}</RequestId>
+  </ResponseMetadata>
+</ModifyListenerAttributesResponse>"""

other_langs/terraform/elb/listener.tf

@@ -0,0 +1,15 @@
+resource "aws_lb_listener" "front_end" {
+  load_balancer_arn = aws_lb.this.arn
+  port              = "80"
+  protocol          = "HTTP"
+
+  default_action {
+    type = "redirect"
+
+    redirect {
+      port        = "443"
+      protocol    = "HTTPS"
+      status_code = "HTTP_301"
+    }
+  }
+}

tests/test_elbv2/test_elbv2_listener_attributes.py

@@ -0,0 +1,48 @@
+import boto3
+import pytest
+
+from moto import mock_aws
+
+
+@mock_aws
+@pytest.mark.parametrize(
+    "name,default_value,new_value", [("tcp.idle_timeout.seconds", "350", "360")]
+)
+def test_modify_listener_attributes(name, default_value, new_value):
+    elbv2 = boto3.client("elbv2", "us-east-1")
+
+    ec2 = boto3.resource("ec2", region_name="us-east-1")
+
+    vpc = ec2.create_vpc(CidrBlock="172.28.7.0/24", InstanceTenancy="default")
+    subnet1 = ec2.create_subnet(
+        VpcId=vpc.id, CidrBlock="172.28.7.192/26", AvailabilityZone="us-east-1a"
+    )
+
+    response = elbv2.create_load_balancer(
+        Name="my-lb", Subnets=[subnet1.id], Scheme="internal"
+    )
+    load_balancer_arn = response["LoadBalancers"][0]["LoadBalancerArn"]
+
+    response = elbv2.create_target_group(
+        Name="a-target", Protocol="HTTPS", Port=8443, VpcId=vpc.id
+    )
+    target_group = response["TargetGroups"][0]
+    target_group_arn = target_group["TargetGroupArn"]
+
+    listener_arn = elbv2.create_listener(
+        LoadBalancerArn=load_balancer_arn,
+        Protocol="HTTPS",
+        Port=443,
+        DefaultActions=[{"Type": "forward", "TargetGroupArn": target_group_arn}],
+    )["Listeners"][0]["ListenerArn"]
+
+    attrs = elbv2.describe_listener_attributes(ListenerArn=listener_arn)
+    assert attrs["Attributes"] == [{"Key": name, "Value": default_value}]
+
+    attrs = elbv2.modify_listener_attributes(
+        ListenerArn=listener_arn, Attributes=[{"Key": name, "Value": new_value}]
+    )
+    assert attrs["Attributes"] == [{"Key": name, "Value": new_value}]
+
+    attrs = elbv2.describe_listener_attributes(ListenerArn=listener_arn)
+    assert attrs["Attributes"] == [{"Key": name, "Value": new_value}]