Repository: getmoto/moto (python)

Issue:
Implement state transitions for ECS tasks
![task-lifecycle](https://github.com/getmoto/moto/assets/16805946/4ae02338-375e-4a82-99bd-77ebe48f3e4e)

* https://docs.aws.amazon.com/AmazonECS/latest/developerguide/task-lifecycle.html
* https://boto3.amazonaws.com/v1/documentation/api/1.28.1/reference/services/ecs/waiter/TasksStopped.html

Seems straightforward so I'll give it a shot 😅.


Pull request:
Implement state transitions for ECS tasks
Closes #6856 

Patch:
diff --git a/docs/docs/configuration/state_transition/models.rst b/docs/docs/configuration/state_transition/models.rst
index b3f3e8dbf..1cfb4e902 100644
--- a/docs/docs/configuration/state_transition/models.rst
+++ b/docs/docs/configuration/state_transition/models.rst
@@ -134,3 +134,16 @@ Transition type: Manual - describe the resource 1 time before the state advances
 Advancement:
 
     Call `boto3.client("transcribe").get_medical_transcription_job(..)`
+
+Service: ECS
+--------------
+
+**Model**: `ecs::task`   :raw-html:`<br />`
+Available states:
+
+    "RUNNING" --> "DEACTIVATING" --> "STOPPING" --> "DEPROVISIONING" --> "STOPPED"
+
+Transition type: Manual - describe the resource 1 time before the state advances  :raw-html:`<br />`
+Advancement:
+
+    Call `boto3.client("ecs").describe_tasks(..)`
diff --git a/moto/ec2/exceptions.py b/moto/ec2/exceptions.py
index 9163cca1f..e978992b1 100644
--- a/moto/ec2/exceptions.py
+++ b/moto/ec2/exceptions.py
@@ -436,6 +436,14 @@ class InvalidParameterValueError(EC2ClientError):
         )
 
 
