Repository: getmoto/moto (python)

Issue:
update_autoscaling_group returns KeyError instead of ValidationError when ASG name is not found
Hello, 

I'm using moto to test a function that updates an autoscaling group. The tests for successful autoscaling group update works as expected. The error produced when an invalid autoscaling group name is provided, is not as expected. 

I expect a `botocore.exceptions.ClientError.ValidationError` when providing an invalid autoscaling group name. Example error.response expected:
```
{'Error': {'Type': 'Sender', 'Code': 'ValidationError', 'Message': 'AutoScalingGroup name not found - null'}, 'ResponseMetadata': {'RequestId': '<redacted>', 'HTTPStatusCode': 400, 'HTTPHeaders': {'x-amzn-requestid': '<redacted>', 'content-type': 'text/xml', 'content-length': '292', 'date': 'Wed, 14 Aug 2024 23:25:56 GMT', 'connection': 'close'}, 'RetryAttempts': 0}}
```

Instead moto throws a `KeyError`. 

Traceback:
```
pytest test/test.py
============================================= test session starts ==============================================
platform darwin -- Python 3.12.3, pytest-8.3.2, pluggy-1.5.0
rootdir: <redacted>
collected 23 items

test/test.py ....F..................                                         [100%]

=================================================== FAILURES ===================================================
______________________________________________ test_scale_asg_dne ______________________________________________

aws_session = Session(region_name='us-east-1')

    def test_scale_asg_dne(aws_session):
        client = aws_session.client("autoscaling")
        try:
>           client.update_auto_scaling_group(
                AutoScalingGroupName="fake-asg",
                MinSize=2,
                MaxSize=2,
                DesiredCapacity=2,
            )

test/test.py:151:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
lib/python3.12/site-packages/botocore/client.py:565: in _api_call
    return self._make_api_call(operation_name, kwargs)
lib/python3.12/site-packages/botocore/client.py:999: in _make_api_call
    http, parsed_response = self._make_request(
lib/python3.12/site-packages/botocore/client.py:1023: in _make_request
    return self._endpoint.make_request(operation_model, request_dict)
lib/python3.12/site-packages/botocore/endpoint.py:119: in make_request
    return self._send_request(request_dict, operation_model)
lib/python3.12/site-packages/botocore/endpoint.py:200: in _send_request
    while self._needs_retry(
lib/python3.12/site-packages/botocore/endpoint.py:352: in _needs_retry
    responses = self._event_emitter.emit(
lib/python3.12/site-packages/botocore/hooks.py:412: in emit
    return self._emitter.emit(aliased_event_name, **kwargs)
lib/python3.12/site-packages/botocore/hooks.py:256: in emit
    return self._emit(event_name, kwargs)
lib/python3.12/site-packages/botocore/hooks.py:239: in _emit
    response = handler(**kwargs)
lib/python3.12/site-packages/botocore/retryhandler.py:207: in __call__
    if self._checker(**checker_kwargs):
lib/python3.12/site-packages/botocore/retryhandler.py:284: in __call__
    should_retry = self._should_retry(
lib/python3.12/site-packages/botocore/retryhandler.py:307: in _should_retry
    return self._checker(
lib/python3.12/site-packages/botocore/retryhandler.py:363: in __call__
    checker_response = checker(
lib/python3.12/site-packages/botocore/retryhandler.py:247: in __call__
    return self._check_caught_exception(
lib/python3.12/site-packages/botocore/retryhandler.py:416: in _check_caught_exception
    raise caught_exception
lib/python3.12/site-packages/botocore/endpoint.py:276: in _do_get_response
    responses = self._event_emitter.emit(event_name, request=request)
lib/python3.12/site-packages/botocore/hooks.py:412: in emit
    return self._emitter.emit(aliased_event_name, **kwargs)
lib/python3.12/site-packages/botocore/hooks.py:256: in emit
    return self._emit(event_name, kwargs)
lib/python3.12/site-packages/botocore/hooks.py:239: in _emit
    response = handler(**kwargs)
lib/python3.12/site-packages/moto/core/botocore_stubber.py:37: in __call__
    response = self.process_request(request)
lib/python3.12/site-packages/moto/core/botocore_stubber.py:84: in process_request
    status, headers, body = method_to_execute(
lib/python3.12/site-packages/moto/core/responses.py:290: in dispatch
    return cls()._dispatch(*args, **kwargs)
lib/python3.12/site-packages/moto/core/responses.py:501: in _dispatch
    return self.call_action()
lib/python3.12/site-packages/moto/utilities/aws_headers.py:44: in _wrapper
    response = f(*args, **kwargs)
lib/python3.12/site-packages/moto/autoscaling/responses.py:19: in call_action
    return super().call_action()
lib/python3.12/site-packages/moto/core/responses.py:587: in call_action
    response = method()
lib/python3.12/site-packages/moto/autoscaling/responses.py:251: in update_auto_scaling_group
    self.autoscaling_backend.update_auto_scaling_group(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <moto.autoscaling.models.AutoScalingBackend object at 0x105cf3500>, name = 'fake-asg'
availability_zones = [], desired_capacity = 2, max_size = 2, min_size = 2, launch_config_name = None
launch_template = {}, vpc_zone_identifier = None, health_check_period = None, health_check_type = None
new_instances_protected_from_scale_in = None

    def update_auto_scaling_group(
        self,
        name: str,
        availability_zones: List[str],
        desired_capacity: Optional[int],
        max_size: Optional[int],
        min_size: Optional[int],
        launch_config_name: str,
        launch_template: Dict[str, Any],
        vpc_zone_identifier: str,
        health_check_period: int,
        health_check_type: str,
        new_instances_protected_from_scale_in: Optional[bool] = None,
    ) -> FakeAutoScalingGroup:
        """
        The parameter DefaultCooldown, PlacementGroup, TerminationPolicies are not yet implemented
        """
        # TODO: Add MixedInstancesPolicy once implemented.
        # Verify only a single launch config-like parameter is provided.
        if launch_config_name and launch_template:
            raise ValidationError(
                "Valid requests must contain either LaunchTemplate, LaunchConfigurationName "
                "or MixedInstancesPolicy parameter."
            )

>       group = self.autoscaling_groups[name]
E       KeyError: 'fake-asg'

lib/python3.12/site-packages/moto/autoscaling/models.py:1175: KeyError
```

