Add support for deletion of AWS snapshot backups

Adds support for deletion of AWS snapshot backups by implementing the
delete_snapshot_backup function in AwsCloudSnapshotInterface.

Closes BAR-25.

barman/cloud_providers/aws_s3.py

@@ -774,13 +774,40 @@ def take_snapshot_backup(self, backup_info, instance_identifier, volumes):
             account_id=snapshot_resp["OwnerId"],
         )
 
+    def _delete_snapshot(self, snapshot_id):
+        """
+        Delete the specified snapshot.
+
+        :param str snapshot_id: The ID of the snapshot to be deleted.
+        """
+        try:
+            self.ec2_client.delete_snapshot(SnapshotId=snapshot_id)
+        except ClientError as exc:
+            error_code = exc.response["Error"]["Code"]
+            # If the snapshot could not be found then deletion is considered successful
+            # otherwise we raise a CloudProviderError
+            if error_code == "InvalidSnapshot.NotFound":
+                logging.warning("Snapshot {} could not be found".format(snapshot_id))
+            else:
+                raise CloudProviderError(
+                    "Deletion of snapshot %s failed with error code %s: %s"
+                    % (snapshot_id, error_code, exc.response["Error"])
+                )
+        logging.info("Snapshot %s deleted", snapshot_id)
+
     def delete_snapshot_backup(self, backup_info):
         """
         Delete all snapshots for the supplied backup.
 
         :param barman.infofile.LocalBackupInfo backup_info: Backup information.
         """
-        raise NotImplementedError()
+        for snapshot in backup_info.snapshots_info.snapshots:
+            logging.info(
+                "Deleting snapshot '%s' for backup %s",
+                snapshot.identifier,
+                backup_info.backup_id,
+            )
+            self._delete_snapshot(snapshot.identifier)
 
     def get_attached_volumes(
         self, instance_identifier, disks=None, fail_on_missing=True

doc/manual/28-snapshots.en.md

@@ -92,6 +92,7 @@ The following additional prerequisites apply to snapshot backups on AWS:
 The following permissions are required:
 
 - `ec2:CreateSnapshot`
+- `ec2:DeleteSnapshot`
 - `ec2:DescribeSnapshots`
 - `ec2:DescribeInstances`
 - `ec2:DescribeVolumes`
@@ -105,8 +106,7 @@ backup_method = snapshot
 snapshot_provider = gcp
 ```
 
-Currently Google Cloud Platform (`gcp`) and Microsoft Azure (`azure`) are fully supported.
-Snapshot backups are supported using AWS however *support for deletion of AWS snapshot backups is not yet implemented*.
+Currently Google Cloud Platform (`gcp`), Microsoft Azure (`azure`) and AWS (`aws`) are supported.
 
 The following parameters must be set regardless of cloud provider:
 

tests/test_cloud_snapshot_interface.py

@@ -3231,6 +3231,99 @@ def test_instance_exists_not_found(self, mock_ec2_client):
         # THEN it returns False
         assert resp is False
 
+    def test_delete_snapshot(self, mock_ec2_client, caplog):
+        """Verify that a snapshot can be deleted successfully."""
+        # GIVEN a successful response from the delete snapshot request
+        mock_ec2_client.delete_snapshot.return_value = {}
+        # AND a mock snapshots interface
+        snapshot_interface = AwsCloudSnapshotInterface(region=self.aws_region)
+        # AND log level is info
+        caplog.set_level(logging.INFO)
+
+        # WHEN a snapshot is deleted
+        snapshot_id = "snap-0123"
+        snapshot_interface._delete_snapshot(snapshot_id)
+
+        # THEN delete was called on the client with the expected arguments
+        mock_ec2_client.delete_snapshot.assert_called_once_with(SnapshotId=snapshot_id)
+        # AND a success message was logged
+        assert "Snapshot {} deleted".format(snapshot_id) in caplog.text
+
+    def test_delete_snapshot_not_found(self, mock_ec2_client, caplog):
+        """Verify that a snapshot ID which can't be found is success."""
+        # GIVEN a successful response from the delete snapshot request
+        mock_ec2_client.delete_snapshot.side_effect = (
+            ClientError({"Error": {"Code": "InvalidSnapshot.NotFound"}}, "message"),
+        )
+        # AND a mock snapshots interface
+        snapshot_interface = AwsCloudSnapshotInterface(region=self.aws_region)
+        # AND log level is info
+        caplog.set_level(logging.INFO)
+
+        # WHEN a snapshot is deleted
+        # THEN no exceptions are raised
+        snapshot_id = "snap-0123"
+        snapshot_interface._delete_snapshot(snapshot_id)
+
+        # THEN delete was called on the client with the expected arguments
+        mock_ec2_client.delete_snapshot.assert_called_once_with(SnapshotId=snapshot_id)
+        # AND a success message was logged
+        assert "Snapshot {} deleted".format(snapshot_id) in caplog.text
+        # AND a warning message was logged
+        assert "Snapshot {} could not be found".format(snapshot_id) in caplog.text
+
+    def test_delete_snapshot_failed(self, mock_ec2_client, caplog):
+        """Verify that a failed deletion results in a CloudProviderError."""
+        # GIVEN an unexpected error from the delete snapshot request
+        mock_ec2_client.delete_snapshot.side_effect = (
+            ClientError({"Error": {"Code": "Something.Bad"}}, "message"),
+        )
+        # AND a mock snapshots interface
+        snapshot_interface = AwsCloudSnapshotInterface(region=self.aws_region)
+
+        # WHEN a snapshot is deleted
+        # THEN a CloudProviderError is raised
+        snapshot_id = "snap-0123"
+        with pytest.raises(CloudProviderError) as exc:
+            snapshot_interface._delete_snapshot(snapshot_id)
+
+        # AND the exception has the expected message
+        expected_message = "Deletion of snapshot {} failed with error code {}".format(
+            snapshot_id, "Something.Bad"
+        )
+        assert expected_message in str(exc.value)
+
+    @pytest.mark.parametrize(
+        "snapshots_list",
+        (
+            [],
+            [mock.Mock(identifier="snap-0123")],
+            [mock.Mock(identifier="snap-0123"), mock.Mock(identifier="snap0124")],
+        ),
+    )
+    def test_delete_snapshot_backup(self, snapshots_list, mock_ec2_client, caplog):
+        """Verify that all snapshots for a backup are deleted."""
+        # GIVEN a backup_info specifying zero or more snapshots
+        backup_info = mock.Mock(
+            backup_id=self.backup_id,
+            snapshots_info=mock.Mock(snapshots=snapshots_list),
+        )
+        # AND log level is info
+        caplog.set_level(logging.INFO)
+        # AND the snapshot delete requests are successful
+        mock_ec2_client.delete_snapshot.return_value = {}
+        # AND a new AwsCloudSnapshotInterface
+        snapshot_interface = AwsCloudSnapshotInterface(region=self.aws_region)
+
+        # WHEN delete_snapshot_backup is called
+        snapshot_interface.delete_snapshot_backup(backup_info)
+
+        # THEN delete_snapshot was called for each snapshot
+        expected_calls = [
+            mock.call(SnapshotId=snapshot.identifier) for snapshot in snapshots_list
+        ]
+        mock_ec2_client.delete_snapshot.assert_has_calls(expected_calls)
+
 
 class TestAwsVolumeMetadata(object):
     """Verify behaviour of AwsVolumeMetadata."""