Repository: getmoto/moto
Pull request #8036 (2024-08-25 17:49:39)

S3: Improve Lock behaviour
Changes the AWS parity tests to run with `loadscope == file`, meaning all tests in a file still run syncronously
Implements `Logs: cancel_export_task()`
S3: Adds support for LockMode=GOVERNANCE
S3: Adds validation that a bucket is versioned before calling `put_object_lock_configuration`
S3: Adds validation to the parameters of `put_object_lock_configuration`
S3: delete_objects() now returns the `DeleteMarkerVersionId`-attribute
SNS: Removes `aws_ verified` ready for SNS test, as it started failing

Followup to #8024 

Should fix the AWS parity tests.

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/getmoto/moto/pull/8036?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 94.41%. Comparing base [(`af18374`)](https://app.codecov.io/gh/getmoto/moto/commit/af18374d38d4a8b13b18b85f0c8f71122a39ad72?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) to head [(`5b004a4`)](https://app.codecov.io/gh/getmoto/moto/commit/5b004a4940f6724c7941b205dbffe5b6f455d520?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).
> Report is 3 commits behind head on master.

<details><summary>Additional details and impacted files</summary>


```diff
@@           Coverage Diff           @@
##           master    #8036   +/-   ##
=======================================
  Coverage   94.40%   94.41%           
=======================================
  Files        1134     1134           
  Lines       96774    96802   +28     
=======================================
+ Hits        91364    91397   +33     
+ Misses       5410     5405    -5     
```

| [Flag](https://app.codecov.io/gh/getmoto/moto/pull/8036/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | Coverage Δ | |
|---|---|---|
| [servertests](https://app.codecov.io/gh/getmoto/moto/pull/8036/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `29.00% <27.77%> (-0.01%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/getmoto/moto/pull/8036/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `94.38% <100.00%> (+<0.01%)` | :arrow_up: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto#carryforward-flags-in-the-pull-request-comment) to find out more.

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/getmoto/moto/pull/8036?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).


Patch:
diff --git a/moto/logs/models.py b/moto/logs/models.py
index c2666af67..3d78a740d 100644
--- a/moto/logs/models.py
+++ b/moto/logs/models.py
@@ -1233,6 +1233,16 @@ class LogsBackend(BaseBackend):
         """
         return self.queries[query_id]
 
+    def cancel_export_task(self, task_id: str) -> None:
+        task = self.export_tasks.get(task_id)
+        if not task:
+            raise ResourceNotFoundException("The specified export task does not exist.")
+        # If the task has already finished, AWS throws an InvalidOperationException
+        #     'The specified export task has already finished'
+        # However, the export task is currently syncronous, meaning it finishes immediately
+        # When we make the Task async, we can also implement the error behaviour
+        task.status = {"code": "CANCELLED", "message": "Cancelled by user"}
+
     def create_export_task(
         self,
         taskName: str,
diff --git a/moto/logs/responses.py b/moto/logs/responses.py
index d213a2742..da45f0472 100644
--- a/moto/logs/responses.py
+++ b/moto/logs/responses.py
@@ -432,6 +432,11 @@ class LogsResponse(BaseResponse):
         query = self.logs_backend.get_query_results(query_id)
         return json.dumps(query.to_result_json())
 
+    def cancel_export_task(self) -> str:
+        task_id = self._get_param("taskId")
+        self.logs_backend.cancel_export_task(task_id)
+        return "{}"
+
     def create_export_task(self) -> str:
         task_id = self.logs_backend.create_export_task(
             logGroupName=self._get_param("logGroupName"),
diff --git a/moto/s3/exceptions.py b/moto/s3/exceptions.py
index 45bdd1572..47c2af819 100644
--- a/moto/s3/exceptions.py
+++ b/moto/s3/exceptions.py
@@ -517,6 +517,13 @@ class InvalidContinuationToken(S3ClientError):
         )
 
 
+class InvalidBucketState(S3ClientError):
+    code = 400
+
+    def __init__(self, msg: str):
+        super().__init__("InvalidBucketState", msg)
+
+
 class InvalidObjectState(BucketError):
     code = 403
 
@@ -537,6 +544,13 @@ class LockNotEnabled(S3ClientError):
         super().__init__("InvalidRequest", "Bucket is missing ObjectLockConfiguration")
 
 
+class MissingRequestBody(S3ClientError):
+    code = 400
+
+    def __init__(self) -> None:
+        super().__init__("MissingRequestBodyError", "Request Body is empty")
+
+
 class AccessDeniedByLock(S3ClientError):
     code = 400
 
diff --git a/moto/s3/models.py b/moto/s3/models.py
index 34a9e8aa9..8af527e9b 100644
--- a/moto/s3/models.py
+++ b/moto/s3/models.py
@@ -45,6 +45,7 @@ from moto.s3.exceptions import (
     EntityTooSmall,
     HeadOnDeleteMarker,
     InvalidBucketName,
+    InvalidBucketState,
     InvalidNotificationDestination,
     InvalidNotificationEvent,
     InvalidObjectState,
@@ -371,12 +372,14 @@ class FakeKey(BaseModel, ManagedState):
         self.value = state["value"]  # type: ignore
         self.lock = threading.Lock()
 
-    @property
-    def is_locked(self) -> bool:
+    def is_locked(self, governance_bypass: bool) -> bool:
         if self.lock_legal_status == "ON":
             return True
 
-        if self.lock_mode == "COMPLIANCE":
+        if self.lock_mode == "GOVERNANCE" and governance_bypass:
+            return False
+
+        if self.lock_mode in ["GOVERNANCE", "COMPLIANCE"]:
             now = utcnow()
             try:
                 until = datetime.datetime.strptime(
@@ -2393,6 +2396,10 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
         years: Optional[int] = None,
     ) -> None:
         bucket = self.get_bucket(bucket_name)
+        if not bucket.is_versioned:
+            raise InvalidBucketState(
+                "Versioning must be 'Enabled' on the bucket to apply a Object Lock configuration"
+            )
 
         if bucket.keys.item_size() > 0:
             raise BucketNeedsToBeNew
@@ -2788,10 +2795,8 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
 
                     for key in bucket.keys.getlist(key_name):
                         if str(key.version_id) == str(version_id):
-                            if (
-                                hasattr(key, "is_locked")
-                                and key.is_locked
-                                and not bypass
+                            if isinstance(key, FakeKey) and key.is_locked(
+                                governance_bypass=bypass
                             ):
                                 raise AccessDeniedByLock
 
@@ -2833,7 +2838,7 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
         bucket_name: str,
         objects: List[Dict[str, Any]],
         bypass_retention: bool = False,
-    ) -> Tuple[List[Tuple[str, Optional[str]]], List[str]]:
+    ) -> Tuple[List[Tuple[str, Optional[str], Optional[str]]], List[str]]:
         deleted = []
         errors = []
         for object_ in objects:
@@ -2841,13 +2846,21 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
             version_id = object_.get("VersionId", None)
 
             try:
-                self.delete_object(
+                success, headers = self.delete_object(
                     bucket_name,
                     key_name,
                     version_id=version_id,
                     bypass=bypass_retention,
                 )
-                deleted.append((key_name, version_id))
+                deleted.append(
+                    (
+                        key_name,
+                        version_id,
+                        headers.get("version-id")
+                        if headers and not version_id
+                        else None,
+                    )
+                )
             except AccessDeniedByLock:
                 errors.append(key_name)
         return deleted, errors
diff --git a/moto/s3/responses.py b/moto/s3/responses.py
index 88355001f..203c41ead 100644
--- a/moto/s3/responses.py
+++ b/moto/s3/responses.py
@@ -4,6 +4,7 @@ import urllib.parse
 from typing import Any, Dict, Iterator, List, Optional, Tuple, Type, Union
 from urllib.parse import parse_qs, unquote, urlencode, urlparse, urlunparse
 from xml.dom import minidom
+from xml.parsers.expat import ExpatError
 
 import xmltodict
 
@@ -46,6 +47,7 @@ from .exceptions import (
     MalformedXML,
     MissingBucket,
     MissingKey,
+    MissingRequestBody,
     MissingVersion,
     NoSystemTags,
     NotAnIntegerException,
@@ -766,7 +768,7 @@ class S3Response(BaseResponse):
         self._authenticate_and_authorize_s3_action(bucket_name=bucket_name)
 
         if "object-lock" in querystring:
-            config = self._lock_config_from_body()
+            config = self._process_lock_config_from_body()
 
             self.backend.put_object_lock_configuration(
                 bucket_name,
@@ -1210,7 +1212,16 @@ class S3Response(BaseResponse):
             deleted, errored = self.backend.delete_objects(
                 bucket_name, objects_to_delete, bypass_retention
             )
-            errors.extend([(err, "AccessDenied", "Access Denied") for err in errored])
+            errors.extend(
+                [
+                    (
+                        err,
+                        "AccessDenied",
+                        "Access Denied because object protected by object lock.",
+                    )
+                    for err in errored
+                ]
+            )
         else:
             deleted = []
             # [(key_name, errorcode, 'error message'), ..]
@@ -1916,6 +1927,14 @@ class S3Response(BaseResponse):
         else:
             return 404, response_headers, ""
 
+    def _process_lock_config_from_body(self) -> Dict[str, Any]:
+        try:
+            return self._lock_config_from_body()
+        except (TypeError, ExpatError):
+            raise MissingRequestBody
+        except KeyError:
+            raise MalformedXML
+
     def _lock_config_from_body(self) -> Dict[str, Any]:
         response_dict: Dict[str, Any] = {
             "enabled": False,
@@ -2711,10 +2730,11 @@ S3_BUCKET_GET_VERSIONS = """<?xml version="1.0" encoding="UTF-8"?>
 
 S3_DELETE_KEYS_RESPONSE = """<?xml version="1.0" encoding="UTF-8"?>
 <DeleteResult xmlns="http://s3.amazonaws.com/doc/2006-03-01">
-{% for k, v in deleted %}
+{% for k, v, dv in deleted %}
 <Deleted>
 <Key>{{k}}</Key>
 {% if v %}<VersionId>{{v}}</VersionId>{% endif %}
+{% if dv %}<DeleteMarkerVersionId>{{ dv }}</DeleteMarkerVersionId><DeleteMarker>true</DeleteMarker>{% endif %}
 </Deleted>
 {% endfor %}
 {% for k,c,m in delete_errors %}


Test patch:
diff --git a/.github/workflows/tests_real_aws.yml b/.github/workflows/tests_real_aws.yml
index f3d204bbb..304867dfd 100644
--- a/.github/workflows/tests_real_aws.yml
+++ b/.github/workflows/tests_real_aws.yml
@@ -44,4 +44,4 @@ jobs:
       env:
         MOTO_TEST_ALLOW_AWS_REQUEST: ${{ true }}
       run: |
-        pytest -sv -n auto tests/test_applicationautoscaling/ tests/test_athena/ tests/test_cloudformation/ tests/test_dynamodb/ tests/test_ec2/ tests/test_events/ tests/test_iam/ tests/test_iot/ tests/test_lakeformation/ tests/test_logs/ tests/test_sqs/ tests/test_ses/ tests/test_s3* tests/test_stepfunctions/ tests/test_sns/ tests/test_timestreamwrite/ -m aws_verified
+        pytest -sv -n --dist loadfile auto tests/test_applicationautoscaling/ tests/test_athena/ tests/test_cloudformation/ tests/test_dynamodb/ tests/test_ec2/ tests/test_events/ tests/test_iam/ tests/test_iot/ tests/test_lakeformation/ tests/test_logs/ tests/test_sqs/ tests/test_ses/ tests/test_s3* tests/test_stepfunctions/ tests/test_sns/ tests/test_timestreamwrite/ -m aws_verified
diff --git a/tests/test_logs/test_export_tasks.py b/tests/test_logs/test_export_tasks.py
index 97146899a..2ca54efca 100644
--- a/tests/test_logs/test_export_tasks.py
+++ b/tests/test_logs/test_export_tasks.py
@@ -118,14 +118,41 @@ def test_create_export_task_happy_path(logs, s3, log_group_name, bucket_name):
         to=to,
         destination=bucket_name,
     )
+    task_id = resp["taskId"]
     # taskId resembles a valid UUID (i.e. a string of 32 hexadecimal digits)
-    assert UUID(resp["taskId"])
+    assert UUID(task_id)
     assert resp["ResponseMetadata"]["HTTPStatusCode"] == 200
 
     # s3 bucket contains indication that permissions were successful
     resp = s3.get_object(Bucket=bucket_name, Key="aws-logs-write-test")
     assert resp["Body"].read() == b"Permission Check Successful"
 
+    try:
+        # ExportTask's can take a long time to succeed
+        # From the docs: https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/S3ExportTasks.html
+        #     > the export task might take anywhere from a few seconds to a few hours
+        #
+        # There can be only one ExportTask active at any point in time
+        # Cancelling this one ensures that there's no Task active
+        # And subsequent tests can create Export Tasks without running into a LimitExceededException
+        logs.cancel_export_task(taskId=task_id)
+    except ClientError as exc:
+        # Because there are no logs, the export task in AWS usually finishes very quickly
+        # Which is fine - we can just ignore that
+        assert (
+            exc.response["Error"]["Message"]
+            == "The specified export task has already finished"
+        )
+
+
+@pytest.mark.aws_verified
+def test_cancel_unknown_export_task(logs):  # pylint: disable=redefined-outer-name
+    with pytest.raises(ClientError) as exc:
+        logs.cancel_export_task(taskId=str(uuid4()))
+    err = exc.value.response["Error"]
+    assert err["Code"] == "ResourceNotFoundException"
+    assert err["Message"] == "The specified export task does not exist."
+
 
 @pytest.mark.aws_verified
 def test_create_export_task_raises_ClientError_when_bucket_not_found(
diff --git a/tests/test_s3/__init__.py b/tests/test_s3/__init__.py
index 55d0f0c7a..032dd0f60 100644
--- a/tests/test_s3/__init__.py
+++ b/tests/test_s3/__init__.py
@@ -3,6 +3,7 @@ from functools import wraps
 from uuid import uuid4
 
 import boto3
+from botocore.exceptions import ClientError
 
 from moto import mock_aws
 from moto.s3.responses import DEFAULT_REGION_NAME
@@ -60,15 +61,23 @@ def s3_aws_verified(func):
 
 
 def empty_bucket(client, bucket_name):
+    # Delete any object lock config, if set before
+    try:
+        client.get_object_lock_configuration(Bucket=bucket_name)
+        kwargs = {"BypassGovernanceRetention": True}
+    except ClientError:
+        # No ObjectLock set
+        kwargs = {}
+
     versions = client.list_object_versions(Bucket=bucket_name).get("Versions", [])
     for key in versions:
         client.delete_object(
-            Bucket=bucket_name, Key=key["Key"], VersionId=key.get("VersionId")
+            Bucket=bucket_name, Key=key["Key"], VersionId=key.get("VersionId"), **kwargs
         )
     delete_markers = client.list_object_versions(Bucket=bucket_name).get(
         "DeleteMarkers", []
     )
     for key in delete_markers:
         client.delete_object(
-            Bucket=bucket_name, Key=key["Key"], VersionId=key.get("VersionId")
+            Bucket=bucket_name, Key=key["Key"], VersionId=key.get("VersionId"), **kwargs
         )
diff --git a/tests/test_s3/test_s3.py b/tests/test_s3/test_s3.py
index 9a9f0c863..c9195d0e9 100644
--- a/tests/test_s3/test_s3.py
+++ b/tests/test_s3/test_s3.py
@@ -436,16 +436,16 @@ def test_delete_versioned_objects():
     assert delete_markers is None
 
 
-@mock_aws
-def test_delete_missing_key():
+@s3_aws_verified
+@pytest.mark.aws_verified
+def test_delete_missing_key(bucket_name=None):
     s3_resource = boto3.resource("s3", region_name=DEFAULT_REGION_NAME)
-    bucket = s3_resource.Bucket("foobar")
-    bucket.create()
+    bucket = s3_resource.Bucket(bucket_name)
 
-    s3_resource.Object("foobar", "key1").put(Body=b"some value")
-    s3_resource.Object("foobar", "key2").put(Body=b"some value")
-    s3_resource.Object("foobar", "key3").put(Body=b"some value")
-    s3_resource.Object("foobar", "key4").put(Body=b"some value")
+    s3_resource.Object(bucket_name, "key1").put(Body=b"some value")
+    s3_resource.Object(bucket_name, "key2").put(Body=b"some value")
+    s3_resource.Object(bucket_name, "key3").put(Body=b"some value")
+    s3_resource.Object(bucket_name, "key4").put(Body=b"some value")
 
     result = bucket.delete_objects(
         Delete={
@@ -1786,28 +1786,36 @@ def test_deleted_versionings_list():
     assert len(listed["Contents"]) == 1
 
 
-@mock_aws
-def test_delete_objects_for_specific_version_id():
+@s3_aws_verified
+@pytest.mark.aws_verified
+def test_delete_objects_for_specific_version_id(bucket_name=None):
     client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
-    client.create_bucket(Bucket="blah")
-    client.put_bucket_versioning(
-        Bucket="blah", VersioningConfiguration={"Status": "Enabled"}
-    )
+    enable_versioning(bucket_name, client)
 
-    client.put_object(Bucket="blah", Key="test1", Body=b"test1a")
-    client.put_object(Bucket="blah", Key="test1", Body=b"test1b")
+    client.put_object(Bucket=bucket_name, Key="test1", Body=b"test1a")
+    client.put_object(Bucket=bucket_name, Key="test1", Body=b"test1b")
 
-    response = client.list_object_versions(Bucket="blah", Prefix="test1")
+    response = client.list_object_versions(Bucket=bucket_name, Prefix="test1")
     id_to_delete = [v["VersionId"] for v in response["Versions"] if v["IsLatest"]][0]
 
     response = client.delete_objects(
-        Bucket="blah", Delete={"Objects": [{"Key": "test1", "VersionId": id_to_delete}]}
+        Bucket=bucket_name,
+        Delete={"Objects": [{"Key": "test1", "VersionId": id_to_delete}]},
     )
     assert response["Deleted"] == [{"Key": "test1", "VersionId": id_to_delete}]
 
-    listed = client.list_objects_v2(Bucket="blah")
+    listed = client.list_objects_v2(Bucket=bucket_name)
     assert len(listed["Contents"]) == 1
 
+    # DeleteObjects without specifying VersionId
+    response = client.delete_objects(
+        Bucket=bucket_name, Delete={"Objects": [{"Key": "test1"}]}
+    )
+    assert "Deleted" in response
+    assert response["Deleted"][0]["DeleteMarker"] is True
+    assert response["Deleted"][0]["DeleteMarkerVersionId"]
+    assert response["Deleted"][0]["Key"] == "test1"
+
 
 @mock_aws
 def test_delete_versioned_bucket():
diff --git a/tests/test_s3/test_s3_bucket_policy.py b/tests/test_s3/test_s3_bucket_policy.py
index 56447fe72..a050a2233 100644
--- a/tests/test_s3/test_s3_bucket_policy.py
+++ b/tests/test_s3/test_s3_bucket_policy.py
@@ -144,6 +144,7 @@ class TestBucketPolicy:
 @pytest.mark.aws_verified
 def test_deny_delete_policy(bucket_name=None):
     client = boto3.client("s3", "us-east-1")
+    resource = boto3.resource("s3", "us-east-1")
 
     policy = {
         "Version": "2012-10-17",
@@ -164,12 +165,11 @@ def test_deny_delete_policy(bucket_name=None):
     with pytest.raises(ClientError):
         client.delete_object(Bucket=bucket_name, Key="obj")
 
-    result = (
-        boto3.resource("s3", "us-east-1").Bucket(bucket_name).objects.all().delete()
-    )
-    assert result[0]["Errors"] == [
-        {"Key": "obj", "Code": "AccessDenied", "Message": "Access Denied"}
-    ]
+    result = resource.Bucket(bucket_name).objects.all().delete()
+    assert result[0]["Errors"][0]["Key"] == "obj"
+    assert result[0]["Errors"][0]["Code"] == "AccessDenied"
+    # Message:
+    # User: {user_arn} is not authorized to perform: s3:DeleteObject on resource: "arn:aws:s3:::{bucket}/{key}" with an explicit deny in a resource-based policy
 
     # Delete Policy to make sure bucket can be emptied during teardown
     client.delete_bucket_policy(Bucket=bucket_name)
diff --git a/tests/test_s3/test_s3_list_object_versions.py b/tests/test_s3/test_s3_list_object_versions.py
index 0ec0466c8..feb2ee89e 100644
--- a/tests/test_s3/test_s3_list_object_versions.py
+++ b/tests/test_s3/test_s3_list_object_versions.py
@@ -1,4 +1,3 @@
-from time import sleep
 from unittest import SkipTest, mock
 from uuid import uuid4
 
@@ -7,6 +6,7 @@ import pytest
 
 from moto import mock_aws, settings
 from moto.s3.responses import DEFAULT_REGION_NAME
+from tests.test_s3.test_s3 import enable_versioning
 
 from . import s3_aws_verified
 
@@ -631,17 +631,6 @@ def test_list_object_versions__sort_order(bucket_name=None):
     assert version_list["DeleteMarkers"][0]["VersionId"] == b_del
 
 
-def enable_versioning(bucket_name, s3_client):
-    s3_client.put_bucket_versioning(
-        Bucket=bucket_name, VersioningConfiguration={"Status": "Enabled"}
-    )
-    # Versioning is not active immediately, so wait until we have confirmation the change has gone through
-    resp = {}
-    while resp.get("Status") != "Enabled":
-        sleep(0.1)
-        resp = s3_client.get_bucket_versioning(Bucket=bucket_name)
-
-
 @mock_aws
 @mock.patch.dict("os.environ", {"MOTO_S3_DEFAULT_MAX_KEYS": "5"})
 def test_list_object_versions_with_custom_limit():
diff --git a/tests/test_s3/test_s3_lock.py b/tests/test_s3/test_s3_lock.py
index 76284d105..bb91a7e3e 100644
--- a/tests/test_s3/test_s3_lock.py
+++ b/tests/test_s3/test_s3_lock.py
@@ -9,67 +9,235 @@ from botocore.config import Config
 from moto import mock_aws
 from moto.core.utils import utcnow
 from moto.s3.responses import DEFAULT_REGION_NAME
+from tests import allow_aws_request
 from tests.test_s3 import s3_aws_verified
+from tests.test_s3.test_s3 import enable_versioning
 
 
 @s3_aws_verified
 @pytest.mark.aws_verified
-def test_locked_object(bucket_name=None):
-    s3_client = boto3.client("s3", config=Config(region_name=DEFAULT_REGION_NAME))
+def test_put_object_lock_on_non_versioned_bucket(bucket_name=None):
+    s3_client = boto3.client("s3", DEFAULT_REGION_NAME)
+
+    with pytest.raises(ClientError) as exc:
+        s3_client.put_object_lock_configuration(
+            Bucket=bucket_name,
+            ObjectLockConfiguration={
+                "ObjectLockEnabled": "Enabled",
+                "Rule": {"DefaultRetention": {"Mode": "COMPLIANCE", "Days": 1}},
+            },
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "InvalidBucketState"
+    assert (
+        err["Message"]
+        == "Versioning must be 'Enabled' on the bucket to apply a Object Lock configuration"
+    )
+
+
+@s3_aws_verified
+@pytest.mark.aws_verified
+def test_put_object_lock_misconfiguration(bucket_name=None):
+    s3_client = boto3.client("s3", DEFAULT_REGION_NAME)
+
+    enable_versioning(bucket_name, s3_client)
+
+    with pytest.raises(ClientError) as exc:
+        s3_client.put_object_lock_configuration(Bucket=bucket_name)
+    err = exc.value.response["Error"]
+    assert err["Code"] == "MissingRequestBodyError"
+    assert err["Message"] == "Request Body is empty"
+
+    with pytest.raises(ClientError) as exc:
+        s3_client.put_object_lock_configuration(
+            Bucket=bucket_name, ObjectLockConfiguration={}
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "MalformedXML"
+
+
+@s3_aws_verified
+@pytest.mark.aws_verified
+@pytest.mark.parametrize(
+    "bypass_governance_retention",
+    [True, False, None],
+    ids=["bypass", "no_bypass", "unspecified"],
+)
+def test_locked_object_governance_mode(bypass_governance_retention, bucket_name=None):
+    s3_client = boto3.client("s3", DEFAULT_REGION_NAME)
 
     key_name = "file.txt"
-    seconds_lock = 2
+    seconds_lock = 10
+
+    enable_versioning(bucket_name, s3_client)
 
-    s3_client.put_bucket_versioning(
-        Bucket=bucket_name, VersioningConfiguration={"Status": "Enabled"}
-    )
     s3_client.put_object_lock_configuration(
         Bucket=bucket_name,
         ObjectLockConfiguration={
             "ObjectLockEnabled": "Enabled",
-            "Rule": {"DefaultRetention": {"Mode": "COMPLIANCE", "Days": 1}},
+            "Rule": {"DefaultRetention": {"Mode": "GOVERNANCE", "Days": 1}},
         },
     )
 
-    until = utcnow() + datetime.timedelta(0, seconds_lock)
+    until = utcnow() + datetime.timedelta(seconds=seconds_lock)
     s3_client.put_object(
         Bucket=bucket_name,
         Body=b"test",
         Key=key_name,
-        ObjectLockMode="COMPLIANCE",
+        ObjectLockMode="GOVERNANCE",
         ObjectLockRetainUntilDate=until,
     )
 
     versions_response = s3_client.list_object_versions(Bucket=bucket_name)
-    version_id = versions_response["Versions"][0]["VersionId"]
+    initial_version_id = versions_response["Versions"][0]["VersionId"]
 
     with pytest.raises(ClientError) as exc:
-        s3_client.delete_object(Bucket=bucket_name, Key=key_name, VersionId=version_id)
+        s3_client.delete_object(
+            Bucket=bucket_name, Key=key_name, VersionId=initial_version_id
+        )
     err = exc.value.response["Error"]
     assert err["Code"] == "AccessDenied"
 
+    kwargs = {}
+    if bypass_governance_retention in [True, False]:
+        kwargs["BypassGovernanceRetention"] = bypass_governance_retention
+
+    # Delete the object without VersionId always succeeds
     response = s3_client.delete_objects(
         Bucket=bucket_name,
         Delete={
             "Objects": [
-                {"Key": key_name, "VersionId": version_id},
+                {"Key": key_name},
             ],
         },
+        **kwargs,
     )
-    assert len(response["Errors"]) == 1
-    assert response["Errors"][0]["Key"] == key_name
-    assert response["Errors"][0]["Code"] == "AccessDenied"
+    assert response["Deleted"][0]["Key"] == key_name
+    deleted_version_id = response["Deleted"][0]["DeleteMarkerVersionId"]
 
+    # Delete any version id only succeeds if BypassGovernanceRetention=true
     response = s3_client.delete_objects(
         Bucket=bucket_name,
         Delete={
             "Objects": [
-                {"Key": key_name, "VersionId": version_id},
+                {"Key": key_name, "VersionId": initial_version_id},
             ],
         },
-        BypassGovernanceRetention=True,
+        **kwargs,
     )
-    assert {"Key": key_name, "VersionId": version_id} in response["Deleted"]
+    if bypass_governance_retention:
+        assert "Deleted" in response
+
+        deleted_version_id = response["Deleted"][0]["VersionId"]
+
+        response = s3_client.delete_objects(
+            Bucket=bucket_name,
+            Delete={
+                "Objects": [
+                    {"Key": key_name, "VersionId": deleted_version_id},
+                ],
+            },
+            **kwargs,
+        )
+        assert response["Deleted"] == [
+            {"Key": key_name, "VersionId": deleted_version_id}
+        ]
+
+    else:
+        # BypassGovernanceMode is either unspecified or False
+        assert "Errors" in response
+        assert response["Errors"][0]["Code"] == "AccessDenied"
+        assert response["Errors"][0]["Key"] == key_name
+        assert (
+            response["Errors"][0]["Message"]
+            == "Access Denied because object protected by object lock."
+        )
+
+        # We know we couldn't delete the initial version ID
+        # Can we delete the DeleteVersion
+        response = s3_client.delete_objects(
+            Bucket=bucket_name,
+            Delete={
+                "Objects": [
+                    {"Key": key_name, "VersionId": deleted_version_id},
+                ],
+            },
+            **kwargs,
+        )
+
+
+@s3_aws_verified
+@pytest.mark.aws_verified
+@pytest.mark.parametrize(
+    "bypass_governance_retention",
+    [True, False, None],
+    ids=["bypass", "no_bypass", "unspecified"],
+)
+def test_locked_object_compliance_mode(bypass_governance_retention, bucket_name=None):
+    s3_client = boto3.client("s3", DEFAULT_REGION_NAME)
+
+    key_name = "file.txt"
+    seconds_lock = 5 if allow_aws_request() else 1
+
+    enable_versioning(bucket_name, s3_client)
+
+    s3_client.put_object_lock_configuration(
+        Bucket=bucket_name,
+        ObjectLockConfiguration={
+            "ObjectLockEnabled": "Enabled",
+            "Rule": {"DefaultRetention": {"Mode": "GOVERNANCE", "Days": 1}},
+        },
+    )
+
+    until = utcnow() + datetime.timedelta(seconds=seconds_lock)
+    s3_client.put_object(
+        Bucket=bucket_name,
+        Body=b"test",
+        Key=key_name,
+        ObjectLockMode="COMPLIANCE",
+        ObjectLockRetainUntilDate=until,
+    )
+
+    versions_response = s3_client.list_object_versions(Bucket=bucket_name)
+    initial_version_id = versions_response["Versions"][0]["VersionId"]
+
+    kwargs = {}
+    if bypass_governance_retention in [True, False]:
+        kwargs["BypassGovernanceRetention"] = bypass_governance_retention
+
+    # Delete the object without VersionId always succeeds
+    response = s3_client.delete_objects(
+        Bucket=bucket_name,
+        Delete={
+            "Objects": [
+                {"Key": key_name},
+            ],
+        },
+        **kwargs,
+    )
+    assert response["Deleted"][0]["Key"] == key_name
+
+    # Delete any version id never succeeds in COMPLIANCE mode
+    response = s3_client.delete_objects(
+        Bucket=bucket_name,
+        Delete={
+            "Objects": [
+                {"Key": key_name, "VersionId": initial_version_id},
+            ],
+        },
+        **kwargs,
+    )
+    assert "Errors" in response
+    assert response["Errors"][0]["Code"] == "AccessDenied"
+    assert response["Errors"][0]["Key"] == key_name
+    assert (
+        response["Errors"][0]["Message"]
+        == "Access Denied because object protected by object lock."
+    )
+
+    from time import sleep
+
+    sleep(seconds_lock)
 
 
 @mock_aws
@@ -78,7 +246,7 @@ def test_fail_locked_object():
     key_name = "file.txt"
     seconds_lock = 2
 
-    s3_client = boto3.client("s3", config=Config(region_name=DEFAULT_REGION_NAME))
+    s3_client = boto3.client("s3", DEFAULT_REGION_NAME)
 
     s3_client.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=False)
     until = utcnow() + datetime.timedelta(0, seconds_lock)
@@ -137,14 +305,23 @@ def test_put_object_lock():
     s3_client.delete_bucket(Bucket=bucket_name)
 
 
-@mock_aws
-def test_put_object_legal_hold():
-    s3_client = boto3.client("s3", config=Config(region_name=DEFAULT_REGION_NAME))
+@s3_aws_verified
+@pytest.mark.aws_verified
+def test_put_object_legal_hold(bucket_name=None):
+    s3_client = boto3.client("s3", DEFAULT_REGION_NAME)
 
-    bucket_name = "put-legal-bucket"
     key_name = "file.txt"
 
-    s3_client.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=True)
+    s3_client.create_bucket(Bucket=bucket_name)
+    enable_versioning(bucket_name, s3_client)
+
+    s3_client.put_object_lock_configuration(
+        Bucket=bucket_name,
+        ObjectLockConfiguration={
+            "ObjectLockEnabled": "Enabled",
+            "Rule": {"DefaultRetention": {"Mode": "GOVERNANCE", "Days": 1}},
+        },
+    )
 
     s3_client.put_object(Bucket=bucket_name, Body=b"test", Key=key_name)
 
@@ -158,24 +335,29 @@ def test_put_object_legal_hold():
         LegalHold={"Status": "ON"},
     )
 
-    deleted = False
-    try:
-        s3_client.delete_object(Bucket=bucket_name, Key=key_name, VersionId=version_id)
-        deleted = True
-    except ClientError as exc:
-        assert exc.response["Error"]["Code"] == "AccessDenied"
-
-    assert deleted is False
+    with pytest.raises(ClientError) as exc:
+        s3_client.delete_object(
+            Bucket=bucket_name,
+            Key=key_name,
+            VersionId=version_id,
+            BypassGovernanceRetention=True,
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "AccessDenied"
 
-    # cleaning
+    # Deletion works when LegalHold is OFF
     s3_client.put_object_legal_hold(
         Bucket=bucket_name,
         Key=key_name,
         VersionId=version_id,
         LegalHold={"Status": "OFF"},
     )
-    s3_client.delete_object(Bucket=bucket_name, Key=key_name, VersionId=version_id)
-    s3_client.delete_bucket(Bucket=bucket_name)
+    s3_client.delete_object(
+        Bucket=bucket_name,
+        Key=key_name,
+        VersionId=version_id,
+        BypassGovernanceRetention=True,
+    )
 
 
 @mock_aws
diff --git a/tests/test_sns/test_application_boto3.py b/tests/test_sns/test_application_boto3.py
index 84ac1eb96..58faa4f46 100644
--- a/tests/test_sns/test_application_boto3.py
+++ b/tests/test_sns/test_application_boto3.py
@@ -474,9 +474,21 @@ def test_publish_to_disabled_platform_endpoint(api_key=None):
             conn.delete_platform_application(PlatformApplicationArn=application_arn)
 
 
-@pytest.mark.aws_verified
 @sns_aws_verified
 def test_publish_to_deleted_platform_endpoint(api_key=None):
+    """
+    This used to run against AWS, but they have changed the API, and this currently throws an exception:
+        Invalid parameter: Attributes Reason: Platform credentials are invalid
+
+    Need to change this test accordingly
+
+    https://docs.aws.amazon.com/sns/latest/dg/sns-send-custom-platform-specific-payloads-mobile-devices.html
+
+    > Amazon SNS now supports Firebase Cloud Messaging (FCM) HTTP v1 API for sending mobile push notifications to Android devices.
+    >
+    > March 26, 2024 – Amazon SNS supports FCM HTTP v1 API for Apple devices and Webpush destinations.
+    > We recommend that you migrate your existing mobile push applications to the latest FCM HTTP v1 API on or before June 1, 2024 to avoid application disruption.
+    """
     conn = boto3.client("sns", region_name="us-east-1")
     platform_name = str(uuid4())[0:6]
     topic_name = "topic_" + str(uuid4())[0:6]