add negative test and validation

localstack · bentsku · Jul 14, 2023 · Jul 13, 2023 · Jul 13, 2023 · Jul 14, 2023
commit 49f744d9cabe8a876b864535f767a30474f09ba0
diff --git a/localstack/services/s3/provider.py b/localstack/services/s3/provider.py
@@ -7,6 +7,7 @@
 from zoneinfo import ZoneInfo
 
 import moto.s3.responses as moto_s3_responses
+from botocore.utils import InvalidArnException
 
 from localstack import config
 from localstack.aws.accounts import get_aws_account_id
@@ -164,6 +165,7 @@
     is_key_expired,
     is_valid_canonical_id,
     serialize_expiration_header,
+    validate_dict_fields,
     validate_kms_key_id,
     verify_checksum,
 )
@@ -1920,7 +1922,84 @@ def validate_website_configuration(website_config: WebsiteConfiguration) -> None
 def validate_inventory_configuration(
     config_id: InventoryId, inventory_configuration: InventoryConfiguration
 ):
-    pass
+    """
+    Validate the Inventory Configuration following AWS docs
+    Validation order is XML then `Id` then S3DestinationBucket
+    https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutBucketInventoryConfiguration.html
+    https://docs.aws.amazon.com/AmazonS3/latest/userguide/storage-inventory.html
+    :param config_id: the passed Id parameter passed to the provider method
+    :param inventory_configuration: InventoryConfiguration
+    :raises MalformedXML: when the file doesn't follow the basic structure/required fields
+    :raises IdMismatch: if the `Id` parameter is different from the `Id` field from the configuration
+    :raises InvalidS3DestinationBucket: if S3 bucket is not provided as an ARN
+    :return: None
+    """
+    required_root_fields = {"Destination", "Id", "IncludedObjectVersions", "IsEnabled", "Schedule"}
+    optional_root_fields = {"Filter", "OptionalFields"}
+
+    if not validate_dict_fields(
+        inventory_configuration, required_root_fields, optional_root_fields
+    ):
+        raise MalformedXML()
+
+    required_s3_bucket_dest_fields = {"Bucket", "Format"}
+    optional_s3_bucket_dest_fields = {"AccountId", "Encryption", "Prefix"}
+
+    if not (
+        s3_bucket_destination := inventory_configuration["Destination"].get("S3BucketDestination")
+        ) or not validate_dict_fields(
+        s3_bucket_destination, required_s3_bucket_dest_fields, optional_s3_bucket_dest_fields
+    ):
+        raise MalformedXML()
+
+    if inventory_configuration["Destination"]["S3BucketDestination"]["Format"] not in (
+        "CSV",
+        "ORC",
+        "Parquet",
+    ):
+        raise MalformedXML()
+
+    if not (frequency := inventory_configuration["Schedule"].get("Frequency")) or frequency not in (
+        "Daily",
+        "Weekly",
+    ):
+        raise MalformedXML()
+
+    if inventory_configuration["IncludedObjectVersions"] not in ("All", "Current"):
+        raise MalformedXML()
+
+    possible_optional_fields = {
+        "Size",
+        "LastModifiedDate",
+        "StorageClass",
+        "ETag",
+        "IsMultipartUploaded",
+        "ReplicationStatus",
+        "EncryptionStatus",
+        "ObjectLockRetainUntilDate",
+        "ObjectLockMode",
+        "ObjectLockLegalHoldStatus",
+        "IntelligentTieringAccessTier",
+        "BucketKeyStatus",
+        "ChecksumAlgorithm",
+    }
+    if (opt_fields := inventory_configuration.get("OptionalFields")) and set(
+        opt_fields
+    ) - possible_optional_fields:
+        raise MalformedXML()
+
+    if inventory_configuration.get("Id") != config_id:
+        raise CommonServiceException(
+            code="IdMismatch", message="Document ID does not match the specified configuration ID."
+        )
+
+    bucket_arn = inventory_configuration["Destination"]["S3BucketDestination"]["Bucket"]
+    try:
+        arns.parse_arn(bucket_arn)
+    except InvalidArnException:
+        raise CommonServiceException(
+            code="InvalidS3DestinationBucket", message="Invalid bucket ARN."
+        )
 
 
 def is_object_expired(

diff --git a/localstack/services/s3/utils.py b/localstack/services/s3/utils.py
@@ -411,3 +411,19 @@ def parse_expiration_header(
 
     except (IndexError, ValueError, KeyError):
         return None, None
+
+
+def validate_dict_fields(data: dict, required_fields: set, optional_fields: set):
+    """
+    Validate whether the `data` dict contains at least the required fields and not more than the union of the required
+    and optional fields
+    TODO: we could pass the TypedDict to also use its required/optional properties, but it could be sensitive to
+     mistake/changes in the specs and not always right
+    :param data: the dict we want to validate
+    :param required_fields: a set containing the required fields
+    :param optional_fields: a set containing the optional fields
+    :return: bool, whether the dict is valid or not
+    """
+    return (set_fields := set(data)) >= required_fields and set_fields <= (
+        required_fields | optional_fields
+    )
diff --git a/tests/integration/s3/test_s3.py b/tests/integration/s3/test_s3.py
@@ -5001,13 +5001,13 @@ def test_s3_inventory_report_crud(self, aws_client, s3_create_bucket, snapshot):
             "Id": "test-inventory",
             "Destination": {
                 "S3BucketDestination": {
-                    # "AccountId": "test",  # TODO
                     "Bucket": f"arn:aws:s3:::{dest_bucket}",
                     "Format": "CSV",
                 }
             },
             "IsEnabled": True,
             "IncludedObjectVersions": "All",
+            "OptionalFields": ["Size", "ETag"],
             "Schedule": {"Frequency": "Daily"},
         }
 
@@ -5045,10 +5045,75 @@ def test_s3_inventory_report_crud(self, aws_client, s3_create_bucket, snapshot):
             )
         snapshot.match("get-nonexistent-inv-config", e.value.response)
 
