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

Refactor - Extract CommonModels into separate file (#5121)

Browse Source
This commit is contained in:
Bert Blommers 2022-05-11 20:18:35 +00:00 committed by GitHub
parent 3e31e49c83
commit e49e67aba5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
31 changed files with 218 additions and 219 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
moto/budgets/__init__.py
View File

@ -1,4 +1,5 @@
from .models import budgets_backend
from ..core.models import base_decorator
budgets_backends = {"global": budgets_backend}
mock_budgets = budgets_backend.decorator
mock_budgets = base_decorator(budgets_backends)

2
moto/cloudformation/custom_model.py
View File

@ -2,7 +2,7 @@ import json
import threading
from moto import settings
from moto.core.models import CloudFormationModel
from moto.core import CloudFormationModel
from moto.awslambda import lambda_backends
from uuid import uuid4

5
moto/core/__init__.py
View File

@ -1,5 +1,6 @@
from .models import BaseModel, BaseBackend, get_account_id, ACCOUNT_ID # noqa
from .models import CloudFormationModel, CloudWatchMetricProvider # noqa
from .models import BaseBackend, get_account_id, ACCOUNT_ID # noqa
from .common_models import BaseModel # noqa
from .common_models import CloudFormationModel, CloudWatchMetricProvider # noqa
from .models import patch_client, patch_resource # noqa
from .responses import ActionAuthenticatorMixin

176
moto/core/common_models.py Normal file
View File

@ -0,0 +1,176 @@
from abc import abstractmethod
from .models import InstanceTrackerMeta
class BaseModel(metaclass=InstanceTrackerMeta):
def __new__(cls, *args, **kwargs): # pylint: disable=unused-argument
instance = super(BaseModel, cls).__new__(cls)
cls.instances.append(instance)
return instance
# Parent class for every Model that can be instantiated by CloudFormation
# On subclasses, implement all methods as @staticmethod to ensure correct behaviour of the CF parser
class CloudFormationModel(BaseModel):
@staticmethod
@abstractmethod
def cloudformation_name_type():
# https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-name.html
# This must be implemented as a staticmethod with no parameters
# Return None for resources that do not have a name property
pass
@staticmethod
@abstractmethod
def cloudformation_type():
# This must be implemented as a staticmethod with no parameters
# See for example https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-dynamodb-table.html
return "AWS::SERVICE::RESOURCE"
@classmethod
@abstractmethod
def has_cfn_attr(cls, attr):
# Used for validation
# If a template creates an Output for an attribute that does not exist, an error should be thrown
return True
@classmethod
@abstractmethod
def create_from_cloudformation_json(
cls, resource_name, cloudformation_json, region_name, **kwargs
):
# This must be implemented as a classmethod with parameters:
# cls, resource_name, cloudformation_json, region_name
# Extract the resource parameters from the cloudformation json
# and return an instance of the resource class
pass
@classmethod
@abstractmethod
def update_from_cloudformation_json(
cls, original_resource, new_resource_name, cloudformation_json, region_name
):
# This must be implemented as a classmethod with parameters:
# cls, original_resource, new_resource_name, cloudformation_json, region_name
# Extract the resource parameters from the cloudformation json,
# delete the old resource and return the new one. Optionally inspect
# the change in parameters and no-op when nothing has changed.
pass
@classmethod
@abstractmethod
def delete_from_cloudformation_json(
cls, resource_name, cloudformation_json, region_name
):
# This must be implemented as a classmethod with parameters:
# cls, resource_name, cloudformation_json, region_name
# Extract the resource parameters from the cloudformation json
# and delete the resource. Do not include a return statement.
pass
@abstractmethod
def is_created(self):
# Verify whether the resource was created successfully
# Assume True after initialization
# Custom resources may need time after init before they are created successfully
return True
class ConfigQueryModel:
def __init__(self, backends):
"""Inits based on the resource type's backends (1 for each region if applicable)"""
self.backends = backends
def list_config_service_resources(
self,
resource_ids,
resource_name,
limit,
next_token,
backend_region=None,
resource_region=None,
aggregator=None,
):
"""For AWS Config. This will list all of the resources of the given type and optional resource name and region.
This supports both aggregated and non-aggregated listing. The following notes the difference:
- Non-Aggregated Listing -
This only lists resources within a region. The way that this is implemented in moto is based on the region
for the resource backend.
You must set the `backend_region` to the region that the API request arrived from. resource_region can be set to `None`.
- Aggregated Listing -
This lists resources from all potential regional backends. For non-global resource types, this should collect a full
list of resources from all the backends, and then be able to filter from the resource region. This is because an
aggregator can aggregate resources from multiple regions. In moto, aggregated regions will *assume full aggregation
from all resources in all regions for a given resource type*.
The `backend_region` should be set to `None` for these queries, and the `resource_region` should optionally be set to
the `Filters` region parameter to filter out resources that reside in a specific region.
For aggregated listings, pagination logic should be set such that the next page can properly span all the region backends.
As such, the proper way to implement is to first obtain a full list of results from all the region backends, and then filter
from there. It may be valuable to make this a concatenation of the region and resource name.
:param resource_ids: A list of resource IDs
:param resource_name: The individual name of a resource
:param limit: How many per page
:param next_token: The item that will page on
:param backend_region: The region for the backend to pull results from. Set to `None` if this is an aggregated query.
:param resource_region: The region for where the resources reside to pull results from. Set to `None` if this is a
non-aggregated query.
:param aggregator: If the query is an aggregated query, *AND* the resource has "non-standard" aggregation logic (mainly, IAM),
you'll need to pass aggregator used. In most cases, this should be omitted/set to `None`. See the
conditional logic under `if aggregator` in the moto/iam/config.py for the IAM example.
:return: This should return a list of Dicts that have the following fields:
[
{
'type': 'AWS::The AWS Config data type',
'name': 'The name of the resource',
'id': 'The ID of the resource',
'region': 'The region of the resource -- if global, then you may want to have the calling logic pass in the
aggregator region in for the resource region -- or just us-east-1 :P'
}
, ...
]
"""
raise NotImplementedError()
def get_config_resource(
self, resource_id, resource_name=None, backend_region=None, resource_region=None
):
"""For AWS Config. This will query the backend for the specific resource type configuration.
This supports both aggregated, and non-aggregated fetching -- for batched fetching -- the Config batching requests
will call this function N times to fetch the N objects needing to be fetched.
- Non-Aggregated Fetching -
This only fetches a resource config within a region. The way that this is implemented in moto is based on the region
for the resource backend.
You must set the `backend_region` to the region that the API request arrived from. `resource_region` should be set to `None`.
- Aggregated Fetching -
This fetches resources from all potential regional backends. For non-global resource types, this should collect a full
list of resources from all the backends, and then be able to filter from the resource region. This is because an
aggregator can aggregate resources from multiple regions. In moto, aggregated regions will *assume full aggregation
from all resources in all regions for a given resource type*.
...
:param resource_id:
:param resource_name:
:param backend_region:
:param resource_region:
:return:
"""
raise NotImplementedError()
class CloudWatchMetricProvider(object):
@staticmethod
@abstractmethod
def get_cloudwatch_metrics():
pass

186
moto/core/models.py
View File

@ -6,7 +6,6 @@ import random
import re
import string
import unittest
from abc import abstractmethod
from collections import defaultdict
from io import BytesIO
from types import FunctionType
@ -495,80 +494,6 @@ class InstanceTrackerMeta(type):
return cls
class BaseModel(metaclass=InstanceTrackerMeta):
def __new__(cls, *args, **kwargs): # pylint: disable=unused-argument
instance = super(BaseModel, cls).__new__(cls)
cls.instances.append(instance)
return instance
# Parent class for every Model that can be instantiated by CloudFormation
# On subclasses, implement the two methods as @staticmethod to ensure correct behaviour of the CF parser
class CloudFormationModel(BaseModel):
@staticmethod
@abstractmethod
def cloudformation_name_type():
# https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-name.html
# This must be implemented as a staticmethod with no parameters
# Return None for resources that do not have a name property
pass
@staticmethod
@abstractmethod
def cloudformation_type():
# This must be implemented as a staticmethod with no parameters
# See for example https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-dynamodb-table.html
return "AWS::SERVICE::RESOURCE"
@classmethod
@abstractmethod
def has_cfn_attr(cls, attr):
# Used for validation
# If a template creates an Output for an attribute that does not exist, an error should be thrown
return True
@classmethod
@abstractmethod
def create_from_cloudformation_json(
cls, resource_name, cloudformation_json, region_name, **kwargs
):
# This must be implemented as a classmethod with parameters:
# cls, resource_name, cloudformation_json, region_name
# Extract the resource parameters from the cloudformation json
# and return an instance of the resource class
pass
@classmethod
@abstractmethod
def update_from_cloudformation_json(
cls, original_resource, new_resource_name, cloudformation_json, region_name
):
# This must be implemented as a classmethod with parameters:
# cls, original_resource, new_resource_name, cloudformation_json, region_name
# Extract the resource parameters from the cloudformation json,
# delete the old resource and return the new one. Optionally inspect
# the change in parameters and no-op when nothing has changed.
pass
@classmethod
@abstractmethod
def delete_from_cloudformation_json(
cls, resource_name, cloudformation_json, region_name
):
# This must be implemented as a classmethod with parameters:
# cls, resource_name, cloudformation_json, region_name
# Extract the resource parameters from the cloudformation json
# and delete the resource. Do not include a return statement.
pass
@abstractmethod
def is_created(self):
# Verify whether the resource was created successfully
# Assume True after initialization
# Custom resources may need time after init before they are created successfully
return True
class BaseBackend:
def _reset_model_refs(self):
# Remove all references to the models stored
@ -703,115 +628,11 @@ class BaseBackend:
]
return [endpoint_service]
def decorator(self, func=None):
if settings.TEST_SERVER_MODE:
mocked_backend = ServerModeMockAWS({"global": self})
else:
mocked_backend = MockAWS({"global": self})
if func:
return mocked_backend(func)
else:
return mocked_backend
# def list_config_service_resources(self, resource_ids, resource_name, limit, next_token):
# """For AWS Config. This will list all of the resources of the given type and optional resource name and region"""
# raise NotImplementedError()
class ConfigQueryModel:
def __init__(self, backends):
"""Inits based on the resource type's backends (1 for each region if applicable)"""
self.backends = backends
def list_config_service_resources(
self,
resource_ids,
resource_name,
limit,
next_token,
backend_region=None,
resource_region=None,
aggregator=None,
):
"""For AWS Config. This will list all of the resources of the given type and optional resource name and region.
This supports both aggregated and non-aggregated listing. The following notes the difference:
- Non-Aggregated Listing -
This only lists resources within a region. The way that this is implemented in moto is based on the region
for the resource backend.
You must set the `backend_region` to the region that the API request arrived from. resource_region can be set to `None`.
- Aggregated Listing -
This lists resources from all potential regional backends. For non-global resource types, this should collect a full
list of resources from all the backends, and then be able to filter from the resource region. This is because an
aggregator can aggregate resources from multiple regions. In moto, aggregated regions will *assume full aggregation
from all resources in all regions for a given resource type*.
The `backend_region` should be set to `None` for these queries, and the `resource_region` should optionally be set to
the `Filters` region parameter to filter out resources that reside in a specific region.
For aggregated listings, pagination logic should be set such that the next page can properly span all the region backends.
As such, the proper way to implement is to first obtain a full list of results from all the region backends, and then filter
from there. It may be valuable to make this a concatenation of the region and resource name.
:param resource_ids: A list of resource IDs
:param resource_name: The individual name of a resource
:param limit: How many per page
:param next_token: The item that will page on
:param backend_region: The region for the backend to pull results from. Set to `None` if this is an aggregated query.
:param resource_region: The region for where the resources reside to pull results from. Set to `None` if this is a
non-aggregated query.
:param aggregator: If the query is an aggregated query, *AND* the resource has "non-standard" aggregation logic (mainly, IAM),
you'll need to pass aggregator used. In most cases, this should be omitted/set to `None`. See the
conditional logic under `if aggregator` in the moto/iam/config.py for the IAM example.
:return: This should return a list of Dicts that have the following fields:
[
{
'type': 'AWS::The AWS Config data type',
'name': 'The name of the resource',
'id': 'The ID of the resource',
'region': 'The region of the resource -- if global, then you may want to have the calling logic pass in the
aggregator region in for the resource region -- or just us-east-1 :P'
}
, ...
]
"""
raise NotImplementedError()
def get_config_resource(
self, resource_id, resource_name=None, backend_region=None, resource_region=None
):
"""For AWS Config. This will query the backend for the specific resource type configuration.
This supports both aggregated, and non-aggregated fetching -- for batched fetching -- the Config batching requests
will call this function N times to fetch the N objects needing to be fetched.
- Non-Aggregated Fetching -
This only fetches a resource config within a region. The way that this is implemented in moto is based on the region
for the resource backend.
You must set the `backend_region` to the region that the API request arrived from. `resource_region` should be set to `None`.
- Aggregated Fetching -
This fetches resources from all potential regional backends. For non-global resource types, this should collect a full
list of resources from all the backends, and then be able to filter from the resource region. This is because an
aggregator can aggregate resources from multiple regions. In moto, aggregated regions will *assume full aggregation
from all resources in all regions for a given resource type*.
...
:param resource_id:
:param resource_name:
:param backend_region:
:param resource_region:
:return:
"""
raise NotImplementedError()
class base_decorator:
mock_backend = MockAWS
@ -828,10 +649,3 @@ class base_decorator:
return mocked_backend(func)
else:
return mocked_backend
class CloudWatchMetricProvider(object):
@staticmethod
@abstractmethod
def get_cloudwatch_metrics():
pass

