Admin - Extract MotoAPI functionality to dedicated module (#5055)

MANIFEST.in

@@ -7,7 +7,9 @@ include moto/ec2/resources/instance_type_offerings/*/*.json
 include moto/ec2/resources/amis.json
 include moto/cognitoidp/resources/*.json
 include moto/dynamodb/parsing/reserved_keywords.txt
+include moto/moto_api/_internal/*
 include moto/ssm/resources/*.json
 include moto/support/resources/*.json
 recursive-include moto/moto_server *
 recursive-include tests *
+recursive-exclude tests/terraformtests *

moto/backends.py

@@ -11,7 +11,7 @@ decorators = [
 decorator_functions = [getattr(moto, f) for f in decorators]
 BACKENDS = {f.boto3_name: (f.name, f.backend) for f in decorator_functions}
 BACKENDS["dynamodb_v20111205"] = ("dynamodb_v20111205", "dynamodb_backends")
-BACKENDS["moto_api"] = ("core", "moto_api_backends")
+BACKENDS["moto_api"] = ("moto_api._internal", "moto_api_backends")
 BACKENDS["instance_metadata"] = ("instance_metadata", "instance_metadata_backends")
 BACKENDS["s3bucket_path"] = ("s3", "s3_backends")
 

moto/core/__init__.py

@@ -1,9 +1,8 @@
-from .models import BaseModel, BaseBackend, moto_api_backend, ACCOUNT_ID  # noqa
+from .models import BaseModel, BaseBackend, ACCOUNT_ID  # noqa
 from .models import CloudFormationModel, CloudWatchMetricProvider  # noqa
 from .models import patch_client, patch_resource  # noqa
 from .responses import ActionAuthenticatorMixin
 
-moto_api_backends = {"global": moto_api_backend}
 set_initial_no_auth_action_count = (
     ActionAuthenticatorMixin.set_initial_no_auth_action_count
 )

moto/core/models.py

@@ -39,7 +39,7 @@ class BaseMockAWS:
 
     def __init__(self, backends):
         from moto.instance_metadata import instance_metadata_backend
-        from moto.core import moto_api_backend
+        from moto.moto_api._internal.models import moto_api_backend
 
         self.backends = backends
 
@@ -822,23 +822,8 @@ class base_decorator:
             return mocked_backend
 
 
-class MotoAPIBackend(BaseBackend):
-    def reset(self):
-        import moto.backends as backends
-
-        for name, backends_ in backends.loaded_backends():
-            if name == "moto_api":
-                continue
-            for backend in backends_.values():
-                backend.reset()
-        self.__init__()
-
-
 class CloudWatchMetricProvider(object):
     @staticmethod
     @abstractmethod
     def get_cloudwatch_metrics():
         pass
-
-
-moto_api_backend = MotoAPIBackend()

moto/core/responses.py

@@ -810,69 +810,6 @@ class BaseResponse(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
         return True
 
 
-class MotoAPIResponse(BaseResponse):
-    def reset_response(
-        self, request, full_url, headers
-    ):  # pylint: disable=unused-argument
-        if request.method == "POST":
-            from .models import moto_api_backend
-
-            moto_api_backend.reset()
-            return 200, {}, json.dumps({"status": "ok"})
-        return 400, {}, json.dumps({"Error": "Need to POST to reset Moto"})
-
-    def reset_auth_response(
-        self, request, full_url, headers
-    ):  # pylint: disable=unused-argument
-        if request.method == "POST":
-            previous_initial_no_auth_action_count = (
-                settings.INITIAL_NO_AUTH_ACTION_COUNT
-            )
-            settings.INITIAL_NO_AUTH_ACTION_COUNT = float(request.data.decode())
-            ActionAuthenticatorMixin.request_count = 0
-            return (
-                200,
-                {},
-                json.dumps(
-                    {
-                        "status": "ok",
-                        "PREVIOUS_INITIAL_NO_AUTH_ACTION_COUNT": str(
-                            previous_initial_no_auth_action_count
-                        ),
-                    }
-                ),
-            )
-        return 400, {}, json.dumps({"Error": "Need to POST to reset Moto Auth"})
-
-    def model_data(self, request, full_url, headers):  # pylint: disable=unused-argument
-        from moto.core.models import model_data
-
-        results = {}
-        for service in sorted(model_data):
-            models = model_data[service]
-            results[service] = {}
-            for name in sorted(models):
-                model = models[name]
-                results[service][name] = []
-                for instance in model.instances:
-                    inst_result = {}
-                    for attr in dir(instance):
-                        if not attr.startswith("_"):
-                            try:
-                                json.dumps(getattr(instance, attr))
-                            except TypeError:
-                                pass
-                            else:
-                                inst_result[attr] = getattr(instance, attr)
-                    results[service][name].append(inst_result)
-        return 200, {"Content-Type": "application/javascript"}, json.dumps(results)
-
-    def dashboard(self, request, full_url, headers):  # pylint: disable=unused-argument
-        from flask import render_template
-
-        return render_template("dashboard.html")
-
-
 class _RecursiveDictRef(object):
     """Store a recursive reference to dict."""
 

moto/core/urls.py

@@ -1,12 +1,3 @@
-from .responses import MotoAPIResponse
+url_bases = []
 
-url_bases = ["https?://motoapi.amazonaws.com"]
+url_paths = {}
-
-response_instance = MotoAPIResponse()
-
-url_paths = {
-    "{0}/moto-api/$": response_instance.dashboard,
-    "{0}/moto-api/data.json": response_instance.model_data,
-    "{0}/moto-api/reset": response_instance.reset_response,
-    "{0}/moto-api/reset-auth": response_instance.reset_auth_response,
-}

moto/moto_api/_internal/__init__.py

@@ -0,0 +1,4 @@
+from .models import moto_api_backend
+
+
+moto_api_backends = {"global": moto_api_backend}

moto/moto_api/_internal/models.py

@@ -0,0 +1,16 @@
+from moto.core.models import BaseBackend
+
+
+class MotoAPIBackend(BaseBackend):
+    def reset(self):
+        import moto.backends as backends
+
+        for name, backends_ in backends.loaded_backends():
+            if name == "moto_api":
+                continue
+            for backend in backends_.values():
+                backend.reset()
+        self.__init__()
+
+
+moto_api_backend = MotoAPIBackend()

moto/moto_api/_internal/responses.py

@@ -0,0 +1,67 @@
+import json
+
+from moto import settings
+from moto.core.responses import ActionAuthenticatorMixin, BaseResponse
+
+
+class MotoAPIResponse(BaseResponse):
+    def reset_response(
+        self, request, full_url, headers
+    ):  # pylint: disable=unused-argument
+        if request.method == "POST":
+            from .models import moto_api_backend
+
+            moto_api_backend.reset()
+            return 200, {}, json.dumps({"status": "ok"})
+        return 400, {}, json.dumps({"Error": "Need to POST to reset Moto"})
+
+    def reset_auth_response(
+        self, request, full_url, headers
+    ):  # pylint: disable=unused-argument
+        if request.method == "POST":
+            previous_initial_no_auth_action_count = (
+                settings.INITIAL_NO_AUTH_ACTION_COUNT
+            )
+            settings.INITIAL_NO_AUTH_ACTION_COUNT = float(request.data.decode())
+            ActionAuthenticatorMixin.request_count = 0
+            return (
+                200,
+                {},
+                json.dumps(
+                    {
+                        "status": "ok",
+                        "PREVIOUS_INITIAL_NO_AUTH_ACTION_COUNT": str(
+                            previous_initial_no_auth_action_count
+                        ),
+                    }
+                ),
+            )
+        return 400, {}, json.dumps({"Error": "Need to POST to reset Moto Auth"})
+
+    def model_data(self, request, full_url, headers):  # pylint: disable=unused-argument
+        from moto.core.models import model_data
+
+        results = {}
+        for service in sorted(model_data):
+            models = model_data[service]
+            results[service] = {}
+            for name in sorted(models):
+                model = models[name]
+                results[service][name] = []
+                for instance in model.instances:
+                    inst_result = {}
+                    for attr in dir(instance):
+                        if not attr.startswith("_"):
+                            try:
+                                json.dumps(getattr(instance, attr))
+                            except TypeError:
+                                pass
+                            else:
+                                inst_result[attr] = getattr(instance, attr)
+                    results[service][name].append(inst_result)
+        return 200, {"Content-Type": "application/javascript"}, json.dumps(results)
+
+    def dashboard(self, request, full_url, headers):  # pylint: disable=unused-argument
+        from flask import render_template
+
+        return render_template("dashboard.html")

moto/moto_api/_internal/urls.py

@@ -0,0 +1,12 @@
+from moto.moto_api._internal.responses import MotoAPIResponse
+
+url_bases = ["https?://motoapi.amazonaws.com"]
+
+response_instance = MotoAPIResponse()
+
+url_paths = {
+    "{0}/moto-api/$": response_instance.dashboard,
+    "{0}/moto-api/data.json": response_instance.model_data,
+    "{0}/moto-api/reset": response_instance.reset_response,
+    "{0}/moto-api/reset-auth": response_instance.reset_auth_response,
+}