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

Techdebt: Various improvements (#7154)

Browse Source
This commit is contained in:
Bert Blommers 2023-12-22 13:01:57 -01:00 committed by GitHub
parent dc18556449
commit f730d59229
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
19 changed files with 144 additions and 91 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.github/workflows/build.yml vendored
View File

@ -40,8 +40,9 @@ jobs:
runs-on: ubuntu-latest
needs: cache
strategy:
fail-fast: false
matrix:
python-version: [3.9]
python-version: [3.9, "3.10", "3.11", "3.12"]
steps:
- uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}

14
.github/workflows/dockertests.yml vendored
View File

@ -8,7 +8,7 @@ jobs:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: [ 3.9 ]
python-version: [ "3.11" ]
steps:
- uses: actions/checkout@v4
@ -41,7 +41,7 @@ jobs:
needs: cache
strategy:
matrix:
python-version: [3.9]
python-version: ["3.11"]
steps:
- uses: actions/checkout@v4
@ -55,7 +55,7 @@ jobs:
run: |
pip install --upgrade build
python -m build
docker run --rm -t --name motoserver -e TEST_SERVER_MODE=true -e AWS_SECRET_ACCESS_KEY=server_secret -e MOTO_PORT=4555 -e AWS_ACCESS_KEY_ID=server_key -v `pwd`:/moto -p 4555:4555 -v /var/run/docker.sock:/var/run/docker.sock python:3.7-buster /moto/scripts/ci_moto_server.sh &
docker run --rm -t --name motoserver -e TEST_SERVER_MODE=true -e AWS_SECRET_ACCESS_KEY=server_secret -e MOTO_PORT=4555 -e AWS_ACCESS_KEY_ID=server_key -v `pwd`:/moto -p 4555:4555 -v /var/run/docker.sock:/var/run/docker.sock python:3.11-slim /moto/scripts/ci_moto_server.sh &
MOTO_PORT=4555 python scripts/ci_wait_for_server.py
- name: Get pip cache dir
id: pip-cache
@ -100,7 +100,7 @@ jobs:
needs: cache
strategy:
matrix:
python-version: [3.9]
python-version: ["3.11"]
steps:
- uses: actions/checkout@v4
@ -115,7 +115,7 @@ jobs:
pip install --upgrade build
python -m build
docker network create -d bridge my-custom-network
docker run --rm -t -e TEST_SERVER_MODE=true -e MOTO_DOCKER_NETWORK_NAME=my-custom-network -e AWS_SECRET_ACCESS_KEY=server_secret -e AWS_ACCESS_KEY_ID=server_key -v `pwd`:/moto -p 5000:5000 --network my-custom-network -v /var/run/docker.sock:/var/run/docker.sock python:3.7-buster /moto/scripts/ci_moto_server.sh &
docker run --rm -t -e TEST_SERVER_MODE=true -e MOTO_DOCKER_NETWORK_NAME=my-custom-network -e AWS_SECRET_ACCESS_KEY=server_secret -e AWS_ACCESS_KEY_ID=server_key -v `pwd`:/moto -p 5000:5000 --network my-custom-network -v /var/run/docker.sock:/var/run/docker.sock python:3.11-slim /moto/scripts/ci_moto_server.sh &
python scripts/ci_wait_for_server.py
- name: Get pip cache dir
id: pip-cache
@ -158,7 +158,7 @@ jobs:
needs: cache
strategy:
matrix:
python-version: [3.9]
python-version: ["3.11"]
steps:
- uses: actions/checkout@v4
@ -172,7 +172,7 @@ jobs:
run: |
pip install --upgrade build
python -m build
docker run --rm -t -e MOTO_DOCKER_NETWORK_MODE=host -e TEST_SERVER_MODE=true -e AWS_SECRET_ACCESS_KEY=server_secret -e MOTO_PORT=4555 -e AWS_ACCESS_KEY_ID=server_key -v `pwd`:/moto -p 4555:4555 -v /var/run/docker.sock:/var/run/docker.sock python:3.7-buster /moto/scripts/ci_moto_server.sh &
docker run --rm -t -e MOTO_DOCKER_NETWORK_MODE=host -e TEST_SERVER_MODE=true -e AWS_SECRET_ACCESS_KEY=server_secret -e MOTO_PORT=4555 -e AWS_ACCESS_KEY_ID=server_key -v `pwd`:/moto -p 4555:4555 -v /var/run/docker.sock:/var/run/docker.sock python:3.11-slim /moto/scripts/ci_moto_server.sh &
MOTO_PORT=4555 python scripts/ci_wait_for_server.py
- name: Get pip cache dir
id: pip-cache

