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

Techdebt: Improve type annotations (#7124)

Browse Source
This commit is contained in:
Bert Blommers 2023-12-14 20:35:36 -01:00 committed by GitHub
parent 43d3f14d08
commit 9c58c689ad
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
21 changed files with 211 additions and 184 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
moto/core/responses.py
View File

@ -6,6 +6,7 @@ import os
import re
from collections import OrderedDict, defaultdict
from typing import (
TYPE_CHECKING,
Any,
Callable,
ClassVar,
@ -41,6 +42,13 @@ log = logging.getLogger(__name__)
JINJA_ENVS: Dict[type, Environment] = {}
if TYPE_CHECKING:
from typing_extensions import ParamSpec
P = ParamSpec("P")
T = TypeVar("T")
ResponseShape = TypeVar("ResponseShape", bound="BaseResponse")
@ -173,13 +181,13 @@ class ActionAuthenticatorMixin(object):
self._authenticate_and_authorize_action(S3IAMRequest, resource)
@staticmethod
def set_initial_no_auth_action_count(initial_no_auth_action_count: int) -> Callable[..., Callable[..., TYPE_RESPONSE]]: # type: ignore[misc]
def set_initial_no_auth_action_count(
initial_no_auth_action_count: int,
) -> "Callable[[Callable[P, T]], Callable[P, T]]":
_test_server_mode_endpoint = settings.test_server_mode_endpoint()
def decorator(
function: Callable[..., TYPE_RESPONSE]
) -> Callable[..., TYPE_RESPONSE]:
def wrapper(*args: Any, **kwargs: Any) -> TYPE_RESPONSE:
def decorator(function: "Callable[P, T]") -> "Callable[P, T]":
def wrapper(*args: "P.args", **kwargs: "P.kwargs") -> T:
if settings.TEST_SERVER_MODE:
response = requests.post(
f"{_test_server_mode_endpoint}/moto-api/reset-auth",

2
setup.cfg
View File

@ -277,7 +277,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, tests/test_core/test_mock_all.py, tests/test_core/test_decorator_calls.py, tests/test_core/test_responses_module.py, tests/test_core/test_mypy.py
files= moto, tests/test_core/
show_column_numbers=True
show_error_codes = True
disable_error_code=abstract

13
tests/test_core/test_account_id_resolution.py
View File

@ -1,4 +1,5 @@
import os
from typing import Dict, Optional
from unittest import SkipTest
import requests
@ -13,7 +14,7 @@ BASE_URL = f"http://localhost:{SERVER_PORT}/"
class TestAccountIdResolution:
def setup_method(self):
def setup_method(self) -> None:
if settings.TEST_SERVER_MODE:
raise SkipTest(
"No point in testing this in ServerMode, as we already start our own server"
@ -21,10 +22,10 @@ class TestAccountIdResolution:
self.server = ThreadedMotoServer(port=SERVER_PORT, verbose=False)
self.server.start()
def teardown_method(self):
def teardown_method(self) -> None:
self.server.stop()
def test_environment_variable_takes_precedence(self):
def test_environment_variable_takes_precedence(self) -> None:
# Verify ACCOUNT ID is standard
resp = self._get_caller_identity()
assert self._get_account_id(resp) == ACCOUNT_ID
@ -51,7 +52,9 @@ class TestAccountIdResolution:
resp = self._get_caller_identity()
assert self._get_account_id(resp) == ACCOUNT_ID
def _get_caller_identity(self, extra_headers=None):
def _get_caller_identity(
self, extra_headers: Optional[Dict[str, str]] = None
) -> requests.Response:
data = "Action=GetCallerIdentity&Version=2011-06-15"
headers = {
"Authorization": "AWS4-HMAC-SHA256 Credential=abcd/20010101/us-east-2/sts/aws4_request, SignedHeaders=host;x-amz-content-sha256;x-amz-date, Signature=...",
@ -61,6 +64,6 @@ class TestAccountIdResolution:
headers.update(extra_headers or {})
return requests.post(f"{BASE_URL}", headers=headers, data=data)
def _get_account_id(self, resp):
def _get_account_id(self, resp: requests.Response) -> str:
data = xmltodict.parse(resp.content)
return data["GetCallerIdentityResponse"]["GetCallerIdentityResult"]["Account"]

139
tests/test_core/test_auth.py
View File

@ -1,4 +1,5 @@
import json
from typing import Any, Dict, Optional
from uuid import uuid4
import boto3
@ -11,16 +12,16 @@ from moto.core import set_initial_no_auth_action_count
@mock_iam
def create_user_with_access_key(user_name="test-user"):
def create_user_with_access_key(user_name: str = "test-user") -> Dict[str, str]:
client = boto3.client("iam", region_name="us-east-1")
client.create_user(UserName=user_name)
return client.create_access_key(UserName=user_name)["AccessKey"]
@mock_iam
def create_user_with_access_key_and_inline_policy(
user_name, policy_document, policy_name="policy1"
):
def create_user_with_access_key_and_inline_policy( # type: ignore[misc]
user_name: str, policy_document: Dict[str, Any], policy_name: str = "policy1"
) -> Dict[str, str]:
client = boto3.client("iam", region_name="us-east-1")
client.create_user(UserName=user_name)
client.put_user_policy(
@ -32,9 +33,9 @@ def create_user_with_access_key_and_inline_policy(
@mock_iam
def create_user_with_access_key_and_attached_policy(
user_name, policy_document, policy_name="policy1"
):
def create_user_with_access_key_and_attached_policy( # type: ignore[misc]
user_name: str, policy_document: Dict[str, Any], policy_name: str = "policy1"
) -> Dict[str, str]:
client = boto3.client("iam", region_name="us-east-1")
client.create_user(UserName=user_name)
policy_arn = client.create_policy(
@ -45,13 +46,13 @@ def create_user_with_access_key_and_attached_policy(
@mock_iam
def create_user_with_access_key_and_multiple_policies(
user_name,
inline_policy_document,
attached_policy_document,
inline_policy_name="policy1",
attached_policy_name="policy1",
):
def create_user_with_access_key_and_multiple_policies( # type: ignore[misc]
user_name: str,
inline_policy_document: Dict[str, Any],
attached_policy_document: Dict[str, Any],
inline_policy_name: str = "policy1",
attached_policy_name: str = "policy1",
) -> Dict[str, str]:
client = boto3.client("iam", region_name="us-east-1")
client.create_user(UserName=user_name)
policy_arn = client.create_policy(
@ -68,8 +69,11 @@ def create_user_with_access_key_and_multiple_policies(
def create_group_with_attached_policy_and_add_user(
user_name, policy_document, group_name="test-group", policy_name=None
):
user_name: str,
policy_document: Dict[str, Any],
group_name: str = "test-group",
policy_name: Optional[str] = None,
) -> None:
if not policy_name:
policy_name = str(uuid4())
client = boto3.client("iam", region_name="us-east-1")
@ -82,8 +86,11 @@ def create_group_with_attached_policy_and_add_user(
def create_group_with_inline_policy_and_add_user(
user_name, policy_document, group_name="test-group", policy_name="policy1"
):
user_name: str,
policy_document: Dict[str, Any],
group_name: str = "test-group",
policy_name: str = "policy1",
) -> None:
client = boto3.client("iam", region_name="us-east-1")
client.create_group(GroupName=group_name)
client.put_group_policy(
@ -95,13 +102,13 @@ def create_group_with_inline_policy_and_add_user(
def create_group_with_multiple_policies_and_add_user(
user_name,
inline_policy_document,
attached_policy_document,
group_name="test-group",
inline_policy_name="policy1",
attached_policy_name=None,
):
user_name: str,
inline_policy_document: Dict[str, Any],
attached_policy_document: Dict[str, Any],
group_name: str = "test-group",
inline_policy_name: str = "policy1",
attached_policy_name: Optional[str] = None,
) -> None:
if not attached_policy_name:
attached_policy_name = str(uuid4())
client = boto3.client("iam", region_name="us-east-1")
@ -121,13 +128,13 @@ def create_group_with_multiple_policies_and_add_user(
@mock_iam
@mock_sts
def create_role_with_attached_policy_and_assume_it(
role_name,
trust_policy_document,
policy_document,
session_name="session1",
policy_name="policy1",
):
def create_role_with_attached_policy_and_assume_it( # type: ignore[misc]
role_name: str,
trust_policy_document: Dict[str, Any],
policy_document: Dict[str, Any],
session_name: str = "session1",
policy_name: str = "policy1",
) -> Dict[str, str]:
iam_client = boto3.client("iam", region_name="us-east-1")
sts_client = boto3.client("sts", region_name="us-east-1")
role_arn = iam_client.create_role(
@ -144,13 +151,13 @@ def create_role_with_attached_policy_and_assume_it(
@mock_iam
@mock_sts
def create_role_with_inline_policy_and_assume_it(
role_name,
trust_policy_document,
policy_document,
session_name="session1",
policy_name="policy1",
):
def create_role_with_inline_policy_and_assume_it( # type: ignore[misc]
role_name: str,
trust_policy_document: Dict[str, Any],
policy_document: Dict[str, Any],
session_name: str = "session1",
policy_name: str = "policy1",
) -> Dict[str, str]:
iam_client = boto3.client("iam", region_name="us-east-1")
sts_client = boto3.client("sts", region_name="us-east-1")
role_arn = iam_client.create_role(
@ -168,7 +175,7 @@ def create_role_with_inline_policy_and_assume_it(
@set_initial_no_auth_action_count(0)
@mock_iam
def test_invalid_client_token_id():
def test_invalid_client_token_id() -> None:
client = boto3.client(
"iam",
region_name="us-east-1",
@ -185,7 +192,7 @@ def test_invalid_client_token_id():
@set_initial_no_auth_action_count(0)
@mock_ec2
def test_auth_failure():
def test_auth_failure() -> None:
client = boto3.client(
"ec2",
region_name="us-east-1",
@ -204,7 +211,7 @@ def test_auth_failure():
@set_initial_no_auth_action_count(2)
@mock_iam
def test_signature_does_not_match():
def test_signature_does_not_match() -> None:
access_key = create_user_with_access_key()
client = boto3.client(
"iam",
@ -224,7 +231,7 @@ def test_signature_does_not_match():
@set_initial_no_auth_action_count(2)
@mock_ec2
def test_auth_failure_with_valid_access_key_id():
def test_auth_failure_with_valid_access_key_id() -> None:
access_key = create_user_with_access_key()
client = boto3.client(
"ec2",
@ -244,7 +251,7 @@ def test_auth_failure_with_valid_access_key_id():
@set_initial_no_auth_action_count(2)
@mock_ec2
def test_access_denied_with_no_policy():
def test_access_denied_with_no_policy() -> None:
user_name = "test-user"
access_key = create_user_with_access_key(user_name)
client = boto3.client(
@ -265,7 +272,7 @@ def test_access_denied_with_no_policy():
@set_initial_no_auth_action_count(3)
@mock_ec2
def test_access_denied_with_not_allowing_policy():
def test_access_denied_with_not_allowing_policy() -> None:
user_name = "test-user"
inline_policy_document = {
"Version": "2012-10-17",
@ -292,7 +299,7 @@ def test_access_denied_with_not_allowing_policy():
@set_initial_no_auth_action_count(3)
@mock_sts
def test_access_denied_explicitly_on_specific_resource():
def test_access_denied_explicitly_on_specific_resource() -> None:
user_name = "test-user"
forbidden_role_arn = f"arn:aws:iam::{ACCOUNT_ID}:role/forbidden_explicitly"
allowed_role_arn = f"arn:aws:iam::{ACCOUNT_ID}:role/allowed_implictly"
@ -333,7 +340,7 @@ def test_access_denied_explicitly_on_specific_resource():
@set_initial_no_auth_action_count(3)
@mock_ec2
def test_access_denied_for_run_instances():
def test_access_denied_for_run_instances() -> None:
# https://github.com/getmoto/moto/issues/2774
# The run-instances method was broken between botocore versions 1.15.8 and 1.15.12
# This was due to the inclusion of '"idempotencyToken":true' in the response, somehow altering the signature and breaking the authentication
@ -366,7 +373,7 @@ def test_access_denied_for_run_instances():
@set_initial_no_auth_action_count(3)
@mock_ec2
def test_access_denied_with_denying_policy():
def test_access_denied_with_denying_policy() -> None:
user_name = "test-user"
inline_policy_document = {
"Version": "2012-10-17",
@ -396,7 +403,7 @@ def test_access_denied_with_denying_policy():
@set_initial_no_auth_action_count(3)
@mock_sts
def test_get_caller_identity_allowed_with_denying_policy():
def test_get_caller_identity_allowed_with_denying_policy() -> None:
user_name = "test-user"
inline_policy_document = {
"Version": "2012-10-17",
@ -421,7 +428,7 @@ def test_get_caller_identity_allowed_with_denying_policy():
@set_initial_no_auth_action_count(3)
@mock_ec2
def test_allowed_with_wildcard_action():
def test_allowed_with_wildcard_action() -> None:
user_name = "test-user"
inline_policy_document = {
"Version": "2012-10-17",
@ -441,7 +448,7 @@ def test_allowed_with_wildcard_action():
@set_initial_no_auth_action_count(4)
@mock_iam
def test_allowed_with_explicit_action_in_attached_policy():
def test_allowed_with_explicit_action_in_attached_policy() -> None:
user_name = "test-user"
attached_policy_document = {
"Version": "2012-10-17",
@ -462,7 +469,7 @@ def test_allowed_with_explicit_action_in_attached_policy():
@set_initial_no_auth_action_count(8)
@mock_s3
@mock_iam
def test_s3_access_denied_with_denying_attached_group_policy():
def test_s3_access_denied_with_denying_attached_group_policy() -> None:
user_name = "test-user"
attached_policy_document = {
"Version": "2012-10-17",
@ -496,7 +503,7 @@ def test_s3_access_denied_with_denying_attached_group_policy():
@set_initial_no_auth_action_count(6)
@mock_s3
@mock_iam
def test_s3_access_denied_with_denying_inline_group_policy():
def test_s3_access_denied_with_denying_inline_group_policy() -> None:
user_name = "test-user"
bucket_name = "test-bucket"
inline_policy_document = {
@ -530,7 +537,7 @@ def test_s3_access_denied_with_denying_inline_group_policy():
@set_initial_no_auth_action_count(10)
@mock_iam
@mock_ec2
def test_access_denied_with_many_irrelevant_policies():
def test_access_denied_with_many_irrelevant_policies() -> None:
user_name = "test-user"
inline_policy_document = {
"Version": "2012-10-17",
@ -581,7 +588,7 @@ def test_access_denied_with_many_irrelevant_policies():
@mock_sts
@mock_ec2
@mock_elbv2
def test_allowed_with_temporary_credentials():
def test_allowed_with_temporary_credentials() -> None:
role_name = "test-role"
trust_policy_document = {
"Version": "2012-10-17",
@ -632,7 +639,7 @@ def test_allowed_with_temporary_credentials():
@mock_iam
@mock_sts
@mock_rds
def test_access_denied_with_temporary_credentials():
def test_access_denied_with_temporary_credentials() -> None:
role_name = "test-role"
session_name = "test-session"
trust_policy_document = {
@ -675,7 +682,7 @@ def test_access_denied_with_temporary_credentials():
@set_initial_no_auth_action_count(3)
@mock_iam
def test_get_user_from_credentials():
def test_get_user_from_credentials() -> None:
user_name = "new-test-user"
inline_policy_document = {
"Version": "2012-10-17",
@ -695,7 +702,7 @@ def test_get_user_from_credentials():
@set_initial_no_auth_action_count(0)
@mock_s3
def test_s3_invalid_access_key_id():
def test_s3_invalid_access_key_id() -> None:
client = boto3.client(
"s3",
region_name="us-east-1",
@ -715,7 +722,7 @@ def test_s3_invalid_access_key_id():
@set_initial_no_auth_action_count(3)
@mock_s3
@mock_iam
def test_s3_signature_does_not_match():
def test_s3_signature_does_not_match() -> None:
bucket_name = "test-bucket"
access_key = create_user_with_access_key()
client = boto3.client(
@ -738,7 +745,7 @@ def test_s3_signature_does_not_match():
@set_initial_no_auth_action_count(7)
@mock_s3
@mock_iam
def test_s3_access_denied_not_action():
def test_s3_access_denied_not_action() -> None:
user_name = "test-user"
bucket_name = "test-bucket"
inline_policy_document = {
@ -773,7 +780,7 @@ def test_s3_access_denied_not_action():
@mock_iam
@mock_sts
@mock_s3
def test_s3_invalid_token_with_temporary_credentials():
def test_s3_invalid_token_with_temporary_credentials() -> None:
role_name = "test-role"
session_name = "test-session"
bucket_name = "test-bucket-888"
@ -876,17 +883,11 @@ def test_allow_key_access_using_resource_arn() -> None:
@set_initial_no_auth_action_count(3)
@mock_ssm
@mock_iam
def test_ssm_service():
def test_ssm_service() -> None:
user_name = "test-user"
policy_doc = {
"Version": "2012-10-17",
"Statement": [
{
"Action": ["ssm:*"],
"Effect": "Allow",
"Resource": ["*"],
},
],
"Statement": [{"Action": ["ssm:*"], "Effect": "Allow", "Resource": ["*"]}],
}
access_key = create_user_with_access_key_and_inline_policy(user_name, policy_doc)

44
tests/test_core/test_backenddict.py
View File

@ -1,6 +1,7 @@
import random
import time
from threading import Thread
from typing import Any, List, Optional
import pytest
@ -12,32 +13,32 @@ from moto.elbv2.models import ELBv2Backend
class ExampleBackend(BaseBackend):
def __init__(self, region_name, account_id):
def __init__(self, region_name: str, account_id: str):
super().__init__(region_name, account_id)
def test_backend_dict_returns_nothing_by_default():
def test_backend_dict_returns_nothing_by_default() -> None:
backend_dict = BackendDict(ExampleBackend, "ebs")
assert list(backend_dict.items()) == []
def test_account_specific_dict_contains_known_regions():
def test_account_specific_dict_contains_known_regions() -> None:
backend_dict = BackendDict(ExampleBackend, "ec2")
assert isinstance(backend_dict["account"]["eu-north-1"], ExampleBackend)
def test_backend_dict_does_not_contain_unknown_regions():
def test_backend_dict_does_not_contain_unknown_regions() -> None:
backend_dict = BackendDict(ExampleBackend, "ec2")
assert "mars-south-1" not in backend_dict["account"]
def test_backend_dict_fails_when_retrieving_unknown_regions():
def test_backend_dict_fails_when_retrieving_unknown_regions() -> None:
backend_dict = BackendDict(ExampleBackend, "ec2")
with pytest.raises(KeyError):
backend_dict["account"]["mars-south-1"] # pylint: disable=pointless-statement
def test_backend_dict_can_retrieve_for_specific_account():
def test_backend_dict_can_retrieve_for_specific_account() -> None:
backend_dict = BackendDict(ExampleBackend, "ec2")
# Random account does not exist
@ -55,12 +56,12 @@ def test_backend_dict_can_retrieve_for_specific_account():
assert regional_backend.account_id == "012345"
def test_backend_dict_can_ignore_boto3_regions():
def test_backend_dict_can_ignore_boto3_regions() -> None:
backend_dict = BackendDict(ExampleBackend, "ec2", use_boto3_regions=False)
assert backend_dict["account"].get("us-east-1") is None
def test_backend_dict_can_specify_additional_regions():
def test_backend_dict_can_specify_additional_regions() -> None:
backend_dict = BackendDict(
ExampleBackend, "ec2", additional_regions=["region1", "global"]
)["123456"]
@ -74,15 +75,15 @@ def test_backend_dict_can_specify_additional_regions():
class TestMultiThreadedAccess:
class SlowExampleBackend(BaseBackend):
def __init__(self, region_name, account_id):
def __init__(self, region_name: str, account_id: str):
super().__init__(region_name, account_id)
time.sleep(0.1)
self.data = []
self.data: List[int] = []
def setup_method(self):
def setup_method(self) -> None:
self.backend = BackendDict(TestMultiThreadedAccess.SlowExampleBackend, "ec2")
def test_access_a_slow_backend_concurrently(self):
def test_access_a_slow_backend_concurrently(self) -> None:
"""None
Usecase that we want to avoid:
@ -103,7 +104,7 @@ class TestMultiThreadedAccess:
Thread 2 adds a new value to the list
"""
def access(random_number):
def access(random_number: int) -> None:
self.backend["123456789012"]["us-east-1"].data.append(random_number)
threads = []
@ -119,7 +120,7 @@ class TestMultiThreadedAccess:
assert len(self.backend["123456789012"]["us-east-1"].data) == 15
def test_backend_dict_can_be_hashed():
def test_backend_dict_can_be_hashed() -> None:
hashes = []
for backend in [ExampleBackend, set, list, BaseBackend]:
hashes.append(BackendDict(backend, "n/a").__hash__())
@ -127,7 +128,7 @@ def test_backend_dict_can_be_hashed():
assert len(set(hashes)) == 4
def test_account_specific_dict_can_be_hashed():
def test_account_specific_dict_can_be_hashed() -> None:
hashes = []
ids = ["01234567912", "01234567911", "01234567913", "000000000000", "0"]
for accnt_id in ids:
@ -137,7 +138,12 @@ def test_account_specific_dict_can_be_hashed():
assert len(set(hashes)) == 5
def _create_asb(account_id, backend=None, use_boto3_regions=False, regions=None):
def _create_asb(
account_id: str,
backend: Any = None,
use_boto3_regions: bool = False,
regions: Optional[List[str]] = None,
) -> Any:
return AccountSpecificBackend(
service_name="ec2",
account_id=account_id,
@ -147,7 +153,7 @@ def _create_asb(account_id, backend=None, use_boto3_regions=False, regions=None)
)
def test_multiple_backends_cache_behaviour():
def test_multiple_backends_cache_behaviour() -> None:
ec2 = BackendDict(EC2Backend, "ec2")
ec2_useast1 = ec2[DEFAULT_ACCOUNT_ID]["us-east-1"]
assert type(ec2_useast1) == EC2Backend
@ -172,7 +178,7 @@ def test_multiple_backends_cache_behaviour():
assert type(as_1) == AutoScalingBackend
def test_backenddict_cache_hits_and_misses():
def test_backenddict_cache_hits_and_misses() -> None:
backend = BackendDict(ExampleBackend, "ebs")
backend.__getitem__.cache_clear()
@ -213,7 +219,7 @@ def test_backenddict_cache_hits_and_misses():
assert backend.__getitem__.cache_info().currsize == 2
def test_asb_cache_hits_and_misses():
def test_asb_cache_hits_and_misses() -> None:
backend = BackendDict(ExampleBackend, "ebs")
acb = backend["accnt_id"]
acb.__getitem__.cache_clear()

4
tests/test_core/test_context_manager.py
View File

@ -4,11 +4,11 @@ from moto import mock_sqs, settings
from tests import DEFAULT_ACCOUNT_ID
def test_context_manager_returns_mock():
def test_context_manager_returns_mock() -> None:
with mock_sqs() as sqs_mock:
conn = boto3.client("sqs", region_name="us-west-1")
conn.create_queue(QueueName="queue1")
if not settings.TEST_SERVER_MODE:
backend = sqs_mock.backends[DEFAULT_ACCOUNT_ID]["us-west-1"]
assert list(backend.queues.keys()) == ["queue1"]
assert list(backend.queues.keys()) == ["queue1"] # type: ignore[attr-defined]

4
tests/test_core/test_docker.py
View File

@ -9,7 +9,7 @@ logger = logging.getLogger(__name__)
@requires_docker
@pytest.mark.order(0)
def test_docker_package_is_available():
def test_docker_package_is_available() -> None:
try:
import docker # noqa: F401 # pylint: disable=unused-import
except ImportError as err:
@ -22,7 +22,7 @@ def test_docker_package_is_available():
@requires_docker
@pytest.mark.order(0)
def test_docker_is_running_and_available():
def test_docker_is_running_and_available() -> None:
import docker
from docker.errors import DockerException

4
tests/test_core/test_ec2_vpc_endpoint_services.py
View File

@ -11,7 +11,7 @@ from moto import mock_ec2
@mock_ec2
def test_describe_vpc_endpoint_services_bad_args():
def test_describe_vpc_endpoint_services_bad_args() -> None:
"""Verify exceptions are raised for bad arguments."""
ec2 = boto3.client("ec2", region_name="us-west-1")
@ -47,7 +47,7 @@ def test_describe_vpc_endpoint_services_bad_args():
@mock_ec2
def test_describe_vpc_default_endpoint_services():
def test_describe_vpc_default_endpoint_services() -> None:
"""Test successfull calls as well as the next_token arg."""
ec2 = boto3.client("ec2", region_name="us-west-1")

4
tests/test_core/test_environ_patching.py
View File

@ -5,13 +5,13 @@ from moto import mock_ec2, mock_s3
KEY = "AWS_ACCESS_KEY_ID"
def test_aws_keys_are_patched():
def test_aws_keys_are_patched() -> None:
with mock_ec2():
patched_value = os.environ[KEY]
assert patched_value == "FOOBARKEY"
def test_aws_keys_can_be_none():
def test_aws_keys_can_be_none() -> None:
"""
Verify that the os.environ[KEY] can be None
Patching the None-value shouldn't be an issue

31
tests/test_core/test_importorder.py
View File

@ -1,3 +1,4 @@
from typing import Any, List
from unittest import SkipTest
import boto3
@ -7,7 +8,7 @@ from moto import mock_s3, settings
@pytest.fixture(scope="function", name="aws_credentials")
def fixture_aws_credentials(monkeypatch):
def fixture_aws_credentials(monkeypatch: Any) -> None: # type: ignore[misc]
"""Mocked AWS Credentials for moto."""
if settings.TEST_SERVER_MODE:
raise SkipTest("No point in testing this in ServerMode.")
@ -18,8 +19,8 @@ def fixture_aws_credentials(monkeypatch):
def test_mock_works_with_client_created_inside(
aws_credentials,
): # pylint: disable=unused-argument
aws_credentials: Any, # pylint: disable=unused-argument
) -> None:
m = mock_s3()
m.start()
client = boto3.client("s3", region_name="us-east-1")
@ -30,8 +31,8 @@ def test_mock_works_with_client_created_inside(
def test_mock_works_with_client_created_outside(
aws_credentials,
): # pylint: disable=unused-argument
aws_credentials: Any, # pylint: disable=unused-argument
) -> None:
# Create the boto3 client first
outside_client = boto3.client("s3", region_name="us-east-1")
@ -50,8 +51,8 @@ def test_mock_works_with_client_created_outside(
def test_mock_works_with_resource_created_outside(
aws_credentials,
): # pylint: disable=unused-argument
aws_credentials: Any, # pylint: disable=unused-argument
) -> None:
# Create the boto3 client first
outside_resource = boto3.resource("s3", region_name="us-east-1")
@ -68,7 +69,7 @@ def test_mock_works_with_resource_created_outside(
m.stop()
def test_patch_can_be_called_on_a_mocked_client():
def test_patch_can_be_called_on_a_mocked_client() -> None:
# start S3 after the mock, ensuring that the client contains our event-handler
m = mock_s3()
m.start()
@ -90,14 +91,14 @@ def test_patch_can_be_called_on_a_mocked_client():
m.stop()
def test_patch_client_does_not_work_for_random_parameters():
def test_patch_client_does_not_work_for_random_parameters() -> None:
from moto.core import patch_client
with pytest.raises(Exception, match="Argument sth should be of type boto3.client"):
patch_client("sth")
patch_client("sth") # type: ignore[arg-type]
def test_patch_resource_does_not_work_for_random_parameters():
def test_patch_resource_does_not_work_for_random_parameters() -> None:
from moto.core import patch_resource
with pytest.raises(
@ -107,16 +108,16 @@ def test_patch_resource_does_not_work_for_random_parameters():
class ImportantBusinessLogic:
def __init__(self):
def __init__(self) -> None:
self._s3 = boto3.client("s3", region_name="us-east-1")
def do_important_things(self):
def do_important_things(self) -> List[str]:
return self._s3.list_buckets()["Buckets"]
def test_mock_works_when_replacing_client(
aws_credentials,
): # pylint: disable=unused-argument
aws_credentials: Any, # pylint: disable=unused-argument
) -> None:
logic = ImportantBusinessLogic()
m = mock_s3()

8
tests/test_core/test_instance_metadata.py
View File

@ -9,13 +9,13 @@ else:
@mock_ec2
def test_latest_meta_data():
def test_latest_meta_data() -> None:
res = requests.get(f"{BASE_URL}/latest/meta-data/")
assert res.content == b"iam"
@mock_ec2
def test_meta_data_iam():
def test_meta_data_iam() -> None:
res = requests.get(f"{BASE_URL}/latest/meta-data/iam")
json_response = res.json()
default_role = json_response["security-credentials"]["default-role"]
@ -26,13 +26,13 @@ def test_meta_data_iam():
@mock_ec2
def test_meta_data_security_credentials():
def test_meta_data_security_credentials() -> None:
res = requests.get(f"{BASE_URL}/latest/meta-data/iam/security-credentials/")
assert res.content == b"default-role"
@mock_ec2
def test_meta_data_default_role():
def test_meta_data_default_role() -> None:
res = requests.get(
f"{BASE_URL}/latest/meta-data/iam/security-credentials/default-role"
)

10
tests/test_core/test_mock_regions.py
View File

@ -8,7 +8,7 @@ from moto import mock_dynamodb, mock_sns, settings
@mock_sns
def test_use_invalid_region():
def test_use_invalid_region() -> None:
if settings.TEST_SERVER_MODE:
raise SkipTest("ServerMode will throw different errors")
client = boto3.client("sns", region_name="any-region")
@ -19,14 +19,14 @@ def test_use_invalid_region():
@mock_sns
@mock.patch.dict(os.environ, {"AWS_DEFAULT_REGION": "us-east-2"})
def test_use_region_from_env():
def test_use_region_from_env() -> None: # type: ignore[misc]
client = boto3.client("sns")
assert client.list_platform_applications()["PlatformApplications"] == []
@mock_sns
@mock.patch.dict(os.environ, {"AWS_DEFAULT_REGION": "any-region"})
def test_use_unknown_region_from_env():
def test_use_unknown_region_from_env() -> None: # type: ignore[misc]
if settings.TEST_SERVER_MODE:
raise SkipTest("Cannot set environemnt variables in ServerMode")
client = boto3.client("sns")
@ -38,7 +38,7 @@ def test_use_unknown_region_from_env():
@mock_sns
@mock.patch.dict(os.environ, {"AWS_DEFAULT_REGION": "any-region"})
@mock.patch.dict(os.environ, {"MOTO_ALLOW_NONEXISTENT_REGION": "trUe"})
def test_use_unknown_region_from_env_but_allow_it():
def test_use_unknown_region_from_env_but_allow_it() -> None: # type: ignore[misc]
if settings.TEST_SERVER_MODE:
raise SkipTest("Cannot set environemnt variables in ServerMode")
client = boto3.client("sns")
@ -47,7 +47,7 @@ def test_use_unknown_region_from_env_but_allow_it():
@mock_dynamodb
@mock.patch.dict(os.environ, {"MOTO_ALLOW_NONEXISTENT_REGION": "trUe"})
def test_use_unknown_region_from_env_but_allow_it__dynamo():
def test_use_unknown_region_from_env_but_allow_it__dynamo() -> None: # type: ignore[misc]
if settings.TEST_SERVER_MODE:
raise SkipTest("Cannot set environemnt variables in ServerMode")
dynamo_db = boto3.resource("dynamodb", region_name="test")

32
tests/test_core/test_moto_api.py
View File

@ -18,7 +18,7 @@ data_url = f"{base_url}/moto-api/data.json"
@mock_sqs
def test_reset_api():
def test_reset_api() -> None:
conn = boto3.client("sqs", region_name="us-west-1")
conn.create_queue(QueueName="queue1")
assert len(conn.list_queues()["QueueUrls"]) == 1
@ -30,7 +30,7 @@ def test_reset_api():
@mock_sqs
def test_data_api():
def test_data_api() -> None:
conn = boto3.client("sqs", region_name="us-west-1")
conn.create_queue(QueueName="queue1")
@ -41,7 +41,7 @@ def test_data_api():
@mock_s3
def test_overwriting_s3_object_still_returns_data():
def test_overwriting_s3_object_still_returns_data() -> None:
if settings.TEST_SERVER_MODE:
raise SkipTest("No point in testing this behaves the same in ServerMode")
s3 = boto3.client("s3", region_name="us-east-1")
@ -53,7 +53,7 @@ def test_overwriting_s3_object_still_returns_data():
@mock_autoscaling
def test_creation_error__data_api_still_returns_thing():
def test_creation_error__data_api_still_returns_thing() -> None:
if settings.TEST_SERVER_MODE:
raise SkipTest("No point in testing this behaves the same in ServerMode")
# Timeline:
@ -78,7 +78,7 @@ def test_creation_error__data_api_still_returns_thing():
from moto.moto_api._internal.urls import response_instance
_, _, x = response_instance.model_data(None, None, None)
_, _, x = response_instance.model_data(None, "None", None)
as_objects = json.loads(x)["autoscaling"]
assert len(as_objects["FakeAutoScalingGroup"]) >= 1
@ -87,7 +87,7 @@ def test_creation_error__data_api_still_returns_thing():
assert "test_asg" in names
def test_model_data_is_emptied_as_necessary():
def test_model_data_is_emptied_as_necessary() -> None:
if settings.TEST_SERVER_MODE:
raise SkipTest("We're only interested in the decorator performance here")
@ -97,24 +97,24 @@ def test_model_data_is_emptied_as_necessary():
# No instances exist, because we have just reset it
for classes_per_service in model_data.values():
for _class in classes_per_service.values():
assert _class.instances == []
assert _class.instances == [] # type: ignore[attr-defined]
with mock_sqs():
# When just starting a mock, it is empty
for classes_per_service in model_data.values():
for _class in classes_per_service.values():
assert _class.instances == []
assert _class.instances == [] # type: ignore[attr-defined]
# After creating a queue, some data will be present
conn = boto3.client("sqs", region_name="us-west-1")
conn.create_queue(QueueName="queue1")
assert len(model_data["sqs"]["Queue"].instances) == 1
assert len(model_data["sqs"]["Queue"].instances) == 1 # type: ignore[attr-defined]
# But after the mock ends, it is empty again
for classes_per_service in model_data.values():
for _class in classes_per_service.values():
assert _class.instances == []
assert _class.instances == [] # type: ignore[attr-defined]
# When we have multiple/nested mocks, the data should still be present after the first mock ends
with mock_sqs():
@ -122,24 +122,24 @@ def test_model_data_is_emptied_as_necessary():
conn.create_queue(QueueName="queue1")
with mock_s3():
# The data should still be here - instances should not reset if another mock is still active
assert len(model_data["sqs"]["Queue"].instances) == 1
assert len(model_data["sqs"]["Queue"].instances) == 1 # type: ignore[attr-defined]
# The data should still be here - the inner mock has exited, but the outer mock is still active
assert len(model_data["sqs"]["Queue"].instances) == 1
assert len(model_data["sqs"]["Queue"].instances) == 1 # type: ignore[attr-defined]
@mock_sqs
class TestModelDataResetForClassDecorator(TestCase):
def setUp(self):
def setUp(self) -> None:
if settings.TEST_SERVER_MODE:
raise SkipTest("We're only interested in the decorator performance here")
# No data is present at the beginning
for classes_per_service in model_data.values():
for _class in classes_per_service.values():
assert _class.instances == []
assert _class.instances == [] # type: ignore[attr-defined]
conn = boto3.client("sqs", region_name="us-west-1")
conn.create_queue(QueueName="queue1")
def test_should_find_bucket(self):
assert len(model_data["sqs"]["Queue"].instances) == 1
def test_should_find_bucket(self) -> None:
assert len(model_data["sqs"]["Queue"].instances) == 1 # type: ignore[attr-defined]

4
tests/test_core/test_nested.py
View File

@ -8,7 +8,7 @@ from tests import EXAMPLE_AMI_ID
class TestNestedDecoratorsBoto3(unittest.TestCase):
@mock_sqs
def setup_sqs_queue(self):
def setup_sqs_queue(self) -> None:
conn = boto3.resource("sqs", region_name="us-east-1")
queue = conn.create_queue(QueueName="some-queue")
@ -18,7 +18,7 @@ class TestNestedDecoratorsBoto3(unittest.TestCase):
assert queue.attributes["ApproximateNumberOfMessages"] == "1"
@mock_ec2
def test_nested(self):
def test_nested(self) -> None:
self.setup_sqs_queue()
conn = boto3.client("ec2", region_name="us-west-2")

39
tests/test_core/test_responses.py
View File

@ -1,9 +1,10 @@
import datetime
from collections import OrderedDict
from gzip import compress as gzip_compress
from typing import Any, Dict
from unittest import SkipTest, mock
from botocore.awsrequest import AWSPreparedRequest
from botocore.awsrequest import AWSPreparedRequest, HTTPHeaders
from freezegun import freeze_time
from moto import settings
@ -11,10 +12,10 @@ from moto.core.responses import AWSServiceSpec, BaseResponse, flatten_json_reque
from moto.s3.responses import S3Response
def test_flatten_json_request_body():
def test_flatten_json_request_body() -> None:
spec = AWSServiceSpec("data/emr/2009-03-31/service-2.json").input_spec("RunJobFlow")
body = {
body: Dict[str, Any] = {
"Name": "cluster",
"Instances": {
"Ec2KeyName": "ec2key",
@ -93,13 +94,15 @@ def test_flatten_json_request_body():
assert props == body["Configurations"][idx]["Properties"]
def test_parse_qs_unicode_decode_error():
def test_parse_qs_unicode_decode_error() -> None:
body = b'{"key": "%D0"}, "C": "#0 = :0"}'
request = AWSPreparedRequest("GET", "http://request", {"foo": "bar"}, body, False)
headers = HTTPHeaders()
headers["foo"] = "bar"
request = AWSPreparedRequest("GET", "http://request", headers, body, False)
BaseResponse().setup_class(request, request.url, request.headers)
def test_get_params():
def test_get_params() -> None:
subject = BaseResponse()
subject.querystring = OrderedDict(
[
@ -150,7 +153,7 @@ def test_get_params():
}
def test_get_dict_list_params():
def test_get_dict_list_params() -> None:
subject = BaseResponse()
subject.querystring = OrderedDict(
[
@ -168,7 +171,7 @@ def test_get_dict_list_params():
assert result == {"VpcSecurityGroupId": ["sg-123", "sg-456", "sg-789"]}
def test_response_environment_preserved_by_type():
def test_response_environment_preserved_by_type() -> None:
"""Ensure Jinja environment is cached by response type."""
class ResponseA(BaseResponse):
@ -214,7 +217,7 @@ def test_response_environment_preserved_by_type():
"moto.core.responses.settings.PRETTIFY_RESPONSES",
new_callable=mock.PropertyMock(return_value=True),
)
def test_jinja_render_prettify(m_env_var):
def test_jinja_render_prettify(m_env_var: Any) -> None: # type: ignore[misc]
if settings.TEST_SERVER_MODE:
raise SkipTest(
"It is not possible to set the environment variable in server mode"
@ -228,12 +231,12 @@ def test_jinja_render_prettify(m_env_var):
assert m_env_var
def test_response_metadata():
def test_response_metadata() -> None:
# Setup
frozen_time = datetime.datetime(
2023, 5, 20, 10, 20, 30, tzinfo=datetime.timezone.utc
)
request = AWSPreparedRequest("GET", "http://request", {}, None, False)
request = AWSPreparedRequest("GET", "http://request", HTTPHeaders(), None, False)
# Execute
with freeze_time(frozen_time):
@ -246,12 +249,14 @@ def test_response_metadata():
assert bc.response_headers["date"] == "Sat, 20 May 2023 10:20:30 GMT"
def test_compression_gzip():
def test_compression_gzip() -> None:
body = '{"key": "%D0"}, "C": "#0 = :0"}'
headers = HTTPHeaders()
headers["Content-Encoding"] = "gzip"
request = AWSPreparedRequest(
"GET",
url="http://request",
headers={"Content-Encoding": "gzip"},
headers=headers,
body=_gzip_compress_body(body),
stream_output=False,
)
@ -261,12 +266,14 @@ def test_compression_gzip():
assert body == response.body
def test_compression_gzip_in_s3():
def test_compression_gzip_in_s3() -> None:
body = b"some random data"
headers = HTTPHeaders()
headers["Content-Encoding"] = "gzip"
request = AWSPreparedRequest(
"GET",
url="http://request",
headers={"Content-Encoding": "gzip"},
headers=headers,
body=body,
stream_output=False,
)
@ -276,6 +283,6 @@ def test_compression_gzip_in_s3():
assert body == response.body.encode("utf-8")
def _gzip_compress_body(body: str):
def _gzip_compress_body(body: str) -> bytes:
assert isinstance(body, str)
return gzip_compress(data=body.encode("utf-8"))

11
tests/test_core/test_server.py
View File

@ -1,9 +1,10 @@
from typing import Any
from unittest.mock import patch
from moto.server import DomainDispatcherApplication, create_backend_app, main
def test_wrong_arguments():
def test_wrong_arguments() -> None:
try:
main(["name", "test1", "test2", "test3"])
assert False, (
@ -15,7 +16,7 @@ def test_wrong_arguments():
@patch("moto.server.run_simple")
def test_right_arguments(run_simple):
def test_right_arguments(run_simple: Any) -> None: # type: ignore[misc]
main(["s3"])
func_call = run_simple.call_args[0]
assert func_call[0] == "127.0.0.1"
@ -23,14 +24,14 @@ def test_right_arguments(run_simple):
@patch("moto.server.run_simple")
def test_port_argument(run_simple):
def test_port_argument(run_simple: Any) -> None: # type: ignore[misc]
main(["s3", "--port", "8080"])
func_call = run_simple.call_args[0]
assert func_call[0] == "127.0.0.1"
assert func_call[1] == 8080
def test_domain_dispatched():
def test_domain_dispatched() -> None:
dispatcher = DomainDispatcherApplication(create_backend_app)
backend_app = dispatcher.get_application(
{"HTTP_HOST": "email.us-east1.amazonaws.com"}
@ -39,7 +40,7 @@ def test_domain_dispatched():
assert keys[0] == "EmailResponse.dispatch"
def test_domain_dispatched_with_service():
def test_domain_dispatched_with_service() -> None:
# If we pass a particular service, always return that.
dispatcher = DomainDispatcherApplication(create_backend_app, service="s3")
backend_app = dispatcher.get_application({"HTTP_HOST": "s3.us-east1.amazonaws.com"})

6
tests/test_core/test_settings.py
View File

@ -10,17 +10,17 @@ Sanity checks for interpretation of the MOTO_ECS_NEW_ARN-variable
"""
def test_default_is_true():
def test_default_is_true() -> None:
assert settings.ecs_new_arn_format() is True
@pytest.mark.parametrize("value", ["TrUe", "true", "invalid", "0", "1"])
def test_anything_but_false_is_true(value):
def test_anything_but_false_is_true(value: str) -> None:
with mock.patch.dict(os.environ, {"MOTO_ECS_NEW_ARN": value}):
assert settings.ecs_new_arn_format() is True
@pytest.mark.parametrize("value", ["False", "false", "faLse"])
def test_only_false_is_false(value):
def test_only_false_is_false(value: str) -> None:
with mock.patch.dict(os.environ, {"MOTO_ECS_NEW_ARN": value}):
assert settings.ecs_new_arn_format() is False

2
tests/test_core/test_socket.py
View File

@ -6,7 +6,7 @@ from moto import mock_dynamodb
class TestSocketPair(unittest.TestCase):
@mock_dynamodb
def test_socket_pair(self):
def test_socket_pair(self) -> None:
a, b = socket.socketpair()
self.assertIsNotNone(a)
self.assertIsNotNone(b)

6
tests/test_core/test_url_base_regex.py
View File

@ -5,7 +5,7 @@ import moto
from moto import mock_s3
service_names = [
(d[5:], "")
d[5:]
for d in dir(moto)
if d.startswith("mock_") and not d == "mock_xray_client" and not d == "mock_all"
]
@ -16,8 +16,8 @@ class TestMockBucketStartingWithServiceName:
https://github.com/getmoto/moto/issues/4099
"""
@pytest.mark.parametrize("service_name,decorator", service_names)
def test_bucketname_starting_with_service_name(self, service_name, decorator):
@pytest.mark.parametrize("service_name", service_names)
def test_bucketname_starting_with_service_name(self, service_name: str) -> None:
decorator = getattr(moto, f"mock_{service_name}")
with decorator():
with mock_s3():

4
tests/test_core/test_url_mapping.py
View File

@ -1,7 +1,7 @@
from moto.core.utils import convert_regex_to_flask_path
def test_flask_path_converting_simple():
def test_flask_path_converting_simple() -> None:
assert convert_regex_to_flask_path("/") == "/"
assert convert_regex_to_flask_path("/$") == "/"
@ -10,7 +10,7 @@ def test_flask_path_converting_simple():
assert convert_regex_to_flask_path("/foo/bar/") == "/foo/bar/"
def test_flask_path_converting_regex():
def test_flask_path_converting_regex() -> None:
assert (
convert_regex_to_flask_path(r"/(?P<key_name>[a-zA-Z0-9\-_]+)")
== r'/<regex("[a-zA-Z0-9\-_]+"):key_name>'

10
tests/test_core/test_utils.py
View File

@ -19,7 +19,7 @@ from moto.core.utils import (
("ListMFADevices", "list_mfa_devices"),
],
)
def test_camelcase_to_underscores(_input, expected):
def test_camelcase_to_underscores(_input: str, expected: str) -> None:
assert camelcase_to_underscores(_input) == expected
@ -27,7 +27,7 @@ def test_camelcase_to_underscores(_input, expected):
"_input,expected",
[("the_new_attribute", "theNewAttribute"), ("attribute", "attribute")],
)
def test_underscores_to_camelcase(_input, expected):
def test_underscores_to_camelcase(_input: str, expected: str) -> None:
assert underscores_to_camelcase(_input) == expected
@ -35,7 +35,7 @@ def test_underscores_to_camelcase(_input, expected):
"_input,expected",
[("TheNewAttribute", "theNewAttribute"), ("Attribute", "attribute")],
)
def test_pascal_to_camelcase(_input, expected):
def test_pascal_to_camelcase(_input: str, expected: str) -> None:
assert pascal_to_camelcase(_input) == expected
@ -43,10 +43,10 @@ def test_pascal_to_camelcase(_input, expected):
"_input,expected",
[("theNewAttribute", "TheNewAttribute"), ("attribute", "Attribute")],
)
def test_camelcase_to_pascal(_input, expected):
def test_camelcase_to_pascal(_input: str, expected: str) -> None:
assert camelcase_to_pascal(_input) == expected
@freeze_time("2015-01-01 12:00:00")
def test_unix_time():
def test_unix_time() -> None: # type: ignore[misc]
assert unix_time() == 1420113600.0