Techdebt: MyPy ElastiCache (#5970)
This commit is contained in:
parent
3a7f10fece
commit
4d6271d1bd
@ -1,3 +1,4 @@
|
||||
from typing import Any
|
||||
from moto.core.exceptions import RESTError
|
||||
|
||||
EXCEPTION_RESPONSE = """<?xml version="1.0"?>
|
||||
@ -15,21 +16,20 @@ class ElastiCacheException(RESTError):
|
||||
|
||||
code = 400
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
def __init__(self, code: str, message: str, **kwargs: Any):
|
||||
kwargs.setdefault("template", "ecerror")
|
||||
self.templates["ecerror"] = EXCEPTION_RESPONSE
|
||||
super().__init__(*args, **kwargs)
|
||||
super().__init__(code, message)
|
||||
|
||||
|
||||
class PasswordTooShort(ElastiCacheException):
|
||||
|
||||
code = 404
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
def __init__(self) -> None:
|
||||
super().__init__(
|
||||
"InvalidParameterValue",
|
||||
message="Passwords length must be between 16-128 characters.",
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
|
||||
@ -37,11 +37,10 @@ class PasswordRequired(ElastiCacheException):
|
||||
|
||||
code = 404
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
def __init__(self) -> None:
|
||||
super().__init__(
|
||||
"InvalidParameterValue",
|
||||
message="No password was provided. If you want to create/update the user without password, please use the NoPasswordRequired flag.",
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
|
||||
@ -49,15 +48,13 @@ class UserAlreadyExists(ElastiCacheException):
|
||||
|
||||
code = 404
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
super().__init__(
|
||||
"UserAlreadyExists", message="User user1 already exists.", **kwargs
|
||||
)
|
||||
def __init__(self) -> None:
|
||||
super().__init__("UserAlreadyExists", message="User user1 already exists.")
|
||||
|
||||
|
||||
class UserNotFound(ElastiCacheException):
|
||||
|
||||
code = 404
|
||||
|
||||
def __init__(self, user_id, **kwargs):
|
||||
super().__init__("UserNotFound", message=f"User {user_id} not found.", **kwargs)
|
||||
def __init__(self, user_id: str):
|
||||
super().__init__("UserNotFound", message=f"User {user_id} not found.")
|
||||
|
@ -1,3 +1,4 @@
|
||||
from typing import List, Optional
|
||||
from moto.core import BaseBackend, BackendDict, BaseModel
|
||||
|
||||
from .exceptions import UserAlreadyExists, UserNotFound
|
||||
@ -6,14 +7,14 @@ from .exceptions import UserAlreadyExists, UserNotFound
|
||||
class User(BaseModel):
|
||||
def __init__(
|
||||
self,
|
||||
account_id,
|
||||
region,
|
||||
user_id,
|
||||
user_name,
|
||||
access_string,
|
||||
engine,
|
||||
no_password_required,
|
||||
passwords=None,
|
||||
account_id: str,
|
||||
region: str,
|
||||
user_id: str,
|
||||
user_name: str,
|
||||
access_string: str,
|
||||
engine: str,
|
||||
no_password_required: bool,
|
||||
passwords: Optional[List[str]] = None,
|
||||
):
|
||||
self.id = user_id
|
||||
self.name = user_name
|
||||
@ -23,7 +24,7 @@ class User(BaseModel):
|
||||
self.no_password_required = no_password_required
|
||||
self.status = "active"
|
||||
self.minimum_engine_version = "6.0"
|
||||
self.usergroupids = []
|
||||
self.usergroupids: List[str] = []
|
||||
self.region = region
|
||||
self.arn = f"arn:aws:elasticache:{self.region}:{account_id}:user:{self.id}"
|
||||
|
||||
@ -31,7 +32,7 @@ class User(BaseModel):
|
||||
class ElastiCacheBackend(BaseBackend):
|
||||
"""Implementation of ElastiCache APIs."""
|
||||
|
||||
def __init__(self, region_name, account_id):
|
||||
def __init__(self, region_name: str, account_id: str):
|
||||
super().__init__(region_name, account_id)
|
||||
self.users = dict()
|
||||
self.users["default"] = User(
|
||||
@ -45,8 +46,14 @@ class ElastiCacheBackend(BaseBackend):
|
||||
)
|
||||
|
||||
def create_user(
|
||||
self, user_id, user_name, engine, passwords, access_string, no_password_required
|
||||
):
|
||||
self,
|
||||
user_id: str,
|
||||
user_name: str,
|
||||
engine: str,
|
||||
passwords: List[str],
|
||||
access_string: str,
|
||||
no_password_required: bool,
|
||||
) -> User:
|
||||
if user_id in self.users:
|
||||
raise UserAlreadyExists
|
||||
user = User(
|
||||
@ -62,7 +69,7 @@ class ElastiCacheBackend(BaseBackend):
|
||||
self.users[user_id] = user
|
||||
return user
|
||||
|
||||
def delete_user(self, user_id):
|
||||
def delete_user(self, user_id: str) -> User:
|
||||
if user_id in self.users:
|
||||
user = self.users[user_id]
|
||||
if user.status == "active":
|
||||
@ -70,7 +77,7 @@ class ElastiCacheBackend(BaseBackend):
|
||||
return user
|
||||
raise UserNotFound(user_id)
|
||||
|
||||
def describe_users(self, user_id):
|
||||
def describe_users(self, user_id: Optional[str]) -> List[User]:
|
||||
"""
|
||||
Only the `user_id` parameter is currently supported.
|
||||
Pagination is not yet implemented.
|
||||
@ -83,7 +90,7 @@ class ElastiCacheBackend(BaseBackend):
|
||||
return [user]
|
||||
else:
|
||||
raise UserNotFound(user_id)
|
||||
return self.users.values()
|
||||
return list(self.users.values())
|
||||
|
||||
|
||||
elasticache_backends = BackendDict(ElastiCacheBackend, "elasticache")
|
||||
|
@ -1,20 +1,20 @@
|
||||
from moto.core.responses import BaseResponse
|
||||
from .exceptions import PasswordTooShort, PasswordRequired
|
||||
from .models import elasticache_backends
|
||||
from .models import elasticache_backends, ElastiCacheBackend
|
||||
|
||||
|
||||
class ElastiCacheResponse(BaseResponse):
|
||||
"""Handler for ElastiCache requests and responses."""
|
||||
|
||||
def __init__(self):
|
||||
def __init__(self) -> None:
|
||||
super().__init__(service_name="elasticache")
|
||||
|
||||
@property
|
||||
def elasticache_backend(self):
|
||||
def elasticache_backend(self) -> ElastiCacheBackend:
|
||||
"""Return backend instance specific for this region."""
|
||||
return elasticache_backends[self.current_account][self.region]
|
||||
|
||||
def create_user(self):
|
||||
def create_user(self) -> str:
|
||||
params = self._get_params()
|
||||
user_id = params.get("UserId")
|
||||
user_name = params.get("UserName")
|
||||
@ -28,24 +28,24 @@ class ElastiCacheResponse(BaseResponse):
|
||||
raise PasswordTooShort
|
||||
access_string = params.get("AccessString")
|
||||
user = self.elasticache_backend.create_user(
|
||||
user_id=user_id,
|
||||
user_name=user_name,
|
||||
engine=engine,
|
||||
user_id=user_id, # type: ignore[arg-type]
|
||||
user_name=user_name, # type: ignore[arg-type]
|
||||
engine=engine, # type: ignore[arg-type]
|
||||
passwords=passwords,
|
||||
access_string=access_string,
|
||||
access_string=access_string, # type: ignore[arg-type]
|
||||
no_password_required=no_password_required,
|
||||
)
|
||||
template = self.response_template(CREATE_USER_TEMPLATE)
|
||||
return template.render(user=user)
|
||||
|
||||
def delete_user(self):
|
||||
def delete_user(self) -> str:
|
||||
params = self._get_params()
|
||||
user_id = params.get("UserId")
|
||||
user = self.elasticache_backend.delete_user(user_id=user_id)
|
||||
user = self.elasticache_backend.delete_user(user_id=user_id) # type: ignore[arg-type]
|
||||
template = self.response_template(DELETE_USER_TEMPLATE)
|
||||
return template.render(user=user)
|
||||
|
||||
def describe_users(self):
|
||||
def describe_users(self) -> str:
|
||||
params = self._get_params()
|
||||
user_id = params.get("UserId")
|
||||
users = self.elasticache_backend.describe_users(user_id=user_id)
|
||||
|
@ -229,7 +229,7 @@ disable = W,C,R,E
|
||||
enable = anomalous-backslash-in-string, arguments-renamed, dangerous-default-value, deprecated-module, function-redefined, import-self, redefined-builtin, redefined-outer-name, reimported, pointless-statement, super-with-arguments, unused-argument, unused-import, unused-variable, useless-else-on-loop, wildcard-import
|
||||
|
||||
[mypy]
|
||||
files= moto/a*,moto/b*,moto/c*,moto/d*,moto/ebs/,moto/ec2,moto/ec2instanceconnect,moto/ecr,moto/ecs,moto/efs,moto/eks,moto/es,moto/moto_api
|
||||
files= moto/a*,moto/b*,moto/c*,moto/d*,moto/ebs/,moto/ec2,moto/ec2instanceconnect,moto/ecr,moto/ecs,moto/efs,moto/eks,moto/elasticache,moto/es,moto/moto_api
|
||||
show_column_numbers=True
|
||||
show_error_codes = True
|
||||
disable_error_code=abstract
|
||||
|
Loading…
Reference in New Issue
Block a user