Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Techdebt: MyPy ES (#5938)

Browse Source
This commit is contained in:
Bert Blommers 2023-02-17 10:36:30 -01:00 committed by GitHub
parent 768a8884ef
commit c2afe19ff1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 61 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
moto/ec2instanceconnect/models.py
View File

@ -3,7 +3,7 @@ from moto.core import BaseBackend, BackendDict
class Ec2InstanceConnectBackend(BaseBackend):
def send_ssh_public_key(self):
def send_ssh_public_key(self) -> str:
return json.dumps(
{"RequestId": "example-2a47-4c91-9700-e37e85162cb6", "Success": True}
)

8
moto/ec2instanceconnect/responses.py
View File

@ -1,14 +1,14 @@
from moto.core.responses import BaseResponse
from .models import ec2instanceconnect_backends
from .models import ec2instanceconnect_backends, Ec2InstanceConnectBackend
class Ec2InstanceConnectResponse(BaseResponse):
def __init__(self):
def __init__(self) -> None:
super().__init__(service_name="ec2-instanceconnect")
@property
def ec2instanceconnect_backend(self):
def ec2instanceconnect_backend(self) -> Ec2InstanceConnectBackend:
return ec2instanceconnect_backends[self.current_account][self.region]
def send_ssh_public_key(self):
def send_ssh_public_key(self) -> str:
return self.ec2instanceconnect_backend.send_ssh_public_key()

6
moto/es/exceptions.py
View File

@ -9,17 +9,17 @@ class ElasticSearchError(JsonRESTError):
class ResourceNotFound(ElasticSearchError):
code = 409
def __init__(self, resource_type, resource_name):
def __init__(self, resource_type: str, resource_name: str):
msg = f"{resource_type} not found: {resource_name}"
super().__init__("ResourceNotFoundException", msg)
class InvalidDomainName(ElasticSearchError):
def __init__(self, domain_name):
def __init__(self, domain_name: str):
msg = f"1 validation error detected: Value '{domain_name}' at 'domainName' failed to satisfy constraint: Member must satisfy regular expression pattern: [a-z][a-z0-9\\-]+"
super().__init__("ValidationException", msg)
class DomainNotFound(ResourceNotFound):
def __init__(self, domain_name):
def __init__(self, domain_name: str):
super().__init__("Domain", domain_name)

79
moto/es/models.py
View File

@ -1,3 +1,4 @@
from typing import Any, Dict, List
from moto.core import BaseBackend, BackendDict, BaseModel
from moto.moto_api._internal import mock_random
from .exceptions import DomainNotFound
@ -6,22 +7,22 @@ from .exceptions import DomainNotFound
class Domain(BaseModel):
def __init__(
self,
region_name,
domain_name,
es_version,
elasticsearch_cluster_config,
ebs_options,
access_policies,
snapshot_options,
vpc_options,
cognito_options,
encryption_at_rest_options,
node_to_node_encryption_options,
advanced_options,
log_publishing_options,
domain_endpoint_options,
advanced_security_options,
auto_tune_options,
region_name: str,
domain_name: str,
es_version: str,
elasticsearch_cluster_config: Dict[str, Any],
ebs_options: Dict[str, Any],
access_policies: Dict[str, Any],
snapshot_options: Dict[str, Any],
vpc_options: Dict[str, Any],
cognito_options: Dict[str, Any],
encryption_at_rest_options: Dict[str, Any],
node_to_node_encryption_options: Dict[str, Any],
advanced_options: Dict[str, Any],
log_publishing_options: Dict[str, Any],
domain_endpoint_options: Dict[str, Any],
advanced_security_options: Dict[str, Any],
auto_tune_options: Dict[str, Any],
):
self.domain_id = mock_random.get_random_hex(8)
self.region_name = region_name
@ -44,10 +45,10 @@ class Domain(BaseModel):
self.auto_tune_options["State"] = "ENABLED"
@property
def arn(self):
def arn(self) -> str:
return f"arn:aws:es:{self.region_name}:domain/{self.domain_id}"
def to_json(self):
def to_json(self) -> Dict[str, Any]:
return {
"DomainId": self.domain_id,
"DomainName": self.domain_name,
@ -76,28 +77,28 @@ class Domain(BaseModel):
class ElasticsearchServiceBackend(BaseBackend):
"""Implementation of ElasticsearchService APIs."""
def __init__(self, region_name, account_id):
def __init__(self, region_name: str, account_id: str):
super().__init__(region_name, account_id)
self.domains = dict()
self.domains: Dict[str, Domain] = dict()
def create_elasticsearch_domain(
self,
domain_name,
elasticsearch_version,
elasticsearch_cluster_config,
ebs_options,
access_policies,
snapshot_options,
vpc_options,
cognito_options,
encryption_at_rest_options,
node_to_node_encryption_options,
advanced_options,
log_publishing_options,
domain_endpoint_options,
advanced_security_options,
auto_tune_options,
):
domain_name: str,
elasticsearch_version: str,
elasticsearch_cluster_config: Dict[str, Any],
ebs_options: Dict[str, Any],
access_policies: Dict[str, Any],
snapshot_options: Dict[str, Any],
vpc_options: Dict[str, Any],
cognito_options: Dict[str, Any],
encryption_at_rest_options: Dict[str, Any],
node_to_node_encryption_options: Dict[str, Any],
advanced_options: Dict[str, Any],
log_publishing_options: Dict[str, Any],
domain_endpoint_options: Dict[str, Any],
advanced_security_options: Dict[str, Any],
auto_tune_options: Dict[str, Any],
) -> Dict[str, Any]:
# TODO: Persist/Return other attributes
new_domain = Domain(
region_name=self.region_name,
@ -120,17 +121,17 @@ class ElasticsearchServiceBackend(BaseBackend):
self.domains[domain_name] = new_domain
return new_domain.to_json()
def delete_elasticsearch_domain(self, domain_name):
def delete_elasticsearch_domain(self, domain_name: str) -> None:
if domain_name not in self.domains:
raise DomainNotFound(domain_name)
del self.domains[domain_name]
def describe_elasticsearch_domain(self, domain_name):
def describe_elasticsearch_domain(self, domain_name: str) -> Dict[str, Any]:
if domain_name not in self.domains:
raise DomainNotFound(domain_name)
return self.domains[domain_name].to_json()
def list_domain_names(self):
def list_domain_names(self) -> List[Dict[str, str]]:
"""
The engine-type parameter is not yet supported.
Pagination is not yet implemented.

22
moto/es/responses.py
View File

@ -1,38 +1,40 @@
import json
import re
from typing import Any
from moto.core.common_types import TYPE_RESPONSE
from moto.core.responses import BaseResponse
from .exceptions import InvalidDomainName
from .models import es_backends
from .models import es_backends, ElasticsearchServiceBackend
class ElasticsearchServiceResponse(BaseResponse):
"""Handler for ElasticsearchService requests and responses."""
def __init__(self):
def __init__(self) -> None:
super().__init__(service_name="elasticsearch")
@property
def es_backend(self):
def es_backend(self) -> ElasticsearchServiceBackend:
"""Return backend instance specific for this region."""
return es_backends[self.current_account][self.region]
@classmethod
def list_domains(cls, request, full_url, headers):
def list_domains(cls, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore
response = ElasticsearchServiceResponse()
response.setup_class(request, full_url, headers)
if request.method == "GET":
return response.list_domain_names()
@classmethod
def domains(cls, request, full_url, headers):
def domains(cls, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore
response = ElasticsearchServiceResponse()
response.setup_class(request, full_url, headers)
if request.method == "POST":
return response.create_elasticsearch_domain()
@classmethod
def domain(cls, request, full_url, headers):
def domain(cls, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore
response = ElasticsearchServiceResponse()
response.setup_class(request, full_url, headers)
if request.method == "DELETE":
@ -40,7 +42,7 @@ class ElasticsearchServiceResponse(BaseResponse):
if request.method == "GET":
return response.describe_elasticsearch_domain()
def create_elasticsearch_domain(self):
def create_elasticsearch_domain(self) -> TYPE_RESPONSE:
params = json.loads(self.body)
domain_name = params.get("DomainName")
if not re.match(r"^[a-z][a-z0-9\-]+$", domain_name):
@ -78,12 +80,12 @@ class ElasticsearchServiceResponse(BaseResponse):
)
return 200, {}, json.dumps({"DomainStatus": domain_status})
def delete_elasticsearch_domain(self):
def delete_elasticsearch_domain(self) -> TYPE_RESPONSE:
domain_name = self.path.split("/")[-1]
self.es_backend.delete_elasticsearch_domain(domain_name=domain_name)
return 200, {}, json.dumps(dict())
def describe_elasticsearch_domain(self):
def describe_elasticsearch_domain(self) -> TYPE_RESPONSE:
domain_name = self.path.split("/")[-1]
if not re.match(r"^[a-z][a-z0-9\-]+$", domain_name):
raise InvalidDomainName(domain_name)
@ -92,6 +94,6 @@ class ElasticsearchServiceResponse(BaseResponse):
)
return 200, {}, json.dumps({"DomainStatus": domain_status})
def list_domain_names(self):
def list_domain_names(self) -> TYPE_RESPONSE:
domain_names = self.es_backend.list_domain_names()
return 200, {}, json.dumps({"DomainNames": domain_names})

2
setup.cfg
View File

@ -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/moto_api
files= moto/a*,moto/b*,moto/c*,moto/d*,moto/ebs/,moto/ec2,moto/ec2instanceconnect,moto/es,moto/moto_api
show_column_numbers=True
show_error_codes = True
disable_error_code=abstract