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

Techdebt: MyPy Moto-API (#5663)

Browse Source
This commit is contained in:
Bert Blommers 2022-11-12 21:43:46 -01:00 committed by GitHub
parent 52892f5481
commit 9eccce8af3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 95 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
moto/moto_api/_internal/managed_state_model.py
View File

@ -28,7 +28,7 @@ class ManagedState:
self._tick += 1
@property
def status(self):
def status(self) -> str:
"""
Transitions the status as appropriate before returning
"""
@ -52,15 +52,15 @@ class ManagedState:
return self._status
@status.setter
def status(self, value):
def status(self, value: str) -> None:
self._status = value
def _get_next_status(self, previous):
def _get_next_status(self, previous: str) -> str:
return next(
(nxt for prev, nxt in self._transitions if previous == prev), previous
)
def _get_last_status(self, previous):
def _get_last_status(self, previous: str) -> str:
next_state = self._get_next_status(previous)
while next_state != previous:
previous = next_state

11
moto/moto_api/_internal/models.py
View File

@ -1,8 +1,9 @@
from moto.core import BaseBackend, DEFAULT_ACCOUNT_ID
from typing import Any, Dict
class MotoAPIBackend(BaseBackend):
def reset(self):
def reset(self) -> None:
region_name = self.region_name
account_id = self.account_id
@ -13,19 +14,19 @@ class MotoAPIBackend(BaseBackend):
continue
for backend in backends_.values():
backend.reset()
self.__init__(region_name, account_id)
self.__init__(region_name, account_id) # type: ignore[misc]
def get_transition(self, model_name):
def get_transition(self, model_name: str) -> Dict[str, Any]:
from moto.moto_api import state_manager
return state_manager.get_transition(model_name)
def set_transition(self, model_name, transition):
def set_transition(self, model_name: str, transition: Dict[str, Any]) -> None:
from moto.moto_api import state_manager
state_manager.set_transition(model_name, transition)
def unset_transition(self, model_name):
def unset_transition(self, model_name: str) -> None:
from moto.moto_api import state_manager
state_manager.unset_transition(model_name)

28
moto/moto_api/_internal/recorder/models.py
View File

@ -5,12 +5,12 @@ import os
import requests
from botocore.awsrequest import AWSPreparedRequest
from typing import Any, Optional
from typing import Any, Optional, Union, Tuple
from urllib.parse import urlparse
class Recorder:
def __init__(self):
def __init__(self) -> None:
self._location = str(os.environ.get("MOTO_RECORDER_FILEPATH", "moto_recording"))
self._os_enabled = bool(os.environ.get("MOTO_ENABLE_RECORDING", False))
self._user_enabled = self._os_enabled
@ -33,15 +33,15 @@ class Recorder:
if body is None:
if isinstance(request, AWSPreparedRequest):
body, body_encoded = self._encode_body(body=request.body)
body_str, body_encoded = self._encode_body(body=request.body)
else:
try:
request_body = None
request_body_size = int(request.headers["Content-Length"])
request_body = request.environ["wsgi.input"].read(request_body_size)
body, body_encoded = self._encode_body(body=request_body)
body_str, body_encoded = self._encode_body(body=request_body)
except (AttributeError, KeyError):
body = ""
body_str = "" # type: ignore[]
body_encoded = False
finally:
if request_body is not None:
@ -49,15 +49,15 @@ class Recorder:
request_body = request_body.encode("utf-8")
request.environ["wsgi.input"] = io.BytesIO(request_body)
else:
body, body_encoded = self._encode_body(body)
entry.update({"body": body, "body_encoded": body_encoded})
body_str, body_encoded = self._encode_body(body)
entry.update({"body": body_str, "body_encoded": body_encoded})
filepath = self._location
with open(filepath, "a+") as file:
file.write(json.dumps(entry))
file.write("\n")
def _encode_body(self, body):
def _encode_body(self, body: Any) -> Tuple[str, bool]:
body_encoded = False
try:
if isinstance(body, io.BytesIO):
@ -69,7 +69,7 @@ class Recorder:
body = None
return body, body_encoded
def reset_recording(self):
def reset_recording(self) -> None:
"""
Resets the recording. This will erase any requests made previously.
"""
@ -77,16 +77,16 @@ class Recorder:
with open(filepath, "w"):
pass
def start_recording(self):
def start_recording(self) -> None:
"""
Start the recording, and append incoming requests to the log.
"""
self._user_enabled = True
def stop_recording(self):
def stop_recording(self) -> None:
self._user_enabled = False
def upload_recording(self, data):
def upload_recording(self, data: Union[str, bytes]) -> None:
"""
Replaces the current log. Remember to replay the recording afterwards.
"""
@ -96,7 +96,7 @@ class Recorder:
with open(filepath, "bw") as file:
file.write(data)
def download_recording(self):
def download_recording(self) -> str:
"""
Download the current recording. The result can be uploaded afterwards.
"""
@ -104,7 +104,7 @@ class Recorder:
with open(filepath, "r") as file:
return file.read()
def replay_recording(self, target_host=None):
def replay_recording(self, target_host: Optional[str] = None) -> None:
"""
Replays the current log, i.e. replay all requests that were made after the recorder was started.
Download the recording if you want to manually verify the correct requests will be replayed.

26
moto/moto_api/_internal/recorder/responses.py
View File

@ -1,31 +1,45 @@
from ... import recorder
from moto.core.common_types import TYPE_RESPONSE
from moto.core.responses import BaseResponse
from typing import Any
class RecorderResponse(BaseResponse):
def reset_recording(self, req, url, headers): # pylint: disable=unused-argument
def reset_recording(
self, req: Any, url: str, headers: Any # pylint: disable=unused-argument
) -> TYPE_RESPONSE:
recorder.reset_recording()
return 200, {}, ""
def start_recording(self, req, url, headers): # pylint: disable=unused-argument
def start_recording(
self, req: Any, url: str, headers: Any # pylint: disable=unused-argument
) -> TYPE_RESPONSE:
recorder.start_recording()
return 200, {}, "Recording is set to True"
def stop_recording(self, req, url, headers): # pylint: disable=unused-argument
def stop_recording(
self, req: Any, url: str, headers: Any # pylint: disable=unused-argument
) -> TYPE_RESPONSE:
recorder.stop_recording()
return 200, {}, "Recording is set to False"
def upload_recording(self, req, url, headers): # pylint: disable=unused-argument
def upload_recording(
self, req: Any, url: str, headers: Any # pylint: disable=unused-argument
) -> TYPE_RESPONSE:
data = req.data
recorder.upload_recording(data)
return 200, {}, ""
def download_recording(self, req, url, headers): # pylint: disable=unused-argument
def download_recording(
self, req: Any, url: str, headers: Any # pylint: disable=unused-argument
) -> TYPE_RESPONSE:
data = recorder.download_recording()
return 200, {}, data
# NOTE: Replaying assumes, for simplicity, that it is the only action
# running against moto at the time. No recording happens while replaying.
def replay_recording(self, req, url, headers): # pylint: disable=unused-argument
def replay_recording(
self, req: Any, url: str, headers: Any # pylint: disable=unused-argument
) -> TYPE_RESPONSE:
recorder.replay_recording(target_host=url)
return 200, {}, ""

57
moto/moto_api/_internal/responses.py
View File

@ -1,13 +1,18 @@
import json
from moto import settings
from moto.core.common_types import TYPE_RESPONSE
from moto.core.responses import ActionAuthenticatorMixin, BaseResponse
from typing import Any, Dict, List
class MotoAPIResponse(BaseResponse):
def reset_response(
self, request, full_url, headers
): # pylint: disable=unused-argument
self,
request: Any, # pylint: disable=unused-argument
full_url: str, # pylint: disable=unused-argument
headers: Any, # pylint: disable=unused-argument
) -> TYPE_RESPONSE:
if request.method == "POST":
from .models import moto_api_backend
@ -16,8 +21,11 @@ class MotoAPIResponse(BaseResponse):
return 400, {}, json.dumps({"Error": "Need to POST to reset Moto"})
def reset_auth_response(
self, request, full_url, headers
): # pylint: disable=unused-argument
self,
request: Any, # pylint: disable=unused-argument
full_url: str, # pylint: disable=unused-argument
headers: Any, # pylint: disable=unused-argument
) -> TYPE_RESPONSE:
if request.method == "POST":
previous_initial_no_auth_action_count = (
settings.INITIAL_NO_AUTH_ACTION_COUNT
@ -38,17 +46,22 @@ class MotoAPIResponse(BaseResponse):
)
return 400, {}, json.dumps({"Error": "Need to POST to reset Moto Auth"})
def model_data(self, request, full_url, headers): # pylint: disable=unused-argument
def model_data(
self,
request: Any, # pylint: disable=unused-argument
full_url: str, # pylint: disable=unused-argument
headers: Any, # pylint: disable=unused-argument
) -> TYPE_RESPONSE:
from moto.core.base_backend import model_data
results = {}
results: Dict[str, Dict[str, List[Any]]] = {}
for service in sorted(model_data):
models = model_data[service]
results[service] = {}
for name in sorted(models):
model = models[name]
results[service][name] = []
for instance in model.instances:
for instance in model.instances: # type: ignore[attr-defined]
inst_result = {}
for attr in dir(instance):
if not attr.startswith("_"):
@ -61,14 +74,22 @@ class MotoAPIResponse(BaseResponse):
results[service][name].append(inst_result)
return 200, {"Content-Type": "application/javascript"}, json.dumps(results)
def dashboard(self, request, full_url, headers): # pylint: disable=unused-argument
def dashboard(
self,
request: Any, # pylint: disable=unused-argument
full_url: str, # pylint: disable=unused-argument
headers: Any, # pylint: disable=unused-argument
) -> str:
from flask import render_template
return render_template("dashboard.html")
def get_transition(
self, request, full_url, headers
): # pylint: disable=unused-argument
self,
request: Any, # pylint: disable=unused-argument
full_url: str, # pylint: disable=unused-argument
headers: Any, # pylint: disable=unused-argument
) -> TYPE_RESPONSE:
from .models import moto_api_backend
qs_dict = dict(
@ -81,8 +102,11 @@ class MotoAPIResponse(BaseResponse):
return 200, {}, json.dumps(resp)
def set_transition(
self, request, full_url, headers
): # pylint: disable=unused-argument
self,
request: Any, # pylint: disable=unused-argument
full_url: str, # pylint: disable=unused-argument
headers: Any, # pylint: disable=unused-argument
) -> TYPE_RESPONSE:
from .models import moto_api_backend
request_body_size = int(headers["Content-Length"])
@ -95,8 +119,11 @@ class MotoAPIResponse(BaseResponse):
return 201, {}, ""
def unset_transition(
self, request, full_url, headers
): # pylint: disable=unused-argument
self,
request: Any, # pylint: disable=unused-argument
full_url: str, # pylint: disable=unused-argument
headers: Any, # pylint: disable=unused-argument
) -> TYPE_RESPONSE:
from .models import moto_api_backend
request_body_size = int(headers["Content-Length"])
@ -107,7 +134,7 @@ class MotoAPIResponse(BaseResponse):
moto_api_backend.unset_transition(model_name)
return 201, {}, ""
def seed(self, req, full_url, headers):
def seed(self, req: Any, full_url: str, headers: Any) -> TYPE_RESPONSE:
self.setup_class(req, full_url, headers)
from . import mock_random

16
moto/moto_api/_internal/state_manager.py
View File

@ -1,13 +1,13 @@
from typing import Any, Dict
from typing import Any, Dict, List
DEFAULT_TRANSITION = {"progression": "immediate"}
class StateManager:
def __init__(self):
self._default_transitions = dict()
self._transitions = dict()
def __init__(self) -> None:
self._default_transitions: Dict[str, Dict[str, Any]] = dict()
self._transitions: Dict[str, Dict[str, Any]] = dict()
def register_default_transition(
self, model_name: str, transition: Dict[str, Any]
@ -18,7 +18,7 @@ class StateManager:
"""
self._default_transitions[model_name] = transition
def set_transition(self, model_name, transition):
def set_transition(self, model_name: str, transition: Dict[str, Any]) -> None:
"""
Set a transition for a specific model. Any transition added here will take precedence over the default transition that was registered.
@ -26,14 +26,14 @@ class StateManager:
"""
self._transitions[model_name] = transition
def unset_transition(self, model_name):
def unset_transition(self, model_name: str) -> None:
"""
Unset (remove) a custom transition that was set. This is a safe and idempotent operation.
The default transition that was registered will not be altered by this operation.
"""
self._transitions.pop(model_name, None)
def get_transition(self, model_name):
def get_transition(self, model_name: str) -> Dict[str, Any]:
"""
Return the configuration for a specific model. This will return a user-specified configuration, a default configuration of none exists, or the default transition if none exists.
"""
@ -43,5 +43,5 @@ class StateManager:
return self._default_transitions[model_name]
return DEFAULT_TRANSITION
def get_registered_models(self):
def get_registered_models(self) -> List[str]:
return list(self._default_transitions.keys())

2
setup.cfg
View File

@ -18,7 +18,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*
files= moto/a*,moto/b*,moto/c*,moto/moto_api
show_column_numbers=True
show_error_codes = True
disable_error_code=abstract