Techdebt: MyPy ElasticBeanstalk (#5972)
This commit is contained in:
parent
5dd155de4b
commit
902fb9bca2
@ -2,10 +2,10 @@ from moto.core.exceptions import RESTError
|
|||||||
|
|
||||||
|
|
||||||
class InvalidParameterValueError(RESTError):
|
class InvalidParameterValueError(RESTError):
|
||||||
def __init__(self, message):
|
def __init__(self, message: str):
|
||||||
super().__init__("InvalidParameterValue", message)
|
super().__init__("InvalidParameterValue", message)
|
||||||
|
|
||||||
|
|
||||||
class ResourceNotFoundException(RESTError):
|
class ResourceNotFoundException(RESTError):
|
||||||
def __init__(self, message):
|
def __init__(self, message: str):
|
||||||
super().__init__("ResourceNotFoundException", message)
|
super().__init__("ResourceNotFoundException", message)
|
||||||
|
@ -1,4 +1,5 @@
|
|||||||
import weakref
|
import weakref
|
||||||
|
from typing import Dict, List
|
||||||
|
|
||||||
from moto.core import BaseBackend, BackendDict, BaseModel
|
from moto.core import BaseBackend, BackendDict, BaseModel
|
||||||
from .exceptions import InvalidParameterValueError, ResourceNotFoundException
|
from .exceptions import InvalidParameterValueError, ResourceNotFoundException
|
||||||
@ -6,7 +7,13 @@ from .utils import make_arn
|
|||||||
|
|
||||||
|
|
||||||
class FakeEnvironment(BaseModel):
|
class FakeEnvironment(BaseModel):
|
||||||
def __init__(self, application, environment_name, solution_stack_name, tags):
|
def __init__(
|
||||||
|
self,
|
||||||
|
application: "FakeApplication",
|
||||||
|
environment_name: str,
|
||||||
|
solution_stack_name: str,
|
||||||
|
tags: Dict[str, str],
|
||||||
|
):
|
||||||
self.application = weakref.proxy(
|
self.application = weakref.proxy(
|
||||||
application
|
application
|
||||||
) # weakref to break circular dependencies
|
) # weakref to break circular dependencies
|
||||||
@ -15,39 +22,41 @@ class FakeEnvironment(BaseModel):
|
|||||||
self.tags = tags
|
self.tags = tags
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def application_name(self):
|
def application_name(self) -> str:
|
||||||
return self.application.application_name
|
return self.application.application_name
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def environment_arn(self):
|
def environment_arn(self) -> str:
|
||||||
resource_path = f"{self.application_name}/{self.environment_name}"
|
resource_path = f"{self.application_name}/{self.environment_name}"
|
||||||
return make_arn(
|
return make_arn(
|
||||||
self.region, self.application.account_id, "environment", resource_path
|
self.region, self.application.account_id, "environment", resource_path
|
||||||
)
|
)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def platform_arn(self):
|
def platform_arn(self) -> str:
|
||||||
return "TODO" # TODO
|
return "TODO" # TODO
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def region(self):
|
def region(self) -> str:
|
||||||
return self.application.region
|
return self.application.region
|
||||||
|
|
||||||
|
|
||||||
class FakeApplication(BaseModel):
|
class FakeApplication(BaseModel):
|
||||||
def __init__(self, backend, application_name):
|
def __init__(self, backend: "EBBackend", application_name: str):
|
||||||
self.backend = weakref.proxy(backend) # weakref to break cycles
|
self.backend = weakref.proxy(backend) # weakref to break cycles
|
||||||
self.application_name = application_name
|
self.application_name = application_name
|
||||||
self.environments = dict()
|
self.environments: Dict[str, FakeEnvironment] = dict()
|
||||||
self.account_id = self.backend.account_id
|
self.account_id = self.backend.account_id
|
||||||
self.region = self.backend.region_name
|
self.region = self.backend.region_name
|
||||||
self.arn = make_arn(
|
self.arn = make_arn(
|
||||||
self.region, self.account_id, "application", self.application_name
|
self.region, self.account_id, "application", self.application_name
|
||||||
)
|
)
|
||||||
|
|
||||||
def create_environment(self, environment_name, solution_stack_name, tags):
|
def create_environment(
|
||||||
|
self, environment_name: str, solution_stack_name: str, tags: Dict[str, str]
|
||||||
|
) -> FakeEnvironment:
|
||||||
if environment_name in self.environments:
|
if environment_name in self.environments:
|
||||||
raise InvalidParameterValueError
|
raise InvalidParameterValueError(message="")
|
||||||
|
|
||||||
env = FakeEnvironment(
|
env = FakeEnvironment(
|
||||||
application=self,
|
application=self,
|
||||||
@ -61,12 +70,14 @@ class FakeApplication(BaseModel):
|
|||||||
|
|
||||||
|
|
||||||
class EBBackend(BaseBackend):
|
class EBBackend(BaseBackend):
|
||||||
def __init__(self, region_name, account_id):
|
def __init__(self, region_name: str, account_id: str):
|
||||||
super().__init__(region_name, account_id)
|
super().__init__(region_name, account_id)
|
||||||
self.applications = dict()
|
self.applications: Dict[str, FakeApplication] = dict()
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def default_vpc_endpoint_service(service_region, zones):
|
def default_vpc_endpoint_service(
|
||||||
|
service_region: str, zones: List[str]
|
||||||
|
) -> List[Dict[str, str]]:
|
||||||
"""Default VPC endpoint service."""
|
"""Default VPC endpoint service."""
|
||||||
return BaseBackend.default_vpc_endpoint_service_factory(
|
return BaseBackend.default_vpc_endpoint_service_factory(
|
||||||
service_region, zones, "elasticbeanstalk"
|
service_region, zones, "elasticbeanstalk"
|
||||||
@ -74,7 +85,7 @@ class EBBackend(BaseBackend):
|
|||||||
service_region, zones, "elasticbeanstalk-health"
|
service_region, zones, "elasticbeanstalk-health"
|
||||||
)
|
)
|
||||||
|
|
||||||
def create_application(self, application_name):
|
def create_application(self, application_name: str) -> FakeApplication:
|
||||||
if application_name in self.applications:
|
if application_name in self.applications:
|
||||||
raise InvalidParameterValueError(
|
raise InvalidParameterValueError(
|
||||||
f"Application {application_name} already exists."
|
f"Application {application_name} already exists."
|
||||||
@ -83,23 +94,31 @@ class EBBackend(BaseBackend):
|
|||||||
self.applications[application_name] = new_app
|
self.applications[application_name] = new_app
|
||||||
return new_app
|
return new_app
|
||||||
|
|
||||||
def create_environment(self, app, environment_name, stack_name, tags):
|
def create_environment(
|
||||||
|
self,
|
||||||
|
app: FakeApplication,
|
||||||
|
environment_name: str,
|
||||||
|
stack_name: str,
|
||||||
|
tags: Dict[str, str],
|
||||||
|
) -> FakeEnvironment:
|
||||||
return app.create_environment(
|
return app.create_environment(
|
||||||
environment_name=environment_name, solution_stack_name=stack_name, tags=tags
|
environment_name=environment_name, solution_stack_name=stack_name, tags=tags
|
||||||
)
|
)
|
||||||
|
|
||||||
def describe_environments(self):
|
def describe_environments(self) -> List[FakeEnvironment]:
|
||||||
envs = []
|
envs = []
|
||||||
for app in self.applications.values():
|
for app in self.applications.values():
|
||||||
for env in app.environments.values():
|
for env in app.environments.values():
|
||||||
envs.append(env)
|
envs.append(env)
|
||||||
return envs
|
return envs
|
||||||
|
|
||||||
def list_available_solution_stacks(self):
|
def list_available_solution_stacks(self) -> None:
|
||||||
# Implemented in response.py
|
# Implemented in response.py
|
||||||
pass
|
pass
|
||||||
|
|
||||||
def update_tags_for_resource(self, resource_arn, tags_to_add, tags_to_remove):
|
def update_tags_for_resource(
|
||||||
|
self, resource_arn: str, tags_to_add: Dict[str, str], tags_to_remove: List[str]
|
||||||
|
) -> None:
|
||||||
try:
|
try:
|
||||||
res = self._find_environment_by_arn(resource_arn)
|
res = self._find_environment_by_arn(resource_arn)
|
||||||
except KeyError:
|
except KeyError:
|
||||||
@ -113,7 +132,7 @@ class EBBackend(BaseBackend):
|
|||||||
for key in tags_to_remove:
|
for key in tags_to_remove:
|
||||||
del res.tags[key]
|
del res.tags[key]
|
||||||
|
|
||||||
def list_tags_for_resource(self, resource_arn):
|
def list_tags_for_resource(self, resource_arn: str) -> Dict[str, str]:
|
||||||
try:
|
try:
|
||||||
res = self._find_environment_by_arn(resource_arn)
|
res = self._find_environment_by_arn(resource_arn)
|
||||||
except KeyError:
|
except KeyError:
|
||||||
@ -122,7 +141,7 @@ class EBBackend(BaseBackend):
|
|||||||
)
|
)
|
||||||
return res.tags
|
return res.tags
|
||||||
|
|
||||||
def _find_environment_by_arn(self, arn):
|
def _find_environment_by_arn(self, arn: str) -> FakeEnvironment:
|
||||||
for app in self.applications.keys():
|
for app in self.applications.keys():
|
||||||
for env in self.applications[app].environments.values():
|
for env in self.applications[app].environments.values():
|
||||||
if env.environment_arn == arn:
|
if env.environment_arn == arn:
|
||||||
|
@ -1,21 +1,21 @@
|
|||||||
from moto.core.responses import BaseResponse
|
from moto.core.responses import BaseResponse
|
||||||
from moto.core.utils import tags_from_query_string
|
from moto.core.utils import tags_from_query_string
|
||||||
from .models import eb_backends
|
from .models import eb_backends, EBBackend
|
||||||
from .exceptions import InvalidParameterValueError
|
from .exceptions import InvalidParameterValueError
|
||||||
|
|
||||||
|
|
||||||
class EBResponse(BaseResponse):
|
class EBResponse(BaseResponse):
|
||||||
def __init__(self):
|
def __init__(self) -> None:
|
||||||
super().__init__(service_name="elasticbeanstalk")
|
super().__init__(service_name="elasticbeanstalk")
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def backend(self):
|
def backend(self) -> EBBackend:
|
||||||
"""
|
"""
|
||||||
:rtype: EBBackend
|
:rtype: EBBackend
|
||||||
"""
|
"""
|
||||||
return eb_backends[self.current_account][self.region]
|
return eb_backends[self.current_account][self.region]
|
||||||
|
|
||||||
def create_application(self):
|
def create_application(self) -> str:
|
||||||
app = self.backend.create_application(
|
app = self.backend.create_application(
|
||||||
application_name=self._get_param("ApplicationName")
|
application_name=self._get_param("ApplicationName")
|
||||||
)
|
)
|
||||||
@ -23,11 +23,11 @@ class EBResponse(BaseResponse):
|
|||||||
template = self.response_template(EB_CREATE_APPLICATION)
|
template = self.response_template(EB_CREATE_APPLICATION)
|
||||||
return template.render(region_name=self.backend.region_name, application=app)
|
return template.render(region_name=self.backend.region_name, application=app)
|
||||||
|
|
||||||
def describe_applications(self):
|
def describe_applications(self) -> str:
|
||||||
template = self.response_template(EB_DESCRIBE_APPLICATIONS)
|
template = self.response_template(EB_DESCRIBE_APPLICATIONS)
|
||||||
return template.render(applications=self.backend.applications.values())
|
return template.render(applications=self.backend.applications.values())
|
||||||
|
|
||||||
def create_environment(self):
|
def create_environment(self) -> str:
|
||||||
application_name = self._get_param("ApplicationName")
|
application_name = self._get_param("ApplicationName")
|
||||||
try:
|
try:
|
||||||
app = self.backend.applications[application_name]
|
app = self.backend.applications[application_name]
|
||||||
@ -47,16 +47,16 @@ class EBResponse(BaseResponse):
|
|||||||
template = self.response_template(EB_CREATE_ENVIRONMENT)
|
template = self.response_template(EB_CREATE_ENVIRONMENT)
|
||||||
return template.render(environment=env, region=self.backend.region_name)
|
return template.render(environment=env, region=self.backend.region_name)
|
||||||
|
|
||||||
def describe_environments(self):
|
def describe_environments(self) -> str:
|
||||||
envs = self.backend.describe_environments()
|
envs = self.backend.describe_environments()
|
||||||
|
|
||||||
template = self.response_template(EB_DESCRIBE_ENVIRONMENTS)
|
template = self.response_template(EB_DESCRIBE_ENVIRONMENTS)
|
||||||
return template.render(environments=envs)
|
return template.render(environments=envs)
|
||||||
|
|
||||||
def list_available_solution_stacks(self):
|
def list_available_solution_stacks(self) -> str:
|
||||||
return EB_LIST_AVAILABLE_SOLUTION_STACKS
|
return EB_LIST_AVAILABLE_SOLUTION_STACKS
|
||||||
|
|
||||||
def update_tags_for_resource(self):
|
def update_tags_for_resource(self) -> str:
|
||||||
resource_arn = self._get_param("ResourceArn")
|
resource_arn = self._get_param("ResourceArn")
|
||||||
tags_to_add = tags_from_query_string(
|
tags_to_add = tags_from_query_string(
|
||||||
self.querystring, prefix="TagsToAdd.member"
|
self.querystring, prefix="TagsToAdd.member"
|
||||||
@ -66,7 +66,7 @@ class EBResponse(BaseResponse):
|
|||||||
|
|
||||||
return EB_UPDATE_TAGS_FOR_RESOURCE
|
return EB_UPDATE_TAGS_FOR_RESOURCE
|
||||||
|
|
||||||
def list_tags_for_resource(self):
|
def list_tags_for_resource(self) -> str:
|
||||||
resource_arn = self._get_param("ResourceArn")
|
resource_arn = self._get_param("ResourceArn")
|
||||||
tags = self.backend.list_tags_for_resource(resource_arn)
|
tags = self.backend.list_tags_for_resource(resource_arn)
|
||||||
|
|
||||||
|
@ -1,4 +1,6 @@
|
|||||||
def make_arn(region, account_id, resource_type, resource_path):
|
def make_arn(
|
||||||
|
region: str, account_id: str, resource_type: str, resource_path: str
|
||||||
|
) -> str:
|
||||||
arn_template = (
|
arn_template = (
|
||||||
"arn:aws:elasticbeanstalk:{region}:{account_id}:{resource_type}/{resource_path}"
|
"arn:aws:elasticbeanstalk:{region}:{account_id}:{resource_type}/{resource_path}"
|
||||||
)
|
)
|
||||||
|
@ -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
|
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]
|
[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/elasticache,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/elasticbeanstalk,moto/es,moto/moto_api
|
||||||
show_column_numbers=True
|
show_column_numbers=True
|
||||||
show_error_codes = True
|
show_error_codes = True
|
||||||
disable_error_code=abstract
|
disable_error_code=abstract
|
||||||
|
Loading…
Reference in New Issue
Block a user