20
docs/docs/contributing/development_tips/new_state_transitions.rst
View File

@ -45,15 +45,6 @@ An example model could look like this:
from moto.moto_api import state_manager
class Backend():
def __init__():
# This is how we register the model, and specify the default transition-behaviour
# Typically this is done when constructing the Backend-class
state_manager.register_default_transition(
# This name should be the same as the name used in NewModel
model_name="new::model",
# Any transition-config is possible - this is a good default option though
transition={"progression": "immediate"},
)
def list_resources():
for ec2_instance in all_resources:
@ -76,3 +67,14 @@ An example model could look like this:
# Make sure that each way (describe, list, get_, ) calls the advance()-method, and the resource can actually progress to the next state
resource.advance()
return resource
Make sure that the model is registered with the StateManager. This can be done in `moto/moto_api/__init__.py`:
.. sourcecode:: python
state_manager.register_default_transition(
# This name should be the same as the name used in NewModel
model_name="new::model",
# Any transition-config is possible - this is a good default option though
transition={"progression": "immediate"},
)

3
moto/__init__.py
View File

@ -254,9 +254,6 @@ class MockAll(ContextDecorator):
mock_all = MockAll
# import logging
# logging.getLogger('boto').setLevel(logging.CRITICAL)
__title__ = "moto"
__version__ = "4.2.13.dev"

5
moto/batch/models.py
View File

