Repository: getmoto/moto

Issue:
responses_mock.remove() causes infinite loop
`moto.core.responses_custom_registry.CustomRegistry` does not implement `remove()` method,  code will fallback to inherited `responses.registries.FirstMatchRegistry`

test env is current moto master (release 5.0.25) using `make init`

here is a test case which fails due to infinite loop
```
import moto
import requests
import responses
from unittest import TestCase


class TestBad(TestCase):

    @responses.activate
    @moto.mock_aws
    def test_bad(self):
        from moto.core.models import responses_mock

        responses_mock.get(url="https://example.com/", body="test")
        resp = requests.get("https://example.com/")
        assert resp.text == "test"

        # causes infinite loop
        responses_mock.remove("GET", "https://example.com/")


if __name__ == "__main__":
    import unittest
    unittest.main()
```

suggested fix
```
    def remove(self, response: responses.BaseResponse) -> list[responses.BaseResponse]:
        removed_responses = []
        registered = self._registered[response.method]
        while response in registered:
            registered.remove(response)
            removed_responses.append(response)
        return removed_responses
```

Hints from issue comments:
@ralphie0112358: also, .replace is not implemented.  suggested implementation
```
def replace(self, response: responses.BaseResponse) -> responses.BaseResponse:
    registered = self._registered[response.method]
    try:
        index = registered.index(response)
    except ValueError:
        raise ValueError(f"Response is not registered for URL {response.url}")
    registered[index] = response
    return response
```@bblommers: Hi @ralphie0112358, thanks for raising this! I've opened a PR to fix this.

Solution patch:
diff --git a/moto/core/responses_custom_registry.py b/moto/core/responses_custom_registry.py
index 64c9422df..13db30a06 100644
--- a/moto/core/responses_custom_registry.py
+++ b/moto/core/responses_custom_registry.py
@@ -32,6 +32,23 @@ class CustomRegistry(responses.registries.FirstMatchRegistry):
         self._registered[response.method].append(response)
         return response
 
+    def replace(self, response: responses.BaseResponse) -> responses.BaseResponse:
+        registered = self._registered[response.method]
+        try:
+            index = registered.index(response)
+        except ValueError:
+            raise ValueError(f"Response is not registered for URL {response.url}")
+        registered[index] = response
+        return response
+
+    def remove(self, response: responses.BaseResponse) -> List[responses.BaseResponse]:
+        removed_responses = []
+        registered = self._registered[response.method]
+        while response in registered:
+            registered.remove(response)
+            removed_responses.append(response)
+        return removed_responses
+
     def reset(self) -> None:
         self._registered.clear()
 
diff --git a/moto/elasticache/exceptions.py b/moto/elasticache/exceptions.py
index 876406b4e..0266a151f 100644
--- a/moto/elasticache/exceptions.py
+++ b/moto/elasticache/exceptions.py
@@ -43,6 +43,26 @@ class PasswordRequired(ElastiCacheException):
         )
 
 
+class InvalidParameterValueException(ElastiCacheException):
+    code = 404
+
+    def __init__(self, message: str) -> None:
+        super().__init__(
+            "InvalidParameterValue",
+            message=message,
+        )
+
+
+class InvalidParameterCombinationException(ElastiCacheException):
+    code = 404
+
+    def __init__(self, message: str) -> None:
+        super().__init__(
+            "InvalidParameterCombination",
+            message=message,
+        )
+
+
 class UserAlreadyExists(ElastiCacheException):
     code = 404
 
diff --git a/moto/elasticache/models.py b/moto/elasticache/models.py
index ffade5875..6861d1bdd 100644
--- a/moto/elasticache/models.py
+++ b/moto/elasticache/models.py
@@ -12,10 +12,12 @@ from .exceptions import (
     CacheClusterAlreadyExists,
     CacheClusterNotFound,
     InvalidARNFault,
+    InvalidParameterCombinationException,
+    InvalidParameterValueException,
     UserAlreadyExists,
     UserNotFound,
 )
-from .utils import PAGINATION_MODEL
+from .utils import PAGINATION_MODEL, AuthenticationTypes
 
 
 class User(BaseModel):
@@ -29,10 +31,12 @@ class User(BaseModel):
         engine: str,
         no_password_required: bool,
         passwords: Optional[List[str]] = None,
