Repository: getmoto/moto
Pull request #6789 (2023-09-08 07:23:52)

Batch: Using enum for job status
- Adding Enum for Batch job status and some utility methods.
- Adding unit tests for the methods. 
- Replacing some codes where job status is hard coded.


Patch:
diff --git a/moto/athena/responses.py b/moto/athena/responses.py
index e8b21e188..5bdc38c60 100644
--- a/moto/athena/responses.py
+++ b/moto/athena/responses.py
@@ -55,11 +55,15 @@ class AthenaResponse(BaseResponse):
     def get_query_execution(self) -> str:
         exec_id = self._get_param("QueryExecutionId")
         execution = self.athena_backend.get_query_execution(exec_id)
+        ddl_commands = ("ALTER", "CREATE", "DESCRIBE", "DROP", "MSCK", "SHOW")
+        statement_type = "DML"
+        if execution.query.upper().startswith(ddl_commands):
+            statement_type = "DDL"
         result = {
             "QueryExecution": {
                 "QueryExecutionId": exec_id,
                 "Query": execution.query,
-                "StatementType": "DDL",
+                "StatementType": statement_type,
                 "ResultConfiguration": execution.config,
                 "QueryExecutionContext": execution.context,
                 "Status": {
diff --git a/moto/batch/models.py b/moto/batch/models.py
index a0ef1c80f..ae9e6449f 100644
--- a/moto/batch/models.py
+++ b/moto/batch/models.py
@@ -24,6 +24,7 @@ from .utils import (
     make_arn_for_job,
     make_arn_for_task_def,
     lowercase_first_key,
+    JobStatus,
 )
 from moto.ec2.exceptions import InvalidSubnetIdError
 from moto.ec2.models.instance_types import INSTANCE_TYPES as EC2_INSTANCE_TYPES
@@ -487,12 +488,7 @@ class Job(threading.Thread, BaseModel, DockerModel, ManagedState):
         ManagedState.__init__(
             self,
             "batch::job",
-            [
-                ("SUBMITTED", "PENDING"),
-                ("PENDING", "RUNNABLE"),
-                ("RUNNABLE", "STARTING"),
-                ("STARTING", "RUNNING"),
-            ],
+            JobStatus.status_transitions(),
         )
 
         self.job_name = name
@@ -539,8 +535,9 @@ class Job(threading.Thread, BaseModel, DockerModel, ManagedState):
         }
         if self.job_stopped_reason is not None:
             result["statusReason"] = self.job_stopped_reason
-        if result["status"] not in ["SUBMITTED", "PENDING", "RUNNABLE", "STARTING"]:
-            result["startedAt"] = datetime2int_milliseconds(self.job_started_at)
+        if self.status is not None:
+            if JobStatus.is_job_already_started(self.status):
+                result["startedAt"] = datetime2int_milliseconds(self.job_started_at)
         if self.job_stopped:
             result["stoppedAt"] = datetime2int_milliseconds(self.job_stopped_at)
             if self.exit_code is not None:
@@ -638,7 +635,7 @@ class Job(threading.Thread, BaseModel, DockerModel, ManagedState):
             containers: List[docker.models.containers.Container] = []
 
             self.advance()
-            while self.status == "SUBMITTED":
+            while self.status == JobStatus.SUBMITTED:
                 # Wait until we've moved onto state 'PENDING'
                 sleep(0.5)
 
@@ -730,7 +727,7 @@ class Job(threading.Thread, BaseModel, DockerModel, ManagedState):
                     )
 
             self.advance()
-            while self.status == "PENDING":
+            while self.status == JobStatus.PENDING:
                 # Wait until the state is no longer pending, but 'RUNNABLE'
                 sleep(0.5)
             # TODO setup ecs container instance
@@ -765,12 +762,12 @@ class Job(threading.Thread, BaseModel, DockerModel, ManagedState):
 
             log_config = docker.types.LogConfig(type=docker.types.LogConfig.types.JSON)
             self.advance()
-            while self.status == "RUNNABLE":
+            while self.status == JobStatus.RUNNABLE:
                 # Wait until the state is no longer runnable, but 'STARTING'
                 sleep(0.5)
 
             self.advance()
-            while self.status == "STARTING":
+            while self.status == JobStatus.STARTING:
                 # Wait until the state is no longer runnable, but 'RUNNING'
                 sleep(0.5)
 
@@ -898,7 +895,7 @@ class Job(threading.Thread, BaseModel, DockerModel, ManagedState):
         # The describe-method needs them immediately when status is set
         self.job_stopped = True
         self.job_stopped_at = datetime.datetime.now()
-        self.status = "SUCCEEDED" if success else "FAILED"
+        self.status = JobStatus.SUCCEEDED if success else JobStatus.FAILED
         self._stop_attempt()
 
     def _start_attempt(self) -> None:
@@ -934,9 +931,9 @@ class Job(threading.Thread, BaseModel, DockerModel, ManagedState):
             for dependent_id in dependent_ids:
                 if dependent_id in self.all_jobs:
                     dependent_job = self.all_jobs[dependent_id]
-                    if dependent_job.status == "SUCCEEDED":
+                    if dependent_job.status == JobStatus.SUCCEEDED:
                         successful_dependencies.add(dependent_id)
-                    if dependent_job.status == "FAILED":
+                    if dependent_job.status == JobStatus.FAILED:
                         logger.error(
                             f"Terminating job {self.name} due to failed dependency {dependent_job.name}"
                         )
@@ -1038,7 +1035,7 @@ class BatchBackend(BaseBackend):
 
     def reset(self) -> None:
         for job in self._jobs.values():
-            if job.status not in ("FAILED", "SUCCEEDED"):
+            if job.status not in (JobStatus.FAILED, JobStatus.SUCCEEDED):
                 job.stop = True
                 # Try to join
                 job.join(0.2)
@@ -1776,15 +1773,7 @@ class BatchBackend(BaseBackend):
         if job_queue is None:
             raise ClientException(f"Job queue {job_queue_name} does not exist")
 
-        if job_status is not None and job_status not in (
-            "SUBMITTED",
-            "PENDING",
-            "RUNNABLE",
-            "STARTING",
-            "RUNNING",
-            "SUCCEEDED",
-            "FAILED",
-        ):
+        if job_status is not None and job_status not in JobStatus.job_statuses():
             raise ClientException(
                 "Job status is not one of SUBMITTED | PENDING | RUNNABLE | STARTING | RUNNING | SUCCEEDED | FAILED"
             )
@@ -1821,7 +1810,9 @@ class BatchBackend(BaseBackend):
 
         job = self.get_job_by_id(job_id)
         if job is not None:
-            if job.status in ["SUBMITTED", "PENDING", "RUNNABLE"]:
+            if job.status is None:
+                return
+            if JobStatus.is_job_before_starting(job.status):
                 job.terminate(reason)
             # No-Op for jobs that have already started - user has to explicitly terminate those
 
diff --git a/moto/batch/utils.py b/moto/batch/utils.py
index 6df059dca..bb6e02746 100644
--- a/moto/batch/utils.py
+++ b/moto/batch/utils.py
@@ -1,4 +1,7 @@
-from typing import Any, Dict
+from enum import Enum
+from typing import Any, Dict, List, Optional, Tuple
+
+from .exceptions import ValidationError
 
 
 def make_arn_for_compute_env(account_id: str, name: str, region_name: str) -> str:
@@ -34,3 +37,55 @@ def lowercase_first_key(some_dict: Dict[str, Any]) -> Dict[str, Any]:
             new_dict[new_key] = value
 
     return new_dict
+
+
+def validate_job_status(target_job_status: str, valid_job_statuses: List[str]) -> None:
+    if target_job_status not in valid_job_statuses:
+        raise ValidationError(
+            (
+                "1 validation error detected: Value at 'current_status' failed "
+                "to satisfy constraint: Member must satisfy enum value set: {valid_statues}"
+            ).format(valid_statues=valid_job_statuses)
+        )
+
+
+class JobStatus(str, Enum):
+    SUBMITTED = "SUBMITTED"
+    PENDING = "PENDING"
+    RUNNABLE = "RUNNABLE"
+    STARTING = "STARTING"
+    RUNNING = "RUNNING"
+    SUCCEEDED = "SUCCEEDED"
+    FAILED = "FAILED"
+
+    @classmethod
+    def job_statuses(self) -> List[str]:
+        return sorted([item.value for item in JobStatus])
+
+    @classmethod
+    def is_job_already_started(self, current_status: str) -> bool:
+        validate_job_status(current_status, JobStatus.job_statuses())
+        return current_status not in [
+            JobStatus.SUBMITTED,
+            JobStatus.PENDING,
+            JobStatus.RUNNABLE,
+            JobStatus.STARTING,
+        ]
+
+    @classmethod
+    def is_job_before_starting(self, current_status: str) -> bool:
+        validate_job_status(current_status, JobStatus.job_statuses())
+        return current_status in [
+            JobStatus.SUBMITTED,
+            JobStatus.PENDING,
+            JobStatus.RUNNABLE,
+        ]
+
+    @classmethod
+    def status_transitions(self) -> List[Tuple[Optional[str], str]]:
+        return [
+            (JobStatus.SUBMITTED.value, JobStatus.PENDING.value),
+            (JobStatus.PENDING.value, JobStatus.RUNNABLE.value),
+            (JobStatus.RUNNABLE.value, JobStatus.STARTING),
+            (JobStatus.STARTING.value, JobStatus.RUNNING.value),
+        ]
diff --git a/moto/core/responses.py b/moto/core/responses.py
index 467f75a87..67c2cb6cd 100644
--- a/moto/core/responses.py
+++ b/moto/core/responses.py
@@ -149,6 +149,7 @@ class ActionAuthenticatorMixin(object):
                 method=self.method,  # type: ignore[attr-defined]
                 path=path,
                 data=self.data,  # type: ignore[attr-defined]
+                body=self.body,  # type: ignore[attr-defined]
                 headers=self.headers,  # type: ignore[attr-defined]
             )
             iam_request.check_signature()
