Opensearch Ingestion Service Implementation (#8160)

docs/docs/services/osis.rst

@@ -0,0 +1,33 @@
+.. _implementedservice_osis:
+
+.. |start-h3| raw:: html
+
+    <h3>
+
+.. |end-h3| raw:: html
+
+    </h3>
+
+====
+osis
+====
+
+.. autoclass:: moto.osis.models.OpenSearchIngestionBackend
+
+|start-h3| Implemented features for this service |end-h3|
+
+- [X] create_pipeline
+- [X] delete_pipeline
+- [X] get_pipeline
+- [ ] get_pipeline_blueprint
+- [ ] get_pipeline_change_progress
+- [ ] list_pipeline_blueprints
+- [X] list_pipelines
+- [X] list_tags_for_resource
+- [X] start_pipeline
+- [X] stop_pipeline
+- [X] tag_resource
+- [X] untag_resource
+- [X] update_pipeline
+- [ ] validate_pipeline
+

moto/backend_index.py

@@ -128,6 +128,7 @@
     ("opensearchserverless", re.compile("https?://aoss\\.(.+)\\.amazonaws\\.com")),
     ("opsworks", re.compile("https?://opsworks\\.us-east-1\\.amazonaws.com")),
     ("organizations", re.compile("https?://organizations\\.(.+)\\.amazonaws\\.com")),
+    ("osis", re.compile("https?://osis\\.(.+)\\.amazonaws\\.com")),
     ("panorama", re.compile("https?://panorama\\.(.+)\\.amazonaws.com")),
     ("personalize", re.compile("https?://personalize\\.(.+)\\.amazonaws\\.com")),
     ("pinpoint", re.compile("https?://pinpoint\\.(.+)\\.amazonaws\\.com")),

moto/backends.py

@@ -101,6 +101,7 @@
     from moto.opensearchserverless.models import OpenSearchServiceServerlessBackend
     from moto.opsworks.models import OpsWorksBackend
     from moto.organizations.models import OrganizationsBackend
+    from moto.osis.models import OpenSearchIngestionBackend
     from moto.personalize.models import PersonalizeBackend
     from moto.pinpoint.models import PinpointBackend
     from moto.polly.models import PollyBackend
@@ -280,6 +281,7 @@ def get_service_from_url(url: str) -> Optional[str]:
     "Literal['opensearchserverless']",
     "Literal['opsworks']",
     "Literal['organizations']",
+    "Literal['osis']",
     "Literal['personalize']",
     "Literal['pinpoint']",
     "Literal['polly']",
@@ -592,6 +594,10 @@ def get_backend(
 @overload
 def get_backend(name: "Literal['opsworks']") -> "BackendDict[OpsWorksBackend]": ...
 @overload
+def get_backend(
+    name: "Literal['osis']",
+) -> "BackendDict[OpenSearchIngestionBackend]": ...
+@overload
 def get_backend(
     name: "Literal['organizations']",
 ) -> "BackendDict[OrganizationsBackend]": ...

moto/ec2/models/vpcs.py

@@ -349,7 +349,7 @@ def __init__(
         client_token: Optional[str] = None,
         security_group_ids: Optional[List[str]] = None,
         tags: Optional[Dict[str, str]] = None,
-        private_dns_enabled: Optional[str] = None,
+        private_dns_enabled: Optional[bool] = None,
         destination_prefix_list_id: Optional[str] = None,
     ):
         self.ec2_backend = ec2_backend
@@ -906,7 +906,7 @@ def create_vpc_endpoint(
         client_token: Optional[str] = None,
         security_group_ids: Optional[List[str]] = None,
         tags: Optional[Dict[str, str]] = None,
-        private_dns_enabled: Optional[str] = None,
+        private_dns_enabled: Optional[bool] = None,
     ) -> VPCEndPoint:
         vpc_endpoint_id = random_vpc_ep_id()
 

moto/moto_api/__init__.py

@@ -24,6 +24,9 @@
 state_manager.register_default_transition(
     model_name="glue::job_run", transition={"progression": "immediate"}
 )
+state_manager.register_default_transition(
+    model_name="osis::pipeline", transition={"progression": "immediate"}
+)
 state_manager.register_default_transition(
     "s3::keyrestore", transition={"progression": "immediate"}
 )

moto/osis/__init__.py

@@ -0,0 +1 @@
+from .models import osis_backends  # noqa: F401

moto/osis/exceptions.py

@@ -0,0 +1,58 @@
+import json
+from typing import List, Optional
+
+from moto.core.exceptions import JsonRESTError
+
+
+class OpensearchIngestionExceptions(JsonRESTError):
+    pass
+
+
+class PipelineAlreadyExistsException(OpensearchIngestionExceptions):
+    def __init__(self, pipeline_name: str):
+        super().__init__(
+            "ResourceAlreadyExistsException",
+            f"Pipeline with given name {pipeline_name} already exists",
+        )
+        self.description = json.dumps({"message": self.message})
+
+
+class PipelineInvalidStateException(OpensearchIngestionExceptions):
+    def __init__(
+        self, action: str, valid_states: List[str], current_state: Optional[str]
+    ):
+        super().__init__(
+            "ConflictException",
+            f"Only pipelines with one of the following statuses are eligible for {action}: {valid_states}. The current status is {current_state}.",
+        )
+        self.description = json.dumps({"message": self.message})
+
+
+class PipelineNotFoundException(OpensearchIngestionExceptions):
+    def __init__(self, pipeline_name: str):
+        super().__init__(
+            "ResourceNotFoundException", f"Pipeline {pipeline_name} could not be found."
+        )
+        self.description = json.dumps({"message": self.message})
+
+
+class PipelineValidationException(OpensearchIngestionExceptions):
+    def __init__(self, message: str):
+        super().__init__("ValidationException", message)
+        self.description = json.dumps({"message": self.message})
+
+
+class InvalidVPCOptionsException(OpensearchIngestionExceptions):
+    def __init__(self, message: str) -> None:
+        super().__init__("ValidationException", f"Invalid VpcOptions: {message}")
+        self.description = json.dumps({"message": self.message})
+
+
+class SubnetNotFoundException(InvalidVPCOptionsException):
+    def __init__(self, subnet_id: str) -> None:
+        super().__init__(f"The subnet ID {subnet_id} does not exist")
+
+
+class SecurityGroupNotFoundException(InvalidVPCOptionsException):
+    def __init__(self, sg_id: str) -> None:
+        super().__init__(f"The security group {sg_id} does not exist")