+        authentication_type: Optional[str] = None,
     ):
         self.id = user_id
         self.name = user_name
         self.engine = engine
+
         self.passwords = passwords or []
         self.access_string = access_string
         self.no_password_required = no_password_required
@@ -41,6 +45,7 @@ class User(BaseModel):
         self.usergroupids: List[str] = []
         self.region = region
         self.arn = f"arn:{get_partition(self.region)}:elasticache:{self.region}:{account_id}:user:{self.id}"
+        self.authentication_type = authentication_type
 
 
 class CacheCluster(BaseModel):
@@ -161,9 +166,34 @@ class ElastiCacheBackend(BaseBackend):
         passwords: List[str],
         access_string: str,
         no_password_required: bool,
+        authentication_type: str,  # contain it to the str in the enums TODO
     ) -> User:
         if user_id in self.users:
             raise UserAlreadyExists
+
+        if authentication_type not in AuthenticationTypes._value2member_map_:
+            raise InvalidParameterValueException(
+                f"Input Authentication type: {authentication_type} is not in the allowed list: [password,no-password-required,iam]"
+            )
+
+        if (
+            no_password_required
+            and authentication_type != AuthenticationTypes.NOPASSWORD
+        ):
+            raise InvalidParameterCombinationException(
+                f"No password required flag is true but provided authentication type is {authentication_type}"
+            )
+
+        if passwords and authentication_type != AuthenticationTypes.PASSWORD:
+            raise InvalidParameterCombinationException(
+                f"Password field is not allowed with authentication type: {authentication_type}"
+            )
+
+        if not passwords and authentication_type == AuthenticationTypes.PASSWORD:
+            raise InvalidParameterCombinationException(
+                "A user with Authentication Mode: password, must have at least one password"
+            )
+
         user = User(
             account_id=self.account_id,
             region=self.region_name,
@@ -173,6 +203,7 @@ class ElastiCacheBackend(BaseBackend):
             passwords=passwords,
             access_string=access_string,
             no_password_required=no_password_required,
+            authentication_type=authentication_type,
         )
         self.users[user_id] = user
         return user
diff --git a/moto/elasticache/responses.py b/moto/elasticache/responses.py
index 225d2b7c8..82d6ecc0b 100644
--- a/moto/elasticache/responses.py
+++ b/moto/elasticache/responses.py
@@ -1,7 +1,12 @@
 from moto.core.responses import BaseResponse
 
-from .exceptions import PasswordRequired, PasswordTooShort
+from .exceptions import (
+    InvalidParameterCombinationException,
+    InvalidParameterValueException,
+    PasswordTooShort,
+)
 from .models import ElastiCacheBackend, elasticache_backends
+from .utils import AuthenticationTypes
 
 
 class ElastiCacheResponse(BaseResponse):
@@ -21,12 +26,41 @@ class ElastiCacheResponse(BaseResponse):
         user_name = params.get("UserName")
         engine = params.get("Engine")
         passwords = params.get("Passwords", [])
-        no_password_required = self._get_bool_param("NoPasswordRequired", False)
-        password_required = not no_password_required
-        if password_required and not passwords:
-            raise PasswordRequired
+        no_password_required = self._get_bool_param("NoPasswordRequired")
+        authentication_mode = params.get("AuthenticationMode")
+        authentication_type = "null"
+
+        if no_password_required is not None:
+            authentication_type = (
+                AuthenticationTypes.NOPASSWORD.value
+                if no_password_required
+                else AuthenticationTypes.PASSWORD.value
+            )
+
+        if passwords:
+            authentication_type = AuthenticationTypes.PASSWORD.value
+
+        if authentication_mode:
+            for key in authentication_mode.keys():
+                if key not in ["Type", "Passwords"]:
+                    raise InvalidParameterValueException(
+                        f'Unknown parameter in AuthenticationMode: "{key}", must be one of: Type, Passwords'
+                    )
+
+            authentication_type = authentication_mode.get("Type")
+            authentication_passwords = authentication_mode.get("Passwords", [])
+
+            if passwords and authentication_passwords:
+                raise InvalidParameterCombinationException(
+                    "Passwords provided via multiple arguments. Use only one argument"
+                )
+
+            # if passwords is empty, then we can use the authentication_passwords
+            passwords = passwords if passwords else authentication_passwords
+
         if any([len(p) < 16 for p in passwords]):
             raise PasswordTooShort