diff --git a/moto/iam/access_control.py b/moto/iam/access_control.py
index 905dcf971..70b506260 100644
--- a/moto/iam/access_control.py
+++ b/moto/iam/access_control.py
@@ -174,6 +174,7 @@ class IAMRequestBase(object, metaclass=ABCMeta):
         method: str,
         path: str,
         data: Dict[str, str],
+        body: bytes,
         headers: Dict[str, str],
     ):
         log.debug(
@@ -183,6 +184,7 @@ class IAMRequestBase(object, metaclass=ABCMeta):
         self._method = method
         self._path = path
         self._data = data
+        self._body = body
         self._headers = headers
         credential_scope = self._get_string_between(
             "Credential=", ",", self._headers["Authorization"]
@@ -190,13 +192,14 @@ class IAMRequestBase(object, metaclass=ABCMeta):
         credential_data = credential_scope.split("/")
         self._region = credential_data[2]
         self._service = credential_data[3]
+        action_from_request = self._action_from_request()
         self._action = (
             self._service
             + ":"
             + (
-                self._data["Action"][0]
-                if isinstance(self._data["Action"], list)
-                else self._data["Action"]
+                action_from_request[0]
+                if isinstance(action_from_request, list)
+                else action_from_request
             )
         )
         try:
@@ -208,6 +211,11 @@ class IAMRequestBase(object, metaclass=ABCMeta):
         except CreateAccessKeyFailure as e:
             self._raise_invalid_access_key(e.reason)
 
+    def _action_from_request(self) -> str:
+        if "X-Amz-Target" in self._headers:
+            return self._headers["X-Amz-Target"].split(".")[-1]
+        return self._data["Action"]
+
     def check_signature(self) -> None:
         original_signature = self._get_string_between(
             "Signature=", ",", self._headers["Authorization"]
@@ -267,7 +275,10 @@ class IAMRequestBase(object, metaclass=ABCMeta):
         ).split(";")
         headers = self._create_headers_for_aws_request(signed_headers, self._headers)
         request = AWSRequest(
-            method=self._method, url=self._path, data=self._data, headers=headers
+            method=self._method,
+            url=self._path,
+            data=self._body or self._data,
+            headers=headers,
         )
         request.context["timestamp"] = headers["X-Amz-Date"]
 
diff --git a/moto/kms/models.py b/moto/kms/models.py
index db6c20ae1..34a24ff91 100644
--- a/moto/kms/models.py
+++ b/moto/kms/models.py
@@ -322,14 +322,16 @@ class KmsBackend(BaseBackend):
     #
     # In our implementation with just create a copy of all the properties once without any protection from change,
     # as the exact implementation is currently infeasible.
-    def replicate_key(self, key_id: str, replica_region: str) -> None:
+    def replicate_key(self, key_id: str, replica_region: str) -> Key:
         # Using copy() instead of deepcopy(), as the latter results in exception:
         #    TypeError: cannot pickle '_cffi_backend.FFI' object
         # Since we only update top level properties, copy() should suffice.
         replica_key = copy(self.keys[key_id])
         replica_key.region = replica_region
+        replica_key.arn = replica_key.arn.replace(self.region_name, replica_region)
         to_region_backend = kms_backends[self.account_id][replica_region]
         to_region_backend.keys[replica_key.id] = replica_key
+        return replica_key
 
     def update_key_description(self, key_id: str, description: str) -> None:
         key = self.keys[self.get_key_id(key_id)]
diff --git a/moto/kms/responses.py b/moto/kms/responses.py
index acd08181c..a6de11e0f 100644
--- a/moto/kms/responses.py
+++ b/moto/kms/responses.py
@@ -128,11 +128,17 @@ class KmsResponse(BaseResponse):
         )
         return json.dumps(key.to_dict())
 
-    def replicate_key(self) -> None:
+    def replicate_key(self) -> str:
         key_id = self._get_param("KeyId")
         self._validate_key_id(key_id)
         replica_region = self._get_param("ReplicaRegion")
-        self.kms_backend.replicate_key(key_id, replica_region)
+        replica_key = self.kms_backend.replicate_key(key_id, replica_region)
+        return json.dumps(
+            {
+                "ReplicaKeyMetadata": replica_key.to_dict()["KeyMetadata"],
+                "ReplicaPolicy": replica_key.generate_default_policy(),
+            }
+        )
 
     def update_key_description(self) -> str:
         """https://docs.aws.amazon.com/kms/latest/APIReference/API_UpdateKeyDescription.html"""
diff --git a/moto/s3/models.py b/moto/s3/models.py
index 3d67bcf5b..ea70f7caa 100644
--- a/moto/s3/models.py
+++ b/moto/s3/models.py
@@ -54,7 +54,7 @@ from .cloud_formation import cfn_to_api_encryption, is_replacement_update
 from . import notifications
 from .select_object_content import parse_query
 from .utils import _VersionedKeyStore, CaseInsensitiveDict
-from .utils import ARCHIVE_STORAGE_CLASSES, STORAGE_CLASS
+from .utils import ARCHIVE_STORAGE_CLASSES, STORAGE_CLASS, LOGGING_SERVICE_PRINCIPAL
 from ..events.notifications import send_notification as events_send_notification
 from ..settings import get_s3_default_key_buffer_size, S3_UPLOAD_PART_MIN_SIZE
 from ..settings import s3_allow_crossdomain_access
@@ -1196,22 +1196,38 @@ class FakeBucket(CloudFormationModel):
     def delete_cors(self) -> None:
         self.cors = []
 
-    def set_logging(
-        self, logging_config: Optional[Dict[str, Any]], bucket_backend: "S3Backend"
-    ) -> None:
-        if not logging_config:
-            self.logging = {}
-            return
+    @staticmethod
+    def _log_permissions_enabled_policy(
+        target_bucket: "FakeBucket", target_prefix: Optional[str]
+    ) -> bool:
+        target_bucket_policy = target_bucket.policy
+        if target_bucket_policy:
+            target_bucket_policy_json = json.loads(target_bucket_policy.decode())
+            for stmt in target_bucket_policy_json["Statement"]:
+                if (
+                    stmt.get("Principal", {}).get("Service")
+                    != LOGGING_SERVICE_PRINCIPAL
+                ):
+                    continue
+                if stmt.get("Effect", "") != "Allow":
+                    continue
+                if "s3:PutObject" not in stmt.get("Action", []):
+                    continue
+                if (
+                    stmt.get("Resource")
+                    != f"arn:aws:s3:::{target_bucket.name}/{target_prefix if target_prefix else ''}*"
+                    and stmt.get("Resource") != f"arn:aws:s3:::{target_bucket.name}/*"
+                    and stmt.get("Resource") != f"arn:aws:s3:::{target_bucket.name}"
+                ):
+                    continue
+                return True
 
-        # Target bucket must exist in the same account (assuming all moto buckets are in the same account):
-        if not bucket_backend.buckets.get(logging_config["TargetBucket"]):
-            raise InvalidTargetBucketForLogging(
-                "The target bucket for logging does not exist."
-            )
+        return False
 
-        # Does the target bucket have the log-delivery WRITE and READ_ACP permissions?
+    @staticmethod
+    def _log_permissions_enabled_acl(target_bucket: "FakeBucket") -> bool:
         write = read_acp = False
-        for grant in bucket_backend.buckets[logging_config["TargetBucket"]].acl.grants:  # type: ignore
+        for grant in target_bucket.acl.grants:  # type: ignore
             # Must be granted to: http://acs.amazonaws.com/groups/s3/LogDelivery
             for grantee in grant.grantees:
                 if grantee.uri == "http://acs.amazonaws.com/groups/s3/LogDelivery":
@@ -1226,20 +1242,39 @@ class FakeBucket(CloudFormationModel):
                         or "FULL_CONTROL" in grant.permissions
                     ):
                         read_acp = True
