Update FSx Lustre integration test to test manual backups

Author: siddharthsalot

tests/integration-tests/tests/storage/test_fsx_lustre.py

@@ -14,6 +14,7 @@
 
 import boto3
 import pytest
+from botocore.exceptions import ClientError
 from retrying import retry
 
 import utils
@@ -87,11 +88,18 @@ def test_fsx_lustre_backup(
 ):
     """
     Test FSx Lustre backup feature. As part of this test, following steps are performed
-    - Create a cluster with FSx automatic backups feature enabled.
-    - Mount the file system and create a test file in it.
-    - Wait for automatic backup to be created.
-    - Restore a cluster from the automatic backup taken in step 3. Verify whether test file
-      created in step 2 exists in the restored file system.
+    1. Create a cluster with FSx automatic backups feature enabled.
+    2. Mount the file system and create a test file in it.
+    3. Wait for automatic backup to be created.
+    4. Create a manual FSx Lustre backup of the file system.
+    5. Delete the cluster.
+    6. Verify whether automatic backup is deleted. NOTE: FSx team is planning to change this
+       behavior to retain automatic backups upon filesystem deletion. The test case should
+       be update when this change is in place.
+    7. Restore a cluster from the manual backup taken in step 4. Verify whether test file
+       created in step 2 exists in the restored file system.
+    8. Delete manual backup created in step 4.
+
     """
     mount_dir = "/fsx_mount_dir"
     utc_now_plus_15 = datetime.datetime.utcnow() + datetime.timedelta(minutes=15)
@@ -112,12 +120,21 @@ def test_fsx_lustre_backup(
     # Create a text file in the mount directory.
     create_backup_test_file(remote_command_executor, mount_dir)
 
-    # Wait for the creation of automatic backup
-    backup = monitor_fs_backup(remote_command_executor, fsx_fs_id, region)
+    # Wait for the creation of automatic backup and assert if it is in available state.
+    automatic_backup = monitor_automatic_backup_creation(remote_command_executor, fsx_fs_id, region)
+
+    # Create a manual FSx Lustre backup using boto3 client.
+    manual_backup = create_manual_fs_backup(remote_command_executor, fsx_fs_id, region)
+
+    # Delete original cluster.
+    cluster.delete()
+
+    # Verify whether automatic backup is also deleted along with the cluster.
+    _test_automatic_backup_deletion(remote_command_executor, automatic_backup, region)
 
     # Restore backup into a new cluster
     cluster_config_restore = pcluster_config_reader(
-        config_file="pcluster_restore_fsx.config.ini", mount_dir=mount_dir, fsx_backup_id=backup.get("BackupId"),
+        config_file="pcluster_restore_fsx.config.ini", mount_dir=mount_dir, fsx_backup_id=manual_backup.get("BackupId"),
     )
 
     cluster_restore = clusters_factory(cluster_config_restore)
@@ -130,6 +147,9 @@ def test_fsx_lustre_backup(
     # Validate whether text file created in the original file system is present in the restored file system.
     _test_restore_from_backup(remote_command_executor_restore, mount_dir)
 
+    # Test deletion of manual backup
+    _test_delete_manual_backup(remote_command_executor, manual_backup, region)
+
 
 def _test_fsx_lustre_correctly_mounted(remote_command_executor, mount_dir, os, region, fsx_fs_id):
     logging.info("Testing fsx lustre is correctly mounted")
@@ -275,7 +295,7 @@ def create_backup_test_file(remote_command_executor, mount_dir):
     assert_that(result.stdout).is_equal_to("FSx Lustre Backup test file")
 
 
-def monitor_fs_backup(remote_command_executor, fsx_fs_id, region):
+def monitor_automatic_backup_creation(remote_command_executor, fsx_fs_id, region):
     logging.info("Monitoring automatic backup for FSx Lustre file system: {fs_id}".format(fs_id=fsx_fs_id))
     fsx = boto3.client("fsx", region_name=region)
     backup = poll_on_automatic_backup_creation(fsx_fs_id, fsx)
@@ -298,7 +318,45 @@ def poll_on_automatic_backup_creation(fsx_fs_id, fsx):
     return backup
 
 
+def _test_automatic_backup_deletion(remote_command_executor, automatic_backup, region):
+    backup_id = automatic_backup.get("BackupId")
+    logging.info("Verifying whether automatic backup '{0}' was deleted".format(backup_id))
+    error_message = "Backup '{backup_id}' does not exist.".format(backup_id=backup_id)
+    fsx = boto3.client("fsx", region_name=region)
+    with pytest.raises(ClientError, match=error_message):
+        return fsx.describe_backups(BackupIds=[backup_id])
+
+
+def create_manual_fs_backup(remote_command_executor, fsx_fs_id, region):
+    logging.info("Create manual backup for FSx Lustre file system: {fs_id}".format(fs_id=fsx_fs_id))
+    fsx = boto3.client("fsx", region_name=region)
+    backup = fsx.create_backup(FileSystemId=fsx_fs_id).get("Backup")
+    backup = poll_on_manual_backup_creation(backup, fsx)
+    assert_that(backup.get("Lifecycle")).is_equal_to("AVAILABLE")
+    return backup
+
+
+@retry(
+    retry_on_result=lambda result: result.get("Lifecycle") in ["CREATING"],
+    wait_fixed=seconds(5),
+    stop_max_delay=minutes(7),
+)
+def poll_on_manual_backup_creation(backup, fsx):
+    logging.info(
+        "Backup {backup_id}: {status}".format(backup_id=backup.get("BackupId"), status=backup.get("Lifecycle"))
+    )
+    return fsx.describe_backups(BackupIds=[backup.get("BackupId")]).get("Backups")[0]
+
+
 def _test_restore_from_backup(remote_command_executor, mount_dir):
     logging.info("Testing fsx lustre correctly restored from backup")
     result = remote_command_executor.run_remote_command("cat {mount_dir}/file_to_backup".format(mount_dir=mount_dir))
     assert_that(result.stdout).is_equal_to("FSx Lustre Backup test file")
+
+
+def _test_delete_manual_backup(remote_command_executor, backup, region):
+    backup_id = backup.get("BackupId")
+    logging.info("Testing deletion of manual backup {0}".format(backup_id))
+    fsx = boto3.client("fsx", region_name=region)
+    response = fsx.delete_backup(BackupId=backup_id)
+    assert_that(response.get("Lifecycle")).is_equal_to("DELETED")