Reproducing issue, run `pytest test.py`:
```
# test.py
import boto3
import botocore
import pytest
from moto import mock_aws

@pytest.fixture(scope="function")
def aws_credentials():
    """Mocked AWS Credentials"""
    os.environ["AWS_ACCESS_KEY_ID"] = "testing"
    os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"
    os.environ["AWS_SESSION_TOKEN"] = "testing"
    os.environ["AWS_DEFAULT_REGION"] = "us-east-1"

@pytest.fixture(scope="function")
def aws_session(aws_credentials):
    """Mocked AWS Session"""
    with mock_aws():
        yield boto3.Session(aws_credentials)

def test_scale_asg_dne(aws_session):
    client = aws_session.client("autoscaling")
    try:
        client.update_auto_scaling_group(
            AutoScalingGroupName="fake-asg",
            MinSize=2,
            MaxSize=2,
            DesiredCapacity=2,
        ) 
   # throws a KeyError instead of ValidationError
    except botocore.exceptions.ClientError as err:
      if err.response["Error"]["Code"] == "ValidationError":
        print("A validation error occurred")
```

Environment details:
 - moto==5.0.12 installed with pip
 - using mocks. boto3==1.34.155, botocore==1.34.155
 - pytest==8.3.2
 - Python 3.12.3

Thank you!

Pull request description:
Autoscaling: update_group() now validates that the Group exists
Fixes #7980 

Patch:
diff --git a/moto/autoscaling/models.py b/moto/autoscaling/models.py
index c1265c795..7ff98b502 100644
--- a/moto/autoscaling/models.py
+++ b/moto/autoscaling/models.py
@@ -1171,6 +1171,8 @@ class AutoScalingBackend(BaseBackend):
                 "Valid requests must contain either LaunchTemplate, LaunchConfigurationName "
                 "or MixedInstancesPolicy parameter."
             )