-
                     break
 
-        if not write or not read_acp:
+        return write and read_acp
+
+    def set_logging(
+        self, logging_config: Optional[Dict[str, Any]], bucket_backend: "S3Backend"
+    ) -> None:
+        if not logging_config:
+            self.logging = {}
+            return
+
+        # Target bucket must exist in the same account (assuming all moto buckets are in the same account):
+        target_bucket = bucket_backend.buckets.get(logging_config["TargetBucket"])
+        if not target_bucket:
+            raise InvalidTargetBucketForLogging(
+                "The target bucket for logging does not exist."
+            )
+
+        target_prefix = self.logging.get("TargetPrefix", None)
+        has_policy_permissions = self._log_permissions_enabled_policy(
+            target_bucket=target_bucket, target_prefix=target_prefix
+        )
+        has_acl_permissions = self._log_permissions_enabled_acl(
+            target_bucket=target_bucket
+        )
+        if not (has_policy_permissions or has_acl_permissions):
             raise InvalidTargetBucketForLogging(
-                "You must give the log-delivery group WRITE and READ_ACP"
-                " permissions to the target bucket"
+                "You must either provide the necessary permissions to the logging service using a bucket "
+                "policy or give the log-delivery group WRITE and READ_ACP permissions to the target bucket"
             )
 
         # Buckets must also exist within the same region:
-        if (
-            bucket_backend.buckets[logging_config["TargetBucket"]].region_name
-            != self.region_name
-        ):
+        if target_bucket.region_name != self.region_name:
             raise CrossLocationLoggingProhibitted()
 
         # Checks pass -- set the logging config:
diff --git a/moto/s3/utils.py b/moto/s3/utils.py
index 13af61618..5eab429c1 100644
--- a/moto/s3/utils.py
+++ b/moto/s3/utils.py
@@ -36,6 +36,7 @@ STORAGE_CLASS = [
     "ONEZONE_IA",
     "INTELLIGENT_TIERING",
 ] + ARCHIVE_STORAGE_CLASSES
+LOGGING_SERVICE_PRINCIPAL = "logging.s3.amazonaws.com"
 
 
 def bucket_name_from_url(url: str) -> Optional[str]:  # type: ignore


Test patch:
diff --git a/tests/test_athena/test_athena.py b/tests/test_athena/test_athena.py
index 1f3644061..af2973f06 100644
--- a/tests/test_athena/test_athena.py
+++ b/tests/test_athena/test_athena.py
@@ -140,7 +140,7 @@ def test_get_query_execution():
     #
     assert details["QueryExecutionId"] == exex_id
     assert details["Query"] == query
-    assert details["StatementType"] == "DDL"
+    assert details["StatementType"] == "DML"
     assert details["ResultConfiguration"]["OutputLocation"] == location
     assert details["QueryExecutionContext"]["Database"] == database
     assert details["Status"]["State"] == "SUCCEEDED"