3
moto/dynamodb_v20111205/__init__.py
View File

@ -1,4 +1,5 @@
from .models import dynamodb_backend
from ..core.models import base_decorator
"""
An older API version of DynamoDB.
@ -6,4 +7,4 @@ Please see the corresponding tests (tests/test_dynamodb_v20111205) on how to inv
"""
dynamodb_backends = {"global": dynamodb_backend}
mock_dynamodb = dynamodb_backend.decorator
mock_dynamodb = base_decorator(dynamodb_backend)

2
moto/ec2/models/core.py
View File

@ -1,4 +1,4 @@
from moto.core.models import BaseModel
from moto.core import BaseModel
from ..exceptions import FilterNotImplementedError

3
moto/ec2/models/elastic_block_store.py
View File

@ -1,5 +1,4 @@
from moto.core import get_account_id
from moto.core.models import CloudFormationModel
from moto.core import get_account_id, CloudFormationModel
from moto.kms import kms_backends
from moto.packages.boto.ec2.blockdevicemapping import BlockDeviceType
from ..exceptions import (

3
moto/ec2/models/elastic_network_interfaces.py
View File

@ -1,5 +1,4 @@
from moto.core import get_account_id
from moto.core.models import CloudFormationModel
from moto.core import get_account_id, CloudFormationModel
from ..exceptions import InvalidNetworkAttachmentIdError, InvalidNetworkInterfaceIdError
from .core import TaggedEC2Resource
from .security_groups import SecurityGroup

2
moto/ec2/models/flow_logs.py
View File

@ -1,6 +1,6 @@
import itertools
from collections import defaultdict
from moto.core.models import CloudFormationModel
from moto.core import CloudFormationModel
from ..exceptions import (
FlowLogAlreadyExists,
InvalidAggregationIntervalParameterError,

2
moto/ec2/models/iam_instance_profile.py
View File

@ -1,5 +1,5 @@
from moto.core import get_account_id
from moto.core.models import CloudFormationModel
from moto.core import CloudFormationModel
from ..exceptions import (
IncorrectStateIamProfileAssociationError,
InvalidAssociationIDIamProfileAssociationError,

2
moto/ec2/models/instances.py
View File

@ -4,7 +4,7 @@ from collections import OrderedDict
from datetime import datetime
from moto.core import get_account_id
from moto.core.models import CloudFormationModel
from moto.core import CloudFormationModel
from moto.core.utils import camelcase_to_underscores
from moto.packages.boto.ec2.blockdevicemapping import BlockDeviceMapping
from moto.packages.boto.ec2.instance import Instance as BotoInstance, Reservation

2
moto/ec2/models/nat_gateways.py
View File

@ -1,6 +1,6 @@
from datetime import datetime
from moto.core.models import CloudFormationModel
from moto.core import CloudFormationModel
from moto.core.utils import iso_8601_datetime_with_milliseconds
from .core import TaggedEC2Resource
from ..utils import random_nat_gateway_id, random_private_ip

3
moto/ec2/models/security_groups.py
View File

@ -3,8 +3,7 @@ import itertools
import json
from collections import defaultdict
from moto.core import get_account_id
from moto.core.models import CloudFormationModel
from moto.core import get_account_id, CloudFormationModel
from moto.core.utils import aws_api_matches
from ..exceptions import (
DependencyViolationError,

3
moto/ec2/models/spot_requests.py
View File

@ -1,6 +1,7 @@
from collections import defaultdict
from moto.core.models import Model, CloudFormationModel
from moto.core.common_models import CloudFormationModel
from moto.core.models import Model
from moto.packages.boto.ec2.launchspecification import LaunchSpecification
from moto.packages.boto.ec2.spotinstancerequest import (
SpotInstanceRequest as BotoSpotRequest,

2
moto/ec2/models/subnets.py
View File

@ -3,7 +3,7 @@ import itertools
from collections import defaultdict
from moto.core import get_account_id
from moto.core.models import CloudFormationModel
from moto.core import CloudFormationModel
from ..exceptions import (
GenericInvalidParameterValueError,
InvalidAvailabilityZoneError,

2
moto/ec2/models/vpc_peering_connections.py
View File

@ -1,6 +1,6 @@
import weakref
from collections import defaultdict
from moto.core.models import CloudFormationModel
from moto.core import CloudFormationModel
from ..exceptions import (
InvalidVPCPeeringConnectionIdError,
InvalidVPCPeeringConnectionStateTransitionError,

2
moto/ec2/models/vpc_service_configuration.py
View File

@ -1,4 +1,4 @@
from moto.core.models import CloudFormationModel
from moto.core import CloudFormationModel
from moto.core.utils import get_random_hex
from .core import TaggedEC2Resource
from ..exceptions import UnknownVpcEndpointService

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

@ -5,7 +5,7 @@ from collections import defaultdict
from operator import itemgetter
from moto.core import get_account_id
from moto.core.models import CloudFormationModel
from moto.core import CloudFormationModel
from .core import TaggedEC2Resource
from ..exceptions import (
CidrLimitExceeded,

3
moto/glue/__init__.py
View File

@ -1,4 +1,5 @@
from .models import glue_backend
from ..core.models import base_decorator
glue_backends = {"global": glue_backend}
mock_glue = glue_backend.decorator
mock_glue = base_decorator(glue_backends)

3
moto/iam/__init__.py
View File

@ -1,4 +1,5 @@
from .models import iam_backend
from ..core.models import base_decorator
iam_backends = {"global": iam_backend}
mock_iam = iam_backend.decorator
mock_iam = base_decorator(iam_backends)

2
moto/iam/config.py
View File

@ -1,7 +1,7 @@
import json
import boto3
from moto.core.exceptions import InvalidNextTokenException
from moto.core.models import ConfigQueryModel
from moto.core.common_models import ConfigQueryModel
from moto.iam import iam_backends

2
moto/logs/models.py
View File

@ -3,7 +3,7 @@ import uuid
from datetime import datetime, timedelta
from moto.core import get_account_id, BaseBackend, BaseModel
from moto.core.models import CloudFormationModel
from moto.core import CloudFormationModel
from moto.core.utils import unix_time_millis, BackendDict
from moto.utilities.paginator import paginate
from moto.logs.metric_filters import MetricFilters

3
moto/route53/__init__.py
View File

@ -1,4 +1,5 @@
from .models import route53_backend
from ..core.models import base_decorator
route53_backends = {"global": route53_backend}
mock_route53 = route53_backend.decorator
mock_route53 = base_decorator(route53_backends)

3
moto/s3/__init__.py
View File

@ -1,4 +1,5 @@
from .models import s3_backend
from ..core.models import base_decorator
s3_backends = {"global": s3_backend}
mock_s3 = s3_backend.decorator
mock_s3 = base_decorator(s3_backends)

2
moto/s3/config.py
View File

@ -1,7 +1,7 @@
import json
from moto.core.exceptions import InvalidNextTokenException
from moto.core.models import ConfigQueryModel
from moto.core.common_models import ConfigQueryModel
from moto.s3 import s3_backends

2
moto/s3control/config.py
View File

@ -5,7 +5,7 @@ import time
from boto3 import Session
from moto.core.exceptions import InvalidNextTokenException
from moto.core.models import ConfigQueryModel
from moto.core.common_models import ConfigQueryModel
from moto.s3control import s3control_backends
from moto.s3.models import get_moto_s3_account_id

3
moto/sagemaker/__init__.py
View File

@ -1,4 +1,5 @@
from .models import sagemaker_backends
from ..core.models import base_decorator
sagemaker_backend = sagemaker_backends["us-east-1"]
mock_sagemaker = sagemaker_backend.decorator
mock_sagemaker = base_decorator(sagemaker_backends)

3
moto/ses/__init__.py
View File

@ -1,4 +1,5 @@
from .models import ses_backend
from ..core.models import base_decorator
ses_backends = {"global": ses_backend}
mock_ses = ses_backend.decorator
mock_ses = base_decorator(ses_backends)

3
moto/sts/__init__.py
View File

@ -1,4 +1,5 @@
from .models import sts_backend
from ..core.models import base_decorator
sts_backends = {"global": sts_backend}
mock_sts = sts_backend.decorator
mock_sts = base_decorator(sts_backends)

3
moto/transcribe/__init__.py
View File

@ -1,4 +1,5 @@
from .models import transcribe_backends
from ..core.models import base_decorator
transcribe_backend = transcribe_backends["us-east-1"]
mock_transcribe = transcribe_backend.decorator
mock_transcribe = base_decorator(transcribe_backends)