+class InvalidParameterValueErrorPeeringAttachment(EC2ClientError):
+    def __init__(self, operation: str, transit_gateway_attachment_id: str):
+        super().__init__(
+            "InvalidParameterValue",
+            f"Cannot {operation} {transit_gateway_attachment_id} as the source of the peering request.",
+        )
+
+
 class InvalidParameterValueErrorTagSpotFleetRequest(EC2ClientError):
     def __init__(self, resource_type: str):
         super().__init__(
diff --git a/moto/ec2/models/transit_gateway_attachments.py b/moto/ec2/models/transit_gateway_attachments.py
index 0bc535085..108f80ef5 100644
--- a/moto/ec2/models/transit_gateway_attachments.py
+++ b/moto/ec2/models/transit_gateway_attachments.py
@@ -1,8 +1,11 @@
-from typing import Any, Dict, List, Optional
+from collections import defaultdict
+import weakref
+from typing import Any, Dict, List, Optional, Iterator
 from moto.core.utils import iso_8601_datetime_with_milliseconds, utcnow
 from moto.utilities.utils import merge_multiple_dicts, filter_resources
 from .core import TaggedEC2Resource
 from .vpc_peering_connections import PeeringConnectionStatus
+from ..exceptions import InvalidParameterValueErrorPeeringAttachment
 from ..utils import random_transit_gateway_attachment_id, describe_tag_filter
 
 
@@ -99,8 +102,20 @@ class TransitGatewayPeeringAttachment(TransitGatewayAttachment):
 
 
 class TransitGatewayAttachmentBackend:
+    backend_refs = defaultdict(set)  # type: ignore
+
     def __init__(self) -> None:
         self.transit_gateway_attachments: Dict[str, TransitGatewayAttachment] = {}
+        self.backend_refs[self.__class__].add(weakref.ref(self))
+
+    @classmethod
+    def _get_peering_attachment_backend_refs(
+        cls,
+    ) -> Iterator["TransitGatewayAttachmentBackend"]:
+        for backend_ref in cls.backend_refs[cls]:
+            backend = backend_ref()
+            if backend is not None:
+                yield backend
 
     def create_transit_gateway_vpn_attachment(
         self,
@@ -276,11 +291,24 @@ class TransitGatewayAttachmentBackend:
             tags=tags,
             region_name=self.region_name,  # type: ignore[attr-defined]
         )
-        transit_gateway_peering_attachment.status.accept()
-        transit_gateway_peering_attachment.state = "available"
+
         self.transit_gateway_attachments[
             transit_gateway_peering_attachment.id
         ] = transit_gateway_peering_attachment
+
+        # If the peer is not same as the current account or region, create attachment in peer backend
+        if self.account_id != peer_account_id or self.region_name != peer_region:  # type: ignore[attr-defined]
+            for backend in self._get_peering_attachment_backend_refs():
+                if (
+                    backend.account_id == peer_account_id  # type: ignore[attr-defined]
+                    and backend.region_name == peer_region  # type: ignore[attr-defined]
+                ):
+                    backend.transit_gateway_attachments[
+                        transit_gateway_peering_attachment.id
+                    ] = transit_gateway_peering_attachment
+
+        transit_gateway_peering_attachment.status.pending()
+        transit_gateway_peering_attachment.state = "pendingAcceptance"
         return transit_gateway_peering_attachment
 
     def describe_transit_gateway_peering_attachments(
@@ -319,6 +347,23 @@ class TransitGatewayAttachmentBackend:
         transit_gateway_attachment = self.transit_gateway_attachments[
             transit_gateway_attachment_id
         ]
+
+        requester_account_id = transit_gateway_attachment.requester_tgw_info["ownerId"]  # type: ignore[attr-defined]
+        requester_region_name = transit_gateway_attachment.requester_tgw_info["region"]  # type: ignore[attr-defined]
+        accepter_account_id = transit_gateway_attachment.accepter_tgw_info["ownerId"]  # type: ignore[attr-defined]
+        accepter_region_name = transit_gateway_attachment.accepter_tgw_info["region"]  # type: ignore[attr-defined]
+
+        # For cross-account peering, must be accepted by the accepter
+        if requester_account_id != accepter_account_id and self.account_id != accepter_account_id:  # type: ignore[attr-defined]
+            raise InvalidParameterValueErrorPeeringAttachment(
+                "accept", transit_gateway_attachment_id
+            )
+
+        if requester_region_name != accepter_region_name and self.region_name != accepter_region_name:  # type: ignore[attr-defined]
+            raise InvalidParameterValueErrorPeeringAttachment(
+                "accept", transit_gateway_attachment_id
+            )
+
         transit_gateway_attachment.state = "available"
         # Bit dodgy - we just assume that we act on a TransitGatewayPeeringAttachment
         # We could just as easily have another sub-class of TransitGatewayAttachment on our hands, which does not have a status-attribute
@@ -331,6 +376,22 @@ class TransitGatewayAttachmentBackend:
         transit_gateway_attachment = self.transit_gateway_attachments[
             transit_gateway_attachment_id
         ]
+
+        requester_account_id = transit_gateway_attachment.requester_tgw_info["ownerId"]  # type: ignore[attr-defined]
+        requester_region_name = transit_gateway_attachment.requester_tgw_info["region"]  # type: ignore[attr-defined]
+        accepter_account_id = transit_gateway_attachment.accepter_tgw_info["ownerId"]  # type: ignore[attr-defined]
+        accepter_region_name = transit_gateway_attachment.requester_tgw_info["region"]  # type: ignore[attr-defined]
+
+        if requester_account_id != accepter_account_id and self.account_id != accepter_account_id:  # type: ignore[attr-defined]
+            raise InvalidParameterValueErrorPeeringAttachment(
+                "reject", transit_gateway_attachment_id
+            )
+
+        if requester_region_name != accepter_region_name and self.region_name != accepter_region_name:  # type: ignore[attr-defined]
+            raise InvalidParameterValueErrorPeeringAttachment(
+                "reject", transit_gateway_attachment_id
+            )
+
         transit_gateway_attachment.state = "rejected"
         transit_gateway_attachment.status.reject()  # type: ignore[attr-defined]
         return transit_gateway_attachment
diff --git a/moto/ecs/models.py b/moto/ecs/models.py
index b05c3749f..86da09461 100644
--- a/moto/ecs/models.py
+++ b/moto/ecs/models.py
@@ -11,7 +11,10 @@ from moto.core.utils import unix_time, pascal_to_camelcase, remap_nested_keys
 
 from ..ec2.utils import random_private_ip
 from moto.ec2 import ec2_backends
+from moto.moto_api import state_manager
 from moto.moto_api._internal import mock_random
+from moto.moto_api._internal.managed_state_model import ManagedState
+
 from .exceptions import (
     EcsClientException,
     ServiceNotFoundException,
@@ -334,7 +337,7 @@ class TaskDefinition(BaseObject, CloudFormationModel):
             return original_resource
 
 
-class Task(BaseObject):
+class Task(BaseObject, ManagedState):
     def __init__(
         self,
         cluster: Cluster,
@@ -348,11 +351,28 @@ class Task(BaseObject):
         tags: Optional[List[Dict[str, str]]] = None,
         networking_configuration: Optional[Dict[str, Any]] = None,
     ):
+        # Configure ManagedState
+        # https://docs.aws.amazon.com/AmazonECS/latest/developerguide/task-lifecycle.html
+        super().__init__(
+            model_name="ecs::task",
+            transitions=[
+                # We start in RUNNING state in order not to break existing tests.
+                # ("PROVISIONING", "PENDING"),
+                # ("PENDING", "ACTIVATING"),
+                # ("ACTIVATING", "RUNNING"),
+                ("RUNNING", "DEACTIVATING"),
+                ("DEACTIVATING", "STOPPING"),
+                ("STOPPING", "DEPROVISIONING"),
+                ("DEPROVISIONING", "STOPPED"),
+                # There seems to be race condition, where the waiter expects the task to be in
+                # STOPPED state, but it is already in DELETED state.
+                # ("STOPPED", "DELETED"),
+            ],
+        )
         self.id = str(mock_random.uuid4())
         self.cluster_name = cluster.name
         self.cluster_arn = cluster.arn
         self.container_instance_arn = container_instance_arn
-        self.last_status = "RUNNING"
         self.desired_status = "RUNNING"
         self.task_definition_arn = task_definition.arn
         self.overrides = overrides or {}
@@ -401,6 +421,14 @@ class Task(BaseObject):
                 }
             )
 
+    @property
+    def last_status(self) -> Optional[str]:
+        return self.status  # managed state
+
+    @last_status.setter
+    def last_status(self, value: Optional[str]) -> None:
+        self.status = value
+
     @property
     def task_arn(self) -> str:
         if self._backend.enable_long_arn_for_name(name="taskLongArnFormat"):
@@ -411,6 +439,7 @@ class Task(BaseObject):
     def response_object(self) -> Dict[str, Any]:  # type: ignore[misc]
         response_object = self.gen_response_object()
         response_object["taskArn"] = self.task_arn
+        response_object["lastStatus"] = self.last_status
         return response_object
 
 
@@ -929,6 +958,11 @@ class EC2ContainerServiceBackend(BaseBackend):
         self.services: Dict[str, Service] = {}
         self.container_instances: Dict[str, Dict[str, ContainerInstance]] = {}
 
+        state_manager.register_default_transition(
+            model_name="ecs::task",
+            transition={"progression": "manual", "times": 1},
+        )
+
     @staticmethod
     def default_vpc_endpoint_service(service_region: str, zones: List[str]) -> List[Dict[str, Any]]:  # type: ignore[misc]
         """Default VPC endpoint service."""
@@ -1438,6 +1472,7 @@ class EC2ContainerServiceBackend(BaseBackend):
                     or task.task_arn in tasks
                     or any(task_id in task for task in tasks)
                 ):