diff --git a/tests/test_batch/test_utils.py b/tests/test_batch/test_utils.py
new file mode 100644
index 000000000..0c203ff1e
--- /dev/null
+++ b/tests/test_batch/test_utils.py
@@ -0,0 +1,70 @@
+import pytest
+
+from moto.batch.exceptions import ValidationError
+
+from moto.batch.utils import JobStatus
+
+
+@pytest.mark.parametrize(
+    "job_already_started, job_status",
+    [
+        (None, "InvalidJobStatus"),
+        (False, "SUBMITTED"),
+        (False, "PENDING"),
+        (False, "RUNNABLE"),
+        (False, "STARTING"),
+        (True, "RUNNING"),
+        (True, "SUCCEEDED"),
+        (True, "FAILDED"),
+    ],
+)
+def test_JobStatus_is_job_already_sarted(job_already_started, job_status):
+    if job_status not in JobStatus.job_statuses():
+        with pytest.raises(ValidationError) as e:
+            _ = JobStatus.is_job_already_started("InvalidJobStatus")
+        assert (
+            e.value.message
+            == "1 validation error detected: Value at 'current_status' failed to satisfy constraint: Member must satisfy enum value set: ['FAILED', 'PENDING', 'RUNNABLE', 'RUNNING', 'STARTING', 'SUBMITTED', 'SUCCEEDED']"
+        )
+        return
+
+    assert JobStatus.is_job_already_started(job_status) is job_already_started
+
+
+@pytest.mark.parametrize(
+    "job_before_starting, job_status",
+    [
+        (None, "InvalidJobStatus"),
+        (True, "SUBMITTED"),
+        (True, "PENDING"),
+        (True, "RUNNABLE"),
+        (False, "STARTING"),
+        (False, "RUNNING"),
+        (False, "SUCCEEDED"),
+        (False, "FAILDED"),
+    ],
+)
+def test_JobStatus_is_job_before_starting(job_before_starting, job_status):
+    if job_status not in JobStatus.job_statuses():
+        with pytest.raises(ValidationError) as e:
+            _ = JobStatus.is_job_before_starting("InvalidJobStatus")
+        assert (
+            e.value.message
+            == "1 validation error detected: Value at 'current_status' failed to satisfy constraint: Member must satisfy enum value set: ['FAILED', 'PENDING', 'RUNNABLE', 'RUNNING', 'STARTING', 'SUBMITTED', 'SUCCEEDED']"
+        )
+        return
+
+    assert JobStatus.is_job_before_starting(job_status) is job_before_starting
+
+
+def test_JobStatus_status_transitions():
+    for before_status, after_status in JobStatus.status_transitions():
+        if before_status == JobStatus.SUBMITTED:
+            assert after_status == JobStatus.PENDING
+        elif before_status == JobStatus.PENDING:
+            assert after_status == JobStatus.RUNNABLE
+        elif before_status == JobStatus.RUNNABLE:
+            assert after_status == JobStatus.STARTING
+        else:
+            assert before_status == JobStatus.STARTING
+            assert after_status == JobStatus.RUNNING
diff --git a/tests/test_core/test_auth.py b/tests/test_core/test_auth.py
index 8c5f8fa04..b2a051112 100644
--- a/tests/test_core/test_auth.py
+++ b/tests/test_core/test_auth.py
@@ -5,7 +5,7 @@ from botocore.exceptions import ClientError
 
 import pytest
 
-from moto import mock_iam, mock_ec2, mock_s3, mock_sts, mock_elbv2, mock_rds
+from moto import mock_iam, mock_ec2, mock_s3, mock_sts, mock_ssm, mock_elbv2, mock_rds
 from moto.core import set_initial_no_auth_action_count
 from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
 from uuid import uuid4