-    # TODO
+    @pytest.mark.aws_validated
     def test_s3_put_inventory_report_exceptions(self, aws_client, s3_create_bucket, snapshot):
-        # INVALID ARN FOR DEST BUCKET
-        pass
+        snapshot.add_transformer(snapshot.transform.resource_name())
+        src_bucket = s3_create_bucket()
+        dest_bucket = s3_create_bucket()
+        config_id = "test-inventory"
+
+        def _get_config():
+            return {
+                "Id": config_id,
+                "Destination": {
+                    "S3BucketDestination": {
+                        "Bucket": f"arn:aws:s3:::{dest_bucket}",
+                        "Format": "CSV",
+                    }
+                },
+                "IsEnabled": True,
+                "IncludedObjectVersions": "All",
+                "Schedule": {"Frequency": "Daily"},
+            }
+
+        def _put_bucket_inventory_configuration(inventory_configuration):
+            aws_client.s3.put_bucket_inventory_configuration(
+                Bucket=src_bucket,
+                Id=config_id,
+                InventoryConfiguration=inventory_configuration,
+            )
+
+        # put an inventory config with a wrong ID
+        with pytest.raises(ClientError) as e:
+            inv_config = _get_config()
+            inv_config["Id"] = config_id + "wrong"
+            _put_bucket_inventory_configuration(inv_config)
+        snapshot.match("wrong-id", e.value.response)
+
+        # set the Destination Bucket only as the name and not the ARN
+        with pytest.raises(ClientError) as e:
+            inv_config = _get_config()
+            inv_config["Destination"]["S3BucketDestination"]["Bucket"] = dest_bucket
+            _put_bucket_inventory_configuration(inv_config)
+        snapshot.match("wrong-destination-arn", e.value.response)
+
+        # set the wrong Destination Format (should be CSV/ORC/Parquet)
+        with pytest.raises(ClientError) as e:
+            inv_config = _get_config()
+            inv_config["Destination"]["S3BucketDestination"]["Format"] = "WRONG-FORMAT"
+            _put_bucket_inventory_configuration(inv_config)
+        snapshot.match("wrong-destination-format", e.value.response)
+
+        # set the wrong Schedule Frequency (should be Daily/Weekly)
+        with pytest.raises(ClientError) as e:
+            inv_config = _get_config()
+            inv_config["Schedule"]["Frequency"] = "Hourly"
+            _put_bucket_inventory_configuration(inv_config)
+        snapshot.match("wrong-schedule-frequency", e.value.response)
+
+        # set the wrong IncludedObjectVersions (should be All/Current)
+        with pytest.raises(ClientError) as e:
+            inv_config = _get_config()
+            inv_config["IncludedObjectVersions"] = "Wrong"
+            _put_bucket_inventory_configuration(inv_config)
+        snapshot.match("wrong-object-versions", e.value.response)
+
+        # set wrong OptionalFields
+        with pytest.raises(ClientError) as e:
+            inv_config = _get_config()
+            inv_config["OptionalFields"] = ["TestField"]
+            _put_bucket_inventory_configuration(inv_config)
+        snapshot.match("wrong-optional-field", e.value.response)
 
 
 class TestS3MultiAccounts:

diff --git a/tests/integration/s3/test_s3.snapshot.json b/tests/integration/s3/test_s3.snapshot.json
@@ -8348,7 +8348,7 @@
     }
   },
   "tests/integration/s3/test_s3.py::TestS3::test_s3_inventory_report_crud": {
-    "recorded-date": "13-07-2023, 18:24:52",
+    "recorded-date": "14-07-2023, 00:15:43",
     "recorded-content": {
       "put-inventory-config": {
         "ResponseMetadata": {
@@ -8368,6 +8368,10 @@
             "Id": "test-inventory",
             "IncludedObjectVersions": "All",
             "IsEnabled": true,
+            "OptionalFields": [
+              "Size",
+              "ETag"
+            ],
             "Schedule": {
               "Frequency": "Daily"
             }
@@ -8390,6 +8394,10 @@
           "Id": "test-inventory",
           "IncludedObjectVersions": "All",
           "IsEnabled": true,
+          "OptionalFields": [
+            "Size",
+            "ETag"
+          ],
           "Schedule": {
             "Frequency": "Daily"
           }
@@ -8423,5 +8431,70 @@
         }
       }
     }
+  },
+  "tests/integration/s3/test_s3.py::TestS3::test_s3_put_inventory_report_exceptions": {
+    "recorded-date": "13-07-2023, 23:26:28",
+    "recorded-content": {
+      "wrong-id": {
+        "Error": {
+          "Code": "IdMismatch",
+          "Message": "Document ID does not match the specified configuration ID."
+        },
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 400
+        }
+      },
+      "wrong-destination-arn": {
+        "Error": {
+          "Code": "InvalidS3DestinationBucket",
+          "Message": "Invalid bucket ARN."
+        },
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 400
+        }
+      },
+      "wrong-destination-format": {
+        "Error": {
+          "Code": "MalformedXML",
+          "Message": "The XML you provided was not well-formed or did not validate against our published schema"
+        },
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 400
+        }
+      },
+      "wrong-schedule-frequency": {
+        "Error": {
+          "Code": "MalformedXML",
+          "Message": "The XML you provided was not well-formed or did not validate against our published schema"
+        },
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 400
+        }
+      },
+      "wrong-object-versions": {
+        "Error": {
+          "Code": "MalformedXML",
+          "Message": "The XML you provided was not well-formed or did not validate against our published schema"
+        },
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 400
+        }
+      },
+      "wrong-optional-field": {
+        "Error": {
+          "Code": "MalformedXML",
+          "Message": "The XML you provided was not well-formed or did not validate against our published schema"
+        },
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 400
+        }
+      }
+    }
   }
 }
diff --git a/tests/unit/test_s3.py b/tests/unit/test_s3.py
@@ -729,6 +729,44 @@ def test_serialize_expiration_header(self, rule_id, lifecycle_exp, last_modified
         )
         assert serialized_header == header
 
+    @pytest.mark.parametrize(
+        "data, required, optional, result",
+        [
+            (
+                {"field1": "", "field2": "", "field3": ""},
+                {"field1"},
+                {"field2", "field3"},
+                True,
+            ),
+            (
+                {"field1": ""},
+                {"field1"},
+                {"field2", "field3"},
+                True,
+            ),
+            (
+                {"field1": "", "field2": "", "field3": ""},  # field3 is not a field
+                {"field1"},
+                {"field2"},
+                False,
+            ),
+            (
+                {"field2": ""},  # missing field1
+                {"field1"},
+                {"field2"},
+                False,
+            ),
+            (
+                {"field3": ""},  # missing field1 and field3 is not a field
+                {"field1"},
+                {"field2"},
+                False,
+            ),
+        ],
+    )
+    def test_validate_dict_fields(self, data, required, optional, result):
+        assert s3_utils_asf.validate_dict_fields(data, required, optional) == result
+
 
 class TestS3PresignedUrlAsf:
     """