+                    task.advance()
                     response.append(task)
         if "TAGS" in (include or []):
             return response
diff --git a/setup.cfg b/setup.cfg
index 94d1f2796..8d4c9bcf0 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -237,6 +237,7 @@ universal=1
 markers =
     network: requires network connection
     requires_docker: requires running docker
+    aws_verified: Verified against AWS, and should be able to run against AWS
 
 [coverage:run]
 relative_files = True


Test patch:
diff --git a/.github/workflows/tests_real_aws.yml b/.github/workflows/tests_real_aws.yml
new file mode 100644
index 000000000..b3fa3303a
--- /dev/null
+++ b/.github/workflows/tests_real_aws.yml
@@ -0,0 +1,45 @@
+name: Execute tests against AWS
+on:
+  schedule:
+    - cron: '00 6 * * 0'
+  workflow_dispatch:
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    permissions:
+      id-token: write
+
+    steps:
+    - uses: actions/checkout@v4
+      with:
+        fetch-depth: 0
+    - name: Set up Python
+      uses: actions/setup-python@v4
+      with:
+        python-version: "3.11"
+    - name: Get pip cache dir
+      id: pip-cache
+      run: |
+        echo "dir=$(pip cache dir)" >> $GITHUB_OUTPUT
+    - name: pip cache
+      uses: actions/cache@v3
+      with:
+        path: ${{ steps.pip-cache.outputs.dir }}
+        key: pip-3.11-${{ hashFiles('**/setup.cfg') }}
+    - name: Update pip
+      run: |
+        python -m pip install --upgrade pip
+    - name: Install project dependencies
+      run: |
+        pip install -r requirements-dev.txt
+    - name: Configure AWS
+      uses: aws-actions/configure-aws-credentials@v4
+      with:
+        aws-region: us-east-1
+        role-to-assume: arn:aws:iam::682283128318:role/GithubActionsRole
+    - name: Test with pytest
+      env:
+        MOTO_TEST_ALLOW_AWS_REQUEST: ${{ true }}
+      run: |
+        pytest -sv tests/test_s3 -m aws_verified
diff --git a/tests/test_ec2/test_transit_gateway_peering_attachments.py b/tests/test_ec2/test_transit_gateway_peering_attachments.py
index d61eed48a..e09d4ac36 100644
--- a/tests/test_ec2/test_transit_gateway_peering_attachments.py
+++ b/tests/test_ec2/test_transit_gateway_peering_attachments.py
@@ -1,7 +1,12 @@
+import os
+
 import boto3