@@ -831,3 +831,30 @@ def test_allow_key_access_using_resource_arn() -> None:
     s3_client.put_object(Bucket="my_bucket", Key="keyname", Body=b"test")
     with pytest.raises(ClientError):
         s3_client.put_object(Bucket="my_bucket", Key="unknown", Body=b"test")
+
+
+@set_initial_no_auth_action_count(3)
+@mock_ssm
+@mock_iam
+def test_ssm_service():
+    user_name = "test-user"
+    policy_doc = {
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Action": ["ssm:*"],
+                "Effect": "Allow",
+                "Resource": ["*"],
+            },
+        ],
+    }
+    access_key = create_user_with_access_key_and_inline_policy(user_name, policy_doc)
+
+    ssmc = boto3.client(
+        "ssm",
+        region_name="us-east-1",
+        aws_access_key_id=access_key["AccessKeyId"],
+        aws_secret_access_key=access_key["SecretAccessKey"],
+    )
+
+    ssmc.put_parameter(Name="test", Value="value", Type="String")
diff --git a/tests/test_kms/test_kms_boto3.py b/tests/test_kms/test_kms_boto3.py
index b2db4817a..bd2a9f7f3 100644
--- a/tests/test_kms/test_kms_boto3.py
+++ b/tests/test_kms/test_kms_boto3.py
@@ -171,12 +171,16 @@ def test_replicate_key():
         to_region_client.describe_key(KeyId=key_id)
 
     with mock.patch.object(rsa, "generate_private_key", return_value=""):
