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

Techdebt: MyPy S (#6261)

Browse Source
This commit is contained in:
Bert Blommers 2023-04-26 22:20:28 +00:00 committed by GitHub
parent 37f1456747
commit f38babb026
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
24 changed files with 480 additions and 406 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
moto/sdb/exceptions.py
View File

@ -1,4 +1,5 @@
"""Exceptions raised by the sdb service."""
from typing import Any
from moto.core.exceptions import RESTError
@ -18,7 +19,7 @@ SDB_ERROR = """<?xml version="1.0"?>
class InvalidParameterError(RESTError):
code = 400
def __init__(self, **kwargs):
def __init__(self, **kwargs: Any):
kwargs.setdefault("template", "sdb_error")
self.templates["sdb_error"] = SDB_ERROR
kwargs["error_type"] = "InvalidParameterValue"
@ -28,7 +29,7 @@ class InvalidParameterError(RESTError):
class InvalidDomainName(InvalidParameterError):
code = 400
def __init__(self, domain_name):
def __init__(self, domain_name: str):
super().__init__(
message=f"Value ({domain_name}) for parameter DomainName is invalid. "
)
@ -37,7 +38,7 @@ class InvalidDomainName(InvalidParameterError):
class UnknownDomainName(RESTError):
code = 400
def __init__(self, **kwargs):
def __init__(self, **kwargs: Any):
kwargs.setdefault("template", "sdb_error")
self.templates["sdb_error"] = SDB_ERROR
kwargs["error_type"] = "NoSuchDomain"

43
moto/sdb/models.py
View File

@ -1,23 +1,24 @@
"""SimpleDBBackend class with methods for supported APIs."""
import re
from collections import defaultdict
from moto.core import BaseBackend, BackendDict, BaseModel
from threading import Lock
from typing import Any, Dict, List, Iterable, Optional
from moto.core import BaseBackend, BackendDict, BaseModel
from .exceptions import InvalidDomainName, UnknownDomainName
class FakeItem(BaseModel):
def __init__(self):
self.attributes = []
def __init__(self) -> None:
self.attributes: List[Dict[str, Any]] = []
self.lock = Lock()
def get_attributes(self, names):
def get_attributes(self, names: Optional[List[str]]) -> List[Dict[str, Any]]:
if not names:
return self.attributes
return [attr for attr in self.attributes if attr["name"] in names]
def put_attributes(self, attributes):
def put_attributes(self, attributes: List[Dict[str, Any]]) -> None:
# Replacing attributes involves quite a few loops
# Lock this, so we know noone else touches this list while we're operating on it
with self.lock:
@ -26,56 +27,58 @@ class FakeItem(BaseModel):
self._remove_attributes(attr["name"])
self.attributes.append(attr)
def _remove_attributes(self, name):
def _remove_attributes(self, name: str) -> None:
self.attributes = [attr for attr in self.attributes if attr["name"] != name]
class FakeDomain(BaseModel):
def __init__(self, name):
def __init__(self, name: str):
self.name = name
self.items = defaultdict(FakeItem)
self.items: Dict[str, FakeItem] = defaultdict(FakeItem)
def get(self, item_name, attribute_names):
def get(self, item_name: str, attribute_names: List[str]) -> List[Dict[str, Any]]:
item = self.items[item_name]
return item.get_attributes(attribute_names)
def put(self, item_name, attributes):
def put(self, item_name: str, attributes: List[Dict[str, Any]]) -> None:
item = self.items[item_name]
item.put_attributes(attributes)
class SimpleDBBackend(BaseBackend):
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, FakeDomain] = dict()
def create_domain(self, domain_name):
def create_domain(self, domain_name: str) -> None:
self._validate_domain_name(domain_name)
self.domains[domain_name] = FakeDomain(name=domain_name)
def list_domains(self):
def list_domains(self) -> Iterable[str]:
"""
The `max_number_of_domains` and `next_token` parameter have not been implemented yet - we simply return all domains.
"""
return self.domains.keys()
def delete_domain(self, domain_name):
def delete_domain(self, domain_name: str) -> None:
self._validate_domain_name(domain_name)
# Ignore unknown domains - AWS does the same
self.domains.pop(domain_name, None)
def _validate_domain_name(self, domain_name):
def _validate_domain_name(self, domain_name: str) -> None:
# Domain Name needs to have at least 3 chars
# Can only contain characters: a-z, A-Z, 0-9, '_', '-', and '.'
if not re.match("^[a-zA-Z0-9-_.]{3,}$", domain_name):
raise InvalidDomainName(domain_name)
def _get_domain(self, domain_name):
def _get_domain(self, domain_name: str) -> FakeDomain:
if domain_name not in self.domains:
raise UnknownDomainName()
return self.domains[domain_name]
def get_attributes(self, domain_name, item_name, attribute_names):
def get_attributes(
self, domain_name: str, item_name: str, attribute_names: List[str]
) -> List[Dict[str, Any]]:
"""
Behaviour for the consistent_read-attribute is not yet implemented
"""
@ -83,7 +86,9 @@ class SimpleDBBackend(BaseBackend):
domain = self._get_domain(domain_name)
return domain.get(item_name, attribute_names)
def put_attributes(self, domain_name, item_name, attributes):
def put_attributes(
self, domain_name: str, item_name: str, attributes: List[Dict[str, Any]]
) -> None:
"""
Behaviour for the expected-attribute is not yet implemented.
"""

16
moto/sdb/responses.py
View File

@ -1,33 +1,33 @@
from moto.core.responses import BaseResponse
from .models import sdb_backends
from .models import sdb_backends, SimpleDBBackend
class SimpleDBResponse(BaseResponse):
def __init__(self):
def __init__(self) -> None:
super().__init__(service_name="sdb")
@property
def sdb_backend(self):
def sdb_backend(self) -> SimpleDBBackend:
return sdb_backends[self.current_account][self.region]
def create_domain(self):
def create_domain(self) -> str:
domain_name = self._get_param("DomainName")
self.sdb_backend.create_domain(domain_name=domain_name)
template = self.response_template(CREATE_DOMAIN_TEMPLATE)
return template.render()
def delete_domain(self):
def delete_domain(self) -> str:
domain_name = self._get_param("DomainName")
self.sdb_backend.delete_domain(domain_name=domain_name)
template = self.response_template(DELETE_DOMAIN_TEMPLATE)
return template.render()
def list_domains(self):
def list_domains(self) -> str:
domain_names = self.sdb_backend.list_domains()
template = self.response_template(LIST_DOMAINS_TEMPLATE)
return template.render(domain_names=domain_names, next_token=None)
def get_attributes(self):
def get_attributes(self) -> str:
domain_name = self._get_param("DomainName")
item_name = self._get_param("ItemName")
attribute_names = self._get_multi_param("AttributeName.")
@ -39,7 +39,7 @@ class SimpleDBResponse(BaseResponse):
template = self.response_template(GET_ATTRIBUTES_TEMPLATE)
return template.render(attributes=attributes)
def put_attributes(self):
def put_attributes(self) -> str:
domain_name = self._get_param("DomainName")
item_name = self._get_param("ItemName")
attributes = self._get_list_prefix("Attribute")

9
moto/server.py
View File

@ -3,6 +3,7 @@ import os
import signal
import sys
import warnings
from typing import Any, List, Optional
from werkzeug.serving import run_simple
@ -15,11 +16,11 @@ from moto.moto_server.threaded_moto_server import ( # noqa # pylint: disable=un
)
def signal_handler(signum, frame): # pylint: disable=unused-argument
def signal_handler(signum: Any, frame: Any) -> None: # pylint: disable=unused-argument
sys.exit(0)
def main(argv=None):
def main(argv: Optional[List[str]] = None) -> None:
argv = argv or sys.argv[1:]
parser = argparse.ArgumentParser()
@ -79,9 +80,9 @@ def main(argv=None):
# Wrap the main application
main_app = DomainDispatcherApplication(create_backend_app, service=args.service)
main_app.debug = True
main_app.debug = True # type: ignore
ssl_context = None
ssl_context: Any = None
if args.ssl_key and args.ssl_cert:
ssl_context = (args.ssl_cert, args.ssl_key)
elif args.ssl:

8
moto/servicediscovery/exceptions.py
View File

@ -3,20 +3,20 @@ from moto.core.exceptions import JsonRESTError
class OperationNotFound(JsonRESTError):
def __init__(self):
def __init__(self) -> None:
super().__init__("OperationNotFound", "")
class NamespaceNotFound(JsonRESTError):
def __init__(self, ns_id):
def __init__(self, ns_id: str):
super().__init__("NamespaceNotFound", f"{ns_id}")
class ServiceNotFound(JsonRESTError):
def __init__(self, ns_id):
def __init__(self, ns_id: str):
super().__init__("ServiceNotFound", f"{ns_id}")
class ConflictingDomainExists(JsonRESTError):
def __init__(self, vpc_id):
def __init__(self, vpc_id: str):
super().__init__("ConflictingDomainExists", f"{vpc_id}")

141
moto/servicediscovery/models.py
View File

@ -1,4 +1,5 @@
import string
from typing import Any, Dict, Iterable, List, Optional
from moto.core import BaseBackend, BackendDict, BaseModel
from moto.core.utils import unix_time
@ -13,7 +14,7 @@ from .exceptions import (
)
def random_id(size):
def random_id(size: int) -> str:
return "".join(
[random.choice(string.ascii_lowercase + string.digits) for _ in range(size)]
)
@ -22,17 +23,16 @@ def random_id(size):
class Namespace(BaseModel):
def __init__(
self,
account_id,
region,
name,
ns_type,
creator_request_id,
description,
dns_properties,
http_properties,
vpc=None,
account_id: str,
region: str,
name: str,
ns_type: str,
creator_request_id: str,
description: str,
dns_properties: Dict[str, Any],
http_properties: Dict[str, Any],
vpc: Optional[str] = None,
):
super().__init__()
self.id = f"ns-{random_id(20)}"
self.arn = f"arn:aws:servicediscovery:{region}:{account_id}:namespace/{self.id}"
self.name = name
@ -45,7 +45,7 @@ class Namespace(BaseModel):
self.created = unix_time()
self.updated = unix_time()
def to_json(self):
def to_json(self) -> Dict[str, Any]:
return {
"Arn": self.arn,
"Id": self.id,
@ -65,31 +65,30 @@ class Namespace(BaseModel):
class Service(BaseModel):
def __init__(
self,
account_id,
region,
name,
namespace_id,
description,
creator_request_id,
dns_config,
health_check_config,
health_check_custom_config,
service_type,
account_id: str,
region: str,
name: str,
namespace_id: str,
description: str,
creator_request_id: str,
dns_config: Dict[str, Any],
health_check_config: Dict[str, Any],
health_check_custom_config: Dict[str, int],
service_type: str,
):
super().__init__()
self.id = f"srv-{random_id(8)}"
self.arn = f"arn:aws:servicediscovery:{region}:{account_id}:service/{self.id}"
self.name = name
self.namespace_id = namespace_id
self.description = description
self.creator_request_id = creator_request_id
self.dns_config = dns_config
self.dns_config: Optional[Dict[str, Any]] = dns_config
self.health_check_config = health_check_config
self.health_check_custom_config = health_check_custom_config
self.service_type = service_type
self.created = unix_time()
def update(self, details):
def update(self, details: Dict[str, Any]) -> None:
if "Description" in details:
self.description = details["Description"]
if "DnsConfig" in details:
@ -104,7 +103,7 @@ class Service(BaseModel):
if "HealthCheckConfig" in details:
self.health_check_config = details["HealthCheckConfig"]
def to_json(self):
def to_json(self) -> Dict[str, Any]:
return {
"Arn": self.arn,
"Id": self.id,
@ -121,7 +120,7 @@ class Service(BaseModel):
class Operation(BaseModel):
def __init__(self, operation_type, targets):
def __init__(self, operation_type: str, targets: Dict[str, str]):
super().__init__()
self.id = f"{random_id(32)}-{random_id(8)}"
self.status = "SUCCESS"
@ -130,7 +129,7 @@ class Operation(BaseModel):
self.updated = unix_time()
self.targets = targets
def to_json(self, short=False):
def to_json(self, short: bool = False) -> Dict[str, Any]:
if short:
return {"Id": self.id, "Status": self.status}
else:
@ -147,20 +146,26 @@ class Operation(BaseModel):
class ServiceDiscoveryBackend(BaseBackend):
"""Implementation of ServiceDiscovery APIs."""
def __init__(self, region_name, account_id):
def __init__(self, region_name: str, account_id: str):
super().__init__(region_name, account_id)
self.operations = dict()
self.namespaces = dict()
self.services = dict()
self.operations: Dict[str, Operation] = dict()
self.namespaces: Dict[str, Namespace] = dict()
self.services: Dict[str, Service] = dict()
self.tagger = TaggingService()
def list_namespaces(self):
def list_namespaces(self) -> Iterable[Namespace]:
"""
Pagination or the Filters-parameter is not yet implemented
"""
return self.namespaces.values()
def create_http_namespace(self, name, creator_request_id, description, tags):
def create_http_namespace(
self,
name: str,
creator_request_id: str,
description: str,
tags: List[Dict[str, str]],
) -> str:
namespace = Namespace(
account_id=self.account_id,
region=self.region_name,
@ -179,13 +184,12 @@ class ServiceDiscoveryBackend(BaseBackend):
)
return operation_id
def _create_operation(self, op_type, targets):
def _create_operation(self, op_type: str, targets: Dict[str, str]) -> str:
operation = Operation(operation_type=op_type, targets=targets)
self.operations[operation.id] = operation
operation_id = operation.id
return operation_id
return operation.id
def delete_namespace(self, namespace_id):
def delete_namespace(self, namespace_id: str) -> str:
if namespace_id not in self.namespaces:
raise NamespaceNotFound(namespace_id)
del self.namespaces[namespace_id]
@ -194,12 +198,12 @@ class ServiceDiscoveryBackend(BaseBackend):
)
return operation_id
def get_namespace(self, namespace_id):
def get_namespace(self, namespace_id: str) -> Namespace:
if namespace_id not in self.namespaces:
raise NamespaceNotFound(namespace_id)
return self.namespaces[namespace_id]
def list_operations(self):
def list_operations(self) -> Iterable[Operation]:
"""
Pagination or the Filters-argument is not yet implemented
"""
@ -211,23 +215,31 @@ class ServiceDiscoveryBackend(BaseBackend):
}
return self.operations.values()
def get_operation(self, operation_id):
def get_operation(self, operation_id: str) -> Operation:
if operation_id not in self.operations:
raise OperationNotFound()
return self.operations[operation_id]
def tag_resource(self, resource_arn, tags):
def tag_resource(self, resource_arn: str, tags: List[Dict[str, str]]) -> None:
self.tagger.tag_resource(resource_arn, tags)
def untag_resource(self, resource_arn, tag_keys):
def untag_resource(self, resource_arn: str, tag_keys: List[str]) -> None:
self.tagger.untag_resource_using_names(resource_arn, tag_keys)
def list_tags_for_resource(self, resource_arn):
def list_tags_for_resource(
self, resource_arn: str
) -> Dict[str, List[Dict[str, str]]]:
return self.tagger.list_tags_for_resource(resource_arn)
def create_private_dns_namespace(
self, name, creator_request_id, description, vpc, tags, properties
):
self,
name: str,
creator_request_id: str,
description: str,
vpc: str,
tags: List[Dict[str, str]],
properties: Dict[str, Any],
) -> str:
for namespace in self.namespaces.values():
if namespace.vpc == vpc:
raise ConflictingDomainExists(vpc)
@ -253,8 +265,13 @@ class ServiceDiscoveryBackend(BaseBackend):
return operation_id
def create_public_dns_namespace(
self, name, creator_request_id, description, tags, properties
):
self,
name: str,
creator_request_id: str,
description: str,
tags: List[Dict[str, str]],
properties: Dict[str, Any],
) -> str:
dns_properties = (properties or {}).get("DnsProperties", {})
dns_properties["HostedZoneId"] = "hzi"
namespace = Namespace(
@ -277,16 +294,16 @@ class ServiceDiscoveryBackend(BaseBackend):
def create_service(
self,
name,
namespace_id,
creator_request_id,
description,
dns_config,
health_check_config,
health_check_custom_config,
tags,
service_type,
):
name: str,
namespace_id: str,
creator_request_id: str,
description: str,
dns_config: Dict[str, Any],
health_check_config: Dict[str, Any],
health_check_custom_config: Dict[str, Any],
tags: List[Dict[str, str]],
service_type: str,
) -> Service:
service = Service(
account_id=self.account_id,
region=self.region_name,
@ -304,21 +321,21 @@ class ServiceDiscoveryBackend(BaseBackend):
self.tagger.tag_resource(service.arn, tags)
return service
def get_service(self, service_id):
def get_service(self, service_id: str) -> Service:
if service_id not in self.services:
raise ServiceNotFound(service_id)
return self.services[service_id]
def delete_service(self, service_id):
def delete_service(self, service_id: str) -> None:
self.services.pop(service_id, None)
def list_services(self):
def list_services(self) -> Iterable[Service]:
"""
Pagination or the Filters-argument is not yet implemented
"""
return self.services.values()
def update_service(self, service_id, details):
def update_service(self, service_id: str, details: Dict[str, Any]) -> str:
service = self.get_service(service_id)
service.update(details=details)
operation_id = self._create_operation(

45
moto/servicediscovery/responses.py
View File

@ -1,24 +1,25 @@
"""Handles incoming servicediscovery requests, invokes methods, returns responses."""
import json
from moto.core.common_types import TYPE_RESPONSE
from moto.core.responses import BaseResponse
from .models import servicediscovery_backends
from .models import servicediscovery_backends, ServiceDiscoveryBackend
class ServiceDiscoveryResponse(BaseResponse):
def __init__(self):
def __init__(self) -> None:
super().__init__(service_name="servicediscovery")
@property
def servicediscovery_backend(self):
def servicediscovery_backend(self) -> ServiceDiscoveryBackend:
"""Return backend instance specific for this region."""
return servicediscovery_backends[self.current_account][self.region]
def list_namespaces(self):
def list_namespaces(self) -> TYPE_RESPONSE:
namespaces = self.servicediscovery_backend.list_namespaces()
return 200, {}, json.dumps({"Namespaces": [ns.to_json() for ns in namespaces]})
def create_http_namespace(self):
def create_http_namespace(self) -> str:
params = json.loads(self.body)
name = params.get("Name")
creator_request_id = params.get("CreatorRequestId")
@ -32,7 +33,7 @@ class ServiceDiscoveryResponse(BaseResponse):
)
return json.dumps(dict(OperationId=operation_id))
def delete_namespace(self):
def delete_namespace(self) -> str:
params = json.loads(self.body)
namespace_id = params.get("Id")
operation_id = self.servicediscovery_backend.delete_namespace(
@ -40,7 +41,7 @@ class ServiceDiscoveryResponse(BaseResponse):
)
return json.dumps(dict(OperationId=operation_id))
def list_operations(self):
def list_operations(self) -> TYPE_RESPONSE:
operations = self.servicediscovery_backend.list_operations()
return (
200,
@ -48,7 +49,7 @@ class ServiceDiscoveryResponse(BaseResponse):
json.dumps({"Operations": [o.to_json(short=True) for o in operations]}),
)
def get_operation(self):
def get_operation(self) -> str:
params = json.loads(self.body)
operation_id = params.get("OperationId")
operation = self.servicediscovery_backend.get_operation(
@ -56,7 +57,7 @@ class ServiceDiscoveryResponse(BaseResponse):
)
return json.dumps(dict(Operation=operation.to_json()))
def get_namespace(self):
def get_namespace(self) -> str:
params = json.loads(self.body)
namespace_id = params.get("Id")
namespace = self.servicediscovery_backend.get_namespace(
@ -64,23 +65,23 @@ class ServiceDiscoveryResponse(BaseResponse):
)
return json.dumps(dict(Namespace=namespace.to_json()))
def tag_resource(self):
def tag_resource(self) -> str:
params = json.loads(self.body)
resource_arn = params.get("ResourceARN")
tags = params.get("Tags")
self.servicediscovery_backend.tag_resource(resource_arn=resource_arn, tags=tags)
return json.dumps(dict())
return "{}"
def untag_resource(self):
def untag_resource(self) -> str:
params = json.loads(self.body)
resource_arn = params.get("ResourceARN")
tag_keys = params.get("TagKeys")
self.servicediscovery_backend.untag_resource(
resource_arn=resource_arn, tag_keys=tag_keys
)
return json.dumps(dict())
return "{}"
def list_tags_for_resource(self):
def list_tags_for_resource(self) -> TYPE_RESPONSE:
params = json.loads(self.body)
resource_arn = params.get("ResourceARN")
tags = self.servicediscovery_backend.list_tags_for_resource(
@ -88,7 +89,7 @@ class ServiceDiscoveryResponse(BaseResponse):
)
return 200, {}, json.dumps(tags)
def create_private_dns_namespace(self):
def create_private_dns_namespace(self) -> str:
params = json.loads(self.body)
name = params.get("Name")
creator_request_id = params.get("CreatorRequestId")
@ -106,7 +107,7 @@ class ServiceDiscoveryResponse(BaseResponse):
)
return json.dumps(dict(OperationId=operation_id))
def create_public_dns_namespace(self):
def create_public_dns_namespace(self) -> str:
params = json.loads(self.body)
name = params.get("Name")
creator_request_id = params.get("CreatorRequestId")
@ -122,7 +123,7 @@ class ServiceDiscoveryResponse(BaseResponse):
)
return json.dumps(dict(OperationId=operation_id))
def create_service(self):
def create_service(self) -> str:
params = json.loads(self.body)
name = params.get("Name")
namespace_id = params.get("NamespaceId")
@ -146,23 +147,23 @@ class ServiceDiscoveryResponse(BaseResponse):
)
return json.dumps(dict(Service=service.to_json()))
def get_service(self):
def get_service(self) -> str:
params = json.loads(self.body)
service_id = params.get("Id")
service = self.servicediscovery_backend.get_service(service_id=service_id)
return json.dumps(dict(Service=service.to_json()))
def delete_service(self):
def delete_service(self) -> str:
params = json.loads(self.body)
service_id = params.get("Id")
self.servicediscovery_backend.delete_service(service_id=service_id)
return json.dumps(dict())
return "{}"
def list_services(self):
def list_services(self) -> str:
services = self.servicediscovery_backend.list_services()
return json.dumps(dict(Services=[s.to_json() for s in services]))
def update_service(self):
def update_service(self) -> str:
params = json.loads(self.body)
service_id = params.get("Id")
details = params.get("Service")

6
moto/settings.py
View File

@ -38,7 +38,7 @@ SKIP_REQUIRES_DOCKER = bool(os.environ.get("TESTS_SKIP_REQUIRES_DOCKER", False))
LAMBDA_DATA_DIR = os.environ.get("MOTO_LAMBDA_DATA_DIR", "/tmp/data")
def get_sf_execution_history_type():
def get_sf_execution_history_type() -> str:
"""
Determines which execution history events `get_execution_history` returns
:returns: str representing the type of Step Function Execution Type events should be
@ -97,11 +97,11 @@ def moto_lambda_image() -> str:
return os.environ.get("MOTO_DOCKER_LAMBDA_IMAGE", "mlupin/docker-lambda")
def moto_network_name() -> str:
def moto_network_name() -> Optional[str]:
return os.environ.get("MOTO_DOCKER_NETWORK_NAME")
def moto_network_mode() -> str:
def moto_network_mode() -> Optional[str]:
return os.environ.get("MOTO_DOCKER_NETWORK_MODE")

36
moto/signer/models.py
View File

@ -1,10 +1,18 @@
from typing import Any, Dict, List, Optional
from moto.core import BaseBackend, BackendDict, BaseModel
from moto.moto_api._internal import mock_random
class SigningProfile(BaseModel):
def __init__(
self, account_id, region, name, platform_id, signature_validity_period, tags
self,
account_id: str,
region: str,
name: str,
platform_id: str,
signature_validity_period: Optional[Dict[str, Any]],
tags: Dict[str, str],
):
self.name = name
self.platform_id = platform_id
@ -19,11 +27,11 @@ class SigningProfile(BaseModel):
self.profile_version = mock_random.get_random_hex(10)
self.profile_version_arn = f"{self.arn}/{self.profile_version}"
def cancel(self):
def cancel(self) -> None:
self.status = "Canceled"
def to_dict(self, full=True):
small = {
def to_dict(self, full: bool = True) -> Dict[str, Any]:
small: Dict[str, Any] = {
"arn": self.arn,
"profileVersion": self.profile_version,
"profileVersionArn": self.profile_version_arn,
@ -149,22 +157,22 @@ class SignerBackend(BaseBackend):
},
]
def __init__(self, region_name, account_id):
def __init__(self, region_name: str, account_id: str):
super().__init__(region_name, account_id)
self.signing_profiles: [str, SigningProfile] = dict()
self.signing_profiles: Dict[str, SigningProfile] = dict()
def cancel_signing_profile(self, profile_name) -> None:
def cancel_signing_profile(self, profile_name: str) -> None:
self.signing_profiles[profile_name].cancel()
def get_signing_profile(self, profile_name) -> SigningProfile:
def get_signing_profile(self, profile_name: str) -> SigningProfile:
return self.signing_profiles[profile_name]
def put_signing_profile(
self,
profile_name,
signature_validity_period,
platform_id,
tags,
profile_name: str,
signature_validity_period: Optional[Dict[str, Any]],
platform_id: str,
tags: Dict[str, str],
) -> SigningProfile:
"""
The following parameters are not yet implemented: SigningMaterial, Overrides, SigningParamaters
@ -180,7 +188,7 @@ class SignerBackend(BaseBackend):
self.signing_profiles[profile_name] = profile
return profile
def list_signing_platforms(self):
def list_signing_platforms(self) -> List[Dict[str, Any]]:
"""
Pagination is not yet implemented. The parameters category, partner, target are not yet implemented
"""
@ -189,4 +197,4 @@ class SignerBackend(BaseBackend):
# Using the lambda-regions
# boto3.Session().get_available_regions("signer") still returns an empty list
signer_backends: [str, [str, SignerBackend]] = BackendDict(SignerBackend, "lambda")
signer_backends = BackendDict(SignerBackend, "lambda")

10
moto/signer/responses.py
View File

@ -6,7 +6,7 @@ from .models import signer_backends, SignerBackend
class signerResponse(BaseResponse):
def __init__(self):
def __init__(self) -> None:
super().__init__(service_name="signer")
@property
@ -14,17 +14,17 @@ class signerResponse(BaseResponse):
"""Return backend instance specific for this region."""
return signer_backends[self.current_account][self.region]
def cancel_signing_profile(self):
def cancel_signing_profile(self) -> str:
profile_name = self.path.split("/")[-1]
self.signer_backend.cancel_signing_profile(profile_name=profile_name)
return "{}"
def get_signing_profile(self):
def get_signing_profile(self) -> str:
profile_name = self.path.split("/")[-1]
profile = self.signer_backend.get_signing_profile(profile_name=profile_name)
return json.dumps(profile.to_dict())
def put_signing_profile(self):
def put_signing_profile(self) -> str:
params = json.loads(self.body)
profile_name = self.path.split("/")[-1]
signature_validity_period = params.get("signatureValidityPeriod")
@ -38,6 +38,6 @@ class signerResponse(BaseResponse):
)
return json.dumps(profile.to_dict(full=False))
def list_signing_platforms(self):
def list_signing_platforms(self) -> str:
platforms = self.signer_backend.list_signing_platforms()
return json.dumps(dict(platforms=platforms))

2
moto/ssoadmin/exceptions.py
View File

@ -3,5 +3,5 @@ from moto.core.exceptions import JsonRESTError
class ResourceNotFound(JsonRESTError):
def __init__(self):
def __init__(self) -> None:
super().__init__("ResourceNotFound", "Account not found")

135
moto/ssoadmin/models.py
View File

@ -1,21 +1,22 @@
from .exceptions import ResourceNotFound
from typing import Any, Dict, List
from moto.core import BaseBackend, BackendDict, BaseModel
from moto.core.utils import unix_time
from moto.moto_api._internal import mock_random as random
from moto.utilities.paginator import paginate
from .exceptions import ResourceNotFound
from .utils import PAGINATION_MODEL
class AccountAssignment(BaseModel):
def __init__(
self,
instance_arn,
target_id,
target_type,
permission_set_arn,
principal_type,
principal_id,
instance_arn: str,
target_id: str,
target_type: str,
permission_set_arn: str,
principal_type: str,
principal_id: str,
):
self.request_id = str(random.uuid4())
self.instance_arn = instance_arn
@ -26,8 +27,8 @@ class AccountAssignment(BaseModel):
self.principal_id = principal_id
self.created_date = unix_time()
def to_json(self, include_creation_date=False):
summary = {
def to_json(self, include_creation_date: bool = False) -> Dict[str, Any]:
summary: Dict[str, Any] = {
"TargetId": self.target_id,
"TargetType": self.target_type,
"PermissionSetArn": self.permission_set_arn,
@ -42,12 +43,12 @@ class AccountAssignment(BaseModel):
class PermissionSet(BaseModel):
def __init__(
self,
name,
description,
instance_arn,
session_duration,
relay_state,
tags,
name: str,
description: str,
instance_arn: str,
session_duration: str,
relay_state: str,
tags: List[Dict[str, str]],
):
self.name = name
self.description = description
@ -58,8 +59,8 @@ class PermissionSet(BaseModel):
self.tags = tags
self.created_date = unix_time()
def to_json(self, include_creation_date=False):
summary = {
def to_json(self, include_creation_date: bool = False) -> Dict[str, Any]:
summary: Dict[str, Any] = {
"Name": self.name,
"Description": self.description,
"PermissionSetArn": self.permission_set_arn,
@ -71,7 +72,7 @@ class PermissionSet(BaseModel):
return summary
@staticmethod
def generate_id(instance_arn):
def generate_id(instance_arn: str) -> str:
chars = list(range(10)) + ["a", "b", "c", "d", "e", "f"]
return (
instance_arn
@ -83,20 +84,20 @@ class PermissionSet(BaseModel):
class SSOAdminBackend(BaseBackend):
"""Implementation of SSOAdmin APIs."""
def __init__(self, region_name, account_id):
def __init__(self, region_name: str, account_id: str):
super().__init__(region_name, account_id)
self.account_assignments = list()
self.permission_sets = list()
self.account_assignments: List[AccountAssignment] = list()
self.permission_sets: List[PermissionSet] = list()
def create_account_assignment(
self,
instance_arn,
target_id,
target_type,
permission_set_arn,
principal_type,
principal_id,
):
instance_arn: str,
target_id: str,
target_type: str,
permission_set_arn: str,
principal_type: str,
principal_id: str,
) -> Dict[str, Any]:
assignment = AccountAssignment(
instance_arn,
target_id,
@ -110,13 +111,13 @@ class SSOAdminBackend(BaseBackend):
def delete_account_assignment(
self,
instance_arn,
target_id,
target_type,
permission_set_arn,
principal_type,
principal_id,
):
instance_arn: str,
target_id: str,
target_type: str,
permission_set_arn: str,
principal_type: str,
principal_id: str,
) -> Dict[str, Any]:
account = self._find_account(
instance_arn,
target_id,
@ -130,13 +131,13 @@ class SSOAdminBackend(BaseBackend):
def _find_account(
self,
instance_arn,
target_id,
target_type,
permission_set_arn,
principal_type,
principal_id,
):
instance_arn: str,
target_id: str,
target_type: str,
permission_set_arn: str,
principal_type: str,
principal_id: str,
) -> AccountAssignment:
for account in self.account_assignments:
instance_arn_match = account.instance_arn == instance_arn
target_id_match = account.target_id == target_id
@ -155,7 +156,9 @@ class SSOAdminBackend(BaseBackend):
return account
raise ResourceNotFound
def list_account_assignments(self, instance_arn, account_id, permission_set_arn):
def list_account_assignments(
self, instance_arn: str, account_id: str, permission_set_arn: str
) -> List[Dict[str, Any]]:
"""
Pagination has not yet been implemented
"""
@ -178,13 +181,13 @@ class SSOAdminBackend(BaseBackend):
def create_permission_set(
self,
name,
description,
instance_arn,
session_duration,
relay_state,
tags,
):
name: str,
description: str,
instance_arn: str,
session_duration: str,
relay_state: str,
tags: List[Dict[str, str]],
) -> Dict[str, Any]:
permission_set = PermissionSet(
name,
description,
@ -198,12 +201,12 @@ class SSOAdminBackend(BaseBackend):
def update_permission_set(
self,
instance_arn,
permission_set_arn,
description,
session_duration,
relay_state,
):
instance_arn: str,
permission_set_arn: str,
description: str,
session_duration: str,
relay_state: str,
) -> Dict[str, Any]:
permission_set = self._find_permission_set(
instance_arn,
permission_set_arn,
@ -216,10 +219,8 @@ class SSOAdminBackend(BaseBackend):
return permission_set.to_json(True)
def describe_permission_set(
self,
instance_arn,
permission_set_arn,
):
self, instance_arn: str, permission_set_arn: str
) -> Dict[str, Any]:
permission_set = self._find_permission_set(
instance_arn,
permission_set_arn,
@ -227,10 +228,8 @@ class SSOAdminBackend(BaseBackend):
return permission_set.to_json(True)
def delete_permission_set(
self,
instance_arn,
permission_set_arn,
):
self, instance_arn: str, permission_set_arn: str
) -> Dict[str, Any]:
permission_set = self._find_permission_set(
instance_arn,
permission_set_arn,
@ -239,10 +238,8 @@ class SSOAdminBackend(BaseBackend):
return permission_set.to_json(include_creation_date=True)
def _find_permission_set(
self,
instance_arn,
permission_set_arn,
):
self, instance_arn: str, permission_set_arn: str
) -> PermissionSet:
for permission_set in self.permission_sets:
instance_arn_match = permission_set.instance_arn == instance_arn
permission_set_match = (
@ -253,7 +250,7 @@ class SSOAdminBackend(BaseBackend):
raise ResourceNotFound
@paginate(pagination_model=PAGINATION_MODEL)
def list_permission_sets(self, instance_arn):
def list_permission_sets(self, instance_arn: str) -> List[PermissionSet]: # type: ignore[misc]
permission_sets = []
for permission_set in self.permission_sets:
if permission_set.instance_arn == instance_arn:

24
moto/ssoadmin/responses.py
View File

@ -3,21 +3,21 @@ import json
from moto.core.responses import BaseResponse
from moto.moto_api._internal import mock_random
from .models import ssoadmin_backends
from .models import ssoadmin_backends, SSOAdminBackend
class SSOAdminResponse(BaseResponse):
"""Handler for SSOAdmin requests and responses."""
def __init__(self):
def __init__(self) -> None:
super().__init__(service_name="sso-admin")
@property
def ssoadmin_backend(self):
def ssoadmin_backend(self) -> SSOAdminBackend:
"""Return backend instance specific for this region."""
return ssoadmin_backends[self.current_account][self.region]
def create_account_assignment(self):
def create_account_assignment(self) -> str:
params = json.loads(self.body)
instance_arn = params.get("InstanceArn")
target_id = params.get("TargetId")
@ -37,7 +37,7 @@ class SSOAdminResponse(BaseResponse):
summary["RequestId"] = str(mock_random.uuid4())
return json.dumps({"AccountAssignmentCreationStatus": summary})
def delete_account_assignment(self):
def delete_account_assignment(self) -> str:
params = json.loads(self.body)
instance_arn = params.get("InstanceArn")
target_id = params.get("TargetId")
@ -57,7 +57,7 @@ class SSOAdminResponse(BaseResponse):
summary["RequestId"] = str(mock_random.uuid4())
return json.dumps({"AccountAssignmentDeletionStatus": summary})
def list_account_assignments(self):
def list_account_assignments(self) -> str:
params = json.loads(self.body)
instance_arn = params.get("InstanceArn")
account_id = params.get("AccountId")
@ -69,7 +69,7 @@ class SSOAdminResponse(BaseResponse):
)
return json.dumps({"AccountAssignments": assignments})
def create_permission_set(self):
def create_permission_set(self) -> str:
name = self._get_param("Name")
description = self._get_param("Description")
instance_arn = self._get_param("InstanceArn")
@ -88,7 +88,7 @@ class SSOAdminResponse(BaseResponse):
return json.dumps({"PermissionSet": permission_set})
def delete_permission_set(self):
def delete_permission_set(self) -> str:
params = json.loads(self.body)
instance_arn = params.get("InstanceArn")
permission_set_arn = params.get("PermissionSetArn")
@ -96,8 +96,9 @@ class SSOAdminResponse(BaseResponse):
instance_arn=instance_arn,
permission_set_arn=permission_set_arn,
)
return "{}"
def update_permission_set(self):
def update_permission_set(self) -> str:
instance_arn = self._get_param("InstanceArn")
permission_set_arn = self._get_param("PermissionSetArn")
description = self._get_param("Description")
@ -111,8 +112,9 @@ class SSOAdminResponse(BaseResponse):
session_duration=session_duration,
relay_state=relay_state,
)
return "{}"
def describe_permission_set(self):
def describe_permission_set(self) -> str:
instance_arn = self._get_param("InstanceArn")
permission_set_arn = self._get_param("PermissionSetArn")
@ -122,7 +124,7 @@ class SSOAdminResponse(BaseResponse):
)
return json.dumps({"PermissionSet": permission_set})
def list_permission_sets(self):
def list_permission_sets(self) -> str:
instance_arn = self._get_param("InstanceArn")
max_results = self._get_int_param("MaxResults")
next_token = self._get_param("NextToken")

4
moto/stepfunctions/exceptions.py
View File

@ -35,7 +35,7 @@ class InvalidToken(AWSError):
TYPE = "InvalidToken"
STATUS = 400
def __init__(self, message="Invalid token"):
def __init__(self, message: str = "Invalid token"):
super().__init__(f"Invalid Token: {message}")
@ -43,5 +43,5 @@ class ResourceNotFound(AWSError):
TYPE = "ResourceNotFound"
STATUS = 400
def __init__(self, arn):
def __init__(self, arn: str):
super().__init__(f"Resource not found: '{arn}'")

158
moto/stepfunctions/models.py
View File

@ -2,6 +2,7 @@ import json
import re
from datetime import datetime
from dateutil.tz import tzlocal
from typing import Any, Dict, List, Iterable, Optional, Pattern
from moto.core import BaseBackend, BackendDict, CloudFormationModel
from moto.core.utils import iso_8601_datetime_with_milliseconds
@ -21,19 +22,32 @@ from moto.utilities.paginator import paginate
class StateMachine(CloudFormationModel):
def __init__(self, arn, name, definition, roleArn, tags=None):
def __init__(
self,
arn: str,
name: str,
definition: str,
roleArn: str,
tags: Optional[List[Dict[str, str]]] = None,
):
self.creation_date = iso_8601_datetime_with_milliseconds(datetime.now())
self.update_date = self.creation_date
self.arn = arn
self.name = name
self.definition = definition
self.roleArn = roleArn
self.executions = []
self.tags = []
self.executions: List[Execution] = []
self.tags: List[Dict[str, str]] = []
if tags:
self.add_tags(tags)
def start_execution(self, region_name, account_id, execution_name, execution_input):
def start_execution(
self,
region_name: str,
account_id: str,
execution_name: str,
execution_input: str,
) -> "Execution":
self._ensure_execution_name_doesnt_exist(execution_name)
self._validate_execution_input(execution_input)
execution = Execution(
@ -47,7 +61,7 @@ class StateMachine(CloudFormationModel):
self.executions.append(execution)
return execution
def stop_execution(self, execution_arn):
def stop_execution(self, execution_arn: str) -> "Execution":
execution = next(
(x for x in self.executions if x.execution_arn == execution_arn), None
)
@ -58,14 +72,14 @@ class StateMachine(CloudFormationModel):
execution.stop()
return execution
def _ensure_execution_name_doesnt_exist(self, name):
def _ensure_execution_name_doesnt_exist(self, name: str) -> None:
for execution in self.executions:
if execution.name == name:
raise ExecutionAlreadyExists(
"Execution Already Exists: '" + execution.execution_arn + "'"
)
def _validate_execution_input(self, execution_input):
def _validate_execution_input(self, execution_input: str) -> None:
try:
json.loads(execution_input)
except Exception as ex:
@ -73,13 +87,13 @@ class StateMachine(CloudFormationModel):
"Invalid State Machine Execution Input: '" + str(ex) + "'"
)
def update(self, **kwargs):
def update(self, **kwargs: Any) -> None:
for key, value in kwargs.items():
if value is not None:
setattr(self, key, value)
self.update_date = iso_8601_datetime_with_milliseconds(datetime.now())
def add_tags(self, tags):
def add_tags(self, tags: List[Dict[str, str]]) -> List[Dict[str, str]]:
merged_tags = []
for tag in self.tags:
replacement_index = next(
@ -96,15 +110,15 @@ class StateMachine(CloudFormationModel):
self.tags = merged_tags
return self.tags
def remove_tags(self, tag_keys):
def remove_tags(self, tag_keys: List[str]) -> List[Dict[str, str]]:
self.tags = [tag_set for tag_set in self.tags if tag_set["key"] not in tag_keys]
return self.tags
@property
def physical_resource_id(self):
def physical_resource_id(self) -> str:
return self.arn
def get_cfn_properties(self, prop_overrides):
def get_cfn_properties(self, prop_overrides: Dict[str, Any]) -> Dict[str, Any]:
property_names = [
"DefinitionString",
"RoleArn",
@ -124,7 +138,7 @@ class StateMachine(CloudFormationModel):
return properties
@classmethod
def has_cfn_attr(cls, attr):
def has_cfn_attr(cls, attr: str) -> bool:
return attr in [
"Name",
"DefinitionString",
@ -133,7 +147,7 @@ class StateMachine(CloudFormationModel):
"Tags",
]
def get_cfn_attribute(self, attribute_name):
def get_cfn_attribute(self, attribute_name: str) -> Any:
from moto.cloudformation.exceptions import UnformattedGetAttTemplateException
if attribute_name == "Name":
@ -150,17 +164,22 @@ class StateMachine(CloudFormationModel):
raise UnformattedGetAttTemplateException()
@staticmethod
def cloudformation_name_type():
def cloudformation_name_type() -> str:
return "StateMachine"
@staticmethod
def cloudformation_type():
def cloudformation_type() -> str:
return "AWS::StepFunctions::StateMachine"
@classmethod
def create_from_cloudformation_json(
cls, resource_name, cloudformation_json, account_id, region_name, **kwargs
):
def create_from_cloudformation_json( # type: ignore[misc]
cls,
resource_name: str,
cloudformation_json: Any,
account_id: str,
region_name: str,
**kwargs: Any,
) -> "StateMachine":
properties = cloudformation_json["Properties"]
name = properties.get("StateMachineName", resource_name)
definition = properties.get("DefinitionString", "")
@ -170,19 +189,25 @@ class StateMachine(CloudFormationModel):
return sf_backend.create_state_machine(name, definition, role_arn, tags=tags)
@classmethod
def delete_from_cloudformation_json(cls, resource_name, _, account_id, region_name):
def delete_from_cloudformation_json( # type: ignore[misc]
cls,
resource_name: str,
cloudformation_json: Any,
account_id: str,
region_name: str,
) -> None:
sf_backend = stepfunction_backends[account_id][region_name]
sf_backend.delete_state_machine(resource_name)
@classmethod
def update_from_cloudformation_json(
def update_from_cloudformation_json( # type: ignore[misc]
cls,
original_resource,
new_resource_name,
cloudformation_json,
account_id,
region_name,
):
original_resource: Any,
new_resource_name: str,
cloudformation_json: Any,
account_id: str,
region_name: str,
) -> "StateMachine":
properties = cloudformation_json.get("Properties", {})
name = properties.get("StateMachineName", original_resource.name)
@ -214,12 +239,12 @@ class StateMachine(CloudFormationModel):
class Execution:
def __init__(
self,
region_name,
account_id,
state_machine_name,
execution_name,
state_machine_arn,
execution_input,
region_name: str,
account_id: str,
state_machine_name: str,
execution_name: str,
state_machine_arn: str,
execution_input: str,
):
execution_arn = "arn:aws:states:{}:{}:execution:{}:{}"
execution_arn = execution_arn.format(
@ -235,9 +260,9 @@ class Execution:
if settings.get_sf_execution_history_type() == "SUCCESS"
else "FAILED"
)
self.stop_date = None
self.stop_date: Optional[str] = None
def get_execution_history(self, roleArn):
def get_execution_history(self, roleArn: str) -> List[Dict[str, Any]]:
sf_execution_history_type = settings.get_sf_execution_history_type()
if sf_execution_history_type == "SUCCESS":
return [
@ -334,8 +359,9 @@ class Execution:
},
},
]
return []
def stop(self):
def stop(self) -> None:
self.status = "ABORTED"
self.stop_date = iso_8601_datetime_with_milliseconds(datetime.now())
@ -451,13 +477,19 @@ class StepFunctionBackend(BaseBackend):
"arn:aws:states:[-0-9a-zA-Z]+:(?P<account_id>[0-9]{12}):execution:.+"
)
def __init__(self, region_name, account_id):
def __init__(self, region_name: str, account_id: str):
super().__init__(region_name, account_id)
self.state_machines = []
self.executions = []
self.state_machines: List[StateMachine] = []
self.executions: List[Execution] = []
self._account_id = None
def create_state_machine(self, name, definition, roleArn, tags=None):
def create_state_machine(
self,
name: str,
definition: str,
roleArn: str,
tags: Optional[List[Dict[str, str]]] = None,
) -> StateMachine:
self._validate_name(name)
self._validate_role_arn(roleArn)
arn = f"arn:aws:states:{self.region_name}:{self.account_id}:stateMachine:{name}"
@ -469,11 +501,10 @@ class StepFunctionBackend(BaseBackend):
return state_machine
@paginate(pagination_model=PAGINATION_MODEL)
def list_state_machines(self):
state_machines = sorted(self.state_machines, key=lambda x: x.creation_date)
return state_machines
def list_state_machines(self) -> Iterable[StateMachine]: # type: ignore[misc]
return sorted(self.state_machines, key=lambda x: x.creation_date)
def describe_state_machine(self, arn):
def describe_state_machine(self, arn: str) -> StateMachine:
self._validate_machine_arn(arn)
sm = next((x for x in self.state_machines if x.arn == arn), None)
if not sm:
@ -482,13 +513,15 @@ class StepFunctionBackend(BaseBackend):
)
return sm
def delete_state_machine(self, arn):
def delete_state_machine(self, arn: str) -> None:
self._validate_machine_arn(arn)
sm = next((x for x in self.state_machines if x.arn == arn), None)
if sm:
self.state_machines.remove(sm)
def update_state_machine(self, arn, definition=None, role_arn=None):
def update_state_machine(
self, arn: str, definition: Optional[str] = None, role_arn: Optional[str] = None
) -> StateMachine:
sm = self.describe_state_machine(arn)
updates = {
"definition": definition,
@ -497,23 +530,24 @@ class StepFunctionBackend(BaseBackend):
sm.update(**updates)
return sm
def start_execution(self, state_machine_arn, name=None, execution_input=None):
def start_execution(
self, state_machine_arn: str, name: str, execution_input: str
) -> Execution:
state_machine = self.describe_state_machine(state_machine_arn)
execution = state_machine.start_execution(
return state_machine.start_execution(
region_name=self.region_name,
account_id=self.account_id,
execution_name=name or str(mock_random.uuid4()),
execution_input=execution_input,
)
return execution
def stop_execution(self, execution_arn):
def stop_execution(self, execution_arn: str) -> Execution:
self._validate_execution_arn(execution_arn)
state_machine = self._get_state_machine_for_execution(execution_arn)
return state_machine.stop_execution(execution_arn)
@paginate(pagination_model=PAGINATION_MODEL)
def list_executions(self, state_machine_arn, status_filter=None):
def list_executions(self, state_machine_arn: str, status_filter: Optional[str] = None) -> Iterable[Execution]: # type: ignore[misc]
"""
The status of every execution is set to 'RUNNING' by default.
Set the following environment variable if you want to get a FAILED status back:
@ -530,7 +564,7 @@ class StepFunctionBackend(BaseBackend):
executions = sorted(executions, key=lambda x: x.start_date, reverse=True)
return executions
def describe_execution(self, execution_arn):
def describe_execution(self, execution_arn: str) -> Execution:
"""
The status of every execution is set to 'RUNNING' by default.
Set the following environment variable if you want to get a FAILED status back:
@ -551,7 +585,7 @@ class StepFunctionBackend(BaseBackend):
)
return exctn
def get_execution_history(self, execution_arn):
def get_execution_history(self, execution_arn: str) -> List[Dict[str, Any]]:
"""
A static list of successful events is returned by default.
Set the following environment variable if you want to get a static list of events for a failed execution:
@ -572,61 +606,61 @@ class StepFunctionBackend(BaseBackend):
)
return execution.get_execution_history(state_machine.roleArn)
def list_tags_for_resource(self, arn):
def list_tags_for_resource(self, arn: str) -> List[Dict[str, str]]:
try:
state_machine = self.describe_state_machine(arn)
return state_machine.tags or []
except StateMachineDoesNotExist:
return []
def tag_resource(self, resource_arn, tags):
def tag_resource(self, resource_arn: str, tags: List[Dict[str, str]]) -> None:
try:
state_machine = self.describe_state_machine(resource_arn)
state_machine.add_tags(tags)
except StateMachineDoesNotExist:
raise ResourceNotFound(resource_arn)
def untag_resource(self, resource_arn, tag_keys):
def untag_resource(self, resource_arn: str, tag_keys: List[str]) -> None:
try:
state_machine = self.describe_state_machine(resource_arn)
state_machine.remove_tags(tag_keys)
except StateMachineDoesNotExist:
raise ResourceNotFound(resource_arn)
def _validate_name(self, name):
def _validate_name(self, name: str) -> None:
if any(invalid_char in name for invalid_char in self.invalid_chars_for_name):
raise InvalidName("Invalid Name: '" + name + "'")
if any(name.find(char) >= 0 for char in self.invalid_unicodes_for_name):
raise InvalidName("Invalid Name: '" + name + "'")
def _validate_role_arn(self, role_arn):
def _validate_role_arn(self, role_arn: str) -> None:
self._validate_arn(
arn=role_arn,
regex=self.accepted_role_arn_format,
invalid_msg="Invalid Role Arn: '" + role_arn + "'",
)
def _validate_machine_arn(self, machine_arn):
def _validate_machine_arn(self, machine_arn: str) -> None:
self._validate_arn(
arn=machine_arn,
regex=self.accepted_mchn_arn_format,
invalid_msg="Invalid State Machine Arn: '" + machine_arn + "'",
)
def _validate_execution_arn(self, execution_arn):
def _validate_execution_arn(self, execution_arn: str) -> None:
self._validate_arn(
arn=execution_arn,
regex=self.accepted_exec_arn_format,
invalid_msg="Execution Does Not Exist: '" + execution_arn + "'",
)
def _validate_arn(self, arn, regex, invalid_msg):
def _validate_arn(self, arn: str, regex: Pattern[str], invalid_msg: str) -> None:
match = regex.match(arn)
if not arn or not match:
raise InvalidArn(invalid_msg)
def _get_state_machine_for_execution(self, execution_arn):
def _get_state_machine_for_execution(self, execution_arn: str) -> StateMachine:
state_machine_name = execution_arn.split(":")[6]
state_machine_arn = next(
(x.arn for x in self.state_machines if x.name == state_machine_name), None

37
moto/stepfunctions/responses.py
View File

@ -1,20 +1,21 @@
import json
from moto.core.common_types import TYPE_RESPONSE
from moto.core.responses import BaseResponse
from moto.utilities.aws_headers import amzn_request_id
from .models import stepfunction_backends
from .models import stepfunction_backends, StepFunctionBackend
class StepFunctionResponse(BaseResponse):
def __init__(self):
def __init__(self) -> None:
super().__init__(service_name="stepfunctions")
@property
def stepfunction_backend(self):
def stepfunction_backend(self) -> StepFunctionBackend:
return stepfunction_backends[self.current_account][self.region]
@amzn_request_id
def create_state_machine(self):
def create_state_machine(self) -> TYPE_RESPONSE:
name = self._get_param("name")
definition = self._get_param("definition")
roleArn = self._get_param("roleArn")
@ -29,7 +30,7 @@ class StepFunctionResponse(BaseResponse):
return 200, {}, json.dumps(response)
@amzn_request_id
def list_state_machines(self):
def list_state_machines(self) -> TYPE_RESPONSE:
max_results = self._get_int_param("maxResults")
next_token = self._get_param("nextToken")
results, next_token = self.stepfunction_backend.list_state_machines(
@ -49,12 +50,12 @@ class StepFunctionResponse(BaseResponse):
return 200, {}, json.dumps(response)
@amzn_request_id
def describe_state_machine(self):
def describe_state_machine(self) -> TYPE_RESPONSE:
arn = self._get_param("stateMachineArn")
return self._describe_state_machine(arn)
@amzn_request_id
def _describe_state_machine(self, state_machine_arn):
def _describe_state_machine(self, state_machine_arn: str) -> TYPE_RESPONSE:
state_machine = self.stepfunction_backend.describe_state_machine(
state_machine_arn
)
@ -69,13 +70,13 @@ class StepFunctionResponse(BaseResponse):
return 200, {}, json.dumps(response)
@amzn_request_id
def delete_state_machine(self):
def delete_state_machine(self) -> TYPE_RESPONSE:
arn = self._get_param("stateMachineArn")
self.stepfunction_backend.delete_state_machine(arn)
return 200, {}, json.dumps("{}")
@amzn_request_id
def update_state_machine(self):
def update_state_machine(self) -> TYPE_RESPONSE:
arn = self._get_param("stateMachineArn")
definition = self._get_param("definition")
role_arn = self._get_param("roleArn")
@ -88,28 +89,28 @@ class StepFunctionResponse(BaseResponse):
return 200, {}, json.dumps(response)
@amzn_request_id
def list_tags_for_resource(self):
def list_tags_for_resource(self) -> TYPE_RESPONSE:
arn = self._get_param("resourceArn")
tags = self.stepfunction_backend.list_tags_for_resource(arn)
response = {"tags": tags}
return 200, {}, json.dumps(response)
@amzn_request_id
def tag_resource(self):
def tag_resource(self) -> TYPE_RESPONSE:
arn = self._get_param("resourceArn")
tags = self._get_param("tags", [])
self.stepfunction_backend.tag_resource(arn, tags)
return 200, {}, json.dumps({})
@amzn_request_id
def untag_resource(self):
def untag_resource(self) -> TYPE_RESPONSE:
arn = self._get_param("resourceArn")
tag_keys = self._get_param("tagKeys", [])
self.stepfunction_backend.untag_resource(arn, tag_keys)
return 200, {}, json.dumps({})
@amzn_request_id
def start_execution(self):
def start_execution(self) -> TYPE_RESPONSE:
arn = self._get_param("stateMachineArn")
name = self._get_param("name")
execution_input = self._get_param("input", if_none="{}")
@ -123,7 +124,7 @@ class StepFunctionResponse(BaseResponse):
return 200, {}, json.dumps(response)
@amzn_request_id
def list_executions(self):
def list_executions(self) -> TYPE_RESPONSE:
max_results = self._get_int_param("maxResults")
next_token = self._get_param("nextToken")
arn = self._get_param("stateMachineArn")
@ -151,7 +152,7 @@ class StepFunctionResponse(BaseResponse):
return 200, {}, json.dumps(response)
@amzn_request_id
def describe_execution(self):
def describe_execution(self) -> TYPE_RESPONSE:
arn = self._get_param("executionArn")
execution = self.stepfunction_backend.describe_execution(arn)
response = {
@ -166,20 +167,20 @@ class StepFunctionResponse(BaseResponse):
return 200, {}, json.dumps(response)
@amzn_request_id
def describe_state_machine_for_execution(self):
def describe_state_machine_for_execution(self) -> TYPE_RESPONSE:
arn = self._get_param("executionArn")
execution = self.stepfunction_backend.describe_execution(arn)
return self._describe_state_machine(execution.state_machine_arn)
@amzn_request_id
def stop_execution(self):
def stop_execution(self) -> TYPE_RESPONSE:
arn = self._get_param("executionArn")
execution = self.stepfunction_backend.stop_execution(arn)
response = {"stopDate": execution.stop_date}
return 200, {}, json.dumps(response)
@amzn_request_id
def get_execution_history(self):
def get_execution_history(self) -> TYPE_RESPONSE:
execution_arn = self._get_param("executionArn")
execution_history = self.stepfunction_backend.get_execution_history(
execution_arn

13
moto/stepfunctions/utils.py
View File

@ -1,3 +1,6 @@
from typing import Dict, List
PAGINATION_MODEL = {
"list_executions": {
"input_token": "next_token",
@ -14,11 +17,9 @@ PAGINATION_MODEL = {
}
def cfn_to_api_tags(cfn_tags_entry):
api_tags = [{k.lower(): v for k, v in d.items()} for d in cfn_tags_entry]
return api_tags
def cfn_to_api_tags(cfn_tags_entry: List[Dict[str, str]]) -> List[Dict[str, str]]:
return [{k.lower(): v for k, v in d.items()} for d in cfn_tags_entry]
def api_to_cfn_tags(api_tags):
cfn_tags_entry = [{k.capitalize(): v for k, v in d.items()} for d in api_tags]
return cfn_tags_entry
def api_to_cfn_tags(api_tags: List[Dict[str, str]]) -> List[Dict[str, str]]:
return [{k.capitalize(): v for k, v in d.items()} for d in api_tags]

3
moto/sts/exceptions.py
View File

@ -1,3 +1,4 @@
from typing import Any
from moto.core.exceptions import RESTError
@ -6,5 +7,5 @@ class STSClientError(RESTError):
class STSValidationError(STSClientError):
def __init__(self, *args, **kwargs):
def __init__(self, *args: Any, **kwargs: Any):
super().__init__("ValidationError", *args, **kwargs)

69
moto/sts/models.py
View File

@ -1,10 +1,12 @@
from base64 import b64decode
from typing import Any, Dict, List, Optional, Tuple
import datetime
import re
import xmltodict
from moto.core import BaseBackend, BaseModel, BackendDict
from moto.core.utils import iso_8601_datetime_with_milliseconds
from moto.iam import iam_backends
from moto.iam.models import iam_backends, AccessKey
from moto.sts.utils import (
random_session_token,
DEFAULT_STS_SESSION_DURATION,
@ -13,27 +15,27 @@ from moto.sts.utils import (
class Token(BaseModel):
def __init__(self, duration, name=None):
def __init__(self, duration: int, name: Optional[str] = None):
now = datetime.datetime.utcnow()
self.expiration = now + datetime.timedelta(seconds=duration)
self.name = name
self.policy = None
@property
def expiration_ISO8601(self):
def expiration_ISO8601(self) -> str:
return iso_8601_datetime_with_milliseconds(self.expiration)
class AssumedRole(BaseModel):
def __init__(
self,
account_id,
access_key,
role_session_name,
role_arn,
policy,
duration,
external_id,
account_id: str,
access_key: AccessKey,
role_session_name: str,
role_arn: str,
policy: str,
duration: int,
external_id: str,
):
self.account_id = account_id
self.session_name = role_session_name
@ -48,11 +50,11 @@ class AssumedRole(BaseModel):
self.session_token = random_session_token()
@property
def expiration_ISO8601(self):
def expiration_ISO8601(self) -> str:
return iso_8601_datetime_with_milliseconds(self.expiration)
@property
def user_id(self):
def user_id(self) -> str:
iam_backend = iam_backends[self.account_id]["global"]
try:
role_id = iam_backend.get_role_by_arn(arn=self.role_arn).id
@ -61,31 +63,38 @@ class AssumedRole(BaseModel):
return role_id + ":" + self.session_name
@property
def arn(self):
def arn(self) -> str:
return f"arn:aws:sts::{self.account_id}:assumed-role/{self.role_arn.split('/')[-1]}/{self.session_name}"
class STSBackend(BaseBackend):
def __init__(self, region_name, account_id):
def __init__(self, region_name: str, account_id: str):
super().__init__(region_name, account_id)
self.assumed_roles = []
self.assumed_roles: List[AssumedRole] = []
@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."""
return BaseBackend.default_vpc_endpoint_service_factory(
service_region, zones, "sts"
)
def get_session_token(self, duration):
token = Token(duration=duration)
return token
def get_session_token(self, duration: int) -> Token:
return Token(duration=duration)
def get_federation_token(self, name, duration):
token = Token(duration=duration, name=name)
return token
def get_federation_token(self, name: Optional[str], duration: int) -> Token:
return Token(duration=duration, name=name)
def assume_role(self, role_session_name, role_arn, policy, duration, external_id):
def assume_role(
self,
role_session_name: str,
role_arn: str,
policy: str,
duration: int,
external_id: str,
) -> AssumedRole:
"""
Assume an IAM Role. Note that the role does not need to exist. The ARN can point to another account, providing an opportunity to switch accounts.
"""
@ -103,16 +112,18 @@ class STSBackend(BaseBackend):
account_backend.assumed_roles.append(role)
return role
def get_assumed_role_from_access_key(self, access_key_id):
def get_assumed_role_from_access_key(
self, access_key_id: str
) -> Optional[AssumedRole]:
for assumed_role in self.assumed_roles:
if assumed_role.access_key_id == access_key_id:
return assumed_role
return None
def assume_role_with_web_identity(self, **kwargs):
def assume_role_with_web_identity(self, **kwargs: Any) -> AssumedRole:
return self.assume_role(**kwargs)
def assume_role_with_saml(self, **kwargs):
def assume_role_with_saml(self, **kwargs: Any) -> AssumedRole:
del kwargs["principal_arn"]
saml_assertion_encoded = kwargs.pop("saml_assertion")
saml_assertion_decoded = b64decode(saml_assertion_encoded)
@ -150,7 +161,7 @@ class STSBackend(BaseBackend):
if "duration" not in kwargs:
kwargs["duration"] = DEFAULT_STS_SESSION_DURATION
account_id, access_key = self._create_access_key(role=target_role)
account_id, access_key = self._create_access_key(role=target_role) # type: ignore
kwargs["account_id"] = account_id
kwargs["access_key"] = access_key
@ -160,7 +171,7 @@ class STSBackend(BaseBackend):
self.assumed_roles.append(role)
return role
def get_caller_identity(self, access_key_id):
def get_caller_identity(self, access_key_id: str) -> Tuple[str, str, str]:
assumed_role = self.get_assumed_role_from_access_key(access_key_id)
if assumed_role:
return assumed_role.user_id, assumed_role.arn, assumed_role.account_id
@ -175,7 +186,7 @@ class STSBackend(BaseBackend):
arn = f"arn:aws:sts::{self.account_id}:user/moto"
return user_id, arn, self.account_id
def _create_access_key(self, role):
def _create_access_key(self, role: str) -> Tuple[str, AccessKey]:
account_id_match = re.search(r"arn:aws:iam::([0-9]+).+", role)
if account_id_match:
account_id = account_id_match.group(1)

34
moto/sts/responses.py
View File

@ -1,25 +1,25 @@
from moto.core.responses import BaseResponse
from .exceptions import STSValidationError
from .models import sts_backends
from .models import sts_backends, STSBackend
MAX_FEDERATION_TOKEN_POLICY_LENGTH = 2048
class TokenResponse(BaseResponse):
def __init__(self):
def __init__(self) -> None:
super().__init__(service_name="sts")
@property
def backend(self):
def backend(self) -> STSBackend:
return sts_backends[self.current_account]["global"]
def get_session_token(self):
def get_session_token(self) -> str:
duration = int(self.querystring.get("DurationSeconds", [43200])[0])
token = self.backend.get_session_token(duration=duration)
template = self.response_template(GET_SESSION_TOKEN_RESPONSE)
return template.render(token=token)
def get_federation_token(self):
def get_federation_token(self) -> str:
duration = int(self.querystring.get("DurationSeconds", [43200])[0])
policy = self.querystring.get("Policy", [None])[0]
@ -31,14 +31,14 @@ class TokenResponse(BaseResponse):
f" equal to {MAX_FEDERATION_TOKEN_POLICY_LENGTH}"
)
name = self.querystring.get("Name")[0]
name = self.querystring.get("Name")[0] # type: ignore
token = self.backend.get_federation_token(duration=duration, name=name)
template = self.response_template(GET_FEDERATION_TOKEN_RESPONSE)
return template.render(token=token, account_id=self.current_account)
def assume_role(self):
role_session_name = self.querystring.get("RoleSessionName")[0]
role_arn = self.querystring.get("RoleArn")[0]
def assume_role(self) -> str:
role_session_name = self.querystring.get("RoleSessionName")[0] # type: ignore
role_arn = self.querystring.get("RoleArn")[0] # type: ignore
policy = self.querystring.get("Policy", [None])[0]
duration = int(self.querystring.get("DurationSeconds", [3600])[0])
@ -54,9 +54,9 @@ class TokenResponse(BaseResponse):
template = self.response_template(ASSUME_ROLE_RESPONSE)
return template.render(role=role)
def assume_role_with_web_identity(self):
role_session_name = self.querystring.get("RoleSessionName")[0]
role_arn = self.querystring.get("RoleArn")[0]
def assume_role_with_web_identity(self) -> str:
role_session_name = self.querystring.get("RoleSessionName")[0] # type: ignore
role_arn = self.querystring.get("RoleArn")[0] # type: ignore
policy = self.querystring.get("Policy", [None])[0]
duration = int(self.querystring.get("DurationSeconds", [3600])[0])
@ -72,10 +72,10 @@ class TokenResponse(BaseResponse):
template = self.response_template(ASSUME_ROLE_WITH_WEB_IDENTITY_RESPONSE)
return template.render(role=role)
def assume_role_with_saml(self):
role_arn = self.querystring.get("RoleArn")[0]
principal_arn = self.querystring.get("PrincipalArn")[0]
saml_assertion = self.querystring.get("SAMLAssertion")[0]
def assume_role_with_saml(self) -> str:
role_arn = self.querystring.get("RoleArn")[0] # type: ignore
principal_arn = self.querystring.get("PrincipalArn")[0] # type: ignore
saml_assertion = self.querystring.get("SAMLAssertion")[0] # type: ignore
role = self.backend.assume_role_with_saml(
role_arn=role_arn,
@ -85,7 +85,7 @@ class TokenResponse(BaseResponse):
template = self.response_template(ASSUME_ROLE_WITH_SAML_RESPONSE)
return template.render(role=role)
def get_caller_identity(self):
def get_caller_identity(self) -> str:
template = self.response_template(GET_CALLER_IDENTITY_RESPONSE)
access_key_id = self.get_access_key()

4
moto/sts/utils.py
View File

@ -16,13 +16,13 @@ def random_session_token() -> str:
)
def random_assumed_role_id():
def random_assumed_role_id() -> str:
return (
ACCOUNT_SPECIFIC_ASSUMED_ROLE_ID_PREFIX + _random_uppercase_or_digit_sequence(9)
)
def _random_uppercase_or_digit_sequence(length):
def _random_uppercase_or_digit_sequence(length: int) -> str:
return "".join(
str(random.choice(string.ascii_uppercase + string.digits))
for _ in range(length)

64
moto/support/models.py
View File

@ -4,6 +4,7 @@ from moto.moto_api._internal.managed_state_model import ManagedState
from moto.moto_api._internal import mock_random as random
from moto.utilities.utils import load_resource
import datetime
from typing import Any, Dict, List, Optional
checks_json = "resources/describe_trusted_advisor_checks.json"
@ -11,7 +12,7 @@ ADVISOR_CHECKS = load_resource(__name__, checks_json)
class SupportCase(ManagedState):
def __init__(self, **kwargs):
def __init__(self, **kwargs: Any):
# Configure ManagedState
super().__init__(
"support::case",
@ -56,21 +57,21 @@ class SupportCase(ManagedState):
}
}
def get_datetime(self):
def get_datetime(self) -> str:
return str(datetime.datetime.now().isoformat())
class SupportBackend(BaseBackend):
def __init__(self, region_name, account_id):
def __init__(self, region_name: str, account_id: str):
super().__init__(region_name, account_id)
self.check_status = {}
self.cases = {}
self.check_status: Dict[str, str] = {}
self.cases: Dict[str, SupportCase] = {}
state_manager.register_default_transition(
model_name="support::case", transition={"progression": "manual", "times": 1}
)
def describe_trusted_advisor_checks(self):
def describe_trusted_advisor_checks(self) -> List[Dict[str, Any]]:
"""
The Language-parameter is not yet implemented
"""
@ -78,18 +79,17 @@ class SupportBackend(BaseBackend):
checks = ADVISOR_CHECKS["checks"]
return checks
def refresh_trusted_advisor_check(self, check_id):
def refresh_trusted_advisor_check(self, check_id: str) -> Dict[str, Any]:
self.advance_check_status(check_id)
status = {
return {
"status": {
"checkId": check_id,
"status": self.check_status[check_id],
"millisUntilNextRefreshable": 123,
}
}
return status
def advance_check_status(self, check_id):
def advance_check_status(self, check_id: str) -> None:
"""
Fake an advancement through statuses on refreshing TA checks
"""
@ -111,14 +111,13 @@ class SupportBackend(BaseBackend):
elif self.check_status[check_id] == "abandoned":
self.check_status[check_id] = "none"
def advance_case_status(self, case_id):
def advance_case_status(self, case_id: str) -> None:
"""
Fake an advancement through case statuses
"""
self.cases[case_id].advance()
def advance_case_severity_codes(self, case_id):
def advance_case_severity_codes(self, case_id: str) -> None:
"""
Fake an advancement through case status severities
"""
@ -137,28 +136,26 @@ class SupportBackend(BaseBackend):
elif self.cases[case_id].severity_code == "critical":
self.cases[case_id].severity_code = "low"
def resolve_case(self, case_id):
def resolve_case(self, case_id: str) -> Dict[str, Optional[str]]:
self.advance_case_status(case_id)
resolved_case = {
return {
"initialCaseStatus": self.cases[case_id].status,
"finalCaseStatus": "resolved",
}
return resolved_case
# persist case details to self.cases
def create_case(
self,
subject,
service_code,
severity_code,
category_code,
communication_body,
cc_email_addresses,
language,
attachment_set_id,
):
subject: str,
service_code: str,
severity_code: str,
category_code: str,
communication_body: str,
cc_email_addresses: List[str],
language: str,
attachment_set_id: str,
) -> Dict[str, str]:
"""
The IssueType-parameter is not yet implemented
"""
@ -184,11 +181,11 @@ class SupportBackend(BaseBackend):
def describe_cases(
self,
case_id_list,
include_resolved_cases,
next_token,
include_communications,
):
case_id_list: List[str],
include_resolved_cases: bool,
next_token: Optional[str],
include_communications: bool,
) -> Dict[str, Any]:
"""
The following parameters have not yet been implemented:
DisplayID, AfterTime, BeforeTime, MaxResults, Language
@ -223,10 +220,7 @@ class SupportBackend(BaseBackend):
continue
cases.append(formatted_case)
case_values = {"cases": cases}
case_values.update({"nextToken": next_token})
return case_values
return {"cases": cases, "nextToken": next_token}
support_backends = BackendDict(

16
moto/support/responses.py
View File

@ -1,33 +1,33 @@
from moto.core.responses import BaseResponse
from .models import support_backends
from .models import support_backends, SupportBackend
import json
class SupportResponse(BaseResponse):
def __init__(self):
def __init__(self) -> None:
super().__init__(service_name="support")
@property
def support_backend(self):
def support_backend(self) -> SupportBackend:
return support_backends[self.current_account][self.region]
def describe_trusted_advisor_checks(self):
def describe_trusted_advisor_checks(self) -> str:
checks = self.support_backend.describe_trusted_advisor_checks()
return json.dumps({"checks": checks})
def refresh_trusted_advisor_check(self):
def refresh_trusted_advisor_check(self) -> str:
check_id = self._get_param("checkId")
status = self.support_backend.refresh_trusted_advisor_check(check_id=check_id)
return json.dumps(status)
def resolve_case(self):
def resolve_case(self) -> str:
case_id = self._get_param("caseId")
resolve_case_response = self.support_backend.resolve_case(case_id=case_id)
return json.dumps(resolve_case_response)
def create_case(self):
def create_case(self) -> str:
subject = self._get_param("subject")
service_code = self._get_param("serviceCode")
severity_code = self._get_param("severityCode")
@ -49,7 +49,7 @@ class SupportResponse(BaseResponse):
return json.dumps(create_case_response)
def describe_cases(self):
def describe_cases(self) -> str:
case_id_list = self._get_param("caseIdList")
include_resolved_cases = self._get_param("includeResolvedCases", False)
next_token = self._get_param("nextToken")

2
setup.cfg
View File

@ -239,7 +239,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/e*,moto/f*,moto/g*,moto/i*,moto/k*,moto/l*,moto/m*,moto/n*,moto/o*,moto/p*,moto/q*,moto/r*,moto/s3*,moto/sagemaker,moto/secretsmanager,moto/ses,moto/sqs,moto/ssm,moto/scheduler,moto/swf,moto/sns
files= moto/a*,moto/b*,moto/c*,moto/d*,moto/e*,moto/f*,moto/g*,moto/i*,moto/k*,moto/l*,moto/m*,moto/n*,moto/o*,moto/p*,moto/q*,moto/r*,moto/s*
show_column_numbers=True
show_error_codes = True
disable_error_code=abstract