Repository: getmoto/moto (python)

Issue:
enhancement: add support for `s3.put_object` `IfNoneMatch` option
S3 recently added support for conditional writes: https://aws.amazon.com/about-aws/whats-new/2024/08/amazon-s3-conditional-writes/

https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObject.html#API_PutObject_RequestParameters

Would be cool to be able to test this using moto!

The issue can be tested with the following snippet
```
import boto3
from moto import mock_aws

with mock_aws():
    s3 = boto3.client("s3", region_name="us-east-1")
    s3.create_bucket(Bucket="test_bucket")

    first = s3.put_object(
        Key="test_object",
        Body="test",
        Bucket="test_bucket",
        IfNoneMatch="*",
    )

    second = s3.put_object(
        Key="test_object",
        Body="another_test",
        Bucket="test_bucket",
        IfNoneMatch="*",
    )

assert second['ResponseMetadata']['HTTPStatusCode'] == 412
```

You can see for yourself that this fails (as in the write to S3 succeeded) using `mock_aws()` while it behaves correctly if I use a real S3 client on a real bucket.

Pull request description:
S3: put_object() now support conditional writes
Fixes #8091 

Patch:
diff --git a/moto/s3/responses.py b/moto/s3/responses.py
index 967382100..eb79a3228 100644
--- a/moto/s3/responses.py
+++ b/moto/s3/responses.py
@@ -1694,12 +1694,19 @@ class S3Response(BaseResponse):
         storage_class = self.headers.get("x-amz-storage-class", "STANDARD")
         encryption = self.headers.get("x-amz-server-side-encryption")
         kms_key_id = self.headers.get("x-amz-server-side-encryption-aws-kms-key-id")
+        if_none_match = self.headers.get("If-None-Match")
         bucket_key_enabled = self.headers.get(
             "x-amz-server-side-encryption-bucket-key-enabled"
         )
         if bucket_key_enabled is not None:
             bucket_key_enabled = str(bucket_key_enabled).lower()
 
+        if (
+            if_none_match == "*"
+            and self.backend.get_object(self.bucket_name, key_name) is not None
+        ):
+            raise PreconditionFailed("If-None-Match")
+
         checksum_algorithm, checksum_value = self._get_checksum(response_headers)
 
         bucket = self.backend.get_bucket(self.bucket_name)


Test patch:
diff --git a/tests/test_elbv2/test_elbv2_target_health.py b/tests/test_elbv2/test_elbv2_target_health.py
index 2647839bd..47d68b5e3 100644
--- a/tests/test_elbv2/test_elbv2_target_health.py
+++ b/tests/test_elbv2/test_elbv2_target_health.py
@@ -3,6 +3,7 @@ from uuid import uuid4
 
 import boto3
 import pytest
+from botocore.exceptions import WaiterError
 
 from . import elbv2_aws_verified
 
@@ -313,8 +314,13 @@ sudo systemctl enable httpd"""
 
         if load_balancer_arn:
             conn.delete_load_balancer(LoadBalancerArn=load_balancer_arn)
-            waiter = conn.get_waiter("load_balancers_deleted")
-            waiter.wait(LoadBalancerArns=[load_balancer_arn])
+            try:
+                waiter = conn.get_waiter("load_balancers_deleted")
+                waiter.wait(LoadBalancerArns=[load_balancer_arn])
+            except WaiterError:
+                # This operations fails against botocore 1.35.15 and up
+                # https://github.com/boto/botocore/issues/3252
+                pass
 
     if target_is_aws:
         # Resources take a while to deregister
@@ -423,8 +429,13 @@ def test_describe_unknown_targets(target_is_aws=False):
 
         if load_balancer_arn:
             conn.delete_load_balancer(LoadBalancerArn=load_balancer_arn)
-            waiter = conn.get_waiter("load_balancers_deleted")
-            waiter.wait(LoadBalancerArns=[load_balancer_arn])
+            try:
+                waiter = conn.get_waiter("load_balancers_deleted")
+                waiter.wait(LoadBalancerArns=[load_balancer_arn])
+            except WaiterError:
+                # This operations fails against botocore 1.35.15 and up
+                # https://github.com/boto/botocore/issues/3252
+                pass
 
     if target_is_aws:
         # Resources take a while to deregister
diff --git a/tests/test_s3/test_s3.py b/tests/test_s3/test_s3.py
index 3f057eb72..6b178665f 100644
--- a/tests/test_s3/test_s3.py
+++ b/tests/test_s3/test_s3.py
@@ -526,6 +526,37 @@ def test_key_with_special_characters(key):
     assert resp["Body"].read() == b"value"
 
 
+@s3_aws_verified
+@pytest.mark.aws_verified
+@pytest.mark.parametrize("versioned", [True, False], ids=["versioned", "not versioned"])
+def test_conditional_write(versioned, bucket_name=None):
+    s3 = boto3.client("s3", region_name="us-east-1")
+    if versioned:
+        enable_versioning(bucket_name, s3)
+
+    s3.put_object(
+        Key="test_object",
+        Body="test",
+        Bucket=bucket_name,
+        IfNoneMatch="*",
+    )
+
+    with pytest.raises(ClientError) as exc:
+        s3.put_object(
+            Key="test_object",
+            Body="another_test",
+            Bucket=bucket_name,
+            IfNoneMatch="*",
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "PreconditionFailed"
+    assert (
+        err["Message"]
+        == "At least one of the pre-conditions you specified did not hold"
+    )
+    assert err["Condition"] == "If-None-Match"
+
+
 @mock_aws
 def test_bucket_key_listing_order():
     s3_resource = boto3.resource("s3", region_name=DEFAULT_REGION_NAME)