+        if name not in self.autoscaling_groups:
+            raise ValidationError("AutoScalingGroup name not found - null")
 
         group = self.autoscaling_groups[name]
         group.update(
diff --git a/moto/emr/models.py b/moto/emr/models.py
index fc14a121a..5922d17a2 100644
--- a/moto/emr/models.py
+++ b/moto/emr/models.py
@@ -10,10 +10,9 @@ from moto.emr.exceptions import (
     ResourceNotFoundException,
     ValidationException,
 )
-from moto.utilities.utils import get_partition
+from moto.utilities.utils import CamelToUnderscoresWalker, get_partition
 
 from .utils import (
-    CamelToUnderscoresWalker,
     EmrSecurityGroupManager,
     random_cluster_id,
     random_instance_group_id,
diff --git a/moto/emr/utils.py b/moto/emr/utils.py
index 18f070c85..8b3a08226 100644
--- a/moto/emr/utils.py
+++ b/moto/emr/utils.py
@@ -3,10 +3,7 @@ import re
 import string
 from typing import Any, Dict, Iterator, List, Tuple
 
-from moto.core.utils import (
-    camelcase_to_underscores,
-    iso_8601_datetime_with_milliseconds,
-)
+from moto.core.utils import iso_8601_datetime_with_milliseconds
 from moto.moto_api._internal import mock_random as random
 
 
@@ -129,37 +126,6 @@ class Unflattener:
             return len(container) >= i
 
 
-class CamelToUnderscoresWalker:
-    """A class to convert the keys in dict/list hierarchical data structures from CamelCase to snake_case (underscores)"""
-
-    @staticmethod
-    def parse(x: Any) -> Any:  # type: ignore[misc]
-        if isinstance(x, dict):
-            return CamelToUnderscoresWalker.parse_dict(x)
-        elif isinstance(x, list):
-            return CamelToUnderscoresWalker.parse_list(x)
-        else:
-            return CamelToUnderscoresWalker.parse_scalar(x)
-
-    @staticmethod
-    def parse_dict(x: Dict[str, Any]) -> Dict[str, Any]:  # type: ignore[misc]
-        temp = {}
-        for key in x.keys():
-            temp[camelcase_to_underscores(key)] = CamelToUnderscoresWalker.parse(x[key])
-        return temp
-
-    @staticmethod
-    def parse_list(x: Any) -> Any:  # type: ignore[misc]
-        temp = []
-        for i in x:
-            temp.append(CamelToUnderscoresWalker.parse(i))
-        return temp
-
-    @staticmethod
-    def parse_scalar(x: Any) -> Any:  # type: ignore[misc]
-        return x
-
-
 class ReleaseLabel:
     version_re = re.compile(r"^emr-(\d+)\.(\d+)\.(\d+)$")
 
diff --git a/moto/events/models.py b/moto/events/models.py
index 1efe0e359..9b43ba579 100644
--- a/moto/events/models.py
+++ b/moto/events/models.py
@@ -7,7 +7,7 @@ from collections import OrderedDict
 from enum import Enum, unique
 from json import JSONDecodeError
 from operator import eq, ge, gt, le, lt
-from typing import Any, Dict, List, Optional
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
 
 import requests
 
@@ -29,18 +29,32 @@ from moto.events.exceptions import (
     ValidationException,
 )
 from moto.moto_api._internal import mock_random as random
-from moto.secretsmanager import secretsmanager_backends
 from moto.utilities.arns import parse_arn
 from moto.utilities.paginator import paginate
 from moto.utilities.tagging_service import TaggingService
-from moto.utilities.utils import ARN_PARTITION_REGEX, get_partition
+from moto.utilities.utils import (
+    ARN_PARTITION_REGEX,
+    CamelToUnderscoresWalker,
+    get_partition,
+)
 
 from .utils import _BASE_EVENT_MESSAGE, PAGINATION_MODEL, EventMessageType
 
+if TYPE_CHECKING:
+    from moto.secretsmanager.models import SecretsManagerBackend
+
 # Sentinel to signal the absence of a field for `Exists` pattern matching
 UNDEFINED = object()
 
 
+def get_secrets_manager_backend(
+    account_id: str, region: str
+) -> "SecretsManagerBackend":
+    from moto.secretsmanager import secretsmanager_backends
+
+    return secretsmanager_backends[account_id][region]
+
+
 class Rule(CloudFormationModel):
     def __init__(
         self,
@@ -768,15 +782,23 @@ class Connection(BaseModel):
         self.state = "AUTHORIZED"
 
         connection_id = f"{self.name}/{self.uuid}"
-        secretsmanager_backend = secretsmanager_backends[account_id][region_name]
+        secretsmanager_backend = get_secrets_manager_backend(account_id, region_name)
+        secret_value = {}
+        for key, value in self.auth_parameters.items():
+            if key == "InvocationHttpParameters":
+                secret_value.update({"InvocationHttpParameters": value})
+            else:
+                secret_value.update(value)
+        secret_value = CamelToUnderscoresWalker.parse(secret_value)
+
         secret = secretsmanager_backend.create_secret(
-            name=f"{connection_id}/auth",
-            secret_string=json.dumps(self.auth_parameters),
-            replica_regions=list(),
+            name=f"events!connection/{connection_id}/auth",
+            secret_string=json.dumps(secret_value),
+            replica_regions=[],
             force_overwrite=False,
             secret_binary=None,
             description=f"Auth parameters for Eventbridge connection {connection_id}",
-            tags=list(),
+            tags=[{"Key": "aws:secretsmanager:owningService", "Value": "events"}],
             kms_key_id=None,
             client_request_token=None,
         )
@@ -817,7 +839,6 @@ class Connection(BaseModel):
             - The original response also has:
                 - LastAuthorizedTime (number)
                 - LastModifiedTime (number)
-                - SecretArn (string)
                 - StateReason (string)
             - At the time of implementing this, there was no place where to set/get
             those attributes. That is why they are not in the response.
@@ -1842,6 +1863,16 @@ class EventsBackend(BaseBackend):
         if not connection:
             raise ResourceNotFoundException(f"Connection '{name}' does not exist.")
 
+        # Delete Secret
+        secretsmanager_backend = get_secrets_manager_backend(
+            self.account_id, self.region_name
+        )
+        secretsmanager_backend.delete_secret(
+            secret_id=connection.secret_arn,
+            recovery_window_in_days=None,
+            force_delete_without_recovery=True,
+        )
+
         return connection.describe_short()
 
     def create_api_destination(
diff --git a/moto/secretsmanager/models.py b/moto/secretsmanager/models.py
index 6f38516aa..5a3cf7f05 100644
--- a/moto/secretsmanager/models.py
+++ b/moto/secretsmanager/models.py
@@ -914,7 +914,7 @@ class SecretsManagerBackend(BaseBackend):
     def delete_secret(
         self,
         secret_id: str,
-        recovery_window_in_days: int,
+        recovery_window_in_days: Optional[int],
         force_delete_without_recovery: bool,
     ) -> Tuple[str, str, float]:
         if recovery_window_in_days is not None and (
diff --git a/moto/utilities/utils.py b/moto/utilities/utils.py
index 77fae680e..8fd3b1641 100644
--- a/moto/utilities/utils.py
+++ b/moto/utilities/utils.py
@@ -123,3 +123,36 @@ class LowercaseDict(MutableMapping[str, Any]):
 
     def _keytransform(self, key: str) -> str:
         return key.lower()
+
+
+class CamelToUnderscoresWalker:
+    """A class to convert the keys in dict/list hierarchical data structures from CamelCase to snake_case (underscores)"""
+
+    @staticmethod
+    def parse(x: Any) -> Any:  # type: ignore[misc]
+        if isinstance(x, dict):
+            return CamelToUnderscoresWalker.parse_dict(x)
+        elif isinstance(x, list):
+            return CamelToUnderscoresWalker.parse_list(x)
+        else:
+            return CamelToUnderscoresWalker.parse_scalar(x)
+
+    @staticmethod
+    def parse_dict(x: Dict[str, Any]) -> Dict[str, Any]:  # type: ignore[misc]
+        from moto.core.utils import camelcase_to_underscores
+
+        temp = {}
+        for key in x.keys():
+            temp[camelcase_to_underscores(key)] = CamelToUnderscoresWalker.parse(x[key])
+        return temp
+
+    @staticmethod
+    def parse_list(x: Any) -> Any:  # type: ignore[misc]
+        temp = []
+        for i in x:
+            temp.append(CamelToUnderscoresWalker.parse(i))
+        return temp
+
+    @staticmethod
+    def parse_scalar(x: Any) -> Any:  # type: ignore[misc]
+        return x


Test patch:
diff --git a/tests/test_autoscaling/test_autoscaling.py b/tests/test_autoscaling/test_autoscaling.py
index a052b8640..614b75e79 100644
--- a/tests/test_autoscaling/test_autoscaling.py
+++ b/tests/test_autoscaling/test_autoscaling.py
@@ -7,7 +7,7 @@ from botocore.exceptions import ClientError
 
 from moto import mock_aws
 from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
-from tests import EXAMPLE_AMI_ID
+from tests import EXAMPLE_AMI_ID, aws_verified
 
 from .utils import setup_instance_with_networking, setup_networking
 
@@ -651,6 +651,22 @@ def test_update_autoscaling_group_max_size_desired_capacity_change():
     assert len(group["Instances"]) == 5
 
 
+@aws_verified
+@pytest.mark.aws_verified
+def test_update_unknown_group():
+    client = boto3.client("autoscaling", region_name="us-east-1")
+    with pytest.raises(ClientError) as exc:
+        client.update_auto_scaling_group(
+            AutoScalingGroupName="fake-asg",
+            MinSize=2,
+            MaxSize=2,
+            DesiredCapacity=2,
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "ValidationError"
+    assert err["Message"] == "AutoScalingGroup name not found - null"
+
+
 @mock_aws
 def test_autoscaling_describe_policies():
     mocked_networking = setup_networking()
diff --git a/tests/test_events/test_events.py b/tests/test_events/test_events.py
index 995cac188..b2fe8dd78 100644
--- a/tests/test_events/test_events.py
+++ b/tests/test_events/test_events.py
@@ -3,6 +3,8 @@ import random
 import unittest
 import warnings
 from datetime import datetime, timezone
+from time import sleep
+from uuid import uuid4
 
 import boto3
 import pytest
@@ -11,6 +13,7 @@ from botocore.exceptions import ClientError
 from moto import mock_aws, settings
 from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
 from moto.core.utils import iso_8601_datetime_without_milliseconds
+from tests import aws_verified
 
 RULES = [
     {"Name": "test1", "ScheduleExpression": "rate(5 minutes)"},
@@ -2312,6 +2315,95 @@ def test_create_and_update_connection():
     assert "CreationTime" in description
 
 
+@aws_verified
+@pytest.mark.aws_verified
+@pytest.mark.parametrize(
+    "auth_type,auth_parameters",
+    [
+        (
+            "API_KEY",
+            {"ApiKeyAuthParameters": {"ApiKeyName": "test", "ApiKeyValue": "test"}},
+        ),
+        ("BASIC", {"BasicAuthParameters": {"Username": "un", "Password": "pw"}}),
+    ],
+    ids=["auth_params", "basic_auth_params"],
+)
+@pytest.mark.parametrize(
+    "with_headers", [True, False], ids=["with_headers", "without_headers"]
+)
+def test_kms_key_is_created(auth_type, auth_parameters, with_headers):
+    client = boto3.client("events", "us-east-1")
+    secrets = boto3.client("secretsmanager", "us-east-1")
+    sts = boto3.client("sts", "us-east-1")
+
+    name = f"event_{str(uuid4())[0:6]}"
+    account_id = sts.get_caller_identity()["Account"]
+    connection_deleted = False
+    if with_headers:
+        auth_parameters["InvocationHttpParameters"] = {
+            "HeaderParameters": [
+                {"Key": "k1", "Value": "v1", "IsValueSecret": True},
+                {"Key": "k2", "Value": "v2", "IsValueSecret": False},
+            ]
+        }
+    else:
+        auth_parameters.pop("InvocationHttpParameters", None)
+
+    client.create_connection(
+        Name=name,
+        AuthorizationType=auth_type,
+        AuthParameters=auth_parameters,
+    )
+    try:
+        description = client.describe_connection(Name=name)
+        secret_arn = description["SecretArn"]
+        assert secret_arn.startswith(
+            f"arn:aws:secretsmanager:us-east-1:{account_id}:secret:events!connection/{name}/"
+        )
+
+        secret = secrets.describe_secret(SecretId=secret_arn)
+        assert secret["Name"].startswith(f"events!connection/{name}/")
+        assert secret["Tags"] == [
+            {"Key": "aws:secretsmanager:owningService", "Value": "events"}
+        ]
+
+        if auth_type == "BASIC":
+            expected_secret = {"username": "un", "password": "pw"}
+        else:
+            expected_secret = {"api_key_name": "test", "api_key_value": "test"}
+        if with_headers:
+            expected_secret["invocation_http_parameters"] = {
+                "header_parameters": [
+                    {"key": "k1", "value": "v1", "is_value_secret": True},
+                    {"key": "k2", "value": "v2", "is_value_secret": False},
+                ]
+            }
+
+        secret_value = secrets.get_secret_value(SecretId=secret_arn)
+        assert json.loads(secret_value["SecretString"]) == expected_secret
+
+        client.delete_connection(Name=name)
+        connection_deleted = True
+
+        secret_deleted = False
+        attempts = 0
+        while not secret_deleted and attempts < 5:
+            try:
+                attempts += 1
+                secrets.describe_secret(SecretId=secret_arn)
+                sleep(1)
+            except ClientError as e:
+                secret_deleted = (
+                    e.response["Error"]["Code"] == "ResourceNotFoundException"
+                )
+
+        if not secret_deleted:
+            assert False, f"Should have automatically deleted secret {secret_arn}"
+    finally:
+        if not connection_deleted:
+            client.delete_connection(Name=name)
+
+
 @mock_aws
 def test_update_unknown_connection():
     client = boto3.client("events", "eu-north-1")