@ -22,7 +22,6 @@ from moto.ecs.models import EC2ContainerServiceBackend, ecs_backends
from moto.iam.exceptions import IAMNotFoundException
from moto.iam.models import IAMBackend, iam_backends
from moto.logs.models import LogsBackend, logs_backends
from moto.moto_api import state_manager
from moto.moto_api._internal import mock_random
from moto.moto_api._internal.managed_state_model import ManagedState
from moto.utilities.docker_utilities import DockerModel
@ -1014,10 +1013,6 @@ class BatchBackend(BaseBackend):
self._jobs: Dict[str, Job] = {}
self._scheduling_policies: Dict[str, SchedulingPolicy] = {}
state_manager.register_default_transition(
"batch::job", transition={"progression": "manual", "times": 1}
)
@property
def iam_backend(self) -> IAMBackend:
"""

5
moto/cloudfront/models.py
View File

@ -3,7 +3,6 @@ from typing import Any, Dict, Iterable, List, Optional, Tuple
from moto.core import BackendDict, BaseBackend, BaseModel
from moto.core.utils import iso_8601_datetime_with_milliseconds
from moto.moto_api import state_manager
from moto.moto_api._internal import mock_random as random
from moto.moto_api._internal.managed_state_model import ManagedState
from moto.utilities.tagging_service import TaggingService
@ -271,10 +270,6 @@ class CloudFrontBackend(BaseBackend):
self.origin_access_controls: Dict[str, OriginAccessControl] = dict()
self.tagger = TaggingService()
state_manager.register_default_transition(
"cloudfront::distribution", transition={"progression": "manual", "times": 1}
)
def create_distribution(
self, distribution_config: Dict[str, Any], tags: List[Dict[str, str]]
) -> Tuple[Distribution, str, str]:

14
moto/cognitoidp/models.py
View File

@ -2162,6 +2162,10 @@ class RegionAgnosticBackend:
# Without authentication-header, we lose the context of which region the request was send to
# This backend will cycle through all backends as a workaround
def __init__(self, account_id: str, region_name: str):
self.account_id = account_id
self.region_name = region_name
def _find_backend_by_access_token(self, access_token: str) -> CognitoIdpBackend:
for account_specific_backends in cognitoidp_backends.values():
for region, backend in account_specific_backends.items():
@ -2170,7 +2174,7 @@ class RegionAgnosticBackend:
for p in backend.user_pools.values():
if access_token in p.access_tokens:
return backend
return backend
return cognitoidp_backends[self.account_id][self.region_name]
def _find_backend_for_clientid(self, client_id: str) -> CognitoIdpBackend:
for account_specific_backends in cognitoidp_backends.values():
@ -2180,7 +2184,7 @@ class RegionAgnosticBackend:
for p in backend.user_pools.values():
if client_id in p.clients:
return backend
return backend
return cognitoidp_backends[self.account_id][self.region_name]
def sign_up(
self,
@ -2237,7 +2241,9 @@ cognitoidp_backends = BackendDict(CognitoIdpBackend, "cognito-idp")
# Hack to help moto-server process requests on localhost, where the region isn't
# specified in the host header. Some endpoints (change password, confirm forgot
# password) have no authorization header from which to extract the region.
def find_account_region_by_value(key: str, value: str) -> Tuple[str, str]:
def find_account_region_by_value(
key: str, value: str, fallback: Tuple[str, str]
) -> Tuple[str, str]:
for account_id, account_specific_backend in cognitoidp_backends.items():
for region, backend in account_specific_backend.items():
for user_pool in backend.user_pools.values():
@ -2249,4 +2255,4 @@ def find_account_region_by_value(key: str, value: str) -> Tuple[str, str]:
# If we can't find the `client_id` or `access_token`, we just pass
# back a default backend region, which will raise the appropriate
# error message (e.g. NotAuthorized or NotFound).
return account_id, region
return fallback

32
moto/cognitoidp/responses.py
View File

@ -14,13 +14,14 @@ from .models import (
find_account_region_by_value,
)
region_agnostic_backend = RegionAgnosticBackend()
class CognitoIdpResponse(BaseResponse):
def __init__(self) -> None:
super().__init__(service_name="cognito-idp")
def _get_region_agnostic_backend(self) -> RegionAgnosticBackend:
return RegionAgnosticBackend(self.current_account, self.region)
@property
def parameters(self) -> Dict[str, Any]: # type: ignore[misc]
return json.loads(self.body)
@ -353,7 +354,7 @@ class CognitoIdpResponse(BaseResponse):
def get_user(self) -> str:
access_token = self._get_param("AccessToken")
user = region_agnostic_backend.get_user(access_token=access_token)
user = self._get_region_agnostic_backend().get_user(access_token=access_token)
return json.dumps(user.to_json(extended=True, attributes_key="UserAttributes"))
def list_users(self) -> str:
@ -454,7 +455,8 @@ class CognitoIdpResponse(BaseResponse):
client_id = self._get_param("ClientId")
challenge_name = self._get_param("ChallengeName")
challenge_responses = self._get_param("ChallengeResponses")
auth_result = region_agnostic_backend.admin_respond_to_auth_challenge(
backend = self._get_region_agnostic_backend()
auth_result = backend.admin_respond_to_auth_challenge(
session, client_id, challenge_name, challenge_responses
)
@ -465,7 +467,7 @@ class CognitoIdpResponse(BaseResponse):
client_id = self._get_param("ClientId")
challenge_name = self._get_param("ChallengeName")
challenge_responses = self._get_param("ChallengeResponses")
auth_result = region_agnostic_backend.respond_to_auth_challenge(
auth_result = self._get_region_agnostic_backend().respond_to_auth_challenge(
session, client_id, challenge_name, challenge_responses
)
@ -474,7 +476,9 @@ class CognitoIdpResponse(BaseResponse):
def forgot_password(self) -> str:
client_id = self._get_param("ClientId")
username = self._get_param("Username")
account, region = find_account_region_by_value("client_id", client_id)
account, region = find_account_region_by_value(
"client_id", client_id, fallback=(self.current_account, self.region)
)
confirmation_code, response = cognitoidp_backends[account][
region
].forgot_password(client_id, username)
@ -492,7 +496,9 @@ class CognitoIdpResponse(BaseResponse):
username = self._get_param("Username")
password = self._get_param("Password")
confirmation_code = self._get_param("ConfirmationCode")
account, region = find_account_region_by_value("client_id", client_id)
account, region = find_account_region_by_value(
"client_id", client_id, fallback=(self.current_account, self.region)
)
cognitoidp_backends[account][region].confirm_forgot_password(
client_id, username, password, confirmation_code
)
@ -503,7 +509,9 @@ class CognitoIdpResponse(BaseResponse):
access_token = self._get_param("AccessToken")
previous_password = self._get_param("PreviousPassword")
proposed_password = self._get_param("ProposedPassword")
account, region = find_account_region_by_value("access_token", access_token)
account, region = find_account_region_by_value(
"access_token", access_token, fallback=(self.current_account, self.region)
)
cognitoidp_backends[account][region].change_password(
access_token, previous_password, proposed_password
)
@ -573,7 +581,7 @@ class CognitoIdpResponse(BaseResponse):
client_id = self._get_param("ClientId")
username = self._get_param("Username")
password = self._get_param("Password")
user = region_agnostic_backend.sign_up(
user = self._get_region_agnostic_backend().sign_up(
client_id=client_id,
username=username,
password=password,
@ -589,7 +597,9 @@ class CognitoIdpResponse(BaseResponse):
def confirm_sign_up(self) -> str:
client_id = self._get_param("ClientId")
username = self._get_param("Username")
region_agnostic_backend.confirm_sign_up(client_id=client_id, username=username)
self._get_region_agnostic_backend().confirm_sign_up(
client_id=client_id, username=username
)
return ""
def initiate_auth(self) -> str:
@ -597,7 +607,7 @@ class CognitoIdpResponse(BaseResponse):
auth_flow = self._get_param("AuthFlow")
auth_parameters = self._get_param("AuthParameters")
auth_result = region_agnostic_backend.initiate_auth(
auth_result = self._get_region_agnostic_backend().initiate_auth(
client_id, auth_flow, auth_parameters
)

5
moto/dax/models.py
View File

@ -3,7 +3,6 @@ from typing import Any, Dict, Iterable, List
from moto.core import BackendDict, BaseBackend, BaseModel
from moto.core.utils import unix_time
from moto.moto_api import state_manager
from moto.moto_api._internal import mock_random as random
from moto.moto_api._internal.managed_state_model import ManagedState
from moto.utilities.paginator import paginate
@ -167,10 +166,6 @@ class DAXBackend(BaseBackend):
self._clusters: Dict[str, DaxCluster] = dict()
self._tagger = TaggingService()
state_manager.register_default_transition(
model_name="dax::cluster", transition={"progression": "manual", "times": 4}
)
@property
def clusters(self) -> Dict[str, DaxCluster]:
self._clusters = {

47
moto/ec2/models/vpcs.py
View File

@ -36,6 +36,50 @@ from ..utils import (
from .availability_zones_and_regions import RegionsAndZonesBackend
from .core import TaggedEC2Resource
# We used to load the entirety of Moto into memory, and check every module if it's supported
# But having a fixed list is much more performant
# Maintaining it is more difficult, but the contents of this list does not change very often
IMPLEMENTED_ENDPOINT_SERVICES = [
"acm",
"applicationautoscaling",
"athena",
"autoscaling",
"lambda",
"cloudformation",
"cloudwatch",
"codecommit",
"codepipeline",
"config",
"datasync",
"dms",
"ds",
"dynamodb",
"ec2",
"ecr",
"ecs",
"elasticbeanstalk",
"elbv2",
"emr",
"events",
"firehose",
"glue",
"iot",
"kinesis",
"kms",
"logs",
"rds",
"redshift",
"route53resolver",
"s3",
"sagemaker",
"secretsmanager",
"sns",
"sqs",
"ssm",
"sts",
"transcribe",
"xray",
]
MAX_NUMBER_OF_ENDPOINT_SERVICES_RESULTS = 1000
DEFAULT_VPC_ENDPOINT_SERVICES: List[Dict[str, str]] = []
@ -725,7 +769,8 @@ class VPCBackend:
from moto import backends # pylint: disable=import-outside-toplevel
for _backends in backends.service_backends():
for implemented_service in IMPLEMENTED_ENDPOINT_SERVICES:
_backends = backends.get_backend(implemented_service) # type: ignore[call-overload]
account_backend = _backends[account_id]
if region in account_backend:
service = account_backend[region].default_vpc_endpoint_service(

6
moto/ecs/models.py
View File

@ -9,7 +9,6 @@ from moto.core import BackendDict, BaseBackend, BaseModel, CloudFormationModel
from moto.core.exceptions import JsonRESTError
from moto.core.utils import pascal_to_camelcase, remap_nested_keys, unix_time
from moto.ec2 import ec2_backends
from moto.moto_api import state_manager
from moto.moto_api._internal import mock_random
from moto.moto_api._internal.managed_state_model import ManagedState
@ -962,11 +961,6 @@ class EC2ContainerServiceBackend(BaseBackend):
self.services: Dict[str, Service] = {}
self.container_instances: Dict[str, Dict[str, ContainerInstance]] = {}
state_manager.register_default_transition(
model_name="ecs::task",
transition={"progression": "manual", "times": 1},
)
@staticmethod
def default_vpc_endpoint_service(service_region: str, zones: List[str]) -> List[Dict[str, Any]]: # type: ignore[misc]
"""Default VPC endpoint service."""

5
moto/glue/models.py
View File

@ -7,7 +7,6 @@ from typing import Any, Dict, List, Optional
from moto.core import BackendDict, BaseBackend, BaseModel
from moto.core.utils import unix_time, utcnow
from moto.moto_api import state_manager
from moto.moto_api._internal import mock_random
from moto.moto_api._internal.managed_state_model import ManagedState
@ -113,10 +112,6 @@ class GlueBackend(BaseBackend):
self.num_schemas = 0
self.num_schema_versions = 0
state_manager.register_default_transition(
model_name="glue::job_run", transition={"progression": "immediate"}
)
@staticmethod
def default_vpc_endpoint_service(
service_region: str, zones: List[str]

37
moto/moto_api/__init__.py
View File

@ -6,6 +6,43 @@ Use this manager to configure how AWS models transition between states. (initial
"""
state_manager = _internal.state_manager.StateManager()
"""
Default transitions across Moto
"""
state_manager.register_default_transition(
"batch::job", transition={"progression": "manual", "times": 1}
)
state_manager.register_default_transition(
"cloudfront::distribution", transition={"progression": "manual", "times": 1}
)
state_manager.register_default_transition(
model_name="dax::cluster", transition={"progression": "manual", "times": 4}
)
state_manager.register_default_transition(
model_name="ecs::task", transition={"progression": "manual", "times": 1}
)
state_manager.register_default_transition(
model_name="glue::job_run", transition={"progression": "immediate"}
)
state_manager.register_default_transition(
"s3::keyrestore", transition={"progression": "immediate"}
)
state_manager.register_default_transition(
model_name="support::case", transition={"progression": "manual", "times": 1}
)
state_manager.register_default_transition(
"transcribe::vocabulary", transition={"progression": "manual", "times": 1}
)
state_manager.register_default_transition(
"transcribe::medicalvocabulary", transition={"progression": "manual", "times": 1}
)
state_manager.register_default_transition(
"transcribe::transcriptionjob", transition={"progression": "manual", "times": 1}
)
state_manager.register_default_transition(
"transcribe::medicaltranscriptionjob",
transition={"progression": "manual", "times": 1},
)
""""
Recorder, used to record calls to Moto and replay them later

5
moto/s3/models.py
View File

@ -29,7 +29,6 @@ from moto.core.utils import (
unix_time_millis,
utcnow,
)
from moto.moto_api import state_manager
from moto.moto_api._internal import mock_random as random
from moto.moto_api._internal.managed_state_model import ManagedState
from moto.s3.exceptions import (
@ -1615,10 +1614,6 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
self.buckets: Dict[str, FakeBucket] = {}
self.tagger = TaggingService()
state_manager.register_default_transition(
"s3::keyrestore", transition={"progression": "immediate"}
)
def reset(self) -> None:
# For every key and multipart, Moto opens a TemporaryFile to write the value of those keys
# Ensure that these TemporaryFile-objects are closed, and leave no filehandles open

5
moto/support/models.py
View File

@ -2,7 +2,6 @@ import datetime
from typing import Any, Dict, List, Optional
from moto.core import BackendDict, BaseBackend
from moto.moto_api import state_manager
from moto.moto_api._internal import mock_random as random
from moto.moto_api._internal.managed_state_model import ManagedState
from moto.utilities.utils import load_resource
@ -67,10 +66,6 @@ class SupportBackend(BaseBackend):
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) -> List[Dict[str, Any]]:
"""
The Language-parameter is not yet implemented