+import pytest
+from botocore.exceptions import ClientError
+
 from moto import mock_ec2, settings
 from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
-from unittest import SkipTest
+from unittest import SkipTest, mock
 
 
 @mock_ec2
@@ -89,7 +94,7 @@ def test_describe_transit_gateway_peering_attachment_by_filters():
     )["TransitGatewayPeeringAttachments"]
     assert [a["TransitGatewayAttachmentId"] for a in find_3] == [attchmnt3]
 
-    filters = [{"Name": "state", "Values": ["available"]}]
+    filters = [{"Name": "state", "Values": ["pendingAcceptance"]}]
     find_all = retrieve_all_attachments(ec2, filters)
     all_ids = [a["TransitGatewayAttachmentId"] for a in find_all]
     assert attchmnt1 in all_ids
@@ -105,7 +110,7 @@ def test_describe_transit_gateway_peering_attachment_by_filters():
 
     find_available = ec2.describe_transit_gateway_peering_attachments(
         TransitGatewayAttachmentIds=[attchmnt1, attchmnt2],
-        Filters=[{"Name": "state", "Values": ["available"]}],
+        Filters=[{"Name": "state", "Values": ["pendingAcceptance"]}],
     )["TransitGatewayPeeringAttachments"]
     assert [a["TransitGatewayAttachmentId"] for a in find_available] == [attchmnt1]
 
@@ -119,7 +124,9 @@ def test_create_and_accept_transit_gateway_peering_attachment():
     gateway_id2 = ec2.create_transit_gateway(Description="my second gateway")[
         "TransitGateway"
     ]["TransitGatewayId"]
