add some unit tests

plugins/modules/ec2_vpc_net.py

@@ -237,27 +237,26 @@ def get_vpc_id(connection, module: AnsibleAWSModule) -> Optional[str]:
     with a CIDR, it will check for matching tags to determine if it is a match
     otherwise it will assume the VPC does not exist and thus return None.
     """
-    vpc_id = module.params.get("vpc_id")
-    if not vpc_id:
-        name = module.params.get("name")
-        cidr_block = module.params.get("cidr_block")
-        multi = module.params.get("multi_ok")
-        vpc_filters = ansible_dict_to_boto3_filter_list({"tag:Name": name, "cidr-block": cidr_block})
+    name = module.params.get("name")
+    cidr_block = module.params.get("cidr_block")
+    multi_ok = module.params.get("multi_ok")
+    vpc_filters = ansible_dict_to_boto3_filter_list({"tag:Name": name, "cidr-block": cidr_block})
+    matching_vpcs = describe_vpcs(connection, Filters=vpc_filters)
+    # If an exact matching using a list of CIDRs isn't found, check for a match with the first CIDR as is documented for C(cidr_block)
+    if not matching_vpcs:
+        vpc_filters = ansible_dict_to_boto3_filter_list({"tag:Name": name, "cidr-block": [cidr_block[0]]})
         matching_vpcs = describe_vpcs(connection, Filters=vpc_filters)
-        # If an exact matching using a list of CIDRs isn't found, check for a match with the first CIDR as is documented for C(cidr_block)
-        if not matching_vpcs:
-            vpc_filters = ansible_dict_to_boto3_filter_list({"tag:Name": name, "cidr-block": [cidr_block[0]]})
-            matching_vpcs = describe_vpcs(connection, Filters=vpc_filters)
-
-        if len(matching_vpcs) == 1:
-            vpc_id = matching_vpcs[0]["VpcId"]
-        elif len(matching_vpcs) > 1 and not multi:
+
+    vpc_id = None
+    if not multi_ok and matching_vpcs:
+        if len(matching_vpcs) > 1:
             module.fail_json(
                 msg=(
                     f"Currently there are {len(matching_vpcs)} VPCs that have the same name and CIDR block you specified."
                     " If you would like to create the VPC anyway please pass True to the multi_ok param."
                 )
             )
+        vpc_id = matching_vpcs[0]["VpcId"]
     return vpc_id
 
 
@@ -343,7 +342,7 @@ def create_vpc_net(
     # Wait for the VPC to enter an 'Available' State
     wait_for_vpc(module, connection, waiter_name="vpc_available", max_attempts=30, VpcIds=[vpc_obj["VpcId"]])
 
-    return vpc_obj["Vpc"]
+    return vpc_obj
 
 
 def wait_for_vpc_attribute(connection, module, vpc_id, attribute, expected_value):
@@ -473,8 +472,8 @@ def update_ipv6_cidrs(connection, module, vpc_obj, vpc_id, ipv6_cidr):
     return True
 
 
-def update_cidrs(connection, module, vpc_obj, vpc_id, cidr_block, purge_cidrs):
-    if cidr_block is None:
+def update_cidrs(connection, module, vpc_obj, cidr_block, purge_cidrs):
+    if not cidr_block:
         return False, None
 
     associated_cidrs = dict(
@@ -494,26 +493,24 @@ def update_cidrs(connection, module, vpc_obj, vpc_id, cidr_block, purge_cidrs):
     if not cidrs_to_add and not cidrs_to_remove:
         return False, None
 
-    if module.check_mode:
-        return True, list(desired_cidrs)
-
-    for cidr in cidrs_to_add:
-        try:
-            associate_vpc_cidr_block(connection, vpc_id=vpc_id, CidrBlock=cidr)
-        except AnsibleEC2Error as e:
-            module.fail_json_aws(e, f"Unable to associate CIDR {cidr}.")
-
-    for cidr in cidrs_to_remove:
-        try:
-            disassociate_vpc_cidr_block(connection, associated_cidrs[cidr])
-        except AnsibleEC2Error as e:
-            module.fail_json_aws(
-                e,
-                (
-                    f"Unable to disassociate {associated_cidrs[cidr]}. You must detach or delete all gateways and resources"
-                    " that are associated with the CIDR block before you can disassociate it."
-                ),
-            )
+    if not module.check_mode:
+        for cidr in cidrs_to_add:
+            try:
+                associate_vpc_cidr_block(connection, vpc_id=vpc_obj["VpcId"], CidrBlock=cidr)
+            except AnsibleEC2Error as e:
+                module.fail_json_aws(e, f"Unable to associate CIDR {cidr}.")
+
+        for cidr in cidrs_to_remove:
+            try:
+                disassociate_vpc_cidr_block(connection, associated_cidrs[cidr])
+            except AnsibleEC2Error as e:
+                module.fail_json_aws(
+                    e,
+                    (
+                        f"Unable to disassociate {associated_cidrs[cidr]}. You must detach or delete all gateways and resources"
+                        " that are associated with the CIDR block before you can disassociate it."
+                    ),
+                )
     return True, list(desired_cidrs)
 
 
@@ -610,7 +607,7 @@ def ensure_present(connection, module: AnsibleAWSModule, vpc_id: Optional[str])
         changed |= update_ipv6_cidrs(connection, module, vpc_obj, vpc_id, ipv6_cidr)
         changed |= update_vpc_tags(connection, module, vpc_id, tags, name, purge_tags)
 
-    cidrs_changed, desired_cidrs = update_cidrs(connection, module, vpc_obj, vpc_id, cidr_block, purge_cidrs)
+    cidrs_changed, desired_cidrs = update_cidrs(connection, module, vpc_obj, cidr_block, purge_cidrs)
     changed |= cidrs_changed
     changed |= update_dhcp_opts(connection, module, vpc_obj, dhcp_id)
     changed |= update_dns_enabled(connection, module, vpc_id, dns_support)
@@ -669,7 +666,7 @@ def main():
 
     connection = module.client("ec2")
     try:
-        vpc_id = get_vpc_id(connection, module)
+        vpc_id = module.params.get("vpc_id") or get_vpc_id(connection, module)
         if state == "present":
             ensure_present(connection, module, vpc_id)
         else:

tests/unit/plugins/modules/test_ec2_vpc_net.py

@@ -0,0 +1,265 @@
+# (c) 2024 Red Hat Inc.
+#
+# This file is part of Ansible
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from unittest.mock import ANY
+from unittest.mock import MagicMock
+from unittest.mock import call
+from unittest.mock import patch
+
+import pytest
+
+from ansible_collections.amazon.aws.plugins.modules import ec2_vpc_net
+
+module_name = "ansible_collections.amazon.aws.plugins.modules.ec2_vpc_net"
+
+
+@pytest.fixture
+def ansible_module():
+    module = MagicMock()
+    module.check_mode = False
+    module.params = {}
+    module.fail_json.side_effect = SystemExit(1)
+    module.fail_json_aws.side_effect = SystemExit(1)
+
+    return module
+
+
+@patch(module_name + ".ansible_dict_to_boto3_filter_list")
+@patch(module_name + ".describe_vpcs")
+def test_get_vpc_id_no_id(m_describe_vpcs, m_ansible_dict_to_boto3_filter_list, ansible_module):
+    connection = MagicMock()
+    cidr_block = MagicMock()
+    vpc_id = MagicMock()
+
+    m_ansible_dict_to_boto3_filter_list.return_value = MagicMock()
+    ansible_module.params.update({"cidr_block": cidr_block})
+    m_describe_vpcs.return_value = [{"VpcId": vpc_id}]
+
+    assert ec2_vpc_net.get_vpc_id(connection, ansible_module) == vpc_id
+    m_ansible_dict_to_boto3_filter_list.assert_called_once_with(
+        {"tag:Name": ansible_module.params.get("name"), "cidr-block": cidr_block}
+    )
+    m_describe_vpcs.assert_called_once_with(connection, Filters=m_ansible_dict_to_boto3_filter_list.return_value)
+
+
+@pytest.mark.parametrize("multi_ok", [True, False])
+@patch(module_name + ".ansible_dict_to_boto3_filter_list")
+@patch(module_name + ".describe_vpcs")
+def test_get_vpc_id_multiple(m_describe_vpcs, m_ansible_dict_to_boto3_filter_list, ansible_module, multi_ok):
+    connection = MagicMock()
+    cidr_block = MagicMock()
+    vpc_id = MagicMock()
+    another_vpc_id = MagicMock()
+
+    m_ansible_dict_to_boto3_filter_list.return_value = MagicMock()
+    ansible_module.params.update({"cidr_block": cidr_block, "multi_ok": multi_ok})
+    m_describe_vpcs.return_value = [{"VpcId": vpc_id}, {"VpcId": another_vpc_id}]
+
+    if multi_ok:
+        assert not ec2_vpc_net.get_vpc_id(connection, ansible_module)
+    else:
+        with pytest.raises(SystemExit):
+            ec2_vpc_net.get_vpc_id(connection, ansible_module)
+
+    m_ansible_dict_to_boto3_filter_list.assert_called_once_with(
+        {"tag:Name": ansible_module.params.get("name"), "cidr-block": cidr_block}
+    )
+    m_describe_vpcs.assert_called_once_with(connection, Filters=m_ansible_dict_to_boto3_filter_list.return_value)
+
+
+@patch(module_name + ".ansible_dict_to_boto3_filter_list")
+@patch(module_name + ".describe_vpcs")
+def test_get_vpc_id_cidr_block_list(m_describe_vpcs, m_ansible_dict_to_boto3_filter_list, ansible_module):
+    connection = MagicMock()
+    cidr_block = [MagicMock(), MagicMock()]
+    vpc_id = MagicMock()
+    vpc_filters = [MagicMock(), MagicMock()]
+
+    m_ansible_dict_to_boto3_filter_list.side_effect = vpc_filters
+    ansible_module.params.update({"cidr_block": cidr_block})
+    m_describe_vpcs.side_effect = [[], [{"VpcId": vpc_id}]]
+
+    assert ec2_vpc_net.get_vpc_id(connection, ansible_module) == vpc_id
+    m_ansible_dict_to_boto3_filter_list.assert_has_calls(
+        [
+            call({"tag:Name": ansible_module.params.get("name"), "cidr-block": cidr_block}),
+            call({"tag:Name": ansible_module.params.get("name"), "cidr-block": [cidr_block[0]]}),
+        ]
+    )
+    m_describe_vpcs.assert_has_calls(
+        [
+            call(connection, Filters=vpc_filters[0]),
+            call(connection, Filters=vpc_filters[1]),
+        ]
+    )
+
+
+@pytest.mark.parametrize("cidr_block", [None, []])
+@patch(module_name + ".disassociate_vpc_cidr_block")
+@patch(module_name + ".associate_vpc_cidr_block")
+def test_update_cidrs_no_cidr_block(
+    m_associate_vpc_cidr_block, m_disassociate_vpc_cidr_block, ansible_module, cidr_block
+):
+    connection = MagicMock()
+    vpc_obj = {}
+    cidr_block = []
+    changed, desired_cidrs = ec2_vpc_net.update_cidrs(connection, ansible_module, vpc_obj, cidr_block, ANY)
+    assert not changed
+    assert desired_cidrs is None
+
+    m_associate_vpc_cidr_block.assert_not_called()
+    m_disassociate_vpc_cidr_block.assert_not_called()
+
+
+@pytest.mark.parametrize("check_mode", [True, False])
+@pytest.mark.parametrize(
+    "associated_cidrs, cidr_block, purge_cidrs, add, remove, expected",
+    [
+        (
+            [
+                {
+                    "AssociationId": "vpc-cidr-assoc-001",
+                    "CidrBlock": "10.1.0.0/24",
+                    "CidrBlockState": {"State": "associated"},
+                },
+            ],
+            ["10.1.0.0/24"],
+            True,
+            [],
+            [],
+            (False, None),
+        ),
+        (
+            [
+                {
+                    "AssociationId": "vpc-cidr-assoc-001",
+                    "CidrBlock": "10.1.0.0/24",
+                    "CidrBlockState": {"State": "associated"},
+                },
+            ],
+            ["10.1.0.0/24"],
+            False,
+            [],
+            [],
+            (False, None),
+        ),
+        (
+            [
+                {
+                    "AssociationId": "vpc-cidr-assoc-001",
+                    "CidrBlock": "10.1.0.0/24",
+                    "CidrBlockState": {"State": "associated"},
+                },
+                {
+                    "AssociationId": "vpc-cidr-assoc-001",
+                    "CidrBlock": "10.2.0.0/24",
+                    "CidrBlockState": {"State": "failed"},
+                },
+            ],
+            ["10.1.0.0/24"],
+            True,
+            [],
+            [],
+            (True, ["10.1.0.0/24"]),
+        ),
+        (
+            [
+                {
+                    "AssociationId": "vpc-cidr-assoc-001",
+                    "CidrBlock": "10.1.0.0/24",
+                    "CidrBlockState": {"State": "associated"},
+                },
+                {
+                    "AssociationId": "vpc-cidr-assoc-001",
+                    "CidrBlock": "10.2.0.0/24",
+                    "CidrBlockState": {"State": "failed"},
+                },
+            ],
+            ["10.1.0.0/24"],
+            False,
+            [],
+            [],
+            (False, None),
+        ),
+        (
+            [
+                {
+                    "AssociationId": "vpc-cidr-assoc-001",
+                    "CidrBlock": "10.1.0.0/24",
+                    "CidrBlockState": {"State": "associated"},
+                },
+                {
+                    "AssociationId": "vpc-cidr-assoc-001",
+                    "CidrBlock": "10.2.0.0/24",
+                    "CidrBlockState": {"State": "disassociating"},
+                },
+            ],
+            ["10.2.0.0/24"],
+            False,
+            ["10.2.0.0/24"],
+            [],
+            (True, ["10.1.0.0/24", "10.2.0.0/24"]),
+        ),
+        (
+            [
+                {
+                    "AssociationId": "vpc-cidr-assoc-001",
+                    "CidrBlock": "10.1.0.0/24",
+                    "CidrBlockState": {"State": "associated"},
+                },
+                {
+                    "AssociationId": "vpc-cidr-assoc-001",
+                    "CidrBlock": "10.2.0.0/24",
+                    "CidrBlockState": {"State": "disassociating"},
+                },
+            ],
+            ["10.2.0.0/24"],
+            True,
+            ["10.2.0.0/24"],
+            ["vpc-cidr-assoc-001"],
+            (True, ["10.2.0.0/24"]),
+        ),
+    ],
+)
+@patch(module_name + ".disassociate_vpc_cidr_block")
+@patch(module_name + ".associate_vpc_cidr_block")
+def test_update_cidrs_no_purge(
+    m_associate_vpc_cidr_block,
+    m_disassociate_vpc_cidr_block,
+    ansible_module,
+    check_mode,
+    associated_cidrs,
+    cidr_block,
+    purge_cidrs,
+    add,
+    remove,
+    expected,
+):
+    connection = MagicMock()
+    vpc_obj = {"VpcId": MagicMock(), "CidrBlockAssociationSet": associated_cidrs}
+    ansible_module.check_mode = check_mode
+    changed, desired_cidrs = ec2_vpc_net.update_cidrs(connection, ansible_module, vpc_obj, cidr_block, purge_cidrs)
+    assert expected[0] == changed
+
+    def sorted_list(x):
+        return x if not x else sorted(x)
+
+    assert sorted_list(expected[1]) == sorted_list(desired_cidrs)
+
+    if not expected[0] or check_mode:
+        m_associate_vpc_cidr_block.assert_not_called()
+        m_disassociate_vpc_cidr_block.assert_not_called()
+    else:
+        if add:
+            m_associate_vpc_cidr_block.assert_has_calls(
+                [call(connection, vpc_id=vpc_obj["VpcId"], CidrBlock=cidr) for cidr in add], any_order=True
+            )
+        else:
+            m_associate_vpc_cidr_block.assert_not_called()
+
+        if remove:
+            m_disassociate_vpc_cidr_block.assert_has_calls(
+                [call(connection, association_id) for association_id in remove], any_order=False
+            )