diff --git a/plugins/modules/rds_cluster_snapshot.py b/plugins/modules/rds_cluster_snapshot.py
new file mode 100644
index 00000000000..09077c9638b
--- /dev/null
+++ b/plugins/modules/rds_cluster_snapshot.py
@@ -0,0 +1,372 @@
+#!/usr/bin/python
+# Copyright (c) 2014 Ansible Project
+# Copyright (c) 2021 Alina Buzachis (@alinabuzachis)
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+
+DOCUMENTATION = r'''
+---
+module: rds_cluster_snapshot
+version_added: 4.0.0
+short_description: Manage Amazon RDS snapshots of DB clusters
+description:
+  - Create, modify and delete RDS snapshots of DB clusters.
+options:
+  state:
+    description:
+      - Specify the desired state of the snapshot.
+    default: present
+    choices: [ 'present', 'absent']
+    type: str
+  db_cluster_snapshot_identifier:
+    description:
+      - The identifier of the DB cluster snapshot.
+    required: true
+    aliases:
+      - snapshot_id
+      - id
+      - snapshot_name
+    type: str
+  db_cluster_identifier:
+    description:
+      - The identifier of the DB cluster to create a snapshot for.
+      - Required when I(state=present).
+    aliases:
+      - cluster_id
+      - cluster_name
+    type: str
+  source_db_cluster_snapshot_identifier:
+     description:
+      - The identifier of the DB cluster snapshot to copy.
+      - If the source snapshot is in the same AWS region as the copy, specify the snapshot's identifier.
+      - If the source snapshot is in a different AWS region as the copy, specify the snapshot's ARN.
+     aliases:
+      - source_id
+      - source_snapshot_id
+     type: str
+  source_region:
+    description:
+      - The region that contains the snapshot to be copied.
+    type: str
+  copy_tags:
+    description:
+      - Whether to copy all tags from I(source_db_cluster_snapshot_identifier) to I(db_cluster_snapshot_identifier).
+    type: bool
+    default: False
+  wait:
+    description:
+      - Whether or not to wait for snapshot creation or deletion.
+    type: bool
+    default: false
+  wait_timeout:
+    description:
+      - How long before wait gives up, in seconds.
+    default: 300
+    type: int
+notes:
+  - Retrieve the information about a specific DB cluster or list the DB cluster snapshots for a specific DB cluster
+    can de done using M(community.aws.rds_snapshot_info).
+author:
+  - Alina Buzachis (@alinabuzachis)
+extends_documentation_fragment:
+  - amazon.aws.aws
+  - amazon.aws.ec2
+  - amazon.aws.tags
+'''
+
+EXAMPLES = r'''
+- name: Create a DB cluster snapshot
+  community.aws.rds_cluster_snapshot:
+    db_cluster_identifier: "{{ cluster_id }}"
+    db_cluster_snapshot_identifier: new-cluster-snapshot
+
+- name: Delete a DB cluster snapshot
+  community.aws.rds_cluster_snapshot:
+    db_cluster_snapshot_identifier: new-cluster-snapshot
+    state: absent
+
+- name: Copy snapshot from a different region and copy its tags
+  community.aws.rds_cluster_snapshot:
+    id: new-database-snapshot-copy
+    region: us-east-1
+    source_id: "{{ snapshot.db_snapshot_arn }}"
+    source_region: us-east-2
+    copy_tags: yes
+'''
+
+RETURN = r'''
+availability_zone:
+  description: Availability zone of the database from which the snapshot was created.
+  returned: always
+  type: str
+  sample: us-west-2a
+db_cluster_snapshot_identifier:
+  description: Specifies the identifier for the DB cluster snapshot.
+  returned: always
+  type: str
+  sample: ansible-test-16638696-test-snapshot
+db_cluster_identifier:
+  description: Specifies the DB cluster identifier of the DB cluster that this DB cluster snapshot was created from.
+  returned: always
+  type: str
+  sample: ansible-test-16638696
+snapshot_create_time:
+  description: Provides the time when the snapshot was taken, in Universal Coordinated Time (UTC).
+  returned: always
+  type: str
+  sample: '2019-06-15T10:46:23.776000+00:00'
+engine:
+  description: Specifies the name of the database engine for this DB cluster snapshot.
+  returned: always
+  type: str
+  sample: "aurora"
+engine_mode:
+  description: Provides the engine mode of the database engine for this DB cluster snapshot.
+  returned: always
+  type: str
+  sample: "5.6.mysql_aurora.1.22.5"
+allocated_storage:
+  description: Specifies the allocated storage size in gibibytes (GiB).
+  returned: always
+  type: int
+  sample: 20
+status:
+  description: Specifies the status of this DB cluster snapshot.
+  returned: always
+  type: str
+  sample: available
+port:
+  description: Port on which the database is listening.
+  returned: always
+  type: int
+  sample: 3306
+vpc_id:
+  description: ID of the VPC in which the DB lives.
+  returned: always
+  type: str
+  sample: vpc-09ff232e222710ae0
+cluster_create_time:
+  description: Specifies the time when the DB cluster was created, in Universal Coordinated Time (UTC).
+  returned: always
+  type: str
+  sample: '2019-06-15T10:15:56.221000+00:00'
+master_username:
+  description: Provides the master username for this DB cluster snapshot.
+  returned: always
+  type: str
+  sample: test
+engine_version:
+  description: Version of the cluster from which the snapshot was created.
+  returned: always
+  type: str
+  sample: "5.6.mysql_aurora.1.22.5"
+license_model:
+  description: Provides the license model information for this DB cluster snapshot.
+  returned: always
+  type: str
+  sample: general-public-license
+snapshot_type:
+  description: How the snapshot was created (always manual for this module!).
+  returned: always
+  type: str
+  sample: manual
+percent_progress:
+  description: Specifies the percentage of the estimated data that has been transferred.
+  returned: always
+  type: int
+  sample: 100
+storage_encrypted:
+  description: Specifies whether the DB cluster snapshot is encrypted.
+  returned: always
+  type: bool
+  sample: false
+kms_key_id:
+  description: The Amazon Web Services KMS key identifier is the key ARN, key ID, alias ARN, or alias name for the KMS key.
+  returned: always
+  type: str
+db_cluster_snapshot_arn:
+  description: Amazon Resource Name for the snapshot.
+  returned: always
+  type: str
+  sample: arn:aws:rds:us-west-2:123456789012:snapshot:ansible-test-16638696-test-snapshot
+source_db_cluster_snapshot_arn:
+  description: If the DB cluster snapshot was copied from a source DB cluster snapshot, the ARN for the source DB cluster snapshot, otherwise, null.
+  returned: always
+  type: str
+  sample: null
+iam_database_authentication_enabled:
+  description: Whether IAM database authentication is enabled.
+  returned: always
+  type: bool
+  sample: false
+tag_list:
+  description: A list of tags.
+  returned: always
+  type: list
+  sample: []
+tags:
+  description: Tags applied to the snapshot.
+  returned: always
+  type: complex
+  contains: {}
+'''
+
+try:
+    import botocore
+except ImportError:
+    pass  # caught by AnsibleAWSModule
+
+from ansible_collections.amazon.aws.plugins.module_utils.core import AnsibleAWSModule
+from ansible_collections.amazon.aws.plugins.module_utils.core import is_boto3_error_code
+from ansible_collections.amazon.aws.plugins.module_utils.ec2 import AWSRetry
+from ansible_collections.amazon.aws.plugins.module_utils.ec2 import camel_dict_to_snake_dict
+from ansible_collections.amazon.aws.plugins.module_utils.ec2 import ansible_dict_to_boto3_tag_list
+from ansible_collections.amazon.aws.plugins.module_utils.rds import get_tags
+from ansible_collections.amazon.aws.plugins.module_utils.rds import ensure_tags
+from ansible_collections.amazon.aws.plugins.module_utils.rds import call_method
+from ansible_collections.amazon.aws.plugins.module_utils.core import get_boto3_client_method_parameters
+from ansible_collections.amazon.aws.plugins.module_utils.rds import get_rds_method_attribute
+from ansible_collections.amazon.aws.plugins.module_utils.rds import arg_spec_to_rds_params
+
+
+def get_snapshot(snapshot_id):
+    try:
+        snapshot = client.describe_db_cluster_snapshots(DBClusterSnapshotIdentifier=snapshot_id, aws_retry=True)["DBClusterSnapshots"][0]
+        snapshot["Tags"] = get_tags(client, module, snapshot["DBClusterSnapshotArn"])
+    except is_boto3_error_code("DBClusterSnapshotNotFound"):
+        return {}
+    except is_boto3_error_code("DBClusterSnapshotNotFoundFault"):  # pylint: disable=duplicate-except
+        return {}
+    except (botocore.exceptions.BotoCoreError, botocore.exceptions.ClientError) as e:  # pylint: disable=duplicate-except
+        module.fail_json_aws(e, msg="Couldn't get snapshot {0}".format(snapshot_id))
+    return snapshot
+
+
+def get_parameters(parameters, method_name):
+    if method_name == 'copy_db_cluster_snapshot':
+        parameters['TargetDBClusterSnapshotIdentifier'] = module.params['db_cluster_snapshot_identifier']
+
+    required_options = get_boto3_client_method_parameters(client, method_name, required=True)
+    if any(parameters.get(k) is None for k in required_options):
+        module.fail_json(msg='To {0} requires the parameters: {1}'.format(
+            get_rds_method_attribute(method_name, module).operation_description, required_options))
+    options = get_boto3_client_method_parameters(client, method_name)
+    parameters = dict((k, v) for k, v in parameters.items() if k in options and v is not None)
+
+    return parameters
+
+
+def ensure_snapshot_absent():
+    snapshot_name = module.params.get("db_cluster_snapshot_identifier")
+    params = {"DBClusterSnapshotIdentifier": snapshot_name}
+    changed = False
+
+    snapshot = get_snapshot(snapshot_name)
+    if not snapshot:
+        module.exit_json(changed=changed)
+    elif snapshot and snapshot["Status"] != "deleting":
+        snapshot, changed = call_method(client, module, "delete_db_cluster_snapshot", params)
+
+    module.exit_json(changed=changed)
+
+
+def copy_snapshot(params):
+    changed = False
+    snapshot_id = module.params.get('db_cluster_snapshot_identifier')
+    snapshot = get_snapshot(snapshot_id)
+
+    if not snapshot:
+        method_params = get_parameters(params, 'copy_db_cluster_snapshot')
+        if method_params.get('Tags'):
+            method_params['Tags'] = ansible_dict_to_boto3_tag_list(method_params['Tags'])
+        result, changed = call_method(client, module, 'copy_db_cluster_snapshot', method_params)
+
+    return changed
+
+
+def ensure_snapshot_present(params):
+    source_id = module.params.get('source_db_cluster_snapshot_identifier')
+    snapshot_name = module.params.get("db_cluster_snapshot_identifier")
+    changed = False
+
+    snapshot = get_snapshot(snapshot_name)
+
+    # Copy snapshot
+    if source_id:
+        changed |= copy_snapshot(params)
+
+    # Create snapshot
+    elif not snapshot:
+        changed |= create_snapshot(params)
+
+    # Snapshot exists and we're not creating a copy - modify exising snapshot
+    else:
+        changed |= modify_snapshot()
+
+    snapshot = get_snapshot(snapshot_name)
+    module.exit_json(changed=changed, **camel_dict_to_snake_dict(snapshot, ignore_list=['Tags']))
+
+
+def create_snapshot(params):
+    method_params = get_parameters(params, 'create_db_cluster_snapshot')
+    if method_params.get('Tags'):
+        method_params['Tags'] = ansible_dict_to_boto3_tag_list(method_params['Tags'])
+    snapshot, changed = call_method(client, module, 'create_db_cluster_snapshot', method_params)
+
+    return changed
+
+
+def modify_snapshot():
+    # TODO - add other modifications aside from purely tags
+    changed = False
+    snapshot_id = module.params.get('db_cluster_snapshot_identifier')
+    snapshot = get_snapshot(snapshot_id)
+
+    if module.params.get('tags'):
+        changed |= ensure_tags(client, module, snapshot['DBClusterSnapshotArn'], snapshot['Tags'], module.params['tags'], module.params['purge_tags'])
+
+    return changed
+
+
+def main():
+    global client
+    global module
+
+    argument_spec = dict(
+        state=dict(type='str', choices=['present', 'absent'], default='present'),
+        db_cluster_snapshot_identifier=dict(type='str', aliases=['id', 'snapshot_id', 'snapshot_name'], required=True),
+        db_cluster_identifier=dict(type='str', aliases=['cluster_id', 'cluster_name']),
+        source_db_cluster_snapshot_identifier=dict(type='str', aliases=['source_id', 'source_snapshot_id']),
+        wait=dict(type='bool', default=False),
+        wait_timeout=dict(type='int', default=300),
+        tags=dict(type='dict', aliases=['resource_tags']),
+        purge_tags=dict(type='bool', default=True),
+        copy_tags=dict(type='bool', default=False),
+        source_region=dict(type='str'),
+    )
+
+    module = AnsibleAWSModule(
+        argument_spec=argument_spec,
+        supports_check_mode=True,
+    )
+
+    retry_decorator = AWSRetry.jittered_backoff(retries=10)
+    try:
+        client = module.client('rds', retry_decorator=retry_decorator)
+    except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e:
+        module.fail_json_aws(e, msg="Failed to connect to AWS.")
+
+    state = module.params.get("state")
+
+    if state == "absent":
+        ensure_snapshot_absent()
+    elif state == "present":
+        params = arg_spec_to_rds_params(dict((k, module.params[k]) for k in module.params if k in argument_spec))
+        ensure_snapshot_present(params)
+
+
+if __name__ == '__main__':
+    main()
diff --git a/tests/integration/targets/rds_cluster_snapshot/aliases b/tests/integration/targets/rds_cluster_snapshot/aliases
new file mode 100644
index 00000000000..e9ea741b75a
--- /dev/null
+++ b/tests/integration/targets/rds_cluster_snapshot/aliases
@@ -0,0 +1,3 @@
+cloud/aws
+
+rds_snapshot_info
diff --git a/tests/integration/targets/rds_cluster_snapshot/defaults/main.yml b/tests/integration/targets/rds_cluster_snapshot/defaults/main.yml
new file mode 100644
index 00000000000..268ab154f20
--- /dev/null
+++ b/tests/integration/targets/rds_cluster_snapshot/defaults/main.yml
@@ -0,0 +1,13 @@
+---
+# defaults file for rds_cluster_snapshot
+_resource_prefix: 'ansible-test-{{ tiny_prefix }}'
+
+# Create RDS cluster
+cluster_id: '{{ _resource_prefix }}-rds-cluster'
+username: 'testrdsusername'
+password: "{{ lookup('password', 'dev/null length=12 chars=ascii_letters,digits') }}"
+engine: 'aurora'
+port: 3306
+
+# Create snapshot
+snapshot_id: '{{ _resource_prefix }}-rds-cluster-snapshot'
diff --git a/tests/integration/targets/rds_cluster_snapshot/tasks/main.yml b/tests/integration/targets/rds_cluster_snapshot/tasks/main.yml
new file mode 100644
index 00000000000..f3bd583ed31
--- /dev/null
+++ b/tests/integration/targets/rds_cluster_snapshot/tasks/main.yml
@@ -0,0 +1,481 @@
+---
+- module_defaults:
+    group/aws:
+      region: "{{ aws_region }}"
+      aws_access_key: "{{ aws_access_key }}"
+      aws_secret_key: "{{ aws_secret_key }}"
+      security_token: "{{ security_token | default(omit) }}"
+  collections:
+    - community.aws
+    - amazon.aws
+
+  block:
+  - name: Create a source DB cluster
+    rds_cluster:
+      id: "{{ cluster_id }}"
+      state: present
+      engine: "{{ engine}}"
+      backup_retention_period: 1
+      username: "{{ username }}"
+      password: "{{ password }}"
+      preferred_backup_window: "01:15-01:45"
+    register: _result_create_source_db_cluster
+
+  - assert:
+      that:
+      - _result_create_source_db_cluster.changed
+      - "'allocated_storage' in _result_create_source_db_cluster"
+      - _result_create_source_db_cluster.allocated_storage == 1
+      - "'cluster_create_time' in _result_create_source_db_cluster"
+      - _result_create_source_db_cluster.copy_tags_to_snapshot == false
+      - "'db_cluster_arn' in _result_create_source_db_cluster"
+      - _result_create_source_db_cluster.db_cluster_identifier == "{{ cluster_id }}"
+      - "'db_cluster_parameter_group' in _result_create_source_db_cluster"
+      - "'db_cluster_resource_id' in _result_create_source_db_cluster"
+      - "'endpoint' in _result_create_source_db_cluster"
+      - "'engine' in _result_create_source_db_cluster"
+      - _result_create_source_db_cluster.engine == "{{ engine }}"
+      - "'engine_mode' in _result_create_source_db_cluster"
+      - _result_create_source_db_cluster.engine_mode == "provisioned"
+      - "'engine_version' in _result_create_source_db_cluster"
+      - "'master_username' in _result_create_source_db_cluster"
+      - _result_create_source_db_cluster.master_username == "{{ username }}"
+      - "'port' in _result_create_source_db_cluster"
+      - "_result_create_source_db_cluster.port == {{ port }}"
+      - "'status' in _result_create_source_db_cluster"
+      - _result_create_source_db_cluster.status == "available"
+      - "'tags' in _result_create_source_db_cluster"
+      - "'vpc_security_groups' in _result_create_source_db_cluster"
+
+  - name: Get all RDS snapshots for the existing DB cluster
+    rds_snapshot_info:
+      db_cluster_identifier: "{{ cluster_id }}"
+    register: _result_cluster_snapshot_info
+
+  - assert:
+      that:
+      - _result_cluster_snapshot_info is successful
+      - _result_cluster_snapshot_info.cluster_snapshots | length == 0
+
+  - name: Take a snapshot of the existing DB cluster (CHECK_MODE)
+    rds_cluster_snapshot:
+      state: present
+      db_cluster_identifier: "{{ cluster_id }}"
+      db_cluster_snapshot_identifier: "{{ snapshot_id }}"
+    check_mode: true
+    register: _result_cluster_snapshot
+
+  - assert:
+      that:
+      - _result_cluster_snapshot.changed
+
+  - name: Take a snapshot of the existing DB cluster
+    rds_cluster_snapshot:
+      state: present
+      db_cluster_identifier: "{{ cluster_id }}"
+      db_cluster_snapshot_identifier: "{{ snapshot_id }}"
+      wait: true
+    register: _result_cluster_snapshot
+
+  - assert:
+      that:
+      - _result_cluster_snapshot.changed
+      - "'allocated_storage' in _result_cluster_snapshot"
+      - "'cluster_create_time' in _result_cluster_snapshot"
+      - "'db_cluster_identifier' in _result_cluster_snapshot"
+      - _result_cluster_snapshot.db_cluster_identifier == "{{ cluster_id }}"
+      - "'db_cluster_snapshot_identifier' in _result_cluster_snapshot"
+      - _result_cluster_snapshot.db_cluster_snapshot_identifier == "{{ snapshot_id }}"
+      - "'db_cluster_snapshot_arn' in _result_cluster_snapshot"
+      - "'engine' in _result_cluster_snapshot"
+      - _result_cluster_snapshot.engine == "{{ engine }}"
+      # - "'engine_mode' in _result_cluster_snapshot"
+      # - _result_cluster_snapshot.engine_mode == "provisioned"
+      - "'engine_version' in _result_cluster_snapshot"
+      - "'iam_database_authentication_enabled' in _result_cluster_snapshot"
+      - "'license_model' in _result_cluster_snapshot"
+      - "'master_username' in _result_cluster_snapshot"
+      - _result_cluster_snapshot.master_username == "{{ username }}"
+      - "'snapshot_create_time' in _result_cluster_snapshot"
+      - "'snapshot_type' in _result_cluster_snapshot"
+      - "'status' in _result_cluster_snapshot"
+      - _result_create_source_db_cluster.status == "available"
+      - "'storage_encrypted' in _result_cluster_snapshot"
+      - "'tags' in _result_cluster_snapshot"
+      - "'vpc_id' in _result_cluster_snapshot"
+
+  - name: Get information about the existing DB snapshot
+    rds_snapshot_info:
+      db_cluster_snapshot_identifier: "{{ snapshot_id }}"
+    register: _result_cluster_snapshot_info
+
+  - assert:
+      that:
+      - _result_cluster_snapshot_info is successful
+      - _result_cluster_snapshot_info.cluster_snapshots[0].db_cluster_identifier == "{{ cluster_id }}"
+      - _result_cluster_snapshot_info.cluster_snapshots[0].db_cluster_snapshot_identifier == "{{ snapshot_id }}"
+
+  - name: Get info of the existing DB cluster
+    rds_cluster_info:
+      cluster_id: "{{ cluster_id }}"
+    register: result_cluster_info
+
+  - assert:
+      that:
+      - result_cluster_info is successful
+
+  - name: Create another source DB cluster
+    rds_cluster:
+      id: "{{ cluster_id }}-b"
+      state: present
+      engine: "{{ engine}}"
+      backup_retention_period: 1
+      username: "{{ username }}"
+      password: "{{ password }}"
+      preferred_backup_window: "01:15-01:45"
+    register: _result_create_source_db_cluster
+
+  - assert:
+      that:
+      - _result_create_source_db_cluster.changed
+      - "'allocated_storage' in _result_create_source_db_cluster"
+      - _result_create_source_db_cluster.allocated_storage == 1
+      - "'cluster_create_time' in _result_create_source_db_cluster"
+      - _result_create_source_db_cluster.copy_tags_to_snapshot == false
+      - "'db_cluster_arn' in _result_create_source_db_cluster"
+      - _result_create_source_db_cluster.db_cluster_identifier == "{{ cluster_id }}-b"
+      - "'db_cluster_parameter_group' in _result_create_source_db_cluster"
+      - "'db_cluster_resource_id' in _result_create_source_db_cluster"
+      - "'endpoint' in _result_create_source_db_cluster"
+      - "'engine' in _result_create_source_db_cluster"
+      - _result_create_source_db_cluster.engine == "{{ engine }}"
+      - "'engine_mode' in _result_create_source_db_cluster"
+      - _result_create_source_db_cluster.engine_mode == "provisioned"
+      - "'engine_version' in _result_create_source_db_cluster"
+      - "'master_username' in _result_create_source_db_cluster"
+      - _result_create_source_db_cluster.master_username == "{{ username }}"
+      - "'port' in _result_create_source_db_cluster"
+      - "_result_create_source_db_cluster.port == {{ port }}"
+      - "'status' in _result_create_source_db_cluster"
+      - _result_create_source_db_cluster.status == "available"
+      - "'tags' in _result_create_source_db_cluster"
+      - "'vpc_security_groups' in _result_create_source_db_cluster"
+
+  - name: Take another snapshot of the existing DB cluster
+    rds_cluster_snapshot:
+      state: present
+      db_cluster_identifier: "{{ cluster_id }}-b"
+      db_cluster_snapshot_identifier: "{{ snapshot_id }}-b"
+      wait: true
+    register: _result_cluster_snapshot
+
+  - assert:
+      that:
+      - _result_cluster_snapshot.changed
+      - "'allocated_storage' in _result_cluster_snapshot"
+      - "'cluster_create_time' in _result_cluster_snapshot"
+      - "'db_cluster_identifier' in _result_cluster_snapshot"
+      - _result_cluster_snapshot.db_cluster_identifier == "{{ cluster_id }}-b"
+      - "'db_cluster_snapshot_identifier' in _result_cluster_snapshot"
+      - _result_cluster_snapshot.db_cluster_snapshot_identifier == "{{ snapshot_id }}-b"
+      - "'db_cluster_snapshot_arn' in _result_cluster_snapshot"
+      - "'engine' in _result_cluster_snapshot"
+      - _result_cluster_snapshot.engine == "{{ engine }}"
+      # - "'engine_mode' in _result_cluster_snapshot"
+      # - _result_cluster_snapshot.engine_mode == "provisioned"
+      - "'engine_version' in _result_cluster_snapshot"
+      - "'iam_database_authentication_enabled' in _result_cluster_snapshot"
+      - "'license_model' in _result_cluster_snapshot"
+      - "'master_username' in _result_cluster_snapshot"
+      - _result_cluster_snapshot.master_username == "{{ username }}"
+      - "'snapshot_create_time' in _result_cluster_snapshot"
+      - "'snapshot_type' in _result_cluster_snapshot"
+      - "'status' in _result_cluster_snapshot"
+      - _result_create_source_db_cluster.status == "available"
+      - "'storage_encrypted' in _result_cluster_snapshot"
+      - "'tags' in _result_cluster_snapshot"
+      - "'vpc_id' in _result_cluster_snapshot"
+
+  - name: Get all RDS snapshots for the existing DB cluster
+    rds_snapshot_info:
+      db_cluster_identifier: "{{ cluster_id }}-b"
+    register: _result_cluster_snapshot_info
+
+  - assert:
+      that:
+      - _result_cluster_snapshot_info is successful
+      - _result_cluster_snapshot_info.cluster_snapshots | length == 1
+
+  - name: Delete existing DB cluster snapshot (CHECK_MODE)
+    rds_cluster_snapshot:
+      state: absent
+      db_cluster_snapshot_identifier: "{{ snapshot_id }}-b"
+    register: _result_delete_snapshot
+    check_mode: true
+
+  - assert:
+      that:
+      - _result_delete_snapshot.changed
+
+  - name: Delete the existing DB cluster snapshot
+    rds_cluster_snapshot:
+      state: absent
+      db_cluster_snapshot_identifier: "{{ snapshot_id }}-b"
+    register: _result_delete_snapshot
+
+  - assert:
+      that:
+      - _result_delete_snapshot.changed
+
+  - name: Get info of the existing DB cluster
+    rds_cluster_info:
+      cluster_id: "{{ cluster_id }}"
+    register: _result_cluster_info
+
+  - assert:
+      that:
+      - result_cluster_info is successful
+
+  - name: Take another snapshot of the existing DB cluster and assign tags
+    rds_cluster_snapshot:
+      state: present
+      db_cluster_identifier: "{{ cluster_id }}"
+      db_cluster_snapshot_identifier: "{{ snapshot_id }}-b"
+      wait: true
+      tags:
+        tag_one: '{{ snapshot_id }}-b One'
+        "Tag Two": 'two {{ snapshot_id }}-b'
+    register: _result_cluster_snapshot
+
+  - assert:
+      that:
+      - _result_cluster_snapshot.changed
+      - "'allocated_storage' in _result_cluster_snapshot"
+      - "'cluster_create_time' in _result_cluster_snapshot"
+      - "'db_cluster_identifier' in _result_cluster_snapshot"
+      - _result_cluster_snapshot.db_cluster_identifier == "{{ cluster_id }}"
+      - "'db_cluster_snapshot_identifier' in _result_cluster_snapshot"
+      - _result_cluster_snapshot.db_cluster_snapshot_identifier == "{{ snapshot_id }}-b"
+      - "'db_cluster_snapshot_arn' in _result_cluster_snapshot"
+      - "'engine' in _result_cluster_snapshot"
+      - _result_cluster_snapshot.engine == "{{ engine }}"
+      # - "'engine_mode' in _result_cluster_snapshot"
+      # - _result_cluster_snapshot.engine_mode == "provisioned"
+      - "'engine_version' in _result_cluster_snapshot"
+      - "'iam_database_authentication_enabled' in _result_cluster_snapshot"
+      - "'license_model' in _result_cluster_snapshot"
+      - "'master_username' in _result_cluster_snapshot"
+      - _result_cluster_snapshot.master_username == "{{ username }}"
+      - "'snapshot_create_time' in _result_cluster_snapshot"
+      - "'snapshot_type' in _result_cluster_snapshot"
+      - "'status' in _result_cluster_snapshot"
+      - _result_create_source_db_cluster.status == "available"
+      - "'storage_encrypted' in _result_cluster_snapshot"
+      - "'tags' in _result_cluster_snapshot"
+      - _result_cluster_snapshot.tags | length == 2
+      - _result_cluster_snapshot.tags["tag_one"] == "{{ snapshot_id }}-b One"
+      - _result_cluster_snapshot.tags["Tag Two"] == "two {{ snapshot_id }}-b"
+      - "'vpc_id' in _result_cluster_snapshot"
+
+  - name: Attempt to take another snapshot of the existing DB cluster and assign tags (idempotence)
+    rds_cluster_snapshot:
+      state: present
+      db_cluster_identifier: "{{ cluster_id }}"
+      db_cluster_snapshot_identifier: "{{ snapshot_id }}-b"
+      wait: true
+      tags:
+        tag_one: '{{ snapshot_id }}-b One'
+        "Tag Two": 'two {{ snapshot_id }}-b'
+    register: _result_cluster_snapshot
+
+  - assert:
+      that:
+      - not _result_cluster_snapshot.changed
+
+  - name: Take another snapshot of the existing DB cluster and update tags
+    rds_cluster_snapshot:
+      state: present
+      db_cluster_identifier: "{{ cluster_id }}"
+      db_cluster_snapshot_identifier: "{{ snapshot_id }}-b"
+      tags:
+        tag_three: '{{ snapshot_id }}-b Three'
+        "Tag Two": 'two {{ snapshot_id }}-b'
+    register: _result_cluster_snapshot
+
+  - assert:
+      that:
+      - _result_cluster_snapshot.changed
+      - "'allocated_storage' in _result_cluster_snapshot"
+      - "'cluster_create_time' in _result_cluster_snapshot"
+      - "'db_cluster_identifier' in _result_cluster_snapshot"
+      - _result_cluster_snapshot.db_cluster_identifier == "{{ cluster_id }}"
+      - "'db_cluster_snapshot_identifier' in _result_cluster_snapshot"
+      - _result_cluster_snapshot.db_cluster_snapshot_identifier == "{{ snapshot_id }}-b"
+      - "'db_cluster_snapshot_arn' in _result_cluster_snapshot"
+      - "'engine' in _result_cluster_snapshot"
+      - _result_cluster_snapshot.engine == "{{ engine }}"
+      # - "'engine_mode' in _result_cluster_snapshot"
+      # - _result_cluster_snapshot.engine_mode == "provisioned"
+      - "'engine_version' in _result_cluster_snapshot"
+      - "'iam_database_authentication_enabled' in _result_cluster_snapshot"
+      - "'license_model' in _result_cluster_snapshot"
+      - "'master_username' in _result_cluster_snapshot"
+      - _result_cluster_snapshot.master_username == "{{ username }}"
+      - "'snapshot_create_time' in _result_cluster_snapshot"
+      - "'snapshot_type' in _result_cluster_snapshot"
+      - "'status' in _result_cluster_snapshot"
+      - _result_create_source_db_cluster.status == "available"
+      - "'storage_encrypted' in _result_cluster_snapshot"
+      - "'tags' in _result_cluster_snapshot"
+      - _result_cluster_snapshot.tags | length == 2
+      - _result_cluster_snapshot.tags["tag_three"] == "{{ snapshot_id }}-b Three"
+      - _result_cluster_snapshot.tags["Tag Two"] == "two {{ snapshot_id }}-b"
+      - "'vpc_id' in _result_cluster_snapshot"
+
+  - name: Take another snapshot of the existing DB cluster and update tags without purge
+    rds_cluster_snapshot:
+      state: present
+      db_cluster_identifier: "{{ cluster_id }}"
+      db_cluster_snapshot_identifier: "{{ snapshot_id }}-b"
+      purge_tags: false
+      tags:
+        tag_one: '{{ snapshot_id }}-b One'
+    register: _result_cluster_snapshot
+
+  - assert:
+      that:
+      - _result_cluster_snapshot.changed
+      - "'allocated_storage' in _result_cluster_snapshot"
+      - "'cluster_create_time' in _result_cluster_snapshot"
+      - "'db_cluster_identifier' in _result_cluster_snapshot"
+      - _result_cluster_snapshot.db_cluster_identifier == "{{ cluster_id }}"
+      - "'db_cluster_snapshot_identifier' in _result_cluster_snapshot"
+      - _result_cluster_snapshot.db_cluster_snapshot_identifier == "{{ snapshot_id }}-b"
+      - "'db_cluster_snapshot_arn' in _result_cluster_snapshot"
+      - "'engine' in _result_cluster_snapshot"
+      - _result_cluster_snapshot.engine == "{{ engine }}"
+      # - "'engine_mode' in _result_cluster_snapshot"
+      # - _result_cluster_snapshot.engine_mode == "provisioned"
+      - "'engine_version' in _result_cluster_snapshot"
+      - "'iam_database_authentication_enabled' in _result_cluster_snapshot"
+      - "'license_model' in _result_cluster_snapshot"
+      - "'master_username' in _result_cluster_snapshot"
+      - _result_cluster_snapshot.master_username == "{{ username }}"
+      - "'snapshot_create_time' in _result_cluster_snapshot"
+      - "'snapshot_type' in _result_cluster_snapshot"
+      - "'status' in _result_cluster_snapshot"
+      - _result_create_source_db_cluster.status == "available"
+      - "'storage_encrypted' in _result_cluster_snapshot"
+      - "'tags' in _result_cluster_snapshot"
+      - _result_cluster_snapshot.tags | length == 3
+      - _result_cluster_snapshot.tags["tag_one"] == "{{ snapshot_id }}-b One"
+      - _result_cluster_snapshot.tags["Tag Two"] == "two {{ snapshot_id }}-b"
+      - _result_cluster_snapshot.tags["tag_three"] == "{{ snapshot_id }}-b Three"
+      - "'vpc_id' in _result_cluster_snapshot"
+
+  - name: Take another snapshot of the existing DB cluster and do not specify any tag to ensure previous tags are not removed
+    rds_cluster_snapshot:
+      state: present
+      db_cluster_identifier: "{{ cluster_id }}"
+      db_cluster_snapshot_identifier: "{{ snapshot_id }}-b"
+    register: _result_cluster_snapshot
+
+  - assert:
+      that:
+      - not _result_cluster_snapshot.changed
+  
+  # ------------------------------------------------------------------------------------------
+   # Test copying a snapshot
+   ### Copying a DB cluster snapshot from a different region is supported, but not in CI,
+   ### because the aws-terminator only terminates resources in one region.
+  - set_fact:
+      _snapshot_arn: "{{ _result_cluster_snapshot.db_cluster_snapshot_arn }}"
+
+  - name: Copy a DB cluster snapshot (check mode)
+    rds_cluster_snapshot:
+      id: "{{ snapshot_id }}-copy"
+      source_id: "{{ snapshot_id }}-b"
+      copy_tags: yes
+      wait: true
+    register: _result_cluster_copy_snapshot
+    check_mode: yes
+
+  - assert:
+      that:
+        - _result_cluster_copy_snapshot.changed
+
+  - name: Copy a DB cluster snapshot
+    rds_cluster_snapshot:
+      id: "{{ snapshot_id }}-copy"
+      source_id: "{{ snapshot_id }}-b"
+      copy_tags: yes
+      wait: true
+    register: _result_cluster_copy_snapshot
+
+  - assert:
+      that:
+        - _result_cluster_copy_snapshot.changed
+        - _result_cluster_copy_snapshot.db_cluster_identifier == "{{ cluster_id }}"
+        - _result_cluster_copy_snapshot.source_db_cluster_snapshot_arn == "{{ _snapshot_arn }}"
+        - _result_cluster_copy_snapshot.db_cluster_snapshot_identifier == "{{ snapshot_id }}-copy"
+        - "'tags' in _result_cluster_copy_snapshot"
+        - _result_cluster_copy_snapshot.tags | length == 3
+        - _result_cluster_copy_snapshot.tags["tag_one"] == "{{ snapshot_id }}-b One"
+        - _result_cluster_copy_snapshot.tags["Tag Two"] == "two {{ snapshot_id }}-b"
+        - _result_cluster_copy_snapshot.tags["tag_three"] == "{{ snapshot_id }}-b Three"
+
+  - name: Copy a DB cluster snapshot (idempotence - check mode)
+    rds_cluster_snapshot:
+      id: "{{ snapshot_id }}-copy"
+      source_id: "{{ snapshot_id }}-b"
+      copy_tags: yes
+      wait: true
+    register: _result_cluster_copy_snapshot
+    check_mode: yes
+
+  - assert:
+      that:
+        - not _result_cluster_copy_snapshot.changed
+
+  - name: Copy a DB cluster snapshot (idempotence)
+    rds_cluster_snapshot:
+      id: "{{ snapshot_id }}-copy"
+      source_id: "{{ snapshot_id }}-b"
+      copy_tags: yes
+      wait: true
+    register: _result_cluster_copy_snapshot
+
+  - assert:
+      that:
+        - not _result_cluster_copy_snapshot.changed
+        - _result_cluster_copy_snapshot.db_cluster_identifier == "{{ cluster_id }}"
+        - _result_cluster_copy_snapshot.source_db_cluster_snapshot_arn == "{{ _snapshot_arn }}"
+        - _result_cluster_copy_snapshot.db_cluster_snapshot_identifier == "{{ snapshot_id }}-copy"
+        - "'tags' in _result_cluster_copy_snapshot"
+        - _result_cluster_copy_snapshot.tags | length == 3
+        - _result_cluster_copy_snapshot.tags["tag_one"] == "{{ snapshot_id }}-b One"
+        - _result_cluster_copy_snapshot.tags["Tag Two"] == "two {{ snapshot_id }}-b"
+        - _result_cluster_copy_snapshot.tags["tag_three"] == "{{ snapshot_id }}-b Three"
+
+  always:
+  - name: Delete the existing DB cluster snapshots
+    rds_cluster_snapshot:
+      state: absent
+      db_cluster_snapshot_identifier: "{{ item }}"
+    register: _result_delete_snapshot
+    ignore_errors: true
+    loop:
+    - "{{ snapshot_id }}"
+    - "{{ snapshot_id }}-b"
+    - "{{ snapshot_id }}-copy"
+
+  - name: Delete the existing DB cluster without creating a final snapshot
+    rds_cluster:
+      state: absent
+      cluster_id: "{{ item }}"
+      skip_final_snapshot: true
+    register: _result_delete_cluster
+    ignore_errors: true
+    loop:
+    - "{{ cluster_id }}"
+    - "{{ cluster_id }}-b"
diff --git a/tests/integration/targets/rds_cluster_snapshot/vars/main.yml b/tests/integration/targets/rds_cluster_snapshot/vars/main.yml
new file mode 100644
index 00000000000..ed97d539c09
--- /dev/null
+++ b/tests/integration/targets/rds_cluster_snapshot/vars/main.yml
@@ -0,0 +1 @@
+---