-    attchment_id = create_peering_attachment(ec2, gateway_id1, gateway_id2)
+    attchment_id = create_peering_attachment(
+        ec2, gateway_id1, gateway_id2, peer_region="us-west-1"
+    )
 
     ec2.accept_transit_gateway_peering_attachment(
         TransitGatewayAttachmentId=attchment_id
@@ -176,12 +183,86 @@ def test_create_and_delete_transit_gateway_peering_attachment():
     assert attachment["State"] == "deleted"
 
 
-def create_peering_attachment(ec2, gateway_id1, gateway_id2):
+@mock_ec2
+@pytest.mark.parametrize(
+    "account1,account2",
+    [
+        pytest.param("111111111111", "111111111111", id="within account"),
+        pytest.param("111111111111", "222222222222", id="across accounts"),
+    ],
+)
+@pytest.mark.skipif(
+    settings.TEST_SERVER_MODE, reason="Cannot set account ID in server mode"
+)
+def test_transit_gateway_peering_attachments_cross_region(account1, account2):
+    # create transit gateways
+    with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+        ec2_us = boto3.client("ec2", "us-west-1")
+        gateway_us = ec2_us.create_transit_gateway()["TransitGateway"][
+            "TransitGatewayId"
+        ]
+
+    with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}):
+        ec2_eu = boto3.client("ec2", "eu-central-1")
+        gateway_eu = ec2_eu.create_transit_gateway()["TransitGateway"][
+            "TransitGatewayId"
+        ]
+
+    # create peering
+    with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+        attachment_id = create_peering_attachment(
+            ec2_us,
+            gateway_us,
+            gateway_eu,
+            peer_account=account2,
+            peer_region="eu-central-1",
+        )
+
+    # ensure peering can be described by the accepter
+    with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}):
+        response = ec2_eu.describe_transit_gateway_peering_attachments(
+            TransitGatewayAttachmentIds=[attachment_id]
+        )["TransitGatewayPeeringAttachments"][0]
+        assert response["TransitGatewayAttachmentId"] == attachment_id
+        assert response["RequesterTgwInfo"]["OwnerId"] == account1
+        assert response["RequesterTgwInfo"]["Region"] == "us-west-1"
+        assert response["AccepterTgwInfo"]["OwnerId"] == account2
+        assert response["AccepterTgwInfo"]["Region"] == "eu-central-1"
+
+    # ensure accepting in requester account/region raises
+    with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+        with pytest.raises(ClientError) as exc:
+            ec2_us.accept_transit_gateway_peering_attachment(
+                TransitGatewayAttachmentId=attachment_id
+            )
+        assert exc.value.response["Error"]["Code"] == "InvalidParameterValue"
+        assert (
+            exc.value.response["Error"]["Message"]
+            == f"Cannot accept {attachment_id} as the source of the peering request."
+        )
+
+    with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}):
+        # ensure peering can be accepted by the accepter
+        response = ec2_eu.accept_transit_gateway_peering_attachment(
+            TransitGatewayAttachmentId=attachment_id
+        )
+        assert response["TransitGatewayPeeringAttachment"]["State"] == "available"
+
+        # ensure peering can be deleted by the accepter
+        response = ec2_eu.delete_transit_gateway_peering_attachment(
+            TransitGatewayAttachmentId=attachment_id
+        )
+        assert response["TransitGatewayPeeringAttachment"]["State"] == "deleted"
+
+
+def create_peering_attachment(
+    ec2, gateway_id1, gateway_id2, peer_account=ACCOUNT_ID, peer_region="us-east-1"
+):
     return ec2.create_transit_gateway_peering_attachment(
         TransitGatewayId=gateway_id1,
         PeerTransitGatewayId=gateway_id2,
-        PeerAccountId=ACCOUNT_ID,
-        PeerRegion="us-east-1",
+        PeerAccountId=peer_account,
+        PeerRegion=peer_region,
     )["TransitGatewayPeeringAttachment"]["TransitGatewayAttachmentId"]
 
 
diff --git a/tests/test_ecs/test_ecs_boto3.py b/tests/test_ecs/test_ecs_boto3.py
index 912b7be58..250826183 100644
--- a/tests/test_ecs/test_ecs_boto3.py
+++ b/tests/test_ecs/test_ecs_boto3.py
@@ -11,6 +11,7 @@ from unittest import mock, SkipTest
 from uuid import UUID
 
 from moto import mock_ecs, mock_ec2, settings
+from moto.moto_api import state_manager
 from tests import EXAMPLE_AMI_ID
 
 