+
         access_string = params.get("AccessString")
         user = self.elasticache_backend.create_user(
             user_id=user_id,  # type: ignore[arg-type]
@@ -35,6 +69,7 @@ class ElastiCacheResponse(BaseResponse):
             passwords=passwords,
             access_string=access_string,  # type: ignore[arg-type]
             no_password_required=no_password_required,
+            authentication_type=authentication_type,
         )
         template = self.response_template(CREATE_USER_TEMPLATE)
         return template.render(user=user)
@@ -167,7 +202,9 @@ USER_TEMPLATE = """<UserId>{{ user.id }}</UserId>
       {% if user.no_password_required %}
       <Type>no-password</Type>
       {% else %}
-      <Type>password</Type>
+      <Type>{{ user.authentication_type }}</Type>
+      {% endif %}
+      {% if user.passwords %}
       <PasswordCount>{{ user.passwords|length }}</PasswordCount>
       {% endif %}
     </Authentication>
diff --git a/moto/elasticache/utils.py b/moto/elasticache/utils.py
index f0ac58c20..9377e997d 100644
--- a/moto/elasticache/utils.py
+++ b/moto/elasticache/utils.py
@@ -1,3 +1,5 @@
+from enum import Enum
+
 PAGINATION_MODEL = {
     "describe_cache_clusters": {
         "input_token": "marker",
@@ -6,3 +8,9 @@ PAGINATION_MODEL = {
         "unique_attribute": "cache_cluster_id",
     },
 }
+
+
+class AuthenticationTypes(str, Enum):
+    NOPASSWORD = "no-password-required"
+    PASSWORD = "password"
+    IAM = "iam"


Test patch:
diff --git a/tests/test_core/test_request_mocking.py b/tests/test_core/test_request_mocking.py
index 3865ba679..d5ea67d88 100644
--- a/tests/test_core/test_request_mocking.py
+++ b/tests/test_core/test_request_mocking.py
@@ -1,8 +1,12 @@
+from unittest import SkipTest
+
 import boto3
 import pytest
 import requests
+from responses import Response
 
 from moto import mock_aws, settings
+from moto.core.models import responses_mock
 
 
 @mock_aws
@@ -48,3 +52,20 @@ def test_decorator_ordering() -> None:
 
     resp = requests.get(presigned_url)
     assert resp.status_code == 200
+
+
+@mock_aws()
+def test_replace_and_remove_mock() -> None:
+    if not settings.TEST_DECORATOR_MODE:
+        raise SkipTest("Only need to test responses mock in decorator mode")
+    rsp1 = Response(method="GET", url="http://example.com", body="test")
+    responses_mock.add(rsp1)
+
+    assert requests.get("http://example.com/").text == "test"
+
+    rsp2 = Response(method="GET", url="http://example.com", body="test2")
+    responses_mock.replace(rsp2)
+
+    assert requests.get("http://example.com/").text == "test2"
+
+    responses_mock.remove("GET", "https://example.com/")
diff --git a/tests/test_elasticache/test_elasticache.py b/tests/test_elasticache/test_elasticache.py
index d7b47934b..79c387d5a 100644
--- a/tests/test_elasticache/test_elasticache.py
+++ b/tests/test_elasticache/test_elasticache.py
@@ -1,6 +1,6 @@
 import boto3
 import pytest
-from botocore.exceptions import ClientError
+from botocore.exceptions import ClientError, ParamValidationError
 
 from moto import mock_aws
 from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
@@ -89,10 +89,279 @@ def test_create_user_without_password():
     assert err["Code"] == "InvalidParameterValue"
     assert (
         err["Message"]
-        == "No password was provided. If you want to create/update the user without password, please use the NoPasswordRequired flag."
+        == "Input Authentication type: null is not in the allowed list: [password,no-password-required,iam]"
     )
 
 
+@mock_aws
+def test_create_user_with_iam():
+    client = boto3.client("elasticache", region_name="us-east-1")
+    user_id = "user1"
+    resp = client.create_user(
+        UserId=user_id,
+        UserName="User1",
+        Engine="Redis",
+        AccessString="on ~* +@all",
+        AuthenticationMode={"Type": "iam"},
+    )
+
+    assert resp["Status"] == "active"
+    assert resp["Engine"] == "Redis"
+    assert resp["AccessString"] == "on ~* +@all"
+    assert resp["UserGroupIds"] == []
+    assert resp["Authentication"]["Type"] == "iam"
+
+
+@mock_aws
+def test_create_user_invalid_authentication_type():
+    client = boto3.client("elasticache", region_name="us-east-1")
+    with pytest.raises(ClientError) as exc:
+        client.create_user(
+            UserId="user1",
+            UserName="User1",
+            Engine="Redis",
+            AccessString="?",
+            AuthenticationMode={"Type": "invalidtype"},
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "InvalidParameterValue"
+    assert (
+        err["Message"]
+        == "Input Authentication type: invalidtype is not in the allowed list: [password,no-password-required,iam]"
+    )
+
+
+@mock_aws
+def test_create_user_with_iam_with_passwords():
+    # IAM authentication mode should not come with password fields
+    client = boto3.client("elasticache", region_name="us-east-1")
+    with pytest.raises(ClientError) as exc:
+        client.create_user(
+            UserId="user1",
+            UserName="user1",
+            Engine="Redis",
+            AccessString="?",
+            AuthenticationMode={"Type": "iam"},
+            Passwords=["mysecretpassthatsverylong"],
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "InvalidParameterCombination"
+    assert (
+        err["Message"] == "Password field is not allowed with authentication type: iam"
+    )
+
+
+# handled by botocore
+@mock_aws
+def test_create_user_with_invalid_authmode_key():
+    client = boto3.client("elasticache", region_name="us-east-1")
+    with pytest.raises(ParamValidationError) as exc:
+        client.create_user(
+            UserId="user1",
+            UserName="user1",
+            Engine="Redis",
+            AccessString="?",
+            AuthenticationMode={"invalidkey": ""},
+        )
+    assert (
+        exc.value.kwargs["report"]
+        == 'Unknown parameter in AuthenticationMode: "invalidkey", must be one of: Type, Passwords'
+    )
+
+
+# handled by botocore
+@mock_aws
+def test_create_user_with_empty_passwords_with_password():
+    client = boto3.client("elasticache", region_name="us-east-1")
+    with pytest.raises(ParamValidationError) as exc:
+        client.create_user(
+            UserId="user1",
+            UserName="user1",
+            Engine="Redis",
+            AccessString="?",
+            Passwords=[],
+            AuthenticationMode={"Type": "password"},
+        )
+    assert (
+        exc.value.kwargs["report"]
+        == "Invalid length for parameter Passwords, value: 0, valid min length: 1"
+    )
+
+
+# handled by botocore
+@mock_aws
+def test_create_user_with_empty_passwords_with_authmode_password():
+    client = boto3.client("elasticache", region_name="us-east-1")
+    with pytest.raises(ParamValidationError) as exc:
+        client.create_user(
+            UserId="user1",
+            UserName="user1",
+            Engine="Redis",
+            AccessString="?",
+            AuthenticationMode={"Type": "password", "Passwords": []},
+        )
+
+    assert (
+        exc.value.kwargs["report"]
+        == "Invalid length for parameter AuthenticationMode.Passwords, value: 0, valid min length: 1"
+    )
+
+
+@mock_aws
+def test_create_user_authmode_password_with_multiple_password_fields():
+    client = boto3.client("elasticache", region_name="us-east-1")
+    with pytest.raises(ClientError) as exc:
+        client.create_user(
+            UserId="user1",
+            UserName="user1",
+            Engine="Redis",
+            AccessString="on ~* +@all",
+            AuthenticationMode={"Type": "password", "Passwords": ["authmodepassword"]},
+            Passwords=["requestpassword"],
+            NoPasswordRequired=False,
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "InvalidParameterCombination"
+    assert (
+        err["Message"]
+        == "Passwords provided via multiple arguments. Use only one argument"
+    )
+
+
+@mock_aws
+def test_create_user_with_authmode_password_without_passwords():
+    client = boto3.client("elasticache", region_name="us-east-1")
+    with pytest.raises(ClientError) as exc:
+        client.create_user(
+            UserId="user1",
+            UserName="user1",
+            Engine="Redis",
+            AccessString="?",
+            AuthenticationMode={"Type": "password"},
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "InvalidParameterCombination"
+    assert (
+        err["Message"]
+        == "A user with Authentication Mode: password, must have at least one password"
+    )
+
+
+@mock_aws
+def test_create_user_with_authmode_no_password():
+    # AuthenticationMode should be either 'password' or 'iam' or 'no-password'
+    client = boto3.client("elasticache", region_name="us-east-1")
+    user_id = "user1"
+    resp = client.create_user(
+        UserId=user_id,
+        UserName="User1",
+        Engine="Redis",
+        AccessString="on ~* +@all",
+        AuthenticationMode={"Type": "no-password-required"},
+    )
+
+    assert resp["Status"] == "active"
+    assert resp["Engine"] == "Redis"
+    assert resp["AccessString"] == "on ~* +@all"
+    assert resp["UserGroupIds"] == []
+    assert resp["Authentication"]["Type"] == "no-password-required"
+    assert (
+        "PasswordCount" not in resp["Authentication"]
+    )  # even though optional, we don't expect it for no-password-required
+
+
+@mock_aws
+def test_create_user_with_no_password_required_and_authmode_nopassword():
+    user_id = "user1"
+    client = boto3.client("elasticache", region_name="us-east-1")
+    resp = client.create_user(
+        UserId=user_id,
+        UserName="User1",
+        Engine="Redis",
+        AccessString="on ~* +@all",
+        NoPasswordRequired=True,
+        AuthenticationMode={"Type": "no-password-required"},
+    )
+
+    assert resp["Status"] == "active"
+    assert resp["Engine"] == "Redis"
+    assert resp["AccessString"] == "on ~* +@all"
+    assert resp["UserGroupIds"] == []
+    assert resp["Authentication"]["Type"] == "no-password"
+    assert (
+        "PasswordCount" not in resp["Authentication"]
+    )  # even though optional, we don't expect it for no-password-required
+
+
+@mock_aws
+def test_create_user_with_no_password_required_and_authmode_different():
+    for auth_mode in ["password", "iam"]:
+        client = boto3.client("elasticache", region_name="ap-southeast-1")
+        with pytest.raises(ClientError) as exc:
+            client.create_user(
+                UserId="user1",
+                UserName="user1",
+                Engine="Redis",
+                AccessString="on ~* +@all",
+                NoPasswordRequired=True,
+                AuthenticationMode={"Type": auth_mode},
+            )
+        err = exc.value.response["Error"]
+        assert err["Code"] == "InvalidParameterCombination"
+        assert (
+            err["Message"]
+            == f"No password required flag is true but provided authentication type is {auth_mode}"
+        )
+
+
+@mock_aws
+def test_create_user_with_authmode_password():
+    # AuthenticationMode should be either 'password' or 'iam' or 'no-password'
+    client = boto3.client("elasticache", region_name="us-east-1")
+    user_id = "user1"
+    resp = client.create_user(
+        UserId=user_id,
+        UserName="User1",
+        Engine="Redis",
+        AccessString="on ~* +@all",
+        AuthenticationMode={
+            "Type": "password",
+            "Passwords": ["mysecretpassthatsverylong"],
+        },
+    )
+
+    assert resp["Status"] == "active"
+    assert resp["Engine"] == "Redis"
+    assert resp["AccessString"] == "on ~* +@all"
+    assert resp["UserGroupIds"] == []
+    assert resp["Authentication"]["Type"] == "password"
+    assert resp["Authentication"]["PasswordCount"] == 1
+
+
+@mock_aws
+def test_create_user_with_authmode_password_multiple():
+    # AuthenticationMode should be either 'password' or 'iam' or 'no-password'
+    client = boto3.client("elasticache", region_name="us-east-1")
+    user_id = "user1"
+    resp = client.create_user(
+        UserId=user_id,
+        UserName="User1",
+        Engine="Redis",
+        AccessString="on ~* +@all",
+        AuthenticationMode={
+            "Type": "password",
+            "Passwords": ["mysecretpassthatsverylong", "mysecretpassthatsverylong2"],
+        },
+    )
+
+    assert resp["Status"] == "active"
+    assert resp["Engine"] == "Redis"
+    assert resp["AccessString"] == "on ~* +@all"
+    assert resp["UserGroupIds"] == []
+    assert resp["Authentication"]["Type"] == "password"
+    assert resp["Authentication"]["PasswordCount"] == 2
+
+
 @mock_aws
 def test_create_user_twice():
     client = boto3.client("elasticache", region_name="ap-southeast-1")