Add utils for the ec2_vpc_vpn* modules

Signed-off-by: Alina Buzachis <[email protected]>

changelogs/fragments/20240927-ec2-utils.yml

@@ -0,0 +1,2 @@
+minor_changes:
+  - module_utils/ec2 - add utils for the ec2_vpc_vpn* modules (https://github.com/ansible-collections/amazon.aws/pull/2312).

plugins/module_utils/ec2.py

@@ -320,7 +320,6 @@ def disassociate_vpc_cidr_block(client, association_id: str) -> Dict[str, Any]:
 
 # EC2 VPC Peering Connection
 class EC2VpcPeeringErrorHandler(AWSErrorHandler):
-    _CUSTOM_EXCEPTION = AnsibleEC2Error
 
     @classmethod
     def _is_missing(cls):
@@ -361,6 +360,49 @@ def accept_vpc_peering_connection(client, peering_id: str) -> bool:
 def reject_vpc_peering_connection(client, peering_id: str) -> bool:
     client.reject_vpc_peering_connection(VpcPeeringConnectionId=peering_id)
     return True
+        return is_boto3_error_code("InvalidVpnConnectionID.NotFound")
+
+
+# EC2 VPC VPN
+class EC2VpnErrorHandler(AWSErrorHandler):
+    _CUSTOM_EXCEPTION = AnsibleEC2Error
+
+    @classmethod
+    def _is_missing(cls):
+        return is_boto3_error_code("InvalidVpnConnectionID.NotFound")
+
+
+@EC2VpcErrorHandler.list_error_handler("describe vpn connections", [])
+@AWSRetry.jittered_backoff()
+def describe_vpn_connections(client, **params: Dict[str, Any]) -> List[Dict[str, Any]]:
+    # The paginator does not exist for `describe_vpn_connections`
+    return client.describe_vpn_connections(**params)
+
+
+@EC2VpcErrorHandler.common_error_handler("create vpn connection route")
+@AWSRetry.jittered_backoff()
+def create_vpn_connection_route(client, vpn_connection_id: str, route: Dict[str, Any]) -> bool:
+    client.create_vpn_connection_route(VpnConnectionId=vpn_connection_id, DestinationCidrBlock=route)
+    return True
+
+
+@EC2VpcErrorHandler.common_error_handler("delete vpn connection route")
+@AWSRetry.jittered_backoff()
+def delete_vpn_connection_route(client, vpn_connection_id: str, route: Dict[str, Any]) -> bool:
+    client.delete_vpn_connection_route(VpnConnectionId=vpn_connection_id, DestinationCidrBlock=route)
+    return True
+
+
+@EC2VpcErrorHandler.common_error_handler("create vpn connection")
+@AWSRetry.jittered_backoff()
+def create_vpn_connection(client, **params: Dict[str, Any]) -> Dict[str, Any]:
+    return client.create_vpn_connection(**params)
+
+
+@EC2VpcErrorHandler.common_error_handler("delete vpn connection")
+@AWSRetry.jittered_backoff()
+def delete_vpn_connection(client, vpn_connection_id: str) -> Dict[str, Any]:
+    return client.delete_vpn_connection(VpnConnectionId=vpn_connection_id)
 
 
 # EC2 Internet Gateway