@@ -1845,6 +1846,140 @@ def test_run_task():
     assert task["tags"][0].get("value") == "tagValue0"
 
 
+@mock_ec2
+@mock_ecs
+def test_wait_tasks_stopped():
+    if settings.TEST_SERVER_MODE:
+        raise SkipTest("Can't set transition directly in ServerMode")
+
+    state_manager.set_transition(
+        model_name="ecs::task",
+        transition={"progression": "immediate"},
+    )
+
+    client = boto3.client("ecs", region_name="us-east-1")
+    ec2 = boto3.resource("ec2", region_name="us-east-1")
+
+    test_cluster_name = "test_ecs_cluster"
+
+    client.create_cluster(clusterName=test_cluster_name)
+
+    test_instance = ec2.create_instances(
+        ImageId=EXAMPLE_AMI_ID, MinCount=1, MaxCount=1
+    )[0]
+
+    instance_id_document = json.dumps(
+        ec2_utils.generate_instance_identity_document(test_instance)
+    )
+
+    response = client.register_container_instance(
+        cluster=test_cluster_name,
+        instanceIdentityDocument=instance_id_document,
+    )
+
+    client.register_task_definition(
+        family="test_ecs_task",
+        containerDefinitions=[
+            {
+                "name": "hello_world",
+                "image": "docker/hello-world:latest",
+                "cpu": 1024,
+                "memory": 400,
+                "essential": True,
+                "environment": [
+                    {"name": "AWS_ACCESS_KEY_ID", "value": "SOME_ACCESS_KEY"}
+                ],
+                "logConfiguration": {"logDriver": "json-file"},
+            }
+        ],
+    )
+    response = client.run_task(
+        cluster="test_ecs_cluster",
+        overrides={},
+        taskDefinition="test_ecs_task",
+        startedBy="moto",
+    )
+    task_arn = response["tasks"][0]["taskArn"]
+
+    assert len(response["tasks"]) == 1
+    client.get_waiter("tasks_stopped").wait(
+        cluster="test_ecs_cluster",
+        tasks=[task_arn],
+    )
+
+    response = client.describe_tasks(cluster="test_ecs_cluster", tasks=[task_arn])
+    assert response["tasks"][0]["lastStatus"] == "STOPPED"
+
+    state_manager.unset_transition("ecs::task")
+
+
+@mock_ec2
+@mock_ecs
+def test_task_state_transitions():
+    if settings.TEST_SERVER_MODE:
+        raise SkipTest("Can't set transition directly in ServerMode")
+
+    state_manager.set_transition(
+        model_name="ecs::task",
+        transition={"progression": "manual", "times": 1},
+    )
+
+    client = boto3.client("ecs", region_name="us-east-1")
+    ec2 = boto3.resource("ec2", region_name="us-east-1")
+
+    test_cluster_name = "test_ecs_cluster"
+
+    client.create_cluster(clusterName=test_cluster_name)
+
+    test_instance = ec2.create_instances(
+        ImageId=EXAMPLE_AMI_ID, MinCount=1, MaxCount=1
+    )[0]
+
+    instance_id_document = json.dumps(
+        ec2_utils.generate_instance_identity_document(test_instance)
+    )
+
+    response = client.register_container_instance(
+        cluster=test_cluster_name,
+        instanceIdentityDocument=instance_id_document,
+    )
+
+    client.register_task_definition(
+        family="test_ecs_task",
+        containerDefinitions=[
+            {
+                "name": "hello_world",
+                "image": "docker/hello-world:latest",
+                "cpu": 1024,
+                "memory": 400,
+                "essential": True,
+                "environment": [
+                    {"name": "AWS_ACCESS_KEY_ID", "value": "SOME_ACCESS_KEY"}
+                ],
+                "logConfiguration": {"logDriver": "json-file"},
+            }
+        ],
+    )
+
+    response = client.run_task(
+        cluster="test_ecs_cluster",
+        overrides={},
+        taskDefinition="test_ecs_task",
+        startedBy="moto",
+    )
+    task_arn = response["tasks"][0]["taskArn"]
+    assert len(response["tasks"]) == 1
+
+    task_status = response["tasks"][0]["lastStatus"]
+    assert task_status == "RUNNING"
+
+    for status in ("DEACTIVATING", "STOPPING", "DEPROVISIONING", "STOPPED"):
+        response = client.describe_tasks(cluster="test_ecs_cluster", tasks=[task_arn])
+        assert response["tasks"][0]["lastStatus"] == status
+
+    state_manager.unset_transition("ecs::task")
+
+
 @mock_ec2
 @mock_ecs
 def test_run_task_awsvpc_network():
