elbv2 boto3 client - implement new api for ip address type

plugins/module_utils/elbv2.py

@@ -71,8 +71,10 @@ def __init__(self, connection, module):
         if self.elb is not None:
             self.elb_attributes = self.get_elb_attributes()
             self.elb['tags'] = self.get_elb_tags()
+            self.elb_ip_addr_type = self.get_elb_ip_address_type()
         else:
             self.elb_attributes = None
+            self.elb_ip_addr_type = None
 
     def wait_for_status(self, elb_arn):
         """
@@ -107,6 +109,15 @@ def get_elb_attributes(self):
         # Replace '.' with '_' in attribute key names to make it more Ansibley
         return dict((k.replace('.', '_'), v) for k, v in elb_attributes.items())
 
+    def get_elb_ip_address_type(self):
+        """
+        Retrieve load balancer ip address type using describe_load_balancers
+
+        :return:
+        """
+
+        return self.elb.get('IpAddressType', None)
+
     def update_elb_attributes(self):
         """
         Update the elb_attributes parameter
@@ -232,6 +243,21 @@ def update(self):
         self.elb = get_elb(self.connection, self.module, self.module.params.get("name"))
         self.elb['tags'] = self.get_elb_tags()
 
+    def modify_ip_address_type(self, ip_addr_type):
+        """
+        Modify ELB ip address type
+        :return:
+        """
+        if self.elb_ip_addr_type != ip_addr_type:
+            try:
+                AWSRetry.jittered_backoff()(
+                    self.connection.set_ip_address_type
+                )(LoadBalancerArn=self.elb['LoadBalancerArn'], IpAddressType=ip_addr_type)
+            except (BotoCoreError, ClientError) as e:
+                self.module.fail_json_aws(e)
+
+            self.changed = True
+
 
 class ApplicationLoadBalancer(ElasticLoadBalancerV2):
 

tests/unit/module_utils/test_elbv2.py

@@ -8,7 +8,8 @@
 __metaclass__ = type
 
 import ansible_collections.amazon.aws.plugins.module_utils.elbv2 as elbv2
-
+from ansible_collections.amazon.aws.tests.unit.compat import unittest
+from ansible_collections.amazon.aws.tests.unit.compat.mock import MagicMock
 
 one_action = [
     {
@@ -41,3 +42,126 @@ def _prune_secret():
 
 def _sort_actions_one_entry():
     assert elbv2._sort_actions(one_action) == one_action
+
+
+class ElBV2UtilsTestSuite(unittest.TestCase):
+
+    def setUp(self):
+        self.connection = MagicMock(name="connection")
+        self.module = MagicMock(name="module")
+
+        self.module.params = dict()
+
+        self.conn_paginator = MagicMock(name="connection.paginator")
+        self.paginate = MagicMock(name="paginator.paginate")
+
+        self.connection.get_paginator.return_value = self.conn_paginator
+        self.conn_paginator.paginate.return_value = self.paginate
+
+        self.loadbalancer = {
+            "Type": "application",
+            "Scheme": "internet-facing",
+            "IpAddressType": "ipv4",
+            "VpcId": "vpc-3ac0fb5f",
+            "AvailabilityZones": [
+                {
+                    "ZoneName": "us-west-2a",
+                    "SubnetId": "subnet-8360a9e7"
+                },
+                {
+                    "ZoneName": "us-west-2b",
+                    "SubnetId": "subnet-b7d581c0"
+                }
+            ],
+            "CreatedTime": "2016-03-25T21:26:12.920Z",
+            "CanonicalHostedZoneId": "Z2P70J7EXAMPLE",
+            "DNSName": "my-load-balancer-424835706.us-west-2.elb.amazonaws.com",
+            "SecurityGroups": [
+                "sg-5943793c"
+            ],
+            "LoadBalancerName": "my-load-balancer",
+            "State": {
+                "Code": "active"
+            },
+            "LoadBalancerArn": "arn:aws:elasticloadbalancing:us-west-2:123456789012:loadbalancer/app/my-load-balancer/50dc6c495c0c9188"
+        }
+        self.paginate.build_full_result.return_value = {
+            'LoadBalancers': [self.loadbalancer]
+        }
+
+        self.connection.describe_load_balancer_attributes.return_value = {
+            "Attributes": [
+                {
+                    "Value": "false",
+                    "Key": "access_logs.s3.enabled"
+                },
+                {
+                    "Value": "",
+                    "Key": "access_logs.s3.bucket"
+                },
+                {
+                    "Value": "",
+                    "Key": "access_logs.s3.prefix"
+                },
+                {
+                    "Value": "60",
+                    "Key": "idle_timeout.timeout_seconds"
+                },
+                {
+                    "Value": "false",
+                    "Key": "deletion_protection.enabled"
+                },
+                {
+                    "Value": "true",
+                    "Key": "routing.http2.enabled"
+                }
+            ]
+        }
+        self.connection.describe_tags.return_value = {
+            "TagDescriptions": [
+                {
+                    "ResourceArn": "arn:aws:elasticloadbalancing:us-west-2:123456789012:loadbalancer/app/my-load-balancer/50dc6c495c0c9188",
+                    "Tags": [
+                        {
+                            "Value": "ansible",
+                            "Key": "project"
+                        },
+                        {
+                            "Value": "RedHat",
+                            "Key": "company"
+                        }
+                    ]
+                }
+            ]
+        }
+        self.elbv2obj = elbv2.ElasticLoadBalancerV2(self.connection, self.module)
+
+    # Test the simplest case - Read the ip address type
+    def test_get_elb_ip_address_type(self):
+        # Run module
+        return_value = self.elbv2obj.get_elb_ip_address_type()
+        # check that no method was called and this has been retrieved from elb attributes
+        self.connection.describe_load_balancer_attributes.assert_called_once()
+        self.connection.get_paginator.assert_called_once()
+        self.connection.describe_tags.assert_called_once()
+        self.conn_paginator.paginate.assert_called_once()
+        # assert we got the expected value
+        self.assertEqual(return_value, 'ipv4')
+
+    # Test modify_ip_address_type idempotency
+    def test_modify_ip_address_type_idempotency(self):
+        # Run module
+        return_value = self.elbv2obj.modify_ip_address_type("ipv4")
+        # check that no method was called and this has been retrieved from elb attributes
+        self.connection.set_ip_address_type.assert_not_called()
+        # assert we got the expected value
+        self.assertEqual(self.elbv2obj.changed, False)
+
+    # Test modify_ip_address_type
+    def test_modify_ip_address_type_update(self):
+        # Run module
+        return_value = self.elbv2obj.modify_ip_address_type("dualstack")
+        # check that no method was called and this has been retrieved from elb attributes
+        self.connection.set_ip_address_type.assert_called_once()
+        # assert we got the expected value
+        self.assertEqual(self.elbv2obj.changed, True)