-        from_region_client.replicate_key(
+        replica_response = from_region_client.replicate_key(
             KeyId=key_id, ReplicaRegion=region_to_replicate_to
         )
     to_region_client.describe_key(KeyId=key_id)
     from_region_client.describe_key(KeyId=key_id)
 
+    assert "ReplicaKeyMetadata" in replica_response
+    assert region_to_replicate_to in replica_response["ReplicaKeyMetadata"]["Arn"]
+    assert "ReplicaPolicy" in replica_response
+
 
 @mock_kms
 def test_create_key_deprecated_master_custom_key_spec():
diff --git a/tests/test_s3/test_s3_logging.py b/tests/test_s3/test_s3_logging.py
index 456489707..c5a9ee461 100644
--- a/tests/test_s3/test_s3_logging.py
+++ b/tests/test_s3/test_s3_logging.py
@@ -1,8 +1,17 @@
+import json
+
 import boto3
-from botocore.client import ClientError
 import pytest
+from botocore.client import ClientError
+from unittest import SkipTest
+from unittest.mock import patch
+
 
 from moto import mock_s3
+from moto import settings
+from moto.core import DEFAULT_ACCOUNT_ID
+from moto.s3 import s3_backends
+from moto.s3.models import FakeBucket
 from moto.s3.responses import DEFAULT_REGION_NAME
 
 
@@ -261,3 +270,322 @@ def test_log_file_is_created():
     assert any(c for c in contents if bucket_name in c)
     assert any(c for c in contents if "REST.GET.BUCKET" in c)
     assert any(c for c in contents if "REST.PUT.BUCKET" in c)
+
+
+@mock_s3
+def test_invalid_bucket_logging_when_permissions_are_false():
+    if settings.TEST_SERVER_MODE:
+        raise SkipTest("Can't patch permission logic in ServerMode")
+
+    s3_client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+    bucket_name = "mybucket"
+    log_bucket = "logbucket"
+    s3_client.create_bucket(Bucket=bucket_name)
+    s3_client.create_bucket(Bucket=log_bucket)
+    with patch(
+        "moto.s3.models.FakeBucket._log_permissions_enabled_policy", return_value=False
+    ), patch(
+        "moto.s3.models.FakeBucket._log_permissions_enabled_acl", return_value=False
+    ):
+        with pytest.raises(ClientError) as err:
+            s3_client.put_bucket_logging(
+                Bucket=bucket_name,
+                BucketLoggingStatus={
+                    "LoggingEnabled": {"TargetBucket": log_bucket, "TargetPrefix": ""}
+                },
+            )
+        assert err.value.response["Error"]["Code"] == "InvalidTargetBucketForLogging"
+        assert "log-delivery" in err.value.response["Error"]["Message"]
+
+
+@mock_s3
+def test_valid_bucket_logging_when_permissions_are_true():
+    if settings.TEST_SERVER_MODE:
+        raise SkipTest("Can't patch permission logic in ServerMode")
+
+    s3_client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+    bucket_name = "mybucket"
+    log_bucket = "logbucket"
+    s3_client.create_bucket(Bucket=bucket_name)
+    s3_client.create_bucket(Bucket=log_bucket)
+    with patch(
+        "moto.s3.models.FakeBucket._log_permissions_enabled_policy", return_value=True
+    ), patch(
+        "moto.s3.models.FakeBucket._log_permissions_enabled_acl", return_value=True
+    ):
+        s3_client.put_bucket_logging(
+            Bucket=bucket_name,
+            BucketLoggingStatus={
+                "LoggingEnabled": {
+                    "TargetBucket": log_bucket,
+                    "TargetPrefix": f"{bucket_name}/",
+                }
+            },
+        )
+        result = s3_client.get_bucket_logging(Bucket=bucket_name)
+        assert result["LoggingEnabled"]["TargetBucket"] == log_bucket
+        assert result["LoggingEnabled"]["TargetPrefix"] == f"{bucket_name}/"
+
+
+@mock_s3
+def test_bucket_policy_not_set():
+    if settings.TEST_SERVER_MODE:
+        raise SkipTest("Can't patch permission logic in ServerMode")
+
+    s3_client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+    s3_backend = s3_backends[DEFAULT_ACCOUNT_ID]["global"]
+
+    log_bucket = "log_bucket"
+    s3_client.create_bucket(Bucket=log_bucket)
+    log_bucket_obj = s3_backend.get_bucket(log_bucket)
+
+    assert (
+        FakeBucket._log_permissions_enabled_policy(
+            target_bucket=log_bucket_obj, target_prefix=""
+        )
+        is False
+    )
+
+
+@mock_s3
+def test_bucket_policy_principal():
+    if settings.TEST_SERVER_MODE:
+        raise SkipTest("Can't patch permission logic in ServerMode")
+
+    s3_client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+    s3_backend = s3_backends[DEFAULT_ACCOUNT_ID]["global"]
+
+    log_bucket = "log_bucket"
+    s3_client.create_bucket(Bucket=log_bucket)
+    log_bucket_obj = s3_backend.get_bucket(log_bucket)
+
+    invalid_principal_policy = {
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Sid": "S3ServerAccessLogsPolicy",
+                "Effect": "Allow",
+                "Principal": {"Service": "not_logging.s3.amazonaws.com"},
+                "Action": ["s3:PutObject"],
+                "Resource": f"arn:aws:s3:::{log_bucket}/*",
+            }
+        ],
+    }
+    s3_client.put_bucket_policy(
+        Bucket=log_bucket, Policy=json.dumps(invalid_principal_policy)
+    )
+    assert (
+        FakeBucket._log_permissions_enabled_policy(
+            target_bucket=log_bucket_obj, target_prefix=""
+        )
+        is False
+    )
+
+    s3_client.delete_bucket_policy(Bucket=log_bucket)
+    valid_principal_policy = {
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Sid": "S3ServerAccessLogsPolicy",
+                "Effect": "Allow",
+                "Principal": {"Service": "logging.s3.amazonaws.com"},
+                "Action": ["s3:PutObject"],
+                "Resource": f"arn:aws:s3:::{log_bucket}/*",
+            }
+        ],
+    }
+    s3_client.put_bucket_policy(
+        Bucket=log_bucket, Policy=json.dumps(valid_principal_policy)
+    )
+    assert FakeBucket._log_permissions_enabled_policy(
+        target_bucket=log_bucket_obj, target_prefix=""
+    )
+
+
+@mock_s3
+def test_bucket_policy_effect():
+    if settings.TEST_SERVER_MODE:
+        raise SkipTest("Can't patch permission logic in ServerMode")
+
+    s3_client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+    s3_backend = s3_backends[DEFAULT_ACCOUNT_ID]["global"]
+
+    log_bucket = "log_bucket"
+    s3_client.create_bucket(Bucket=log_bucket)
+    log_bucket_obj = s3_backend.get_bucket(log_bucket)
+    deny_effect_policy = {
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Sid": "S3ServerAccessLogsPolicy",
+                "Effect": "Deny",
+                "Principal": {"Service": "logging.s3.amazonaws.com"},
+                "Action": ["s3:PutObject"],
+                "Resource": f"arn:aws:s3:::{log_bucket}/*",
+            }
+        ],
+    }
+    s3_client.put_bucket_policy(
+        Bucket=log_bucket, Policy=json.dumps(deny_effect_policy)
+    )
+    assert (
+        FakeBucket._log_permissions_enabled_policy(
+            target_bucket=log_bucket_obj, target_prefix=""
+        )
+        is False
+    )
+
+    s3_client.delete_bucket_policy(Bucket=log_bucket)
+    allow_effect_policy = {
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Sid": "S3ServerAccessLogsPolicy",
+                "Effect": "Allow",
+                "Principal": {"Service": "logging.s3.amazonaws.com"},
+                "Action": ["s3:PutObject"],
+                "Resource": f"arn:aws:s3:::{log_bucket}/*",
+            }
+        ],
+    }
+    s3_client.put_bucket_policy(
+        Bucket=log_bucket, Policy=json.dumps(allow_effect_policy)
+    )
+    assert FakeBucket._log_permissions_enabled_policy(
+        target_bucket=log_bucket_obj, target_prefix=""
+    )
+
+
+@mock_s3
+def test_bucket_policy_action():
+    if settings.TEST_SERVER_MODE:
+        raise SkipTest("Can't patch permission logic in ServerMode")
+
+    s3_client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+    s3_backend = s3_backends[DEFAULT_ACCOUNT_ID]["global"]
+
+    log_bucket = "log_bucket"
+    s3_client.create_bucket(Bucket=log_bucket)
+    log_bucket_obj = s3_backend.get_bucket(log_bucket)
+    non_put_object_policy = {
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Sid": "S3ServerAccessLogsPolicy",
+                "Effect": "Allow",
+                "Principal": {"Service": "logging.s3.amazonaws.com"},
+                "Action": ["s3:GetObject"],
+                "Resource": f"arn:aws:s3:::{log_bucket}/*",
+            }
+        ],
+    }
+    s3_client.put_bucket_policy(
+        Bucket=log_bucket, Policy=json.dumps(non_put_object_policy)
+    )
+    assert (
+        FakeBucket._log_permissions_enabled_policy(
+            target_bucket=log_bucket_obj, target_prefix=""
+        )
+        is False
+    )
+
+    s3_client.delete_bucket_policy(Bucket=log_bucket)
+    put_object_policy = {
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Sid": "S3ServerAccessLogsPolicy",
+                "Effect": "Allow",
+                "Principal": {"Service": "logging.s3.amazonaws.com"},
+                "Action": ["s3:PutObject"],
+                "Resource": f"arn:aws:s3:::{log_bucket}/*",
+            }
+        ],
+    }
+    s3_client.put_bucket_policy(Bucket=log_bucket, Policy=json.dumps(put_object_policy))
+    assert FakeBucket._log_permissions_enabled_policy(
+        target_bucket=log_bucket_obj, target_prefix=""
+    )
+
+
+@mock_s3
+def test_bucket_policy_resource():
+    if settings.TEST_SERVER_MODE:
+        raise SkipTest("Can't patch permission logic in ServerMode")
+
+    s3_client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+    s3_backend = s3_backends[DEFAULT_ACCOUNT_ID]["global"]
+
+    log_bucket = "log_bucket"
+    s3_client.create_bucket(Bucket=log_bucket)
+    log_bucket_obj = s3_backend.get_bucket(log_bucket)
+    entire_bucket_policy = {
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Sid": "S3ServerAccessLogsPolicy",
+                "Effect": "Allow",
+                "Principal": {"Service": "logging.s3.amazonaws.com"},
+                "Action": ["s3:PutObject"],
+                "Resource": f"arn:aws:s3:::{log_bucket}/*",
+            }
+        ],
+    }
+    s3_client.put_bucket_policy(
+        Bucket=log_bucket, Policy=json.dumps(entire_bucket_policy)
+    )
+    assert FakeBucket._log_permissions_enabled_policy(
+        target_bucket=log_bucket_obj, target_prefix=""
+    )
+    assert FakeBucket._log_permissions_enabled_policy(
+        target_bucket=log_bucket_obj, target_prefix="prefix"
+    )
+
+    s3_client.delete_bucket_policy(Bucket=log_bucket)
+    bucket_level_policy = {
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Sid": "S3ServerAccessLogsPolicy",
+                "Effect": "Allow",
+                "Principal": {"Service": "logging.s3.amazonaws.com"},
+                "Action": ["s3:PutObject"],
+                "Resource": f"arn:aws:s3:::{log_bucket}",
+            }
+        ],
+    }
+    s3_client.put_bucket_policy(
+        Bucket=log_bucket, Policy=json.dumps(bucket_level_policy)
+    )
+    assert FakeBucket._log_permissions_enabled_policy(
+        target_bucket=log_bucket_obj, target_prefix=""
+    )
+    assert FakeBucket._log_permissions_enabled_policy(
+        target_bucket=log_bucket_obj, target_prefix="prefix"
+    )
+
+    s3_client.delete_bucket_policy(Bucket=log_bucket)
+    specfic_prefix_policy = {
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Sid": "S3ServerAccessLogsPolicy",
+                "Effect": "Allow",
+                "Principal": {"Service": "logging.s3.amazonaws.com"},
+                "Action": ["s3:PutObject"],
+                "Resource": f"arn:aws:s3:::{log_bucket}/prefix*",
+            }
+        ],
+    }
+    s3_client.put_bucket_policy(
+        Bucket=log_bucket, Policy=json.dumps(specfic_prefix_policy)
+    )
+    assert (
+        FakeBucket._log_permissions_enabled_policy(
+            target_bucket=log_bucket_obj, target_prefix=""
+        )
+        is False
+    )
+    assert FakeBucket._log_permissions_enabled_policy(
+        target_bucket=log_bucket_obj, target_prefix="prefix"
+    )