diff --git a/tests/test_s3/__init__.py b/tests/test_s3/__init__.py
index e69de29bb..c65c7d84f 100644
--- a/tests/test_s3/__init__.py
+++ b/tests/test_s3/__init__.py
@@ -0,0 +1,69 @@
+import boto3
+import os
+from functools import wraps
+from moto import mock_s3
+from moto.s3.responses import DEFAULT_REGION_NAME
+from uuid import uuid4
+
+
+def s3_aws_verified(func):
+    """
+    Function that is verified to work against AWS.
+    Can be run against AWS at any time by setting:
+      MOTO_TEST_ALLOW_AWS_REQUEST=true
+
+    If this environment variable is not set, the function runs in a `mock_s3` context.
+
+    This decorator will:
+      - Create a bucket
+      - Run the test and pass the bucket_name as an argument
+      - Delete the objects and the bucket itself
+    """
+
+    @wraps(func)
+    def pagination_wrapper():
+        client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+        bucket_name = str(uuid4())
+
+        allow_aws_request = (
+            os.environ.get("MOTO_TEST_ALLOW_AWS_REQUEST", "false").lower() == "true"
+        )
+
+        if allow_aws_request:
+            print(f"Test {func} will create {bucket_name}")
+            resp = create_bucket_and_test(bucket_name, client)
+        else:
+            with mock_s3():
+                resp = create_bucket_and_test(bucket_name, client)
+        return resp
+
+    def create_bucket_and_test(bucket_name, client):
+        client.create_bucket(Bucket=bucket_name)
+        client.put_bucket_tagging(
+            Bucket=bucket_name,
+            Tagging={"TagSet": [{"Key": "environment", "Value": "moto_tests"}]},
+        )
+        try:
+            resp = func(bucket_name)
+        finally:
+            ### CLEANUP ###
+
+            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")
+                )
+            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")
+                )
+            client.delete_bucket(Bucket=bucket_name)
+
+        return resp
+
+    return pagination_wrapper
diff --git a/tests/test_s3/test_s3.py b/tests/test_s3/test_s3.py
index 578c938b5..7b270d0da 100644
--- a/tests/test_s3/test_s3.py
+++ b/tests/test_s3/test_s3.py
@@ -24,6 +24,8 @@ from moto.core.utils import utcnow
 from moto.s3.responses import DEFAULT_REGION_NAME
 import moto.s3.models as s3model
 
+from . import s3_aws_verified
+
 
 class MyModel:
     def __init__(self, name, value, metadata=None):
@@ -1669,13 +1671,11 @@ def test_delete_versioned_bucket():
     client.delete_bucket(Bucket="blah")
 
 
-@mock_s3
-def test_delete_versioned_bucket_returns_metadata():
-    # Verified against AWS
-    name = str(uuid4())
+@pytest.mark.aws_verified
+@s3_aws_verified
+def test_delete_versioned_bucket_returns_metadata(name=None):
     client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
 
-    client.create_bucket(Bucket=name)
     client.put_bucket_versioning(
         Bucket=name, VersioningConfiguration={"Status": "Enabled"}
     )
@@ -2033,11 +2033,11 @@ def test_get_bucket_cors():
     assert len(resp["CORSRules"]) == 2
 
 
-@mock_s3
-def test_delete_bucket_cors():
+@pytest.mark.aws_verified
+@s3_aws_verified
+def test_delete_bucket_cors(bucket_name=None):
     s3_client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
-    bucket_name = "mybucket"
-    s3_client.create_bucket(Bucket=bucket_name)
+
     s3_client.put_bucket_cors(
         Bucket=bucket_name,
         CORSConfiguration={