diff --git a/moto/moto_api/_internal/managed_state_model.py b/moto/moto_api/_internal/managed_state_model.py index 1cf71ab08..2d25be598 100644 --- a/moto/moto_api/_internal/managed_state_model.py +++ b/moto/moto_api/_internal/managed_state_model.py @@ -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 diff --git a/moto/moto_api/_internal/models.py b/moto/moto_api/_internal/models.py index 728b3c3eb..a16b96df6 100644 --- a/moto/moto_api/_internal/models.py +++ b/moto/moto_api/_internal/models.py @@ -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) diff --git a/moto/moto_api/_internal/recorder/models.py b/moto/moto_api/_internal/recorder/models.py index eb3a8b725..63227ed7e 100644 --- a/moto/moto_api/_internal/recorder/models.py +++ b/moto/moto_api/_internal/recorder/models.py @@ -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. diff --git a/moto/moto_api/_internal/recorder/responses.py b/moto/moto_api/_internal/recorder/responses.py index 870427934..667d45d4a 100644 --- a/moto/moto_api/_internal/recorder/responses.py +++ b/moto/moto_api/_internal/recorder/responses.py @@ -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, {}, "" diff --git a/moto/moto_api/_internal/responses.py b/moto/moto_api/_internal/responses.py index eba77ae47..d3a4f40b1 100644 --- a/moto/moto_api/_internal/responses.py +++ b/moto/moto_api/_internal/responses.py @@ -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 diff --git a/moto/moto_api/_internal/state_manager.py b/moto/moto_api/_internal/state_manager.py index f31a0bbe9..7d369de2d 100644 --- a/moto/moto_api/_internal/state_manager.py +++ b/moto/moto_api/_internal/state_manager.py @@ -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()) diff --git a/setup.cfg b/setup.cfg index f9cc6297e..d4e8223e6 100644 --- a/setup.cfg +++ b/setup.cfg @@ -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