17
moto/transcribe/models.py
View File

@ -3,7 +3,6 @@ from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional
from moto.core import BackendDict, BaseBackend, BaseModel
from moto.moto_api import state_manager
from moto.moto_api._internal import mock_random
from moto.moto_api._internal.managed_state_model import ManagedState
@ -480,22 +479,6 @@ class TranscribeBackend(BaseBackend):
self.medical_vocabularies: Dict[str, FakeMedicalVocabulary] = {}
self.vocabularies: Dict[str, FakeVocabulary] = {}
state_manager.register_default_transition(
"transcribe::vocabulary", transition={"progression": "manual", "times": 1}
)
state_manager.register_default_transition(
"transcribe::medicalvocabulary",
transition={"progression": "manual", "times": 1},
)
state_manager.register_default_transition(
"transcribe::transcriptionjob",
transition={"progression": "manual", "times": 1},
)
state_manager.register_default_transition(
"transcribe::medicaltranscriptionjob",
transition={"progression": "manual", "times": 1},
)
@staticmethod
def default_vpc_endpoint_service(
service_region: str, zones: List[str]

2
moto/utilities/utils.py
View File

@ -22,7 +22,7 @@ def load_resource(package: str, resource: str) -> Any:
def load_resource_as_str(package: str, resource: str) -> str:
return load_resource_as_bytes(package, resource).decode("utf-8") # type: ignore
return load_resource_as_bytes(package, resource).decode("utf-8")
def load_resource_as_bytes(package: str, resource: str) -> bytes:

5
tests/test_efs/test_server.py
View File

@ -1,9 +1,10 @@
import re
from unittest import SkipTest
import pytest
import moto.server as server
from moto import mock_ec2, mock_efs
from moto import mock_ec2, mock_efs, settings
FILE_SYSTEMS = "/2015-02-01/file-systems"
MOUNT_TARGETS = "/2015-02-01/mount-targets"
@ -20,6 +21,8 @@ def fixture_aws_credentials(monkeypatch):
@pytest.fixture(scope="function", name="efs_client")
def fixture_efs_client(aws_credentials): # pylint: disable=unused-argument
if not settings.TEST_DECORATOR_MODE:
raise SkipTest("Using server directly - no point in testing ServerMode")
with mock_efs():
yield server.create_backend_app("efs").test_client()

5
tests/test_iam/test_iam.py
View File

@ -1,6 +1,7 @@
import csv
import json
from datetime import datetime
from unittest import SkipTest
from urllib import parse
from uuid import uuid4
@ -3580,6 +3581,8 @@ def test_role_list_config_discovered_resources():
@mock_iam
def test_role_config_dict():
if not settings.TEST_DECORATOR_MODE:
raise SkipTest("Using backend directly - no point in testing ServerMode")
from moto.iam.config import policy_config_query, role_config_query
from moto.iam.utils import random_policy_id, random_role_id
@ -4189,6 +4192,8 @@ def test_policy_list_config_discovered_resources():
@mock_iam
def test_policy_config_dict():
if not settings.TEST_DECORATOR_MODE:
raise SkipTest("Using backend directly - no point in testing ServerMode")
from moto.iam.config import policy_config_query, role_config_query
from moto.iam.utils import random_policy_id