Repository: getmoto/moto
Pull request #8848 (2025-04-24 21:08:18)

[DMS] Add Endpoint Support and Tagging
Addresses #8788 

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/getmoto/moto/pull/8848?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) Report
Attention: Patch coverage is `95.59748%` with `7 lines` in your changes missing coverage. Please review.
> Project coverage is 92.82%. Comparing base [(`fcb513c`)](https://app.codecov.io/gh/getmoto/moto/commit/fcb513cee3ca44e676f83f9ce2a13f4e466c2a89?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) to head [(`224a61e`)](https://app.codecov.io/gh/getmoto/moto/commit/224a61e6cfcb63ff878a92118cb93a88bb183b5c?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).
> Report is 2 commits behind head on master.

| [Files with missing lines](https://app.codecov.io/gh/getmoto/moto/pull/8848?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | Patch % | Lines |
|---|---|---|
| [moto/dms/models.py](https://app.codecov.io/gh/getmoto/moto/pull/8848?src=pr&el=tree&filepath=moto%2Fdms%2Fmodels.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto#diff-bW90by9kbXMvbW9kZWxzLnB5) | 97.29% | [2 Missing :warning: ](https://app.codecov.io/gh/getmoto/moto/pull/8848?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) |
| [moto/dms/responses.py](https://app.codecov.io/gh/getmoto/moto/pull/8848?src=pr&el=tree&filepath=moto%2Fdms%2Fresponses.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto#diff-bW90by9kbXMvcmVzcG9uc2VzLnB5) | 96.61% | [2 Missing :warning: ](https://app.codecov.io/gh/getmoto/moto/pull/8848?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) |
| [moto/resourcegroupstaggingapi/models.py](https://app.codecov.io/gh/getmoto/moto/pull/8848?src=pr&el=tree&filepath=moto%2Fresourcegroupstaggingapi%2Fmodels.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto#diff-bW90by9yZXNvdXJjZWdyb3Vwc3RhZ2dpbmdhcGkvbW9kZWxzLnB5) | 87.50% | [2 Missing :warning: ](https://app.codecov.io/gh/getmoto/moto/pull/8848?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) |
| [moto/dms/exceptions.py](https://app.codecov.io/gh/getmoto/moto/pull/8848?src=pr&el=tree&filepath=moto%2Fdms%2Fexceptions.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto#diff-bW90by9kbXMvZXhjZXB0aW9ucy5weQ==) | 66.66% | [1 Missing :warning: ](https://app.codecov.io/gh/getmoto/moto/pull/8848?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) |

<details><summary>Additional details and impacted files</summary>


```diff
@@           Coverage Diff            @@
##           master    #8848    +/-   ##
========================================
  Coverage   92.82%   92.82%            
========================================
  Files        1266     1266            
  Lines      110646   110803   +157     
========================================
+ Hits       102704   102854   +150     
- Misses       7942     7949     +7     
```

| [Flag](https://app.codecov.io/gh/getmoto/moto/pull/8848/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | Coverage Δ | |
|---|---|---|
| [servertests](https://app.codecov.io/gh/getmoto/moto/pull/8848/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `28.10% <14.46%> (+0.02%)` | :arrow_up: |
| [unittests](https://app.codecov.io/gh/getmoto/moto/pull/8848/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `92.80% <95.59%> (+<0.01%)` | :arrow_up: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto#carryforward-flags-in-the-pull-request-comment) to find out more.

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/getmoto/moto/pull/8848?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).

<details><summary> :rocket: New features to boost your workflow: </summary>

- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>

Patch:
diff --git a/IMPLEMENTATION_COVERAGE.md b/IMPLEMENTATION_COVERAGE.md
index b255bfd4e..73c2e7de9 100644
--- a/IMPLEMENTATION_COVERAGE.md
+++ b/IMPLEMENTATION_COVERAGE.md
@@ -51,7 +51,7 @@
 
 ## amp
 <details>
-<summary>58% implemented</summary>
+<summary>62% implemented</summary>
 
 - [ ] create_alert_manager_definition
 - [X] create_logging_configuration
@@ -68,7 +68,6 @@
 - [X] describe_rule_groups_namespace
 - [ ] describe_scraper
 - [X] describe_workspace
-- [ ] describe_workspace_configuration
 - [ ] get_default_scraper_configuration
 - [X] list_rule_groups_namespaces
 - [ ] list_scrapers
@@ -81,7 +80,6 @@
 - [X] update_logging_configuration
 - [ ] update_scraper
 - [X] update_workspace_alias
-- [ ] update_workspace_configuration
 </details>
 
 ## apigateway
@@ -1758,7 +1756,7 @@
 
 ## comprehend
 <details>
-<summary>12% implemented</summary>
+<summary>30% implemented</summary>
 
 - [ ] batch_detect_dominant_language
 - [ ] batch_detect_entities
@@ -1769,24 +1767,24 @@
 - [ ] classify_document
 - [ ] contains_pii_entities
 - [ ] create_dataset
-- [ ] create_document_classifier
-- [ ] create_endpoint
+- [X] create_document_classifier
+- [X] create_endpoint
 - [X] create_entity_recognizer
-- [ ] create_flywheel
-- [ ] delete_document_classifier
-- [ ] delete_endpoint
+- [X] create_flywheel
+- [X] delete_document_classifier
+- [X] delete_endpoint
 - [X] delete_entity_recognizer
-- [ ] delete_flywheel
+- [X] delete_flywheel
 - [ ] delete_resource_policy
 - [ ] describe_dataset
 - [ ] describe_document_classification_job
-- [ ] describe_document_classifier
+- [X] describe_document_classifier
 - [ ] describe_dominant_language_detection_job
-- [ ] describe_endpoint
+- [X] describe_endpoint
 - [ ] describe_entities_detection_job
 - [X] describe_entity_recognizer
 - [ ] describe_events_detection_job
-- [ ] describe_flywheel
+- [X] describe_flywheel
 - [ ] describe_flywheel_iteration
 - [ ] describe_key_phrases_detection_job
 - [ ] describe_pii_entities_detection_job
@@ -1806,15 +1804,15 @@
 - [ ] list_datasets
 - [ ] list_document_classification_jobs
 - [ ] list_document_classifier_summaries
-- [ ] list_document_classifiers
+- [X] list_document_classifiers
 - [ ] list_dominant_language_detection_jobs
-- [ ] list_endpoints
+- [X] list_endpoints
 - [ ] list_entities_detection_jobs
 - [ ] list_entity_recognizer_summaries
 - [X] list_entity_recognizers
 - [ ] list_events_detection_jobs
 - [ ] list_flywheel_iteration_history
-- [ ] list_flywheels
+- [X] list_flywheels
 - [ ] list_key_phrases_detection_jobs
 - [ ] list_pii_entities_detection_jobs
 - [ ] list_sentiment_detection_jobs
@@ -1826,7 +1824,7 @@
 - [ ] start_dominant_language_detection_job
 - [ ] start_entities_detection_job
 - [ ] start_events_detection_job
-- [ ] start_flywheel_iteration
+- [X] start_flywheel_iteration
 - [ ] start_key_phrases_detection_job
 - [ ] start_pii_entities_detection_job
 - [ ] start_sentiment_detection_job
@@ -1839,11 +1837,11 @@
 - [ ] stop_pii_entities_detection_job
 - [ ] stop_sentiment_detection_job
 - [ ] stop_targeted_sentiment_detection_job
-- [ ] stop_training_document_classifier
+- [X] stop_training_document_classifier
 - [X] stop_training_entity_recognizer
 - [X] tag_resource
 - [X] untag_resource
-- [ ] update_endpoint
+- [X] update_endpoint
 - [ ] update_flywheel
 </details>
 
@@ -2389,14 +2387,13 @@
 
 ## dsql
 <details>
-<summary>18% implemented</summary>
+<summary>20% implemented</summary>
 
 - [X] create_cluster
 - [ ] create_multi_region_clusters
 - [ ] delete_cluster
 - [ ] delete_multi_region_clusters
 - [X] get_cluster
-- [ ] get_vpc_endpoint_service_name
 - [ ] list_clusters
 - [ ] list_tags_for_resource
 - [ ] tag_resource
@@ -8036,7 +8033,7 @@
 
 ## s3tables
 <details>
-<summary>46% implemented</summary>
+<summary>53% implemented</summary>
 
 - [X] create_namespace
 - [X] create_table
@@ -8044,16 +8041,13 @@
 - [X] delete_namespace
 - [X] delete_table
 - [X] delete_table_bucket
-- [ ] delete_table_bucket_encryption
 - [ ] delete_table_bucket_policy
 - [ ] delete_table_policy
 - [X] get_namespace
 - [X] get_table
 - [X] get_table_bucket
-- [ ] get_table_bucket_encryption
 - [ ] get_table_bucket_maintenance_configuration
 - [ ] get_table_bucket_policy
-- [ ] get_table_encryption
 - [ ] get_table_maintenance_configuration
 - [ ] get_table_maintenance_job_status
 - [ ] get_table_metadata_location
@@ -8061,7 +8055,6 @@
 - [X] list_namespaces
 - [X] list_table_buckets
 - [X] list_tables
-- [ ] put_table_bucket_encryption
 - [ ] put_table_bucket_maintenance_configuration
 - [ ] put_table_bucket_policy
 - [ ] put_table_maintenance_configuration
@@ -9467,20 +9460,20 @@
 
 ## timestream-influxdb
 <details>
-<summary>41% implemented</summary>
+<summary>76% implemented</summary>
 
-- [ ] create_db_cluster
+- [X] create_db_cluster
 - [X] create_db_instance
-- [ ] create_db_parameter_group
+- [X] create_db_parameter_group
 - [ ] delete_db_cluster
 - [X] delete_db_instance
-- [ ] get_db_cluster
+- [X] get_db_cluster
 - [X] get_db_instance
-- [ ] get_db_parameter_group
-- [ ] list_db_clusters
+- [X] get_db_parameter_group
+- [X] list_db_clusters
 - [X] list_db_instances
 - [ ] list_db_instances_for_cluster
-- [ ] list_db_parameter_groups
+- [X] list_db_parameter_groups
 - [X] list_tags_for_resource
 - [X] tag_resource
 - [X] untag_resource
diff --git a/docs/docs/services/amp.rst b/docs/docs/services/amp.rst
index 471e3ec1a..016e4dbfb 100644
--- a/docs/docs/services/amp.rst
+++ b/docs/docs/services/amp.rst
@@ -47,7 +47,6 @@ amp
 - [X] describe_rule_groups_namespace
 - [ ] describe_scraper
 - [X] describe_workspace
-- [ ] describe_workspace_configuration
 - [ ] get_default_scraper_configuration
 - [X] list_rule_groups_namespaces
 - [ ] list_scrapers
@@ -68,5 +67,4 @@ amp
         The ClientToken-parameter is not yet implemented
         
 
-- [ ] update_workspace_configuration
 
diff --git a/docs/docs/services/comprehend.rst b/docs/docs/services/comprehend.rst
index 90f194bb8..cd8e710ef 100644
--- a/docs/docs/services/comprehend.rst
+++ b/docs/docs/services/comprehend.rst
@@ -25,28 +25,28 @@ comprehend
 - [ ] classify_document
 - [ ] contains_pii_entities
 - [ ] create_dataset
-- [ ] create_document_classifier
-- [ ] create_endpoint
+- [X] create_document_classifier
+- [X] create_endpoint
 - [X] create_entity_recognizer
   
-        The ClientRequestToken-parameter is not yet implemented
-        
+The ClientRequestToken-parameter is not yet implemented
 
-- [ ] create_flywheel
-- [ ] delete_document_classifier
-- [ ] delete_endpoint
+
+- [X] create_flywheel
+- [X] delete_document_classifier
+- [X] delete_endpoint
 - [X] delete_entity_recognizer
-- [ ] delete_flywheel
+- [X] delete_flywheel
 - [ ] delete_resource_policy
 - [ ] describe_dataset
 - [ ] describe_document_classification_job
-- [ ] describe_document_classifier
+- [X] describe_document_classifier
 - [ ] describe_dominant_language_detection_job
-- [ ] describe_endpoint
+- [X] describe_endpoint
 - [ ] describe_entities_detection_job
 - [X] describe_entity_recognizer
 - [ ] describe_events_detection_job
-- [ ] describe_flywheel
+- [X] describe_flywheel
 - [ ] describe_flywheel_iteration
 - [ ] describe_key_phrases_detection_job
 - [ ] describe_pii_entities_detection_job
@@ -66,20 +66,35 @@ comprehend
 - [ ] list_datasets
 - [ ] list_document_classification_jobs
 - [ ] list_document_classifier_summaries
-- [ ] list_document_classifiers
+- [X] list_document_classifiers
+  
+List document classifiers with optional filtering.
+Pagination is not yet implemented.
+
+
 - [ ] list_dominant_language_detection_jobs
-- [ ] list_endpoints
+- [X] list_endpoints
+  
+List endpoints with optional filtering.
+Pagination is not yet implemented.
+
+
 - [ ] list_entities_detection_jobs
 - [ ] list_entity_recognizer_summaries
 - [X] list_entity_recognizers
   
-        Pagination is not yet implemented.
-        The following filters are not yet implemented: Status, SubmitTimeBefore, SubmitTimeAfter
-        
+Pagination is not yet implemented.
+The following filters are not yet implemented: Status, SubmitTimeBefore, SubmitTimeAfter
+
 
 - [ ] list_events_detection_jobs
 - [ ] list_flywheel_iteration_history
-- [ ] list_flywheels
+- [X] list_flywheels
+  
+List flywheels with optional filtering.
+Pagination is not yet implemented.
+
+
 - [ ] list_key_phrases_detection_jobs
 - [ ] list_pii_entities_detection_jobs
 - [ ] list_sentiment_detection_jobs
@@ -91,7 +106,7 @@ comprehend
 - [ ] start_dominant_language_detection_job
 - [ ] start_entities_detection_job
 - [ ] start_events_detection_job
-- [ ] start_flywheel_iteration
+- [X] start_flywheel_iteration
 - [ ] start_key_phrases_detection_job
 - [ ] start_pii_entities_detection_job
 - [ ] start_sentiment_detection_job
@@ -104,10 +119,10 @@ comprehend
 - [ ] stop_pii_entities_detection_job
 - [ ] stop_sentiment_detection_job
 - [ ] stop_targeted_sentiment_detection_job
-- [ ] stop_training_document_classifier
+- [X] stop_training_document_classifier
 - [X] stop_training_entity_recognizer
 - [X] tag_resource
 - [X] untag_resource
-- [ ] update_endpoint
+- [X] update_endpoint
 - [ ] update_flywheel
 
diff --git a/docs/docs/services/dsql.rst b/docs/docs/services/dsql.rst
index 06a589c17..02bdadf65 100644
--- a/docs/docs/services/dsql.rst
+++ b/docs/docs/services/dsql.rst
@@ -21,7 +21,6 @@ dsql
 - [ ] delete_cluster
 - [ ] delete_multi_region_clusters
 - [X] get_cluster
-- [ ] get_vpc_endpoint_service_name
 - [ ] list_clusters
 - [ ] list_tags_for_resource
 - [ ] tag_resource
diff --git a/docs/docs/services/s3tables.rst b/docs/docs/services/s3tables.rst
index 03e0a72fd..4ad3d48ca 100644
--- a/docs/docs/services/s3tables.rst
+++ b/docs/docs/services/s3tables.rst
@@ -22,16 +22,13 @@ s3tables
 - [X] delete_namespace
 - [X] delete_table
 - [X] delete_table_bucket
-- [ ] delete_table_bucket_encryption
 - [ ] delete_table_bucket_policy
 - [ ] delete_table_policy
 - [X] get_namespace
 - [X] get_table
 - [X] get_table_bucket
-- [ ] get_table_bucket_encryption
 - [ ] get_table_bucket_maintenance_configuration
 - [ ] get_table_bucket_policy
-- [ ] get_table_encryption
 - [ ] get_table_maintenance_configuration
 - [ ] get_table_maintenance_job_status
 - [ ] get_table_metadata_location
@@ -39,7 +36,6 @@ s3tables
 - [X] list_namespaces
 - [X] list_table_buckets
 - [X] list_tables
-- [ ] put_table_bucket_encryption
 - [ ] put_table_bucket_maintenance_configuration
 - [ ] put_table_bucket_policy
 - [ ] put_table_maintenance_configuration
diff --git a/docs/docs/services/timestream-influxdb.rst b/docs/docs/services/timestream-influxdb.rst
index 4b0cf8300..284585aea 100644
--- a/docs/docs/services/timestream-influxdb.rst
+++ b/docs/docs/services/timestream-influxdb.rst
@@ -16,7 +16,7 @@ timestream-influxdb
 
 |start-h3| Implemented features for this service |end-h3|
 
-- [ ] create_db_cluster
+- [X] create_db_cluster
 - [X] create_db_instance
   
         dbParameterGroupIdentifier argument is not yet handled
@@ -27,20 +27,20 @@ timestream-influxdb
         influxAuthParametersSecretArn is not yet handled
         
 
-- [ ] create_db_parameter_group
+- [X] create_db_parameter_group
 - [ ] delete_db_cluster
 - [X] delete_db_instance
-- [ ] get_db_cluster
+- [X] get_db_cluster
 - [X] get_db_instance
-- [ ] get_db_parameter_group
-- [ ] list_db_clusters
+- [X] get_db_parameter_group
+- [X] list_db_clusters
 - [X] list_db_instances
   
         Pagination is not yet implemented
         
 
 - [ ] list_db_instances_for_cluster
-- [ ] list_db_parameter_groups
+- [X] list_db_parameter_groups
 - [X] list_tags_for_resource
 - [X] tag_resource
 - [X] untag_resource
diff --git a/moto/appsync/models.py b/moto/appsync/models.py
index d5a9e39c6..1e6d56c66 100644
--- a/moto/appsync/models.py
+++ b/moto/appsync/models.py
@@ -194,7 +194,8 @@ class GraphqlAPI(BaseModel):
         open_id_connect_config: str,
         lambda_authorizer_config: str,
         visibility: str,
-    ):
+        backend: "AppSyncBackend",
+    ) -> None:
         self.region = region
         self.name = name
         self.api_id = str(mock_random.uuid4())
@@ -213,6 +214,7 @@ class GraphqlAPI(BaseModel):
         self.api_keys: Dict[str, GraphqlAPIKey] = dict()
 
         self.api_cache: Optional[APICache] = None
+        self.backend = backend
 
     def update(
         self,
@@ -321,13 +323,14 @@ class GraphqlAPI(BaseModel):
             "userPoolConfig": self.user_pool_config,
             "xrayEnabled": self.xray_enabled,
             "visibility": self.visibility,
+            "tags": self.backend.list_tags_for_resource(self.arn),
         }
 
 
 class AppSyncBackend(BaseBackend):
     """Implementation of AppSync APIs."""
 
-    def __init__(self, region_name: str, account_id: str):
+    def __init__(self, region_name: str, account_id: str) -> None:
         super().__init__(region_name, account_id)
         self.graphql_apis: Dict[str, GraphqlAPI] = dict()
         self.tagger = TaggingService()
@@ -357,6 +360,7 @@ class AppSyncBackend(BaseBackend):
             open_id_connect_config=open_id_connect_config,
             lambda_authorizer_config=lambda_authorizer_config,
             visibility=visibility,
+            backend=self,
         )
         self.graphql_apis[graphql_api.api_id] = graphql_api
         self.tagger.tag_resource(
diff --git a/moto/backend_index.py b/moto/backend_index.py
index b65523732..6fa4b41e2 100644
--- a/moto/backend_index.py
+++ b/moto/backend_index.py
@@ -26,6 +26,7 @@ backend_url_patterns = [
     ("bedrockagent", re.compile("https?://bedrock-agent\\.(.+)\\.amazonaws\\.com")),
     ("budgets", re.compile("https?://budgets\\.amazonaws\\.com")),
     ("ce", re.compile("https?://ce\\.(.+)\\.amazonaws\\.com")),
+    ("clouddirectory", re.compile("https?://clouddirectory\\.(.+)\\.amazonaws\\.com")),
     ("cloudformation", re.compile("https?://cloudformation\\.(.+)\\.amazonaws\\.com")),
     ("cloudfront", re.compile("https?://cloudfront\\.amazonaws\\.com")),
     ("cloudfront", re.compile("https?://cloudfront\\.(.+)\\.amazonaws\\.com")),
@@ -139,6 +140,10 @@ backend_url_patterns = [
     ),
     ("moto_api._internal", re.compile("https?://motoapi\\.amazonaws\\.com")),
     ("mq", re.compile("https?://mq\\.(.+)\\.amazonaws\\.com")),
+    (
+        "networkfirewall",
+        re.compile("https?://network-firewall\\.(.+)\\.amazonaws\\.com"),
+    ),
     ("networkmanager", re.compile("https?://networkmanager\\.(.+)\\.amazonaws\\.com")),
     ("opensearchserverless", re.compile("https?://aoss\\.(.+)\\.amazonaws\\.com")),
     ("opsworks", re.compile("https?://opsworks\\.us-east-1\\.amazonaws.com")),
@@ -196,6 +201,10 @@ backend_url_patterns = [
     ("secretsmanager", re.compile("https?://secretsmanager\\.(.+)\\.amazonaws\\.com")),
     ("securityhub", re.compile("https?://securityhub\\.(.+)\\.amazonaws\\.com")),
     ("servicecatalog", re.compile("https?://servicecatalog\\.(.+)\\.amazonaws\\.com")),
+    (
+        "servicecatalogappregistry",
+        re.compile("https?://servicecatalog-appregistry\\.(.+)\\.amazonaws\\.com"),
+    ),
     (
         "servicediscovery",
         re.compile("https?://(data-)?servicediscovery\\.(.+)\\.amazonaws\\.com"),
diff --git a/moto/backends.py b/moto/backends.py
index deac05171..6023d9b0e 100644
--- a/moto/backends.py
+++ b/moto/backends.py
@@ -25,6 +25,7 @@ if TYPE_CHECKING:
     from moto.bedrockagent.models import AgentsforBedrockBackend
     from moto.budgets.models import BudgetsBackend
     from moto.ce.models import CostExplorerBackend
+    from moto.clouddirectory.models import CloudDirectoryBackend
     from moto.cloudformation.models import CloudFormationBackend
     from moto.cloudfront.models import CloudFrontBackend
     from moto.cloudtrail.models import CloudTrailBackend
@@ -101,6 +102,7 @@ if TYPE_CHECKING:
     from moto.meteringmarketplace.models import MeteringMarketplaceBackend
     from moto.moto_api._internal.models import MotoAPIBackend
     from moto.mq.models import MQBackend
+    from moto.networkfirewall.models import NetworkFirewallBackend
     from moto.networkmanager.models import NetworkManagerBackend
     from moto.opensearch.models import OpenSearchServiceBackend
     from moto.opensearchserverless.models import OpenSearchServiceServerlessBackend
@@ -134,6 +136,7 @@ if TYPE_CHECKING:
     from moto.scheduler.models import EventBridgeSchedulerBackend
     from moto.sdb.models import SimpleDBBackend
     from moto.secretsmanager.models import SecretsManagerBackend
+    from moto.servicecatalogappregistry.models import AppRegistryBackend
     from moto.servicediscovery.models import ServiceDiscoveryBackend
     from moto.servicequotas.models import ServiceQuotasBackend
     from moto.ses.models import SESBackend
@@ -213,6 +216,7 @@ SERVICE_NAMES = Union[
     "Literal['bedrock-agent']",
     "Literal['budgets']",
     "Literal['ce']",
+    "Literal['clouddirectory']",
     "Literal['cloudformation']",
     "Literal['cloudfront']",
     "Literal['cloudtrail']",
@@ -288,6 +292,7 @@ SERVICE_NAMES = Union[
     "Literal['moto_api']",
     "Literal['mq']",
     "Literal['neptune']",
+    "Literal['networkfirewall']",
     "Literal['networkmanager']",
     "Literal['opensearch']",
     "Literal['opensearchserverless']",
@@ -322,6 +327,7 @@ SERVICE_NAMES = Union[
     "Literal['scheduler']",
     "Literal['sdb']",
     "Literal['secretsmanager']",
+    "Literal['servicecatalogappregistry']",
     "Literal['servicediscovery']",
     "Literal['service-quotas']",
     "Literal['ses']",
@@ -404,6 +410,10 @@ def get_backend(name: "Literal['budgets']") -> "BackendDict[BudgetsBackend]": ..
 @overload
 def get_backend(name: "Literal['ce']") -> "BackendDict[CostExplorerBackend]": ...
 @overload
+def get_backend(
+    name: "Literal['clouddirectory']",
+) -> "BackendDict[CloudDirectoryBackend]": ...
+@overload
 def get_backend(
     name: "Literal['cloudformation']",
 ) -> "BackendDict[CloudFormationBackend]": ...
@@ -608,6 +618,10 @@ def get_backend(name: "Literal['mq']") -> "BackendDict[MQBackend]": ...
 @overload
 def get_backend(name: "Literal['neptune']") -> "BackendDict[RDSBackend]": ...
 @overload
+def get_backend(
+    name: "Literal['networkfirewall']",
+) -> "BackendDict[NetworkFirewallBackend]": ...
+@overload
 def get_backend(
     name: "Literal['networkmanager']",
 ) -> "BackendDict[NetworkManagerBackend]": ...
@@ -714,6 +728,10 @@ def get_backend(
     name: "Literal['secretsmanager']",
 ) -> "BackendDict[SecretsManagerBackend]": ...
 @overload
+def get_backend(
+    name: "Literal['servicecatalogappregistry']",
+) -> "BackendDict[AppRegistryBackend]": ...
+@overload
 def get_backend(
     name: "Literal['servicediscovery']",
 ) -> "BackendDict[ServiceDiscoveryBackend]": ...
diff --git a/moto/clouddirectory/__init__.py b/moto/clouddirectory/__init__.py
new file mode 100644
index 000000000..a5b3b3251
--- /dev/null
+++ b/moto/clouddirectory/__init__.py
@@ -0,0 +1,1 @@
+from .models import CloudDirectoryBackend, clouddirectory_backends  # noqa: F401
diff --git a/moto/clouddirectory/exceptions.py b/moto/clouddirectory/exceptions.py
new file mode 100644
index 000000000..7b2d582ec
--- /dev/null
+++ b/moto/clouddirectory/exceptions.py
@@ -0,0 +1,20 @@
+"""Exceptions raised by the clouddirectory service."""
+
+import json
+
+from moto.core.exceptions import JsonRESTError
+
+
+class ValidationError(JsonRESTError):
+    def __init__(self, message: str):
+        super().__init__("ValidationException", message)
+
+
+class InvalidArnException(JsonRESTError):
+    def __init__(self, resource_id: str):
+        super().__init__("InvalidArnException", "Invalid Arn")
+        body = {
+            "ResourceId": resource_id,
+            "Message": "Invalid Arn",
+        }
+        self.description = json.dumps(body)
diff --git a/moto/clouddirectory/models.py b/moto/clouddirectory/models.py
new file mode 100644
index 000000000..40142e42b
--- /dev/null
+++ b/moto/clouddirectory/models.py
@@ -0,0 +1,96 @@
+"""CloudDirectoryBackend class with methods for supported APIs."""
+
+import datetime
+from typing import Dict, List
+
+from moto.core.base_backend import BackendDict, BaseBackend
+from moto.core.common_models import BaseModel
+from moto.utilities.paginator import paginate
+from moto.utilities.tagging_service import TaggingService
+
+from .exceptions import InvalidArnException
+
+PAGINATION_MODEL = {
+    "list_directories": {
+        "input_token": "next_token",
+        "limit_key": "max_results",
+        "limit_default": 100,
+        "unique_attribute": "directory_arn",
+    }
+}
+
+
+class Directory(BaseModel):
+    def __init__(
+        self, account_id: str, region: str, name: str, schema_arn: str
+    ) -> None:
+        self.name = name
+        self.schema_arn = schema_arn
+        self.directory_arn = (
+            f"arn:aws:clouddirectory:{region}:{account_id}:directory/{name}"
+        )
+        self.state = "ENABLED"
+        self.creation_date_time = datetime.datetime.now()
+        self.object_identifier = f"directory-{name}"
+
+    def to_dict(self) -> Dict[str, str]:
+        return {
+            "Name": self.name,
+            "SchemaArn": self.schema_arn,
+            "DirectoryArn": self.directory_arn,
+            "State": self.state,
+            "CreationDateTime": str(self.creation_date_time),
+            "ObjectIdentifier": self.object_identifier,
+        }
+
+
+class CloudDirectoryBackend(BaseBackend):
+    """Implementation of CloudDirectory APIs."""
+
+    def __init__(self, region_name: str, account_id: str) -> None:
+        super().__init__(region_name, account_id)
+        self.directories: Dict[str, Directory] = {}
+        self.tagger = TaggingService()
+
+    def create_directory(self, name: str, schema_arn: str) -> Directory:
+        directory = Directory(self.account_id, self.region_name, name, schema_arn)
+        self.directories[directory.directory_arn] = directory
+        return directory
+
+    @paginate(pagination_model=PAGINATION_MODEL)
+    def list_directories(self, state: str) -> List[Directory]:
+        directories = list(self.directories.values())
+        if state:
+            directories = [
+                directory for directory in directories if directory.state == state
+            ]
+        return directories
+
+    def tag_resource(self, resource_arn: str, tags: List[Dict[str, str]]) -> None:
+        self.tagger.tag_resource(resource_arn, tags)
+        return
+
+    def untag_resource(self, resource_arn: str, tag_keys: List[str]) -> None:
+        if not isinstance(tag_keys, list):
+            tag_keys = [tag_keys]
+        self.tagger.untag_resource_using_names(resource_arn, tag_keys)
+        return
+
+    def delete_directory(self, directory_arn: str) -> str:
+        directory = self.directories.pop(directory_arn)
+        return directory.directory_arn
+
+    def get_directory(self, directory_arn: str) -> Directory:
+        directory = self.directories.get(directory_arn)
+        if not directory:
+            raise InvalidArnException(directory_arn)
+        return directory
+
+    def list_tags_for_resource(
+        self, resource_arn: str, next_token: str, max_results: int
+    ) -> List[Dict[str, str]]:
+        tags = self.tagger.list_tags_for_resource(resource_arn)["Tags"]
+        return tags
+
+
+clouddirectory_backends = BackendDict(CloudDirectoryBackend, "clouddirectory")
diff --git a/moto/clouddirectory/responses.py b/moto/clouddirectory/responses.py
new file mode 100644
index 000000000..c967791de
--- /dev/null
+++ b/moto/clouddirectory/responses.py
@@ -0,0 +1,93 @@
+"""Handles incoming clouddirectory requests, invokes methods, returns responses."""
+
+import json
+
+from moto.core.responses import BaseResponse
+
+from .models import CloudDirectoryBackend, clouddirectory_backends
+
+
+class CloudDirectoryResponse(BaseResponse):
+    """Handler for CloudDirectory requests and responses."""
+
+    def __init__(self) -> None:
+        super().__init__(service_name="clouddirectory")
+
+    @property
+    def clouddirectory_backend(self) -> "CloudDirectoryBackend":
+        """Return backend instance specific for this region."""
+        return clouddirectory_backends[self.current_account][self.region]
+
+    def create_directory(self) -> str:
+        name = self._get_param("Name")
+        schema_arn = self._get_param("SchemaArn")
+        directory = self.clouddirectory_backend.create_directory(
+            name=name,
+            schema_arn=schema_arn,
+        )
+
+        return json.dumps(
+            dict(
+                DirectoryArn=directory.directory_arn,
+                Name=name,
+                ObjectIdentifier=directory.object_identifier,
+                AppliedSchemaArn=directory.schema_arn,
+            )
+        )
+
+    def list_directories(self) -> str:
+        next_token = self._get_param("NextToken")
+        max_results = self._get_param("MaxResults")
+        state = self._get_param("state")
+        directories, next_token = self.clouddirectory_backend.list_directories(
+            state=state, next_token=next_token, max_results=max_results
+        )
+        directory_list = [directory.to_dict() for directory in directories]
+        return json.dumps(dict(Directories=directory_list, NextToken=next_token))
+
+    def tag_resource(self) -> str:
+        resource_arn = self._get_param("ResourceArn")
+        tags = self._get_param("Tags")
+        self.clouddirectory_backend.tag_resource(
+            resource_arn=resource_arn,
+            tags=tags,
+        )
+        return json.dumps(dict())
+
+    def untag_resource(self) -> str:
+        resource_arn = self._get_param("ResourceArn")
+        tag_keys = self._get_param("TagKeys")
+        self.clouddirectory_backend.untag_resource(
+            resource_arn=resource_arn,
+            tag_keys=tag_keys,
+        )
+        return json.dumps(dict())
+
+    def delete_directory(self) -> str:
+        # Retrieve arn from headers
+        # https://docs.aws.amazon.com/clouddirectory/latest/APIReference/API_DeleteDirectory.html
+        arn = self.headers.get("x-amz-data-partition")
+        directory_arn = self.clouddirectory_backend.delete_directory(
+            directory_arn=arn,
+        )
+        return json.dumps(dict(DirectoryArn=directory_arn))
+
+    def get_directory(self) -> str:
+        # Retrieve arn from headers
+        # https://docs.aws.amazon.com/clouddirectory/latest/APIReference/API_GetDirectory.html
+        arn = self.headers.get("x-amz-data-partition")
+        directory = self.clouddirectory_backend.get_directory(
+            directory_arn=arn,
+        )
+        return json.dumps(dict(Directory=directory.to_dict()))
+
+    def list_tags_for_resource(self) -> str:
+        resource_arn = self._get_param("ResourceArn")
+        next_token = self._get_param("NextToken")
+        max_results = self._get_param("MaxResults")
+        tags = self.clouddirectory_backend.list_tags_for_resource(
+            resource_arn=resource_arn,
+            next_token=next_token,
+            max_results=max_results,
+        )
+        return json.dumps(dict(Tags=tags, NextToken=next_token))
diff --git a/moto/clouddirectory/urls.py b/moto/clouddirectory/urls.py
new file mode 100644
index 000000000..e82051ba4
--- /dev/null
+++ b/moto/clouddirectory/urls.py
@@ -0,0 +1,17 @@
+"""clouddirectory base URL and path."""
+
+from .responses import CloudDirectoryResponse
+
+url_bases = [
+    r"https?://clouddirectory\.(.+)\.amazonaws\.com",
+]
+
+url_paths = {
+    "{0}/amazonclouddirectory/2017-01-11/directory/create$": CloudDirectoryResponse.dispatch,
+    "{0}/amazonclouddirectory/2017-01-11/directory/list$": CloudDirectoryResponse.dispatch,
+    "{0}/amazonclouddirectory/2017-01-11/tags/add$": CloudDirectoryResponse.dispatch,
+    "{0}/amazonclouddirectory/2017-01-11/tags/remove$": CloudDirectoryResponse.dispatch,
+    "{0}/amazonclouddirectory/2017-01-11/directory$": CloudDirectoryResponse.dispatch,
+    "{0}/amazonclouddirectory/2017-01-11/directory/get$": CloudDirectoryResponse.dispatch,
+    "{0}/amazonclouddirectory/2017-01-11/tags$": CloudDirectoryResponse.dispatch,
+}
diff --git a/moto/comprehend/models.py b/moto/comprehend/models.py
index 945004016..4d9c6774f 100644
--- a/moto/comprehend/models.py
+++ b/moto/comprehend/models.py
@@ -1,6 +1,7 @@
 """ComprehendBackend class with methods for supported APIs."""
 
-from typing import Any, Dict, Iterable, List
+import random
+from typing import Any, Dict, Iterable, List, Optional, Tuple
 
 from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
@@ -107,6 +108,125 @@ class EntityRecognizer(BaseModel):
         }
 
 
+class DocumentClassifier(BaseModel):
+    def __init__(
+        self,
+        region_name: str,
+        account_id: str,
+        language_code: str,
+        version_name: str,
+        input_data_config: Dict[str, Any],
+        output_data_config: Dict[str, Any],
+        data_access_role_arn: str,
+        document_classifier_name: str,
+        volume_kms_key_id: str,
+        client_request_token: str,
+        mode: str,
+        vpc_config: Dict[str, List[str]],
+        model_kms_key_id: str,
+        model_policy: str,
+    ):
+        self.name = document_classifier_name
+        self.arn = f"arn:{get_partition(region_name)}:comprehend:{region_name}:{account_id}:document-classifier/{document_classifier_name}/{version_name}"
+        self.language_code = language_code
+        self.version_name = version_name
+        self.input_data_config = input_data_config
+        self.output_data_config = output_data_config
+        self.data_access_role_arn = data_access_role_arn
+        self.volume_kms_key_id = volume_kms_key_id
+        self.client_request_token = client_request_token
+        self.mode = mode
+        self.vpc_config = vpc_config
+        self.model_kms_key_id = model_kms_key_id
+        self.model_policy = model_policy
+        self.status = "TRAINING"
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "DocumentClassifierArn": self.arn,
+            "LanguageCode": self.language_code,
+            "Status": self.status,
+            "InputDataConfig": self.input_data_config,
+            "DataAccessRoleArn": self.data_access_role_arn,
+            "VolumeKmsKeyId": self.volume_kms_key_id,
+            "Mode": self.mode,
+            "VpcConfig": self.vpc_config,
+            "ModelKmsKeyId": self.model_kms_key_id,
+            "ModelPolicy": self.model_policy,
+        }
+
+
+class Endpoint(BaseModel):
+    def __init__(
+        self,
+        endpoint_name: str,
+        region_name: str,
+        account_id: str,
+        model_arn: str,
+        client_request_token: str,
+        data_access_role_arn: str,
+        flywheel_arn: str,
+        desired_inference_units: int,
+    ):
+        self.name = endpoint_name
+        self.arn = f"arn:{get_partition(region_name)}:comprehend:{region_name}:{account_id}:endpoint/{endpoint_name}/{model_arn}"
+        self.model_arn = model_arn
+        self.client_request_token = client_request_token
+        self.data_access_role_arn = data_access_role_arn
+        self.flywheel_arn = flywheel_arn
+        self.desired_inference_units = desired_inference_units
+        self.status = "IN_SERVICE"
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "EndpointArn": self.arn,
+            "ModelArn": self.model_arn,
+            "ClientRequestToken": self.client_request_token,
+            "DataAccessRoleArn": self.data_access_role_arn,
+            "FlywheelArn": self.flywheel_arn,
+            "DesiredInferenceUnits": self.desired_inference_units,
+            "Status": self.status,
+        }
+
+
+class Flywheel(BaseModel):
+    def __init__(
+        self,
+        region_name: str,
+        account_id: str,
+        flywheel_name: str,
+        active_model_arn: str,
+        data_access_role_arn: str,
+        task_config: Dict[str, Any],
+        model_type: str,
+        data_lake_s3_uri: str,
+        data_security_config: Dict[str, Any],
+        client_request_token: str,
+    ):
+        self.name = flywheel_name
+        self.arn = f"arn:{get_partition(region_name)}:comprehend:{region_name}:{account_id}:flywheel/{flywheel_name}"
+        self.active_model_arn = active_model_arn
+        self.data_access_role_arn = data_access_role_arn
+        self.task_config = task_config
+        self.model_type = model_type
+        self.data_lake_s3_uri = data_lake_s3_uri
+        self.data_security_config = data_security_config
+        self.client_request_token = client_request_token
+        self.status = "ACTIVE"
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "FlywheelArn": self.arn,
+            "ActiveModelArn": self.active_model_arn,
+            "DataAccessRoleArn": self.data_access_role_arn,
+            "TaskConfig": self.task_config,
+            "ModelType": self.model_type,
+            "DataLakeS3Uri": self.data_lake_s3_uri,
+            "DataSecurityConfig": self.data_security_config,
+            "ClientRequestToken": self.client_request_token,
+        }
+
+
 class ComprehendBackend(BaseBackend):
     """Implementation of Comprehend APIs."""
 
@@ -132,6 +252,9 @@ class ComprehendBackend(BaseBackend):
         super().__init__(region_name, account_id)
         self.recognizers: Dict[str, EntityRecognizer] = dict()
         self.tagger = TaggingService()
+        self.endpoints: Dict[str, Endpoint] = dict()
+        self.classifiers: Dict[str, DocumentClassifier] = dict()
+        self.flywheels: Dict[str, Flywheel] = dict()
 
     def list_entity_recognizers(
         self, _filter: Dict[str, Any]
@@ -235,5 +358,230 @@ class ComprehendBackend(BaseBackend):
             raise TextSizeLimitExceededException(text_size)
         return CANNED_SENTIMENT_RESPONSE
 
+    def create_document_classifier(
+        self,
+        document_classifier_name: str,
+        version_name: str,
+        data_access_role_arn: str,
+        tags: List[Dict[str, str]],
+        input_data_config: Dict[str, Any],
+        output_data_config: Dict[str, Any],
+        client_request_token: str,
+        language_code: str,
+        volume_kms_key_id: str,
+        vpc_config: Dict[str, List[str]],
+        mode: str,
+        model_kms_key_id: str,
+        model_policy: str,
+    ) -> str:
+        classifier = DocumentClassifier(
+            region_name=self.region_name,
+            account_id=self.account_id,
+            language_code=language_code,
+            version_name=version_name,
+            input_data_config=input_data_config,
+            output_data_config=output_data_config,
+            client_request_token=client_request_token,
+            data_access_role_arn=data_access_role_arn,
+            document_classifier_name=document_classifier_name,
+            volume_kms_key_id=volume_kms_key_id,
+            mode=mode,
+            vpc_config=vpc_config,
+            model_kms_key_id=model_kms_key_id,
+            model_policy=model_policy,
+        )
+        self.classifiers[classifier.arn] = classifier
+        self.tagger.tag_resource(classifier.arn, tags)
+        return classifier.arn
+
+    def create_endpoint(
+        self,
+        endpoint_name: str,
+        model_arn: str,
+        desired_inference_units: int,
+        client_request_token: str,
+        tags: List[Dict[str, str]],
+        data_access_role_arn: str,
+        flywheel_arn: str,
+    ) -> Tuple[str, str]:
+        endpoint = Endpoint(
+            endpoint_name=endpoint_name,
+            region_name=self.region_name,
+            account_id=self.account_id,
+            model_arn=model_arn,
+            client_request_token=client_request_token,
+            data_access_role_arn=data_access_role_arn,
+            flywheel_arn=flywheel_arn,
+            desired_inference_units=desired_inference_units,
+        )
+        self.endpoints[endpoint.arn] = endpoint
+        self.tagger.tag_resource(endpoint.arn, tags)
+        return endpoint.arn, model_arn
+
+    def create_flywheel(
+        self,
+        flywheel_name: str,
+        active_model_arn: str,
+        data_access_role_arn: str,
+        task_config: Dict[str, Any],
+        model_type: str,
+        data_lake_s3_uri: str,
+        data_security_config: Dict[str, Any],
+        client_request_token: str,
+        tags: List[Dict[str, str]],
+    ) -> Tuple[str, str]:
+        flywheel = Flywheel(
+            region_name=self.region_name,
+            account_id=self.account_id,
+            flywheel_name=flywheel_name,
+            active_model_arn=active_model_arn,
+            data_access_role_arn=data_access_role_arn,
+            task_config=task_config,
+            model_type=model_type,
+            data_lake_s3_uri=data_lake_s3_uri,
+            data_security_config=data_security_config,
+            client_request_token=client_request_token,
+        )
+        self.flywheels[flywheel.arn] = flywheel
+        self.tagger.tag_resource(flywheel.arn, tags)
+        return flywheel.arn, active_model_arn
+
+    def describe_document_classifier(
+        self, document_classifier_arn: str
+    ) -> DocumentClassifier:
+        if document_classifier_arn not in self.classifiers:
+            raise ResourceNotFound
+        return self.classifiers[document_classifier_arn]
+
+    def describe_endpoint(self, endpoint_arn: str) -> Endpoint:
+        if endpoint_arn not in self.endpoints:
+            raise ResourceNotFound
+        return self.endpoints[endpoint_arn]
+
+    def describe_flywheel(self, flywheel_arn: str) -> Flywheel:
+        if flywheel_arn not in self.flywheels:
+            raise ResourceNotFound
+        return self.flywheels[flywheel_arn]
+
+    def delete_document_classifier(self, document_classifier_arn: str) -> None:
+        self.classifiers.pop(document_classifier_arn, None)
+
+    def delete_endpoint(self, endpoint_arn: str) -> None:
+        self.endpoints.pop(endpoint_arn, None)
+
+    def delete_flywheel(self, flywheel_arn: str) -> None:
+        self.flywheels.pop(flywheel_arn, None)
+
+    def list_document_classifiers(
+        self,
+        filter: Optional[Dict[str, Any]] = None,
+        next_token: Optional[str] = None,
+        max_results: Optional[int] = None,
+    ) -> Tuple[List[Dict[str, Any]], None]:
+        """
+        List document classifiers with optional filtering.
+        Pagination is not yet implemented.
+        """
+        filter = filter or {}
+
+        if "DocumentClassifierName" in filter:
+            classifiers = [
+                classifier.to_dict()
+                for classifier in self.classifiers.values()
+                if classifier.name == filter["DocumentClassifierName"]
+            ]
+        elif "Status" in filter:
+            classifiers = [
+                classifier.to_dict()
+                for classifier in self.classifiers.values()
+                if classifier.status == filter["Status"]
+            ]
+        else:
+            classifiers = [
+                classifier.to_dict() for classifier in self.classifiers.values()
+            ]
+
+        return classifiers, None
+
+    def list_endpoints(
+        self,
+        filter: Optional[Dict[str, Any]] = None,
+        next_token: Optional[str] = None,
+        max_results: Optional[int] = None,
+    ) -> Tuple[List[Dict[str, Any]], None]:
+        """
+        List endpoints with optional filtering.
+        Pagination is not yet implemented.
+        """
+        filter = filter or {}
+
+        if "ModelArn" in filter:
+            endpoints = [
+                endpoint.to_dict()
+                for endpoint in self.endpoints.values()
+                if endpoint.model_arn == filter["ModelArn"]
+            ]
+        elif "Status" in filter:
+            endpoints = [
+                endpoint.to_dict()
+                for endpoint in self.endpoints.values()
+                if endpoint.status == filter["Status"]
+            ]
+        else:
+            endpoints = [endpoint.to_dict() for endpoint in self.endpoints.values()]
+
+        return endpoints, None
+
+    def list_flywheels(
+        self,
+        filter: Optional[Dict[str, Any]] = None,
+        next_token: Optional[str] = None,
+        max_results: Optional[int] = None,
+    ) -> Tuple[List[Dict[str, Any]], None]:
+        """
+        List flywheels with optional filtering.
+        Pagination is not yet implemented.
+        """
+        # Ensure filter is not None
+        filter = filter or {}
+
+        # Apply filtering based on Status
+        if "Status" in filter:
+            flywheels = [
+                flywheel.to_dict()
+                for flywheel in self.flywheels.values()
+                if flywheel.status == filter["Status"]
+            ]
+        else:
+            flywheels = [flywheel.to_dict() for flywheel in self.flywheels.values()]
+
+        # Return the list of flywheels and a placeholder for next_token
+        return flywheels, None
+
+    def stop_training_document_classifier(self, document_classifier_arn: str) -> None:
+        if document_classifier_arn not in self.classifiers:
+            raise ResourceNotFound
+        classifier = self.describe_document_classifier(document_classifier_arn)
+        if classifier.status == "TRAINING":
+            classifier.status = "STOP_REQUESTED"
+
+    def start_flywheel_iteration(
+        self, flywheel_arn: str, client_request_token: str
+    ) -> Tuple[str, int]:
+        if flywheel_arn not in self.flywheels:
+            raise ResourceNotFound
+        flywheel_iteration_id = int(random.randint(0, 1000000))
+        return flywheel_arn, flywheel_iteration_id
+
+    def update_endpoint(
+        self,
+        endpoint_arn: str,
+        desired_model_arn: str,
+        desired_inference_units: str,
+        desired_data_access_role_arn: str,
+        flywheel_arn: str,
+    ) -> str:
+        return desired_model_arn
+
 
 comprehend_backends = BackendDict(ComprehendBackend, "comprehend")
diff --git a/moto/comprehend/responses.py b/moto/comprehend/responses.py
index b323c1088..bc4b804ad 100644
--- a/moto/comprehend/responses.py
+++ b/moto/comprehend/responses.py
@@ -118,3 +118,232 @@ class ComprehendResponse(BaseResponse):
         language = params.get("LanguageCode")
         resp = self.comprehend_backend.detect_sentiment(text, language)
         return json.dumps(resp)
+
+    def create_document_classifier(self) -> str:
+        params = json.loads(self.body)
+        document_classifier_name = params.get("DocumentClassifierName")
+        version_name = params.get("VersionName")
+        data_access_role_arn = params.get("DataAccessRoleArn")
+        tags = params.get("Tags")
+        input_data_config = params.get("InputDataConfig")
+        output_data_config = params.get("OutputDataConfig")
+        client_request_token = params.get("ClientRequestToken")
+        language_code = params.get("LanguageCode")
+        volume_kms_key_id = params.get("VolumeKmsKeyId")
+        vpc_config = params.get("VpcConfig")
+        mode = params.get("Mode")
+        model_kms_key_id = params.get("ModelKmsKeyId")
+        model_policy = params.get("ModelPolicy")
+
+        document_classifier_arn = self.comprehend_backend.create_document_classifier(
+            document_classifier_name=document_classifier_name,
+            version_name=version_name,
+            data_access_role_arn=data_access_role_arn,
+            tags=tags,
+            input_data_config=input_data_config,
+            output_data_config=output_data_config,
+            client_request_token=client_request_token,
+            language_code=language_code,
+            volume_kms_key_id=volume_kms_key_id,
+            vpc_config=vpc_config,
+            mode=mode,
+            model_kms_key_id=model_kms_key_id,
+            model_policy=model_policy,
+        )
+
+        return json.dumps(dict(DocumentClassifierArn=document_classifier_arn))
+
+    def create_endpoint(self) -> str:
+        params = json.loads(self.body)
+        endpoint_name = params.get("EndpointName")
+        model_arn = params.get("ModelArn")
+        desired_inference_units = params.get("DesiredInferenceUnits")
+        client_request_token = params.get("ClientRequestToken")
+        tags = params.get("Tags")
+        data_access_role_arn = params.get("DataAccessRoleArn")
+        flywheel_arn = params.get("FlywheelArn")
+        endpoint_arn, model_arn = self.comprehend_backend.create_endpoint(
+            endpoint_name=endpoint_name,
+            model_arn=model_arn,
+            desired_inference_units=desired_inference_units,
+            client_request_token=client_request_token,
+            tags=tags,
+            data_access_role_arn=data_access_role_arn,
+            flywheel_arn=flywheel_arn,
+        )
+
+        return json.dumps(dict(EndpointArn=endpoint_arn, ModelArn=model_arn))
+
+    def create_flywheel(self) -> str:
+        params = json.loads(self.body)
+        flywheel_name = params.get("FlywheelName")
+        active_model_arn = params.get("ActiveModelArn")
+        data_access_role_arn = params.get("DataAccessRoleArn")
+        task_config = params.get("TaskConfig")
+        model_type = params.get("ModelType")
+        data_lake_s3_uri = params.get("DataLakeS3Uri")
+        data_security_config = params.get("DataSecurityConfig")
+        client_request_token = params.get("ClientRequestToken")
+        tags = params.get("Tags")
+        flywheel_arn, active_model_arn = self.comprehend_backend.create_flywheel(
+            flywheel_name=flywheel_name,
+            active_model_arn=active_model_arn,
+            data_access_role_arn=data_access_role_arn,
+            task_config=task_config,
+            model_type=model_type,
+            data_lake_s3_uri=data_lake_s3_uri,
+            data_security_config=data_security_config,
+            client_request_token=client_request_token,
+            tags=tags,
+        )
+
+        return json.dumps(
+            dict(FlywheelArn=flywheel_arn, activeModelArn=active_model_arn)
+        )
+
+    def describe_document_classifier(self) -> str:
+        params = json.loads(self.body)
+        document_classifier_arn = params.get("DocumentClassifierArn")
+        document_classifier = self.comprehend_backend.describe_document_classifier(
+            document_classifier_arn=document_classifier_arn,
+        )
+
+        return json.dumps(
+            dict(DocumentClassifierProperties=document_classifier.to_dict())
+        )
+
+    def describe_endpoint(self) -> str:
+        params = json.loads(self.body)
+        endpoint_arn = params.get("EndpointArn")
+        endpoint_properties = self.comprehend_backend.describe_endpoint(
+            endpoint_arn=endpoint_arn,
+        )
+
+        return json.dumps(dict(EndpointProperties=endpoint_properties.to_dict()))
+
+    def describe_flywheel(self) -> str:
+        params = json.loads(self.body)
+        flywheel_arn = params.get("FlywheelArn")
+        flywheel_properties = self.comprehend_backend.describe_flywheel(
+            flywheel_arn=flywheel_arn,
+        )
+
+        return json.dumps(dict(FlywheelProperties=flywheel_properties.to_dict()))
+
+    def delete_document_classifier(self) -> str:
+        params = json.loads(self.body)
+        document_classifier_arn = params.get("DocumentClassifierArn")
+        self.comprehend_backend.delete_document_classifier(
+            document_classifier_arn=document_classifier_arn,
+        )
+
+        return json.dumps(dict())
+
+    def delete_endpoint(self) -> str:
+        params = json.loads(self.body)
+        endpoint_arn = params.get("EndpointArn")
+        self.comprehend_backend.delete_endpoint(
+            endpoint_arn=endpoint_arn,
+        )
+
+        return json.dumps(dict())
+
+    def delete_flywheel(self) -> str:
+        params = json.loads(self.body)
+        flywheel_arn = params.get("FlywheelArn")
+        self.comprehend_backend.delete_flywheel(
+            flywheel_arn=flywheel_arn,
+        )
+
+        return json.dumps(dict())
+
+    def list_document_classifiers(self) -> str:
+        params = json.loads(self.body)
+        filter = params.get("Filter")
+        next_token = params.get("NextToken")
+        max_results = params.get("MaxResults")
+        document_classifier_properties_list, next_token = (
+            self.comprehend_backend.list_document_classifiers(
+                filter=filter,
+                next_token=next_token,
+                max_results=max_results,
+            )
+        )
+
+        return json.dumps(
+            dict(
+                DocumentClassifierPropertiesList=document_classifier_properties_list,
+                NextToken=next_token,
+            )
+        )
+
+    def list_endpoints(self) -> str:
+        params = json.loads(self.body)
+        filter = params.get("Filter")
+        next_token = params.get("NextToken")
+        max_results = params.get("MaxResults")
+        endpoint_properties_list, next_token = self.comprehend_backend.list_endpoints(
+            filter=filter,
+            next_token=next_token,
+            max_results=max_results,
+        )
+
+        return json.dumps(
+            dict(EndpointPropertiesList=endpoint_properties_list, NextToken=next_token)
+        )
+
+    def list_flywheels(self) -> str:
+        params = json.loads(self.body)
+        filter = params.get("Filter")
+        next_token = params.get("NextToken")
+        max_results = params.get("MaxResults")
+        flywheel_summary_list, next_token = self.comprehend_backend.list_flywheels(
+            filter=filter,
+            next_token=next_token,
+            max_results=max_results,
+        )
+
+        return json.dumps(
+            dict(FlywheelSummaryList=flywheel_summary_list, nextToken=next_token)
+        )
+
+    def stop_training_document_classifier(self) -> str:
+        params = json.loads(self.body)
+        document_classifier_arn = params.get("DocumentClassifierArn")
+        self.comprehend_backend.stop_training_document_classifier(
+            document_classifier_arn=document_classifier_arn,
+        )
+
+        return json.dumps(dict())
+
+    def start_flywheel_iteration(self) -> str:
+        params = json.loads(self.body)
+        flywheel_arn = params.get("FlywheelArn")
+        client_request_token = params.get("ClientRequestToken")
+        flywheel_arn, flywheel_iteration_id = (
+            self.comprehend_backend.start_flywheel_iteration(
+                flywheel_arn=flywheel_arn,
+                client_request_token=client_request_token,
+            )
+        )
+
+        return json.dumps(
+            dict(FlywheelArn=flywheel_arn, FlywheelIterationId=flywheel_iteration_id)
+        )
+
+    def update_endpoint(self) -> str:
+        params = json.loads(self.body)
+        endpoint_arn = params.get("EndpointArn")
+        desired_model_arn = params.get("DesiredModelArn")
+        desired_inference_units = params.get("DesiredInferenceUnits")
+        desired_data_access_role_arn = params.get("DesiredDataAccessRoleArn")
+        flywheel_arn = params.get("FlywheelArn")
+        desired_model_arn = self.comprehend_backend.update_endpoint(
+            endpoint_arn=endpoint_arn,
+            desired_model_arn=desired_model_arn,
+            desired_inference_units=desired_inference_units,
+            desired_data_access_role_arn=desired_data_access_role_arn,
+            flywheel_arn=flywheel_arn,
+        )
+
+        return json.dumps(dict(DesiredModelArn=desired_model_arn))
diff --git a/moto/dms/exceptions.py b/moto/dms/exceptions.py
index 6ade6bae2..6ebb4b684 100644
--- a/moto/dms/exceptions.py
+++ b/moto/dms/exceptions.py
@@ -18,3 +18,8 @@ class InvalidResourceStateFault(DmsClientError):
 class ResourceAlreadyExistsFault(DmsClientError):
     def __init__(self, message: str):
         super().__init__("ResourceAlreadyExistsFault", message)
+
+
+class ValidationError(DmsClientError):
+    def __init__(self, message: str):
+        super().__init__("ValidationError", message)
diff --git a/moto/dms/models.py b/moto/dms/models.py
index 342c28748..2f7dedfc0 100644
--- a/moto/dms/models.py
+++ b/moto/dms/models.py
@@ -4,14 +4,16 @@ from typing import Any, Dict, Iterable, List, Optional
 from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
 from moto.core.utils import utcnow
+from moto.utilities.tagging_service import TaggingService
 from moto.utilities.utils import get_partition
 
 from .exceptions import (
     InvalidResourceStateFault,
     ResourceAlreadyExistsFault,
     ResourceNotFoundFault,
+    ValidationError,
 )
-from .utils import filter_tasks
+from .utils import filter_tasks, random_id
 
 
 class DatabaseMigrationServiceBackend(BaseBackend):
@@ -19,6 +21,8 @@ class DatabaseMigrationServiceBackend(BaseBackend):
         super().__init__(region_name, account_id)
         self.replication_tasks: Dict[str, "FakeReplicationTask"] = {}
         self.replication_instances: Dict[str, "FakeReplicationInstance"] = {}
+        self.endpoints: Dict[str, "Endpoint"] = {}
+        self.tagger = TaggingService()
 
     def create_replication_task(
         self,
@@ -146,6 +150,8 @@ class DatabaseMigrationServiceBackend(BaseBackend):
             )
 
         self.replication_instances[replication_instance.arn] = replication_instance
+        if tags:
+            self.tagger.tag_resource(replication_instance.arn, tags)
 
         return replication_instance
 
@@ -192,6 +198,263 @@ class DatabaseMigrationServiceBackend(BaseBackend):
 
         return replication_instances
 
+    def create_endpoint(
+        self,
+        endpoint_identifier: str,
+        endpoint_type: str,
+        engine_name: str,
+        username: str,
+        password: str,
+        server_name: str,
+        port: int,
+        database_name: str,
+        extra_connection_attributes: str,
+        kms_key_id: str,
+        tags: Optional[List[Dict[str, str]]],
+        certificate_arn: str,
+        ssl_mode: str,
+        service_access_role_arn: str,
+        external_table_definition: str,
+        dynamo_db_settings: Optional[Dict[str, Any]],
+        s3_settings: Optional[Dict[str, Any]],
+        dms_transfer_settings: Optional[Dict[str, Any]],
+        mongo_db_settings: Optional[Dict[str, Any]],
+        kinesis_settings: Optional[Dict[str, Any]],
+        kafka_settings: Optional[Dict[str, Any]],
+        elasticsearch_settings: Optional[Dict[str, Any]],
+        neptune_settings: Optional[Dict[str, Any]],
+        redshift_settings: Optional[Dict[str, Any]],
+        postgre_sql_settings: Optional[Dict[str, Any]],
+        my_sql_settings: Optional[Dict[str, Any]],
+        oracle_settings: Optional[Dict[str, Any]],
+        sybase_settings: Optional[Dict[str, Any]],
+        microsoft_sql_server_settings: Optional[Dict[str, Any]],
+        ibm_db2_settings: Optional[Dict[str, Any]],
+        resource_identifier: Optional[str],
+        doc_db_settings: Optional[Dict[str, Any]],
+        redis_settings: Optional[Dict[str, Any]],
+        gcp_my_sql_settings: Optional[Dict[str, Any]],
+        timestream_settings: Optional[Dict[str, Any]],
+    ) -> "Endpoint":
+        if endpoint_type not in ["source", "target"]:
+            raise ValidationError("Invalid endpoint type")
+
+        endpoint = Endpoint(
+            endpoint_identifier=endpoint_identifier,
+            endpoint_type=endpoint_type,
+            engine_name=engine_name,
+            engine_display_name=engine_name,  # Assuming engine_display_name is same as engine_name
+            username=username,
+            password=password,
+            server_name=server_name,
+            port=port,
+            database_name=database_name,
+            extra_connection_attributes=extra_connection_attributes,
+            status="creating",  # Default status
+            kms_key_id=kms_key_id,
+            certificate_arn=certificate_arn,
+            ssl_mode=ssl_mode,
+            service_access_role_arn=service_access_role_arn,
+            external_table_definition=external_table_definition,
+            external_id=None,  # Default value
+            dynamo_db_settings=dynamo_db_settings,
+            s3_settings=s3_settings,
+            dms_transfer_settings=dms_transfer_settings,
+            mongo_db_settings=mongo_db_settings,
+            kinesis_settings=kinesis_settings,
+            kafka_settings=kafka_settings,
+            elasticsearch_settings=elasticsearch_settings,
+            neptune_settings=neptune_settings,
+            redshift_settings=redshift_settings,
+            postgre_sql_settings=postgre_sql_settings,
+            my_sql_settings=my_sql_settings,
+            oracle_settings=oracle_settings,
+            sybase_settings=sybase_settings,
+            microsoft_sql_server_settings=microsoft_sql_server_settings,
+            ibm_db2_settings=ibm_db2_settings,
+            resource_identifier=resource_identifier,
+            doc_db_settings=doc_db_settings,
+            redis_settings=redis_settings,
+            gcp_my_sql_settings=gcp_my_sql_settings,
+            timestream_settings=timestream_settings,
+            account_id=self.account_id,
+            region_name=self.region_name,
+        )
+
+        if tags:
+            self.tagger.tag_resource(endpoint.endpoint_arn, tags)
+
+        self.endpoints[endpoint.endpoint_identifier] = endpoint
+        return endpoint
+
+    # TODO implement pagination
+    def describe_endpoints(
+        self,
+        filters: List[Dict[str, Any]],
+        max_records: Optional[int],
+        marker: Optional[str],
+    ) -> List["Endpoint"]:
+        endpoints = list(self.endpoints.values())
+        filter_map = {
+            "endpoint-arn": "endpoint_arn",
+            "endpoint-type": "endpoint_type",
+            "endpoint-id": "endpoint_identifier",
+            "engine-name": "engine_name",
+        }
+
+        for filter in filters:
+            filter_name = filter.get("Name")
+            filter_values = filter.get("Values", [])
+            if filter_name not in filter_map:
+                raise ValueError(f"Invalid filter name: {filter_name}")
+
+            attribute = filter_map[filter_name]
+            endpoints = [
+                endpoint
+                for endpoint in endpoints
+                if getattr(endpoint, attribute) in filter_values
+            ]
+
+        return endpoints
+
+    def list_tags_for_resource(
+        self, resource_arn_list: List[str]
+    ) -> List[Dict[str, Any]]:
+        result = []
+        for resource_arn in resource_arn_list:
+            tags = self.tagger.get_tag_dict_for_resource(resource_arn)
+            for key, value in tags.items():
+                result.append({"Key": key, "ResourceArn": resource_arn, "Value": value})
+        return result
+
+
+class Endpoint(BaseModel):
+    def __init__(
+        self,
+        endpoint_identifier: str,
+        endpoint_type: str,
+        engine_name: str,
+        engine_display_name: str,
+        username: str,
+        password: str,
+        server_name: str,
+        port: int,
+        database_name: str,
+        extra_connection_attributes: str,
+        status: str,
+        kms_key_id: str,
+        certificate_arn: Optional[str],
+        ssl_mode: Optional[str],
+        service_access_role_arn: Optional[str],
+        external_table_definition: Optional[str],
+        external_id: Optional[str],
+        dynamo_db_settings: Optional[Dict[str, Any]],
+        s3_settings: Optional[Dict[str, Any]],
+        dms_transfer_settings: Optional[Dict[str, Any]],
+        mongo_db_settings: Optional[Dict[str, Any]],
+        kinesis_settings: Optional[Dict[str, Any]],
+        kafka_settings: Optional[Dict[str, Any]],
+        elasticsearch_settings: Optional[Dict[str, Any]],
+        neptune_settings: Optional[Dict[str, Any]],
+        redshift_settings: Optional[Dict[str, Any]],
+        postgre_sql_settings: Optional[Dict[str, Any]],
+        my_sql_settings: Optional[Dict[str, Any]],
+        oracle_settings: Optional[Dict[str, Any]],
+        sybase_settings: Optional[Dict[str, Any]],
+        microsoft_sql_server_settings: Optional[Dict[str, Any]],
+        ibm_db2_settings: Optional[Dict[str, Any]],
+        resource_identifier: Optional[str],
+        doc_db_settings: Optional[Dict[str, Any]],
+        redis_settings: Optional[Dict[str, Any]],
+        gcp_my_sql_settings: Optional[Dict[str, Any]],
+        timestream_settings: Optional[Dict[str, Any]],
+        account_id: str,
+        region_name: str,
+    ):
+        self.endpoint_identifier = endpoint_identifier
+        self.endpoint_type = endpoint_type
+        self.engine_name = engine_name
+        self.engine_display_name = engine_display_name
+        self.username = username
+        self.password = password
+        self.server_name = server_name
+        self.port = port
+        self.database_name = database_name
+        self.extra_connection_attributes = extra_connection_attributes
+        self.status = status
+
+        # In real AWS, KMS key is used to encrypt connection parameters
+        # but just storing it here for now
+        self.kms_key_id = kms_key_id
+
+        self.id = resource_identifier if resource_identifier else random_id(True, 27)
+        self.endpoint_arn = f"arn:{get_partition(region_name)}:dms:{region_name}:{account_id}:endpoint:{self.id}"
+
+        self.certificate_arn = certificate_arn
+        self.ssl_mode = ssl_mode
+        self.service_access_role_arn = service_access_role_arn
+        self.external_table_definition = external_table_definition
+        self.external_id = external_id
+        self.dynamo_db_settings = dynamo_db_settings
+        self.s3_settings = s3_settings
+        self.dms_transfer_settings = dms_transfer_settings
+        self.mongo_db_settings = mongo_db_settings
+        self.kinesis_settings = kinesis_settings
+        self.kafka_settings = kafka_settings
+        self.elasticsearch_settings = elasticsearch_settings
+        self.neptune_settings = neptune_settings
+        self.redshift_settings = redshift_settings
+        self.postgre_sql_settings = postgre_sql_settings
+        self.my_sql_settings = my_sql_settings
+        self.oracle_settings = oracle_settings
+        self.sybase_settings = sybase_settings
+        self.microsoft_sql_server_settings = microsoft_sql_server_settings
+        self.ibm_db2_settings = ibm_db2_settings
+        self.doc_db_settings = doc_db_settings
+        self.redis_settings = redis_settings
+        self.gcp_my_sql_settings = gcp_my_sql_settings
+        self.timestream_settings = timestream_settings
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "EndpointIdentifier": self.endpoint_identifier,
+            "EndpointType": self.endpoint_type,
+            "EngineName": self.engine_name,
+            "EngineDisplayName": self.engine_display_name,
+            "Username": self.username,
+            "ServerName": self.server_name,
+            "Port": self.port,
+            "DatabaseName": self.database_name,
+            "ExtraConnectionAttributes": self.extra_connection_attributes,
+            "Status": self.status,
+            "KmsKeyId": self.kms_key_id,
+            "EndpointArn": self.endpoint_arn,
+            "CertificateArn": self.certificate_arn,
+            "SslMode": self.ssl_mode,
+            "ServiceAccessRoleArn": self.service_access_role_arn,
+            "ExternalTableDefinition": self.external_table_definition,
+            "ExternalId": self.external_id,
+            "DynamoDbSettings": self.dynamo_db_settings,
+            "S3Settings": self.s3_settings,
+            "DmsTransferSettings": self.dms_transfer_settings,
+            "MongoDbSettings": self.mongo_db_settings,
+            "KinesisSettings": self.kinesis_settings,
+            "KafkaSettings": self.kafka_settings,
+            "ElasticsearchSettings": self.elasticsearch_settings,
+            "NeptuneSettings": self.neptune_settings,
+            "RedshiftSettings": self.redshift_settings,
+            "PostgreSQLSettings": self.postgre_sql_settings,
+            "MySQLSettings": self.my_sql_settings,
+            "OracleSettings": self.oracle_settings,
+            "SybaseSettings": self.sybase_settings,
+            "MicrosoftSQLServerSettings": self.microsoft_sql_server_settings,
+            "IBMDb2Settings": self.ibm_db2_settings,
+            "DocDbSettings": self.doc_db_settings,
+            "RedisSettings": self.redis_settings,
+            "GcpMySQLSettings": self.gcp_my_sql_settings,
+            "TimestreamSettings": self.timestream_settings,
+        }
+
 
 class FakeReplicationTask(BaseModel):
     def __init__(
diff --git a/moto/dms/responses.py b/moto/dms/responses.py
index a3cebd586..078f603b5 100644
--- a/moto/dms/responses.py
+++ b/moto/dms/responses.py
@@ -103,7 +103,7 @@ class DatabaseMigrationServiceResponse(BaseResponse):
         multi_az = params.get("MultiAZ")
         engine_version = params.get("EngineVersion")
         auto_minor_version_upgrade = params.get("AutoMinorVersionUpgrade")
-        tags = self._get_list_prefix("Tags.member")
+        tags = params.get("Tags")
         kms_key_id = params.get("KmsKeyId")
         publicly_accessible = params.get("PubliclyAccessible")
         dns_name_servers = params.get("DnsNameServers")
@@ -147,3 +147,122 @@ class DatabaseMigrationServiceResponse(BaseResponse):
 
         # TODO: Add Marker (optional) to the response
         return json.dumps({"ReplicationInstances": instances_dict})
+
+    def create_endpoint(self) -> str:
+        params = json.loads(self.body)
+        endpoint_identifier = params.get("EndpointIdentifier")
+        endpoint_type = params.get("EndpointType")
+        engine_name = params.get("EngineName")
+        username = params.get("Username")
+        password = params.get("Password")
+        server_name = params.get("ServerName")
+        port = params.get("Port")
+        database_name = params.get("DatabaseName")
+        extra_connection_attributes = params.get("ExtraConnectionAttributes")
+        kms_key_id = params.get("KmsKeyId")
+        tags = params.get("Tags")
+        certificate_arn = params.get("CertificateArn")
+        ssl_mode = params.get("SslMode")
+        service_access_role_arn = params.get("ServiceAccessRoleArn")
+        external_table_definition = params.get("ExternalTableDefinition")
+        dynamo_db_settings = params.get("DynamoDbSettings")
+        s3_settings = params.get("S3Settings")
+        dms_transfer_settings = params.get("DmsTransferSettings")
+        mongo_db_settings = params.get("MongoDbSettings")
+        kinesis_settings = params.get("KinesisSettings")
+        kafka_settings = params.get("KafkaSettings")
+        elasticsearch_settings = params.get("ElasticsearchSettings")
+        neptune_settings = params.get("NeptuneSettings")
+        redshift_settings = params.get("RedshiftSettings")
+        postgre_sql_settings = params.get("PostgreSQLSettings")
+        my_sql_settings = params.get("MySQLSettings")
+        oracle_settings = params.get("OracleSettings")
+        sybase_settings = params.get("SybaseSettings")
+        microsoft_sql_server_settings = params.get("MicrosoftSQLServerSettings")
+        ibm_db2_settings = params.get("IBMDb2Settings")
+        resource_identifier = params.get("ResourceIdentifier")
+        doc_db_settings = params.get("DocDbSettings")
+        redis_settings = params.get("RedisSettings")
+        gcp_my_sql_settings = params.get("GcpMySQLSettings")
+        timestream_settings = params.get("TimestreamSettings")
+        endpoint = self.dms_backend.create_endpoint(
+            endpoint_identifier=endpoint_identifier,
+            endpoint_type=endpoint_type,
+            engine_name=engine_name,
+            username=username,
+            password=password,
+            server_name=server_name,
+            port=port,
+            database_name=database_name,
+            extra_connection_attributes=extra_connection_attributes,
+            kms_key_id=kms_key_id,
+            tags=tags,
+            certificate_arn=certificate_arn,
+            ssl_mode=ssl_mode,
+            service_access_role_arn=service_access_role_arn,
+            external_table_definition=external_table_definition,
+            dynamo_db_settings=dynamo_db_settings,
+            s3_settings=s3_settings,
+            dms_transfer_settings=dms_transfer_settings,
+            mongo_db_settings=mongo_db_settings,
+            kinesis_settings=kinesis_settings,
+            kafka_settings=kafka_settings,
+            elasticsearch_settings=elasticsearch_settings,
+            neptune_settings=neptune_settings,
+            redshift_settings=redshift_settings,
+            postgre_sql_settings=postgre_sql_settings,
+            my_sql_settings=my_sql_settings,
+            oracle_settings=oracle_settings,
+            sybase_settings=sybase_settings,
+            microsoft_sql_server_settings=microsoft_sql_server_settings,
+            ibm_db2_settings=ibm_db2_settings,
+            resource_identifier=resource_identifier,
+            doc_db_settings=doc_db_settings,
+            redis_settings=redis_settings,
+            gcp_my_sql_settings=gcp_my_sql_settings,
+            timestream_settings=timestream_settings,
+        )
+
+        return json.dumps(
+            dict(
+                Endpoint={k: v for k, v in endpoint.to_dict().items() if v is not None}
+            )
+        )
+
+    def describe_endpoints(self) -> str:
+        params = json.loads(self.body)
+        filters = params.get("Filters", [])
+        max_records = params.get("MaxRecords")
+        marker = params.get("Marker")
+        endpoints = self.dms_backend.describe_endpoints(
+            filters=filters, max_records=max_records, marker=marker
+        )
+        return json.dumps(
+            dict(
+                Endpoints=[
+                    {k: v for k, v in endpoint.to_dict().items() if v is not None}
+                    for endpoint in endpoints
+                ]
+            )
+        )
+
+    def list_tags_for_resource(self) -> str:
+        params = json.loads(self.body)
+        resource_arn = params.get("ResourceArn")
+        resource_arn_list = params.get("ResourceArnList")
+
+        if resource_arn and resource_arn_list:
+            raise ValueError(
+                "Both ResourceArn and ResourceArnList cannot be specified."
+            )
+        if not resource_arn and not resource_arn_list:
+            raise ValueError(
+                "Either ResourceArn or ResourceArnList should be specified."
+            )
+
+        if resource_arn:
+            tag_list = self.dms_backend.list_tags_for_resource([resource_arn])
+        else:
+            tag_list = self.dms_backend.list_tags_for_resource(resource_arn_list)
+
+        return json.dumps(dict(TagList=tag_list))
diff --git a/moto/dms/utils.py b/moto/dms/utils.py
index 7e69b1ff3..25101b358 100644
--- a/moto/dms/utils.py
+++ b/moto/dms/utils.py
@@ -1,6 +1,17 @@
+import random
+import string
 from typing import Any, Dict, Iterable, List
 
 
+def random_id(uppercase: bool = True, length: int = 13) -> str:
+    ascii_set = string.ascii_uppercase if uppercase else string.ascii_lowercase
+    chars = list(range(10)) + list(ascii_set)
+    resource_id = random.choice(ascii_set) + "".join(
+        str(random.choice(chars)) for _ in range(length - 1)
+    )
+    return resource_id
+
+
 def match_task_arn(task: Dict[str, Any], arns: List[str]) -> bool:
     return task["ReplicationTaskArn"] in arns
 
diff --git a/moto/ec2/exceptions.py b/moto/ec2/exceptions.py
index 26eb36438..4ae7eaaf7 100644
--- a/moto/ec2/exceptions.py
+++ b/moto/ec2/exceptions.py
@@ -59,6 +59,16 @@ class DependencyViolationError(EC2ClientError):
         super().__init__("DependencyViolation", message)
 
 
+class MissingInputError(EC2ClientError):
+    def __init__(self, message: str):
+        super().__init__("MissingInput", message)
+
+
+class InvalidInputError(EC2ClientError):
+    def __init__(self, message: str):
+        super().__init__("InvalidInput", message)
+
+
 class MissingParameterError(EC2ClientError):
     def __init__(self, parameter: str):
         super().__init__(
@@ -114,6 +124,14 @@ class InvalidKeyPairFormatError(EC2ClientError):
         )
 
 
+class VPCIdNotSpecifiedError(EC2ClientError):
+    def __init__(self) -> None:
+        super().__init__(
+            "VPCIdNotSpecified",
+            "No default VPC for this user. GroupName is only supported for EC2-Classic and default VPC.",
+        )
+
+
 class InvalidVPCIdError(EC2ClientError):
     def __init__(self, vpc_id: Any):
         super().__init__("InvalidVpcID.NotFound", f"VpcID {vpc_id} does not exist.")
diff --git a/moto/ec2/models/instances.py b/moto/ec2/models/instances.py
index 953e959c6..b752806a8 100644
--- a/moto/ec2/models/instances.py
+++ b/moto/ec2/models/instances.py
@@ -21,14 +21,17 @@ from moto.packages.boto.ec2.instance import Reservation
 
 from ..exceptions import (
     AvailabilityZoneNotFromRegionError,
+    InvalidInputError,
     InvalidInstanceIdError,
     InvalidInstanceTypeError,
     InvalidParameterCombination,
     InvalidParameterValueErrorUnknownAttribute,
     InvalidSecurityGroupNotFoundError,
     InvalidSubnetIdError,
+    MissingInputError,
     OperationDisableApiStopNotPermitted,
     OperationNotPermitted4,
+    VPCIdNotSpecifiedError,
 )
 from ..utils import (
     convert_tag_spec,
@@ -529,7 +532,26 @@ class Instance(TaggedEC2Resource, BotoInstance, CloudFormationModel):
                 else:
                     # Get default Subnet
                     default_subnets = self.ec2_backend.get_default_subnets()
-                    nic_subnet = default_subnets[self._placement.zone]
+
+                    default_subnet = None
+
+                    if self.subnet_id is None:
+                        default_vpc = self.ec2_backend.get_default_vpc()
+                        if default_vpc is None:
+                            raise VPCIdNotSpecifiedError()
+
+                        if not default_subnets:
+                            raise MissingInputError(
+                                f"No subnets found for the default VPC '{default_vpc.id}'. Please specify a subnet."
+                            )
+
+                        default_subnet = default_subnets.get(self._placement.zone)
+                        if default_subnet is None:
+                            raise InvalidInputError(
+                                f"No default subnet for availability zone: '{self._placement.zone}'."
+                            )
+
+                    nic_subnet = default_subnet or self.subnet_id  # type: ignore[assignment]
 
                 group_ids = nic.get("SecurityGroupId") or []
                 if security_groups:
@@ -595,16 +617,18 @@ class Instance(TaggedEC2Resource, BotoInstance, CloudFormationModel):
 
     def applies(self, filters: List[Dict[str, Any]]) -> bool:
         if filters:
-            applicable = False
+            # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ec2/client/describe_instances.html
+            # Filters Section in the boto3 documentation
+            # If you specify multiple filters, the filters are joined with an AND, and the request returns only results that match all of the specified filters.
             for f in filters:
                 acceptable_values = f["values"]
                 if f["name"] == "instance-state-name":
-                    if self._state.name in acceptable_values:
-                        applicable = True
+                    if self._state.name not in acceptable_values:
+                        return False
                 if f["name"] == "instance-state-code":
-                    if str(self._state.code) in acceptable_values:
-                        applicable = True
-            return applicable
+                    if str(self._state.code) not in acceptable_values:
+                        return False
+            return True
         # If there are no filters, all instances are valid
         return True
 
diff --git a/moto/iam/policy_conditions.py b/moto/iam/policy_conditions.py
index 154bd5ac6..975f5afe6 100644
--- a/moto/iam/policy_conditions.py
+++ b/moto/iam/policy_conditions.py
@@ -25,9 +25,6 @@ class TrustCondition:
         expected_value: List[Union[List[str], str]],
         expected_value_source: List[str],
     ) -> None:
-        if condition not in CONDITION_OPERATIONS:
-            raise NotImplementedError(f"Unsupported condition: {condition}")
-
         self._condition = condition
 
         self._expected_value_source = expected_value_source
@@ -38,7 +35,11 @@ class TrustCondition:
         return self._expected_value_source
 
     def verify_condition(self, actual_values: List[str]) -> bool:
-        verify_action = CONDITION_OPERATIONS[self._condition]
+        verify_action = CONDITION_OPERATIONS.get(self._condition)
+
+        # If the evaluation operation is not supported, ignore condition evaluation
+        if not verify_action:
+            return True
 
         for index, actual_value in enumerate(actual_values):
             if not verify_action(self._expected_value[index], actual_value):
diff --git a/moto/networkfirewall/__init__.py b/moto/networkfirewall/__init__.py
new file mode 100644
index 000000000..d73ec8f7e
--- /dev/null
+++ b/moto/networkfirewall/__init__.py
@@ -0,0 +1,1 @@
+from .models import NetworkFirewallBackend, networkfirewall_backends  # noqa: F401
diff --git a/moto/networkfirewall/exceptions.py b/moto/networkfirewall/exceptions.py
new file mode 100644
index 000000000..8d34ba3ee
--- /dev/null
+++ b/moto/networkfirewall/exceptions.py
@@ -0,0 +1,15 @@
+"""Exceptions raised by the networkfirewall service."""
+
+import json
+
+from moto.core.exceptions import JsonRESTError
+
+
+class ResourceNotFound(JsonRESTError):
+    def __init__(self, resource_id: str):
+        super().__init__("NotFoundException", "Resource not found.")
+        body = {
+            "ResourceId": resource_id,
+            "Message": "Resource not found.",
+        }
+        self.description = json.dumps(body)
diff --git a/moto/networkfirewall/models.py b/moto/networkfirewall/models.py
new file mode 100644
index 000000000..7e8ed45e8
--- /dev/null
+++ b/moto/networkfirewall/models.py
@@ -0,0 +1,167 @@
+"""NetworkFirewallBackend class with methods for supported APIs."""
+
+from typing import Any, Dict, List, Optional
+
+from moto.core.base_backend import BackendDict, BaseBackend
+from moto.core.common_models import BaseModel
+from moto.es.exceptions import ResourceNotFound
+from moto.utilities.paginator import paginate
+from moto.utilities.tagging_service import TaggingService
+
+PAGINATION_MODEL = {
+    "list_firewalls": {
+        "input_token": "next_token",
+        "limit_key": "max_results",
+        "limit_default": 100,
+        "unique_attribute": "arn",
+    },
+}
+
+
+class NetworkFirewallModel(BaseModel):
+    def __init__(
+        self,
+        account_id: str,
+        region_name: str,
+        firewall_name: str,
+        firewall_policy_arn: str,
+        vpc_id: str,
+        subnet_mappings: List[str],
+        delete_protection: bool,
+        subnet_change_protection: bool,
+        firewall_policy_change_protection: bool,
+        description: str,
+        tags: List[Dict[str, str]],
+        encryption_configuration: Dict[str, str],
+        enabled_analysis_types: List[str],
+    ):
+        self.firewall_name = firewall_name
+        self.firewall_policy_arn = firewall_policy_arn
+        self.vpc_id = vpc_id
+        self.subnet_mappings = subnet_mappings
+        self.delete_protection: bool = (
+            delete_protection if delete_protection is not None else True
+        )
+        self.subnet_change_protection: bool = (
+            subnet_change_protection if subnet_change_protection is not None else True
+        )
+        self.firewall_policy_change_protection: bool = (
+            firewall_policy_change_protection
+            if firewall_policy_change_protection is not None
+            else True
+        )
+        self.description = description
+        self.tags = tags
+        self.encryption_configuration = encryption_configuration
+        self.enabled_analysis_types = enabled_analysis_types
+
+        self.arn = f"arn:aws:network-firewall:{region_name}:{account_id}:firewall/{self.firewall_name}"
+
+        self.firewall_status = {
+            "Status": "READY",
+            "ConfigurationSyncStateSummary": "IN_SYNC",
+        }
+        self.logging_configs: List[Dict[str, str]] = []
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "FirewallName": self.firewall_name,
+            "FirewallArn": self.arn,
+            "FirewallPolicyArn": self.firewall_policy_arn,
+            "VpcId": self.vpc_id,
+            "SubnetMappings": self.subnet_mappings,
+            "DeleteProtection": self.delete_protection,
+            "SubnetChangeProtection": self.subnet_change_protection,
+            "FirewallPolicyChangeProtection": self.firewall_policy_change_protection,
+            "Description": self.description,
+            "Tags": self.tags,
+            "EncryptionConfiguration": self.encryption_configuration,
+            "EnabledAnalysisTypes": self.enabled_analysis_types,
+        }
+
+
+class NetworkFirewallBackend(BaseBackend):
+    """Implementation of NetworkFirewall APIs."""
+
+    def __init__(self, region_name: str, account_id: str) -> None:
+        super().__init__(region_name, account_id)
+        self.firewalls: Dict[str, NetworkFirewallModel] = {}
+        self.tagger = TaggingService()
+
+    def create_firewall(
+        self,
+        firewall_name: str,
+        firewall_policy_arn: str,
+        vpc_id: str,
+        subnet_mappings: List[str],
+        delete_protection: bool,
+        subnet_change_protection: bool,
+        firewall_policy_change_protection: bool,
+        description: str,
+        tags: List[Dict[str, str]],
+        encryption_configuration: Dict[str, str],
+        enabled_analysis_types: List[str],
+    ) -> NetworkFirewallModel:
+        firewall = NetworkFirewallModel(
+            self.account_id,
+            self.region_name,
+            firewall_name=firewall_name,
+            firewall_policy_arn=firewall_policy_arn,
+            vpc_id=vpc_id,
+            subnet_mappings=subnet_mappings,
+            delete_protection=delete_protection,
+            subnet_change_protection=subnet_change_protection,
+            firewall_policy_change_protection=firewall_policy_change_protection,
+            description=description,
+            tags=tags,
+            encryption_configuration=encryption_configuration,
+            enabled_analysis_types=enabled_analysis_types,
+        )
+        self.firewalls[firewall.arn] = firewall
+
+        if tags:
+            self.tagger.tag_resource(firewall.arn, tags)
+
+        return firewall
+
+    def _get_firewall(
+        self, firewall_arn: Optional[str], firewall_name: Optional[str]
+    ) -> NetworkFirewallModel:
+        if firewall_arn:
+            if firewall_arn in self.firewalls:
+                return self.firewalls[firewall_arn]
+        for firewall in self.firewalls.values():
+            if firewall.firewall_name == firewall_name:
+                return firewall
+        raise ResourceNotFound("NetworkFirewall", str(firewall_arn or firewall_name))
+
+    def describe_logging_configuration(
+        self, firewall_arn: str, firewall_name: str
+    ) -> NetworkFirewallModel:
+        firewall: NetworkFirewallModel = self._get_firewall(firewall_arn, firewall_name)
+        return firewall
+
+    def update_logging_configuration(
+        self,
+        firewall_arn: str,
+        firewall_name: str,
+        logging_configuration: List[Dict[str, str]],
+    ) -> NetworkFirewallModel:
+        firewall: NetworkFirewallModel = self._get_firewall(firewall_arn, firewall_name)
+        firewall.logging_configs = logging_configuration
+        return firewall
+
+    @paginate(pagination_model=PAGINATION_MODEL)
+    def list_firewalls(self, vpc_ids: List[str]) -> List[NetworkFirewallModel]:
+        firewalls = list(self.firewalls.values())
+        if vpc_ids:
+            firewalls = [fw for fw in firewalls if fw.vpc_id in vpc_ids]
+        return firewalls
+
+    def describe_firewall(
+        self, firewall_name: str, firewall_arn: str
+    ) -> NetworkFirewallModel:
+        return self._get_firewall(firewall_arn, firewall_name)
+
+
+networkfirewall_backends = BackendDict(NetworkFirewallBackend, "network-firewall")
diff --git a/moto/networkfirewall/responses.py b/moto/networkfirewall/responses.py
new file mode 100644
index 000000000..4f7d30391
--- /dev/null
+++ b/moto/networkfirewall/responses.py
@@ -0,0 +1,108 @@
+"""Handles incoming networkfirewall requests, invokes methods, returns responses."""
+
+import json
+
+from moto.core.responses import BaseResponse
+
+from .models import NetworkFirewallBackend, networkfirewall_backends
+
+
+class NetworkFirewallResponse(BaseResponse):
+    """Handler for NetworkFirewall requests and responses."""
+
+    def __init__(self) -> None:
+        super().__init__(service_name="networkfirewall")
+
+    @property
+    def networkfirewall_backend(self) -> NetworkFirewallBackend:
+        """Return backend instance specific for this region."""
+        return networkfirewall_backends[self.current_account][self.region]
+
+    def create_firewall(self) -> str:
+        firewall_name = self._get_param("FirewallName")
+        firewall_policy_arn = self._get_param("FirewallPolicyArn")
+        vpc_id = self._get_param("VpcId")
+        subnet_mappings = self._get_param("SubnetMappings")
+        delete_protection = self._get_param("DeleteProtection")
+        subnet_change_protection = self._get_param("SubnetChangeProtection")
+        firewall_policy_change_protection = self._get_param(
+            "FirewallPolicyChangeProtection"
+        )
+        description = self._get_param("Description")
+        tags = self._get_param("Tags")
+        encryption_configuration = self._get_param("EncryptionConfiguration")
+        enabled_analysis_types = self._get_param("EnabledAnalysisTypes")
+        firewall = self.networkfirewall_backend.create_firewall(
+            firewall_name=firewall_name,
+            firewall_policy_arn=firewall_policy_arn,
+            vpc_id=vpc_id,
+            subnet_mappings=subnet_mappings,
+            delete_protection=delete_protection,
+            subnet_change_protection=subnet_change_protection,
+            firewall_policy_change_protection=firewall_policy_change_protection,
+            description=description,
+            tags=tags,
+            encryption_configuration=encryption_configuration,
+            enabled_analysis_types=enabled_analysis_types,
+        )
+
+        return json.dumps(
+            dict(Firewall=firewall.to_dict(), FirewallStatus=firewall.firewall_status)
+        )
+
+    def describe_logging_configuration(self) -> str:
+        firewall_arn = self._get_param("FirewallArn")
+        firewall_name = self._get_param("FirewallName")
+        firewall = self.networkfirewall_backend.describe_logging_configuration(
+            firewall_arn=firewall_arn,
+            firewall_name=firewall_name,
+        )
+        return json.dumps(
+            dict(
+                FirewallArn=firewall.arn,
+                LoggingConfiguration=firewall.logging_configs,
+            )
+        )
+
+    def update_logging_configuration(self) -> str:
+        firewall_arn = self._get_param("FirewallArn")
+        firewall_name = self._get_param("FirewallName")
+        logging_configuration = self._get_param("LoggingConfiguration")
+        firewall = self.networkfirewall_backend.update_logging_configuration(
+            firewall_arn=firewall_arn,
+            firewall_name=firewall_name,
+            logging_configuration=logging_configuration,
+        )
+        return json.dumps(
+            dict(
+                FirewallArn=firewall.arn,
+                FirewallName=firewall.firewall_name,
+                LoggingConfiguration=firewall.logging_configs,
+            )
+        )
+
+    def list_firewalls(self) -> str:
+        next_token = self._get_param("NextToken")
+        vpc_ids = self._get_param("VpcIds")
+        max_results = self._get_param("MaxResults")
+        firewalls, next_token = self.networkfirewall_backend.list_firewalls(
+            next_token=next_token,
+            vpc_ids=vpc_ids,
+            max_results=max_results,
+        )
+        firewall_list = [fw.to_dict() for fw in firewalls]
+        return json.dumps(dict(nextToken=next_token, Firewalls=firewall_list))
+
+    def describe_firewall(self) -> str:
+        firewall_name = self._get_param("FirewallName")
+        firewall_arn = self._get_param("FirewallArn")
+        firewall = self.networkfirewall_backend.describe_firewall(
+            firewall_name=firewall_name,
+            firewall_arn=firewall_arn,
+        )
+        return json.dumps(
+            dict(
+                Firewall=firewall.to_dict(),
+                FirewallStatus=firewall.firewall_status,
+            )
+        )
diff --git a/moto/networkfirewall/urls.py b/moto/networkfirewall/urls.py
new file mode 100644
index 000000000..4815a8c21
--- /dev/null
+++ b/moto/networkfirewall/urls.py
@@ -0,0 +1,11 @@
+"""networkfirewall base URL and path."""
+
+from .responses import NetworkFirewallResponse
+
+url_bases = [
+    r"https?://network-firewall\.(.+)\.amazonaws\.com",
+]
+
+url_paths = {
+    "{0}/$": NetworkFirewallResponse.dispatch,
+}
diff --git a/moto/organizations/models.py b/moto/organizations/models.py
index cb7057efa..a70abe725 100644
--- a/moto/organizations/models.py
+++ b/moto/organizations/models.py
@@ -257,6 +257,8 @@ class FakePolicy(BaseModel):
             self._arn_format = utils.AI_POLICY_ARN_FORMAT
         elif self.type == "SERVICE_CONTROL_POLICY":
             self._arn_format = utils.SCP_ARN_FORMAT
+        elif self.type == "TAG_POLICY":
+            self._arn_format = utils.TAG_POLICY_ARN_FORMAT
         else:
             raise NotImplementedError(
                 f"The {self.type} policy type has not been implemented"
@@ -778,7 +780,11 @@ class OrganizationsBackend(BaseBackend):
         if not FakePolicy.supported_policy_type(_filter):
             raise InvalidInputException("You specified an invalid value.")
 
-        if _filter not in ["AISERVICES_OPT_OUT_POLICY", "SERVICE_CONTROL_POLICY"]:
+        if _filter not in [
+            "AISERVICES_OPT_OUT_POLICY",
+            "SERVICE_CONTROL_POLICY",
+            "TAG_POLICY",
+        ]:
             raise NotImplementedError(
                 f"The {_filter} policy type has not been implemented"
             )
diff --git a/moto/organizations/utils.py b/moto/organizations/utils.py
index 75560c3ec..37368471c 100644
--- a/moto/organizations/utils.py
+++ b/moto/organizations/utils.py
@@ -15,6 +15,7 @@ SCP_ARN_FORMAT = "arn:{0}:organizations::{1}:policy/{2}/service_control_policy/{
 AI_POLICY_ARN_FORMAT = (
     "arn:{0}:organizations::{1}:policy/{2}/aiservices_opt_out_policy/{3}"
 )
+TAG_POLICY_ARN_FORMAT = "arn:{0}:organizations::{1}:policy/{2}/tag_policy/{3}"
 
 CHARSET = string.ascii_lowercase + string.digits
 ORG_ID_SIZE = 10
diff --git a/moto/resourcegroupstaggingapi/models.py b/moto/resourcegroupstaggingapi/models.py
index 37f6f7cce..2685528d9 100644
--- a/moto/resourcegroupstaggingapi/models.py
+++ b/moto/resourcegroupstaggingapi/models.py
@@ -4,9 +4,11 @@ from moto.acm.models import AWSCertificateManagerBackend, acm_backends
 from moto.appsync.models import AppSyncBackend, appsync_backends
 from moto.awslambda.models import LambdaBackend, lambda_backends
 from moto.backup.models import BackupBackend, backup_backends
+from moto.clouddirectory import CloudDirectoryBackend, clouddirectory_backends
 from moto.cloudfront.models import CloudFrontBackend, cloudfront_backends
 from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.exceptions import RESTError
+from moto.dms.models import DatabaseMigrationServiceBackend, dms_backends
 from moto.dynamodb.models import DynamoDBBackend, dynamodb_backends
 from moto.ec2 import ec2_backends
 from moto.ecs.models import EC2ContainerServiceBackend, ecs_backends
@@ -65,6 +67,10 @@ class ResourceGroupsTaggingAPIBackend(BaseBackend):
     def s3_backend(self) -> S3Backend:
         return s3_backends[self.account_id][self.partition]
 
+    @property
+    def dms_backend(self) -> DatabaseMigrationServiceBackend:
+        return dms_backends[self.account_id][self.region_name]
+
     @property
     def ec2_backend(self) -> Any:  # type: ignore[misc]
         return ec2_backends[self.account_id][self.region_name]
@@ -189,6 +195,12 @@ class ResourceGroupsTaggingAPIBackend(BaseBackend):
             return lexv2models_backends[self.account_id][self.region_name]
         return None
 
+    @property
+    def clouddirectory_backend(self) -> Optional[CloudDirectoryBackend]:
+        if self.region_name in clouddirectory_backends[self.account_id].regions:
+            return clouddirectory_backends[self.account_id][self.region_name]
+        return None
+
     @property
     def cloudfront_backend(self) -> CloudFrontBackend:
         return cloudfront_backends[self.account_id][self.partition]
@@ -296,6 +308,20 @@ class ResourceGroupsTaggingAPIBackend(BaseBackend):
                     continue
                 yield {"ResourceARN": bucket.arn, "Tags": tags}
 
+        # Cloud Directory
+        if self.clouddirectory_backend:
+            if not resource_type_filters or "clouddirectory" in resource_type_filters:
+                clouddirectory_backend = clouddirectory_backends[self.account_id][
+                    self.region_name
+                ]
+                for directory in clouddirectory_backend.directories.values():
+                    tags = clouddirectory_backend.tagger.list_tags_for_resource(
+                        directory.directory_arn
+                    )["Tags"]
+                    if not tags or not tag_filter(tags):
+                        continue
+                    yield {"ResourceARN": f"{directory.directory_arn}", "Tags": tags}
+
         # CloudFormation
         if not resource_type_filters or "cloudformation:stack" in resource_type_filters:
             try:
@@ -328,6 +354,28 @@ class ResourceGroupsTaggingAPIBackend(BaseBackend):
                     continue
                 yield {"ResourceARN": f"{dist.arn}", "Tags": tags}
 
+        # DMS
+        if not resource_type_filters or "dms:endpoint" in resource_type_filters:
+            for endpoint in self.dms_backend.endpoints.values():
+                tags = self.dms_backend.tagger.list_tags_for_resource(
+                    endpoint.endpoint_arn
+                )["Tags"]
+                if not tag_filter(tags):
+                    continue
+                yield {"ResourceARN": f"{endpoint.endpoint_arn}", "Tags": tags}
+
+        if (
+            not resource_type_filters
+            or "dms:replication-instance" in resource_type_filters
+        ):
+            for replication_instance in self.dms_backend.replication_instances.values():
+                tags = self.dms_backend.tagger.list_tags_for_resource(
+                    replication_instance.arn
+                )["Tags"]
+                if not tag_filter(tags):
+                    continue
+                yield {"ResourceARN": f"{replication_instance.arn}", "Tags": tags}
+
         # ECS
         if not resource_type_filters or "ecs:service" in resource_type_filters:
             for service in self.ecs_backend.services.values():
diff --git a/moto/s3/exceptions.py b/moto/s3/exceptions.py
index bfd600f7f..c077ee829 100644
--- a/moto/s3/exceptions.py
+++ b/moto/s3/exceptions.py
@@ -130,6 +130,16 @@ class MissingVersion(S3ClientError):
         super().__init__("NoSuchVersion", "The specified version does not exist.")
 
 
+class MissingInventoryConfig(S3ClientError):
+    code = 404
+
+    def __init__(self) -> None:
+        super().__init__(
+            "NoSuchInventoryConfig",
+            "The specified inventory configuration does not exist.",
+        )
+
+
 class InvalidVersion(S3ClientError):
     code = 400
 
diff --git a/moto/s3/models.py b/moto/s3/models.py
index 6350801e7..3ea17db67 100644
--- a/moto/s3/models.py
+++ b/moto/s3/models.py
@@ -56,6 +56,7 @@ from moto.s3.exceptions import (
     MalformedXML,
     MethodNotAllowed,
     MissingBucket,
+    MissingInventoryConfig,
     MissingKey,
     NoSuchPublicAccessBlockConfiguration,
     NoSuchUpload,
@@ -1067,6 +1068,7 @@ class FakeBucket(CloudFormationModel):
         self.default_lock_days: Optional[int] = 0
         self.default_lock_years: Optional[int] = 0
         self.ownership_rule: Optional[Dict[str, Any]] = None
+        self.inventory_configs: Dict[str, FakeBucketInventoryConfiguration] = {}
         s3_backends.bucket_accounts[name] = (self.partition, account_id)
 
     @property
@@ -1648,6 +1650,24 @@ class FakeBucket(CloudFormationModel):
 class FakeTableStorageBucket(FakeBucket): ...
 
 
+class FakeBucketInventoryConfiguration(BaseModel):
+    def __init__(
+        self,
+        id: str,
+        destination: Dict[str, Any],
+        is_enabled: bool,
+        schedule: Dict[str, Any],
+        filters: Optional[Dict[str, Any]] = None,
+        optional_fields: Optional[List[str]] = None,
+    ):
+        self.id = id
+        self.destination = destination
+        self.is_enabled = is_enabled
+        self.schedule = schedule
+        self.filters = filters
+        self.optional_fields = optional_fields
+
+
 class S3Backend(BaseBackend, CloudWatchMetricProvider):
     """
     Custom S3 endpoints are supported, if you are using a S3-compatible storage solution like Ceph.
@@ -1716,6 +1736,7 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
         self.table_buckets: Dict[str, FakeTableStorageBucket] = {}
         self.tagger = TaggingService()
         self._pagination_tokens: Dict[str, str] = {}
+        self.inventory_configs: Dict[str, FakeBucketInventoryConfiguration] = {}
 
     def reset(self) -> None:
         # For every key and multipart, Moto opens a TemporaryFile to write the value of those keys
@@ -3119,6 +3140,38 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
         # Implementation part of responses.py
         pass
 
+    def put_bucket_inventory_configuration(
+        self, bucket_name: str, inventory_configuration: Dict[str, Any]
+    ) -> None:
+        bucket = self.get_bucket(bucket_name)
+        inv_config = FakeBucketInventoryConfiguration(
+            id=inventory_configuration["Id"],
+            destination=inventory_configuration["Destination"],
+            is_enabled=inventory_configuration["IsEnabled"],
+            schedule=inventory_configuration["Schedule"],
+            filters=inventory_configuration.get("Filter"),
+            optional_fields=inventory_configuration.get("OptionalFields"),
+        )
+
+        bucket.inventory_configs[inv_config.id] = inv_config
+        return
+
+    def get_bucket_inventory_configuration(
+        self, bucket_name: str, id: str, expected_bucket_owner: Optional[str] = None
+    ) -> FakeBucketInventoryConfiguration:
+        bucket = self.get_bucket(bucket_name)
+
+        if id in bucket.inventory_configs:
+            return bucket.inventory_configs[id]
+
+        raise MissingInventoryConfig()
+
+    def list_bucket_inventory_configurations(
+        self, bucket_name: str
+    ) -> List[FakeBucketInventoryConfiguration]:
+        bucket = self.get_bucket(bucket_name)
+        return list(bucket.inventory_configs.values())
+
 
 class S3BackendDict(BackendDict[S3Backend]):
     """
diff --git a/moto/s3/responses.py b/moto/s3/responses.py
index aac4b7d25..48f5220e2 100644
--- a/moto/s3/responses.py
+++ b/moto/s3/responses.py
@@ -353,7 +353,6 @@ class S3Response(BaseResponse):
     ) -> Union[str, TYPE_RESPONSE]:
         querystring = self._get_querystring(request, full_url)
         method = request.method
-
         bucket_name = self.parse_bucket_name_from_url(request, full_url)
         if not bucket_name:
             # If no bucket specified, list all buckets
@@ -555,7 +554,12 @@ class S3Response(BaseResponse):
             return self.get_bucket_accelerate_configuration()
         elif "publicAccessBlock" in querystring:
             return self.get_public_access_block()
-
+        elif "inventory" in querystring:
+            # Only GET includes "id" in the querystring, LIST does not
+            if "id" in querystring:
+                return self.get_bucket_inventory_configuration()
+            else:
+                return self.list_bucket_inventory_configurations()
         elif "versions" in querystring:
             return self.list_object_versions()
         elif "encryption" in querystring:
@@ -886,7 +890,12 @@ class S3Response(BaseResponse):
                 bucket_name, ownership=ownership_rule
             )
             return ""
-
+        elif "inventory" in querystring:
+            inventory_config = self._inventory_config_from_body()
+            self.backend.put_bucket_inventory_configuration(
+                bucket_name, inventory_config
+            )
+            return ""
         else:
             # us-east-1, the default AWS region behaves a bit differently
             # - you should not use any location constraint
@@ -1083,6 +1092,29 @@ class S3Response(BaseResponse):
         template = self.response_template(S3_PUBLIC_ACCESS_BLOCK_CONFIGURATION)
         return template.render(public_block_config=public_block_config)
 
+    def get_bucket_inventory_configuration(self) -> str:
+        config_id = self.querystring["id"][0]
+        inventory_configuration = self.backend.get_bucket_inventory_configuration(
+            bucket_name=self.bucket_name, id=config_id
+        )
+        template = self.response_template(S3_BUCKET_INVENTORY_CONFIGURATION)
+        return template.render(
+            inventory_config=inventory_configuration,
+            s3_bucket_config=inventory_configuration.destination["S3BucketDestination"],
+        )
+
+    def list_bucket_inventory_configurations(self) -> str:
+        inventory_configuration_list = (
+            self.backend.list_bucket_inventory_configurations(
+                bucket_name=self.bucket_name,
+            )
+        )
+        template = self.response_template(LIST_BUCKET_INVENTORY_CONFIGURATIONS_TEMPLATE)
+        # TODO: Add support for pagination/ continuation tokens
+        return template.render(
+            inventory_configuration_list=inventory_configuration_list,
+        )
+
     def _bucket_response_delete(
         self, bucket_name: str, querystring: Dict[str, Any]
     ) -> TYPE_RESPONSE:
@@ -2468,6 +2500,11 @@ class S3Response(BaseResponse):
         config = parsed_xml["ReplicationConfiguration"]
         return config
 
+    def _inventory_config_from_body(self) -> Dict[str, Any]:
+        parsed_xml = xmltodict.parse(self.body)
+        config = parsed_xml["InventoryConfiguration"]
+        return config
+
     def _key_response_delete(
         self, bucket_name: str, query: Dict[str, Any], key_name: str
     ) -> TYPE_RESPONSE:
@@ -3344,6 +3381,97 @@ S3_PUBLIC_ACCESS_BLOCK_CONFIGURATION = """
 </PublicAccessBlockConfiguration>
 """
 
+S3_BUCKET_INVENTORY_CONFIGURATION = """<?xml version="1.0" encoding="UTF-8"?>
+<InventoryConfiguration>
+   <Destination>
+      <S3BucketDestination>
+            {%if s3_bucket_config.get('AccountId') %}
+            <AccountId>{{s3_bucket_config['AccountId']}}</AccountId>
+            {% endif %}
+            <Bucket>{{s3_bucket_config['Bucket']}}</Bucket>
+            <Format>{{s3_bucket_config['Format']}}</Format>
+            {%if s3_bucket_config.get('Prefix') %}
+            <Prefix>{{s3_bucket_config['Prefix']}}</Prefix>
+            {% endif %}
+            {% if s3_bucket_config.get('Encryption') %}
+            <Encryption>
+                ## NOTE boto changes the key SSEKMS to SSE-KMS on put and SSE-KMS to SSEKMS on get
+                {% if s3_bucket_config['Encryption'].get('SSE-KMS') %}
+                <SSE-KMS>
+                    <KeyId>{{s3_bucket_config['Encryption']['SSE-KMS']['KeyId']}}</KeyId>
+                </SSE-KMS>
+                {% else %}
+                <SSES3/>
+                {% endif %}
+            </Encryption>
+            {% endif %}
+      </S3BucketDestination>
+   </Destination>
+   <IsEnabled>{{inventory_config.is_enabled}}</IsEnabled>
+   <Filter>
+      <Prefix>{{inventory_config.filters['Prefix']}}</Prefix>
+   </Filter>
+   <Id>{{inventory_config.id}}</Id>
+   <IncludedObjectVersions>All</IncludedObjectVersions>
+   <OptionalFields>
+        {% for field in inventory_config.optional_fields['Field'] %}
+        <Field>{{ field }}</Field>
+        {% endfor %}
+   </OptionalFields>
+   <Schedule>
+      <Frequency>{{inventory_config.schedule['Frequency']}}</Frequency>
+   </Schedule>
+</InventoryConfiguration>
+"""
+
+LIST_BUCKET_INVENTORY_CONFIGURATIONS_TEMPLATE = """<?xml version="1.0" encoding="UTF-8"?>
+<ListInventoryConfigurationsResult>
+    <IsTruncated>false</IsTruncated>
+    {% for inventory_config in inventory_configuration_list %}
+    <InventoryConfiguration>
+        <Destination>
+            <S3BucketDestination>
+                    {%if inventory_config.destination["S3BucketDestination"].get('AccountId') %}
+                    <AccountId>{{inventory_config.destination["S3BucketDestination"]['AccountId']}}</AccountId>
+                    {% endif %}
+                    <Bucket>{{inventory_config.destination["S3BucketDestination"]['Bucket']}}</Bucket>
+                    <Format>{{inventory_config.destination["S3BucketDestination"]['Format']}}</Format>
+                    {%if inventory_config.destination["S3BucketDestination"].get('Prefix') %}
+                    <Prefix>{{inventory_config.destination["S3BucketDestination"]['Prefix']}}</Prefix>
+                    {% endif %}
+                    {% if inventory_config.destination["S3BucketDestination"].get('Encryption') %}
+                    <Encryption>
+                        ## NOTE boto changes the key SSEKMS to SSE-KMS on put and SSE-KMS to SSEKMS on get
+                        {% if inventory_config.destination["S3BucketDestination"]['Encryption'].get('SSE-KMS') %}
+                        <SSE-KMS>
+                            <KeyId>{{inventory_config.destination["S3BucketDestination"]['Encryption']['SSE-KMS']['KeyId']}}</KeyId>
+                        </SSE-KMS>
+                        {% else %}
+                        <SSES3/>
+                        {% endif %}
+                    </Encryption>
+                    {% endif %}
+            </S3BucketDestination>
+        </Destination>
+        <IsEnabled>{{inventory_config.is_enabled}}</IsEnabled>
+        <Filter>
+            <Prefix>{{inventory_config.filters['Prefix']}}</Prefix>
+        </Filter>
+        <Id>{{inventory_config.id}}</Id>
+        <IncludedObjectVersions>All</IncludedObjectVersions>
+        <OptionalFields>
+                {% for field in inventory_config.optional_fields['Field'] %}
+                <Field>{{ field }}</Field>
+                {% endfor %}
+        </OptionalFields>
+        <Schedule>
+            <Frequency>{{inventory_config.schedule['Frequency']}}</Frequency>
+        </Schedule>
+    </InventoryConfiguration>
+    {% endfor %}
+</ListInventoryConfigurationsResult>
+"""
+
 S3_BUCKET_LOCK_CONFIGURATION = """
 <ObjectLockConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
     {%if lock_enabled %}
diff --git a/moto/servicecatalog/urls.py b/moto/servicecatalog/urls.py
index 785a6308f..a72ae3a41 100644
--- a/moto/servicecatalog/urls.py
+++ b/moto/servicecatalog/urls.py
@@ -1,5 +1,8 @@
 """servicecatalog base URL and path."""
 
+from moto.servicecatalogappregistry.responses import AppRegistryResponse
+from moto.servicecatalogappregistry.urls import url_paths as catalog_paths
+
 from .responses import ServiceCatalogResponse
 
 url_bases = [
@@ -8,4 +11,5 @@ url_bases = [
 
 url_paths = {
     "{0}/$": ServiceCatalogResponse.dispatch,
+    **{path: AppRegistryResponse.dispatch for path in catalog_paths},
 }
diff --git a/moto/servicecatalogappregistry/__init__.py b/moto/servicecatalogappregistry/__init__.py
new file mode 100644
index 000000000..7d72cb3d6
--- /dev/null
+++ b/moto/servicecatalogappregistry/__init__.py
@@ -0,0 +1,1 @@
+from .models import servicecatalogappregistry_backends  # noqa: F401
diff --git a/moto/servicecatalogappregistry/exceptions.py b/moto/servicecatalogappregistry/exceptions.py
new file mode 100644
index 000000000..dff77b7af
--- /dev/null
+++ b/moto/servicecatalogappregistry/exceptions.py
@@ -0,0 +1,15 @@
+"""Exceptions raised by the servicecatalogappregistry service."""
+
+from moto.core.exceptions import JsonRESTError
+
+
+class ResourceNotFoundException(JsonRESTError):
+    def __init__(self, msg: str = "The specified resource does not exist.") -> None:
+        super().__init__("ResourceNotFoundException", msg)
+
+
+class ValidationException(JsonRESTError):
+    def __init__(self) -> None:
+        super().__init__(
+            "ValidationException", "The request has invalid or missing parameters."
+        )
diff --git a/moto/servicecatalogappregistry/models.py b/moto/servicecatalogappregistry/models.py
new file mode 100644
index 000000000..830d2d223
--- /dev/null
+++ b/moto/servicecatalogappregistry/models.py
@@ -0,0 +1,163 @@
+"""AppRegistryBackend class with methods for supported APIs."""
+
+import datetime
+import re
+from typing import Any, Dict, List
+
+from moto.cloudformation.exceptions import ValidationError
+from moto.cloudformation.models import cloudformation_backends
+from moto.core.base_backend import BackendDict, BaseBackend
+from moto.core.common_models import BaseModel
+from moto.moto_api._internal import mock_random
+from moto.resourcegroups.models import FakeResourceGroup
+from moto.servicecatalogappregistry.exceptions import (
+    ResourceNotFoundException,
+    ValidationException,
+)
+from moto.utilities.tagging_service import TaggingService
+from moto.utilities.utils import get_partition
+
+
+class Application(BaseModel):
+    def __init__(
+        self,
+        name: str,
+        description: str,
+        region: str,
+        account_id: str,
+    ):
+        self.id = mock_random.get_random_string(
+            length=27, include_digits=True, lower_case=True
+        )
+        self.arn = f"arn:{get_partition(region)}:servicecatalog:{region}:{account_id}:applications/{self.id}"
+        self.name = name
+        self.description = description
+        self.creationTime = datetime.datetime.now()
+        self.lastUpdateTime = self.creationTime
+        self.tags: Dict[str, str] = dict()
+        self.applicationTag: Dict[str, str] = {"awsApplication": self.arn}
+
+        self.associated_resources: Dict[str, AssociatedResource] = dict()
+
+    def to_json(self) -> Dict[str, Any]:
+        return {
+            "id": self.id,
+            "arn": self.arn,
+            "name": self.name,
+            "description": self.description,
+            "creationTime": str(self.creationTime),
+            "lastUpdateTime": str(self.lastUpdateTime),
+            "tags": self.tags,
+            "applicationTag": self.applicationTag,
+        }
+
+
+class AssociatedResource(BaseBackend):
+    def __init__(
+        self,
+        resource_type: str,
+        resource: str,
+        options: List[str],
+        application: Application,
+        account_id: str,
+        region_name: str,
+    ):
+        if resource_type == "CFN_STACK":
+            self.resource = resource
+            match = re.search(
+                r"^arn:aws:cloudformation:(us(-gov)?|ap|ca|cn|eu|sa)-(central|(north|south)?(east|west)?)-\d:\d{12}:stack/\w[a-zA-Z0-9\-]{0,127}/[a-f0-9]{8}(-[a-f0-9]{4}){3}-[a-f0-9]{12}$",
+                resource,
+            )
+            if match is not None:
+                self.name = resource.split("/")[1]
+            else:
+                self.name = resource
+            cf_backend = cloudformation_backends[account_id][region_name]
+            try:
+                cf_backend.get_stack(self.name)
+            except ValidationError:
+                raise ResourceNotFoundException(
+                    f"No CloudFormation stack called '{self.name}' found"
+                )
+        elif resource_type == "RESOURCE_TAG_VALUE":
+            tags = {
+                "EnableAWSServiceCatalogAppRegistry": "true",
+                "aws:servicecatalog:applicationName": application.name,
+                "aws:servicecatalog:applicationId": application.id,
+                "aws:servicecatalog:applicationArn": application.arn,
+            }
+            new_resource_group = FakeResourceGroup(
+                account_id,
+                region_name,
+                f"AWS_AppRegistry_AppTag_{account_id}-{application.name}",
+                {"Type": "TAG_FILTERS_1_0", "Query": resource},
+                None,
+                tags,
+            )
+            self.resource = new_resource_group.arn
+            self.name = new_resource_group._name
+            self.query = resource
+        else:
+            raise ValidationException
+        self.resource_type = resource_type
+        self.options = options
+
+    def to_json(self) -> Dict[str, Any]:
+        return_dict = {
+            "name": self.name,
+            "arn": self.resource,
+            "resourceType": self.resource_type,
+            "options": self.options,
+        }
+        if self.resource_type == "RESOURCE_TAG_VALUE":
+            return_dict["resourceDetails"] = {"tagValue": self.query}  # type: ignore
+        return return_dict
+
+
+class AppRegistryBackend(BaseBackend):
+    """Implementation of AppRegistry APIs."""
+
+    def __init__(self, region_name: str, account_id: str):
+        super().__init__(region_name, account_id)
+        self.applications: Dict[str, Application] = dict()
+        self.tagger = TaggingService()
+
+    def create_application(
+        self, name: str, description: str, tags: Dict[str, str], client_token: str
+    ) -> Application:
+        app = Application(
+            name,
+            description,
+            region=self.region_name,
+            account_id=self.account_id,
+        )
+        self.applications[app.arn] = app
+        self._tag_resource(app.arn, tags)
+        return app
+
+    def list_applications(self) -> List[Application]:
+        return list(self.applications.values())
+
+    def associate_resource(
+        self, application: str, resource_type: str, resource: str, options: List[str]
+    ) -> Dict[str, Any]:
+        app = self.applications[application]
+        new_resource = AssociatedResource(
+            resource_type, resource, options, app, self.account_id, self.region_name
+        )
+        app.associated_resources[new_resource.resource] = new_resource
+        return {"applicationArn": app.arn, "resourceArn": resource, "options": options}
+
+    def _list_tags_for_resource(self, arn: str) -> Dict[str, str]:
+        return self.tagger.get_tag_dict_for_resource(arn)
+
+    def _tag_resource(self, arn: str, tags: Dict[str, str]) -> None:
+        self.tagger.tag_resource(arn, TaggingService.convert_dict_to_tags_input(tags))
+
+    def _untag_resource(self, arn: str, tag_keys: List[str]) -> None:
+        self.tagger.untag_resource_using_names(arn, tag_keys)
+
+
+servicecatalogappregistry_backends = BackendDict(
+    AppRegistryBackend, "servicecatalog-appregistry"
+)
diff --git a/moto/servicecatalogappregistry/responses.py b/moto/servicecatalogappregistry/responses.py
new file mode 100644
index 000000000..210238acc
--- /dev/null
+++ b/moto/servicecatalogappregistry/responses.py
@@ -0,0 +1,84 @@
+"""Handles incoming servicecatalogappregistry requests, invokes methods, returns responses."""
+
+import json
+from urllib.parse import unquote
+
+from moto.core.responses import BaseResponse
+from moto.servicecatalogappregistry.exceptions import ValidationException
+
+from .models import (
+    Application,
+    AppRegistryBackend,
+    servicecatalogappregistry_backends,
+)
+
+
+class AppRegistryResponse(BaseResponse):
+    """Handler for AppRegistry requests and responses."""
+
+    def __init__(self) -> None:
+        super().__init__(service_name="servicecatalog-appregistry")
+
+    @property
+    def servicecatalogappregistry_backend(self) -> AppRegistryBackend:
+        """Return backend instance specific for this region."""
+        return servicecatalogappregistry_backends[self.current_account][self.region]
+
+    def create_application(self) -> str:
+        name = self._get_param("name")
+        description = self._get_param("description")
+        tags = self._get_param("tags")
+        client_token = self._get_param("clientToken")
+        application = self.servicecatalogappregistry_backend.create_application(
+            name=name,
+            description=description,
+            tags=tags,
+            client_token=client_token,
+        )
+        return json.dumps({"application": application.to_json()})
+
+    def list_applications(self) -> str:
+        applications = self.servicecatalogappregistry_backend.list_applications()
+        json_list = []
+        for app in applications:
+            json_list.append(app.to_json())
+        return json.dumps({"applications": json_list})
+
+    def associate_resource(self) -> str:
+        application = unquote(self._get_param("application"))
+        resource_type = self._get_param("resourceType")
+        resource = unquote(self._get_param("resource"))
+        options = self._get_param("options")
+        app = None
+        app = self._find_app_by_any_value(application)
+        if options is None:
+            options = []
+        new_resource = self.servicecatalogappregistry_backend.associate_resource(
+            app.arn,
+            resource_type,
+            resource,
+            options,
+        )
+        return json.dumps(new_resource)
+
+    def list_associated_resources(self) -> str:
+        application = unquote(self._get_param("application"))
+        app = self._find_app_by_any_value(application)
+        return_list = []
+        for resource in app.associated_resources.values():
+            return_list.append(resource.to_json())
+        return json.dumps({"resources": return_list})
+
+    def _find_app_by_any_value(self, search: str) -> Application:
+        app = None
+        if search in self.servicecatalogappregistry_backend.applications:
+            app = self.servicecatalogappregistry_backend.applications[search]
+        else:
+            for a in self.servicecatalogappregistry_backend.applications.values():
+                if search == a.id:
+                    app = a
+                elif search == a.name:
+                    app = a
+        if app is None:
+            raise ValidationException
+        return app
diff --git a/moto/servicecatalogappregistry/urls.py b/moto/servicecatalogappregistry/urls.py
new file mode 100644
index 000000000..ab52286c6
--- /dev/null
+++ b/moto/servicecatalogappregistry/urls.py
@@ -0,0 +1,13 @@
+"""servicecatalogappregistry base URL and path."""
+
+from .responses import AppRegistryResponse
+
+url_bases = [
+    r"https?://servicecatalog-appregistry\.(.+)\.amazonaws\.com",
+]
+
+url_paths = {
+    "{0}/applications$": AppRegistryResponse.dispatch,
+    "{0}/applications/(?P<application>[^/]+)/resources/(?P<resource_type>[^/]+)/(?P<resource>.+)$": AppRegistryResponse.dispatch,
+    "{0}/applications/(?P<application>.+)/resources$": AppRegistryResponse.dispatch,
+}
diff --git a/moto/timestreaminfluxdb/models.py b/moto/timestreaminfluxdb/models.py
index ce92e2a29..ce819b151 100644
--- a/moto/timestreaminfluxdb/models.py
+++ b/moto/timestreaminfluxdb/models.py
@@ -1,10 +1,11 @@
 """TimestreamInfluxDBBackend class with methods for supported APIs."""
 
 from enum import Enum
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Tuple
 
 from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
+from moto.utilities.paginator import paginate
 from moto.utilities.tagging_service import TaggingService
 
 from .exceptions import (
@@ -14,6 +15,21 @@ from .exceptions import (
 )
 from .utils import random_id, validate_name
 
+PAGINATION_MODEL = {
+    "list_db_parameter_groups": {
+        "input_token": "next_token",
+        "limit_key": "max_results",
+        "limit_default": 100,
+        "unique_attribute": "id",
+    },
+    "list_db_clusters": {
+        "input_token": "next_token",
+        "limit_key": "max_results",
+        "limit_default": 100,
+        "unique_attribute": "id",
+    },
+}
+
 
 class InstanceStatus(str, Enum):
     CREATING = "CREATING"
@@ -54,6 +70,131 @@ class DeploymentType(str, Enum):
     WITH_MULTIAZ_STANDBY = "WITH_MULTIAZ_STANDBY"
 
 
+class ParameterGroup(BaseModel):
+    def __init__(
+        self,
+        name: str,
+        description: Optional[str] = None,
+        parameters: Optional[Dict[str, Any]] = None,
+        region_name: str = "",
+        account_id: str = "",
+    ):
+        self.id = random_id()
+        self.name = name
+        self.description = description or ""
+        self.parameters = parameters or {}
+        self.arn = f"arn:aws:timestream-influxdb:{region_name}:{account_id}:db-parameter-group/{self.id}"
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "id": self.id,
+            "name": self.name,
+            "arn": self.arn,
+            "description": self.description,
+            "parameters": self.parameters,
+        }
+
+    def to_summary_dict(self) -> Dict[str, str]:
+        return {
+            "id": self.id,
+            "name": self.name,
+            "arn": self.arn,
+            "description": self.description,
+        }
+
+
+class Cluster(BaseModel):
+    def __init__(
+        self,
+        name: str,
+        password: str,
+        username: Optional[str] = None,
+        organization: Optional[str] = None,
+        bucket: Optional[str] = None,
+        port: Optional[int] = None,
+        db_parameter_group_identifier: Optional[str] = None,
+        db_instance_type: Optional[str] = None,
+        db_storage_type: Optional[str] = None,
+        allocated_storage: Optional[int] = None,
+        network_type: Optional[str] = None,
+        publicly_accessible: Optional[bool] = None,
+        vpc_subnet_ids: Optional[List[str]] = None,
+        vpc_security_group_ids: Optional[List[str]] = None,
+        deployment_type: Optional[str] = None,
+        failover_mode: Optional[str] = None,
+        log_delivery_configuration: Optional[Dict[str, Any]] = None,
+        region_name: str = "",
+        account_id: str = "",
+        endpoint_id: str = "",
+    ):
+        self.id = random_id()
+        self.name = name
+        self.password = password
+        self.username = username
+        self.organization = organization
+        self.bucket = bucket
+        self.port = port or 8086
+        self.db_parameter_group_identifier = db_parameter_group_identifier
+        self.db_instance_type = db_instance_type or "db.influx.medium"
+        self.db_storage_type = db_storage_type or DBStorageType.InfluxIOIncludedT1
+        self.allocated_storage = allocated_storage or 100
+        self.network_type = network_type or NetworkType.IPV4
+        self.publicly_accessible = publicly_accessible or False
+        self.vpc_subnet_ids = vpc_subnet_ids or ["subnet-default"]
+        self.vpc_security_group_ids = vpc_security_group_ids or ["sg-default"]
+        self.deployment_type = deployment_type or "MULTI_NODE_READ_REPLICAS"
+        self.failover_mode = failover_mode or "AUTOMATIC"
+        self.log_delivery_configuration = log_delivery_configuration or {}
+        self.arn = f"arn:aws:timestream-influxdb:{region_name}:{account_id}:db-cluster/{self.id}"
+        self.endpoint = (
+            f"{self.id}-{endpoint_id}.timestream-influxdb.{region_name}.on.aws"
+        )
+        self.reader_endpoint = (
+            f"{self.id}-{endpoint_id}.reader.timestream-influxdb.{region_name}.on.aws"
+        )
+        self.influx_auth_parameters_secret_arn = f"arn:aws:secretsmanager:{region_name}:{account_id}:secret:timestream-influxdb/{self.id}/auth-params-{random_id(6)}"
+        self.status = "CREATING"
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "id": self.id,
+            "name": self.name,
+            "arn": self.arn,
+            "status": self.status,
+            "endpoint": self.endpoint,
+            "readerEndpoint": self.reader_endpoint,
+            "port": self.port,
+            "deploymentType": self.deployment_type,
+            "dbInstanceType": self.db_instance_type,
+            "networkType": self.network_type,
+            "dbStorageType": self.db_storage_type,
+            "allocatedStorage": self.allocated_storage,
+            "publiclyAccessible": self.publicly_accessible,
+            "dbParameterGroupIdentifier": self.db_parameter_group_identifier,
+            "logDeliveryConfiguration": self.log_delivery_configuration,
+            "influxAuthParametersSecretArn": self.influx_auth_parameters_secret_arn,
+            "vpcSubnetIds": self.vpc_subnet_ids,
+            "vpcSecurityGroupIds": self.vpc_security_group_ids,
+            "failoverMode": self.failover_mode,
+        }
+
+    def to_summary_dict(self) -> Dict[str, Any]:
+        return {
+            "id": self.id,
+            "name": self.name,
+            "arn": self.arn,
+            "status": self.status,
+            "endpoint": self.endpoint,
+            "readerEndpoint": self.reader_endpoint,
+            "port": self.port,
+            "deploymentType": self.deployment_type,
+            "dbInstanceType": self.db_instance_type,
+            "networkType": self.network_type,
+            "dbStorageType": self.db_storage_type,
+            "allocatedStorage": self.allocated_storage,
+        }
+
+
 class DBInstance(BaseModel):
     def __init__(
         self,
@@ -144,6 +285,8 @@ class TimestreamInfluxDBBackend(BaseBackend):
         # https://docs.aws.amazon.com/timestream/latest/developerguide/timestream-for-influxdb.html
         self.endpoint_id: str = random_id(10)
         self.db_instances: Dict[str, DBInstance] = {}
+        self.db_parameter_groups: Dict[str, ParameterGroup] = {}
+        self.db_clusters: Dict[str, Cluster] = {}
         self.tagger = TaggingService()
 
     def create_db_instance(
@@ -270,6 +413,139 @@ class TimestreamInfluxDBBackend(BaseBackend):
     def list_tags_for_resource(self, resource_arn: str) -> Dict[str, str]:
         return self.tagger.get_tag_dict_for_resource(resource_arn)
 
+    def create_db_parameter_group(
+        self,
+        name: str,
+        description: Optional[str] = None,
+        parameters: Optional[Dict[str, Any]] = None,
+        tags: Optional[Dict[str, str]] = None,
+    ) -> ParameterGroup:
+        validate_name(name)
+
+        for param_group in self.db_parameter_groups.values():
+            if param_group.name == name:
+                raise ConflictException(
+                    f"A DB parameter group with the name {name} already exists"
+                )
+
+        param_group = ParameterGroup(
+            name=name,
+            description=description,
+            parameters=parameters,
+            region_name=self.region_name,
+            account_id=self.account_id,
+        )
+
+        self.db_parameter_groups[param_group.id] = param_group
+
+        if tags:
+            self.tag_resource(param_group.arn, tags)
+
+        return param_group
+
+    def get_db_parameter_group(self, identifier: str) -> ParameterGroup:
+        param_group = self.db_parameter_groups.get(identifier)
+        if not param_group:
+            raise ResourceNotFoundException(
+                f"DB parameter group with identifier {identifier} not found"
+            )
+
+        return param_group
+
+    @paginate(pagination_model=PAGINATION_MODEL)
+    def list_db_parameter_groups(self) -> List[Dict[str, str]]:
+        if not self.db_parameter_groups:
+            return []
+
+        return [
+            param_group.to_summary_dict()
+            for param_group in self.db_parameter_groups.values()
+        ]
+
+    @paginate(pagination_model=PAGINATION_MODEL)
+    def list_db_clusters(self) -> List[Dict[str, object]]:
+        if not self.db_clusters:
+            return []
+
+        return [
+            cluster.to_summary_dict()
+            for cluster in self.db_clusters.values()
+            if cluster.status != "DELETED"
+        ]
+
+    def get_db_cluster(self, db_cluster_id: str) -> Cluster:
+        cluster = self.db_clusters.get(db_cluster_id)
+        if not cluster:
+            raise ResourceNotFoundException(
+                f"DB cluster with ID {db_cluster_id} not found"
+            )
+
+        return cluster
+
+    def create_db_cluster(
+        self,
+        name: str,
+        password: str,
+        username: Optional[str] = None,
+        organization: Optional[str] = None,
+        bucket: Optional[str] = None,
+        port: Optional[int] = None,
+        db_parameter_group_identifier: Optional[str] = None,
+        db_instance_type: Optional[str] = None,
+        db_storage_type: Optional[str] = None,
+        allocated_storage: Optional[int] = None,
+        network_type: Optional[str] = None,
+        publicly_accessible: Optional[bool] = None,
+        vpc_subnet_ids: Optional[List[str]] = None,
+        vpc_security_group_ids: Optional[List[str]] = None,
+        deployment_type: Optional[str] = None,
+        failover_mode: Optional[str] = None,
+        log_delivery_configuration: Optional[Dict[str, Any]] = None,
+        tags: Optional[Dict[str, str]] = None,
+    ) -> Tuple[str, str]:
+        validate_name(name)
+
+        for cluster in self.db_clusters.values():
+            if cluster.name == name:
+                raise ConflictException(
+                    f"A DB cluster with the name {name} already exists"
+                )
+
+        new_cluster = Cluster(
+            name=name,
+            password=password,
+            username=username,
+            organization=organization,
+            bucket=bucket,
+            port=port,
+            db_parameter_group_identifier=db_parameter_group_identifier,
+            db_instance_type=db_instance_type,
+            db_storage_type=db_storage_type,
+            allocated_storage=allocated_storage,
+            network_type=network_type,
+            publicly_accessible=publicly_accessible,
+            vpc_subnet_ids=vpc_subnet_ids,
+            vpc_security_group_ids=vpc_security_group_ids,
+            deployment_type=deployment_type,
+            failover_mode=failover_mode,
+            log_delivery_configuration=log_delivery_configuration,
+            region_name=self.region_name,
+            account_id=self.account_id,
+            endpoint_id=self.endpoint_id,
+        )
+
+        new_cluster.status = "AVAILABLE"
+
+        self.db_clusters[new_cluster.id] = new_cluster
+
+        if tags:
+            self.tag_resource(new_cluster.arn, tags)
+
+        return (
+            new_cluster.id,
+            "AVAILABLE",
+        )
+
 
 timestreaminfluxdb_backends = BackendDict(
     TimestreamInfluxDBBackend,
diff --git a/moto/timestreaminfluxdb/responses.py b/moto/timestreaminfluxdb/responses.py
index b6b9dece9..a8c7b7a2f 100644
--- a/moto/timestreaminfluxdb/responses.py
+++ b/moto/timestreaminfluxdb/responses.py
@@ -103,3 +103,119 @@ class TimestreamInfluxDBResponse(BaseResponse):
         arn = params.get("resourceArn")
         tags = self.timestreaminfluxdb_backend.list_tags_for_resource(resource_arn=arn)
         return json.dumps({"tags": tags})
+
+    def create_db_parameter_group(self) -> str:
+        params = json.loads(self.body)
+        name = params.get("name")
+        description = params.get("description")
+        parameters = params.get("parameters")
+        tags = params.get("tags")
+
+        param_group = self.timestreaminfluxdb_backend.create_db_parameter_group(
+            name=name,
+            description=description,
+            parameters=parameters,
+            tags=tags,
+        )
+
+        return json.dumps(param_group.to_dict())
+
+    def get_db_parameter_group(self) -> str:
+        params = json.loads(self.body)
+        identifier = params.get("identifier")
+
+        param_group = self.timestreaminfluxdb_backend.get_db_parameter_group(
+            identifier=identifier,
+        )
+
+        return json.dumps(param_group.to_dict())
+
+    def list_db_parameter_groups(self) -> str:
+        params = json.loads(self.body)
+        next_token = params.get("nextToken")
+        max_results = params.get("maxResults")
+
+        items, next_token = self.timestreaminfluxdb_backend.list_db_parameter_groups(
+            next_token=next_token,
+            max_results=max_results,
+        )
+
+        response = {"items": items}
+        if next_token:
+            response["nextToken"] = next_token
+
+        return json.dumps(response)
+
+    def list_db_clusters(self) -> str:
+        params = json.loads(self.body)
+        next_token = params.get("nextToken")
+        max_results = params.get("maxResults")
+
+        items, next_token = self.timestreaminfluxdb_backend.list_db_clusters(
+            next_token=next_token,
+            max_results=max_results,
+        )
+
+        response = {"items": items}
+        if next_token:
+            response["nextToken"] = next_token
+
+        return json.dumps(response)
+
+    def get_db_cluster(self) -> str:
+        params = json.loads(self.body)
+        db_cluster_id = params.get("dbClusterId")
+
+        cluster = self.timestreaminfluxdb_backend.get_db_cluster(
+            db_cluster_id=db_cluster_id,
+        )
+
+        return json.dumps(cluster.to_dict())
+
+    def create_db_cluster(self) -> str:
+        params = json.loads(self.body)
+        name = params.get("name")
+        username = params.get("username")
+        password = params.get("password")
+        organization = params.get("organization")
+        bucket = params.get("bucket")
+        port = params.get("port")
+        db_parameter_group_identifier = params.get("dbParameterGroupIdentifier")
+        db_instance_type = params.get("dbInstanceType")
+        db_storage_type = params.get("dbStorageType")
+        allocated_storage = params.get("allocatedStorage")
+        network_type = params.get("networkType")
+        publicly_accessible = params.get("publiclyAccessible")
+        vpc_subnet_ids = params.get("vpcSubnetIds")
+        vpc_security_group_ids = params.get("vpcSecurityGroupIds")
+        deployment_type = params.get("deploymentType")
+        failover_mode = params.get("failoverMode")
+        log_delivery_configuration = params.get("logDeliveryConfiguration")
+        tags = params.get("tags")
+
+        db_cluster_id, db_cluster_status = (
+            self.timestreaminfluxdb_backend.create_db_cluster(
+                name=name,
+                username=username,
+                password=password,
+                organization=organization,
+                bucket=bucket,
+                port=port,
+                db_parameter_group_identifier=db_parameter_group_identifier,
+                db_instance_type=db_instance_type,
+                db_storage_type=db_storage_type,
+                allocated_storage=allocated_storage,
+                network_type=network_type,
+                publicly_accessible=publicly_accessible,
+                vpc_subnet_ids=vpc_subnet_ids,
+                vpc_security_group_ids=vpc_security_group_ids,
+                deployment_type=deployment_type,
+                failover_mode=failover_mode,
+                log_delivery_configuration=log_delivery_configuration,
+                tags=tags,
+            )
+        )
+
+        return json.dumps(
+            {"dbClusterId": db_cluster_id, "dbClusterStatus": db_cluster_status}
+        )
diff --git a/scripts/scaffold.py b/scripts/scaffold.py
index e53354f9c..4b98244c7 100755
--- a/scripts/scaffold.py
+++ b/scripts/scaffold.py
@@ -448,6 +448,10 @@ def get_response_restxml_template(service, operation):
         list(client._service_model._service_description["operations"].keys()),
     )
     op_model = client._service_model.operation_model(aws_operation_name)
+    # Resolves https://github.com/getmoto/moto/issues/8726
+    if not hasattr(op_model.output_shape, "output"):
+        # Response has no output
+        return ""
     result_wrapper = op_model._operation_model["output"]["shape"]
     response_wrapper = result_wrapper.replace("Result", "Response")
     metadata = op_model.metadata


Test patch:
diff --git a/.github/workflows/tests_sdk_ruby.yml b/.github/workflows/tests_sdk_ruby.yml
index 7fcf98d8f..b4c1dc743 100644
--- a/.github/workflows/tests_sdk_ruby.yml
+++ b/.github/workflows/tests_sdk_ruby.yml
@@ -13,7 +13,7 @@ jobs:
     steps:
       - uses: actions/checkout@v4
       - name: Set up Ruby ${{ matrix.ruby-version }}
-        uses: ruby/setup-ruby@e5ac7b085f6e63d49c8973eb0c6e04d876b881f1
+        uses: ruby/setup-ruby@ca041f971d66735f3e5ff1e21cc13e2d51e7e535
         with:
           ruby-version: ${{ matrix.ruby-version }}
       - name: Set up Python 3.12
diff --git a/tests/test_appsync/test_appsync.py b/tests/test_appsync/test_appsync.py
index 7c752c562..e34f62aba 100644
--- a/tests/test_appsync/test_appsync.py
+++ b/tests/test_appsync/test_appsync.py
@@ -69,9 +69,11 @@ def test_create_graphql_api_advanced():
 @mock_aws
 def test_get_graphql_api():
     client = boto3.client("appsync", region_name="ap-southeast-1")
-    api_id = client.create_graphql_api(name="api1", authenticationType="API_KEY")[
-        "graphqlApi"
-    ]["apiId"]
+    api_id = client.create_graphql_api(
+        name="api1",
+        authenticationType="API_KEY",
+        tags={"key1": "value1", "key2": "value2"},
+    )["graphqlApi"]["apiId"]
 
     resp = client.get_graphql_api(apiId=api_id)
     assert "graphqlApi" in resp
@@ -81,6 +83,8 @@ def test_get_graphql_api():
     assert "apiId" in api
     assert api["authenticationType"] == "API_KEY"
     assert api["visibility"] == "GLOBAL"
+    assert "tags" in api
+    assert api["tags"] == {"key1": "value1", "key2": "value2"}
 
 
 @mock_aws
@@ -162,10 +166,14 @@ def test_list_graphql_apis():
     assert resp["graphqlApis"] == []
 
     for _ in range(3):
-        client.create_graphql_api(name="api1", authenticationType="API_KEY")
+        client.create_graphql_api(
+            name="api1", authenticationType="API_KEY", tags={"my_key": "value1"}
+        )
 
     resp = client.list_graphql_apis()
     assert len(resp["graphqlApis"]) == 3
+    assert "tags" in resp["graphqlApis"][0]
+    assert resp["graphqlApis"][0]["tags"] == {"my_key": "value1"}
 
 
 @mock_aws
diff --git a/tests/test_clouddirectory/__init__.py b/tests/test_clouddirectory/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/test_clouddirectory/test_clouddirectory.py b/tests/test_clouddirectory/test_clouddirectory.py
new file mode 100644
index 000000000..9aa08fcc8
--- /dev/null
+++ b/tests/test_clouddirectory/test_clouddirectory.py
@@ -0,0 +1,154 @@
+"""Unit tests for clouddirectory-supported APIs."""
+
+import boto3
+
+from moto import mock_aws
+from tests import DEFAULT_ACCOUNT_ID
+
+
+@mock_aws
+def test_create_directory():
+    region = "us-west-2"
+    client = boto3.client("clouddirectory", region_name=region)
+    schema_arn = (
+        f"arn:aws:clouddirectory:{region}:{DEFAULT_ACCOUNT_ID}:directory/test-schema/1"
+    )
+    resp = client.create_directory(SchemaArn=schema_arn, Name="test-directory")
+    assert (
+        resp["DirectoryArn"]
+        == f"arn:aws:clouddirectory:{region}:{DEFAULT_ACCOUNT_ID}:directory/test-directory"
+    )
+    assert resp["Name"] == "test-directory"
+    assert "ObjectIdentifier" in resp
+
+
+@mock_aws
+def test_list_directories():
+    region = "us-west-2"
+    client = boto3.client("clouddirectory", region_name=region)
+    schema_arn = (
+        f"arn:aws:clouddirectory:{region}:{DEFAULT_ACCOUNT_ID}:directory/test-schema/1"
+    )
+    for i in range(3):
+        client.create_directory(SchemaArn=schema_arn, Name=f"test-directory-{i}")
+
+    resp = client.list_directories()
+    assert len(resp["Directories"]) == 3
+
+    directory = resp["Directories"][0]
+    assert "Name" in directory
+    assert "DirectoryArn" in directory
+    assert "State" in directory
+    assert "CreationDateTime" in directory
+
+    # Test pagination
+    resp = client.list_directories(MaxResults=1)
+    assert len(resp["Directories"]) == 1
+    assert "NextToken" in resp
+    resp = client.list_directories(NextToken=resp["NextToken"])
+    assert len(resp["Directories"]) == 2
+
+    # Test filtering by state
+    resp = client.list_directories(state="ENABLED")
+    assert len(resp["Directories"]) == 3
+    assert resp["Directories"][0]["State"] == "ENABLED"
+
+
+@mock_aws
+def test_tag_resource():
+    region = "us-west-2"
+    client = boto3.client("clouddirectory", region_name=region)
+    schema_arn = (
+        f"arn:aws:clouddirectory:{region}:{DEFAULT_ACCOUNT_ID}:directory/test-schema/1"
+    )
+    directory_arn = client.create_directory(
+        SchemaArn=schema_arn, Name="test-directory"
+    )["DirectoryArn"]
+    client.tag_resource(
+        ResourceArn=directory_arn, Tags=[{"Key": "key1", "Value": "value1"}]
+    )
+
+    # Test that the tag was added
+    resp = client.list_tags_for_resource(ResourceArn=directory_arn)
+    assert len(resp["Tags"]) == 1
+    assert resp["Tags"][0]["Key"] == "key1"
+    assert resp["Tags"][0]["Value"] == "value1"
+
+    # Test resourcetaggingapi
+    rtapi = boto3.client("resourcegroupstaggingapi", region_name=region)
+    resp = rtapi.get_resources(ResourceTypeFilters=["clouddirectory"])
+    assert len(resp["ResourceTagMappingList"]) == 1
+    assert resp["ResourceTagMappingList"][0]["ResourceARN"] == directory_arn
+    assert resp["ResourceTagMappingList"][0]["Tags"][0]["Key"] == "key1"
+    assert resp["ResourceTagMappingList"][0]["Tags"][0]["Value"] == "value1"
+
+
+@mock_aws
+def test_untag_resource():
+    region = "us-east-2"
+    client = boto3.client("clouddirectory", region_name=region)
+    schema_arn = (
+        f"arn:aws:clouddirectory:{region}:{DEFAULT_ACCOUNT_ID}:directory/test-schema/1"
+    )
+    directory_arn = client.create_directory(
+        SchemaArn=schema_arn, Name="test-directory"
+    )["DirectoryArn"]
+    client.tag_resource(
+        ResourceArn=directory_arn, Tags=[{"Key": "key1", "Value": "value1"}]
+    )
+    client.tag_resource(
+        ResourceArn=directory_arn, Tags=[{"Key": "key2", "Value": "value2"}]
+    )
+    client.tag_resource(
+        ResourceArn=directory_arn, Tags=[{"Key": "key3", "Value": "value3"}]
+    )
+    resp = client.list_tags_for_resource(ResourceArn=directory_arn)
+    assert len(resp["Tags"]) == 3
+
+    client.untag_resource(ResourceArn=directory_arn, TagKeys=["key2", "key3"])
+    resp = client.list_tags_for_resource(ResourceArn=directory_arn)
+    assert len(resp["Tags"]) == 1
+    assert resp["Tags"][0]["Key"] == "key1"
+
+
+@mock_aws
+def test_delete_directory():
+    region = "us-west-2"
+    client = boto3.client("clouddirectory", region_name=region)
+    schema_arn = (
+        f"arn:aws:clouddirectory:{region}:{DEFAULT_ACCOUNT_ID}:directory/test-schema/1"
+    )
+    directory_arns = []
+    for i in range(3):
+        resp = client.create_directory(SchemaArn=schema_arn, Name=f"test-directory-{i}")
+        directory_arns.append(resp["DirectoryArn"])
+
+    directories = client.list_directories()["Directories"]
+    assert len(directories) == 3
+
+    resp = client.delete_directory(DirectoryArn=directory_arns[1])
+    assert resp["DirectoryArn"] == directory_arns[1]
+
+    directories = client.list_directories()["Directories"]
+    assert len(directories) == 2
+    for directory in directories:
+        assert directory["DirectoryArn"] != directory_arns[1]
+
+
+@mock_aws
+def test_get_directory():
+    region = "us-west-2"
+    client = boto3.client("clouddirectory", region_name=region)
+    schema_arn = (
+        f"arn:aws:clouddirectory:{region}:{DEFAULT_ACCOUNT_ID}:directory/test-schema/1"
+    )
+    directory_arns = []
+    for i in range(3):
+        resp = client.create_directory(SchemaArn=schema_arn, Name=f"test-directory-{i}")
+        directory_arns.append(resp["DirectoryArn"])
+
+    directory = client.get_directory(DirectoryArn=directory_arns[0])["Directory"]
+    assert "Name" in resp
+    assert directory["DirectoryArn"] == directory_arns[0]
+    assert directory["State"] == "ENABLED"
+    assert "CreationDateTime" in directory
diff --git a/tests/test_comprehend/test_comprehend.py b/tests/test_comprehend/test_comprehend.py
index 890797087..fc1bd372a 100644
--- a/tests/test_comprehend/test_comprehend.py
+++ b/tests/test_comprehend/test_comprehend.py
@@ -23,6 +23,45 @@ INPUT_DATA_CONFIG = {
     "EntityTypes": [{"Type": "ENGINEER"}, {"Type": "MANAGER"}],
 }
 
+OUTPUT_DATA_CONFIG = {
+    "S3Uri": "string",
+    "KmsKeyId": "string",
+    "FlywheelStatsS3Prefix": "string",
+}
+
+DOCUMENT_CLASSIFIER_INPUT_DATA_CONFIG = {
+    "DataFormat": "COMPREHEND_CSV",
+    "S3Uri": "s3://fake-bucket/documents.csv",
+    "TestS3Uri": "s3://fake-bucket/test-documents.csv",
+    "LabelDelimiter": ",",
+    "AugmentedManifests": [
+        {
+            "S3Uri": "s3://fake-bucket/augmented-manifest.json",
+            "Split": "TRAIN",
+            "AttributeNames": [
+                "Attribute1",
+                "Attribute2",
+            ],
+            "AnnotationDataS3Uri": "s3://fake-bucket/annotations.json",
+            "SourceDocumentsS3Uri": "s3://fake-bucket/source-documents/",
+            "DocumentType": "PLAIN_TEXT_DOCUMENT",
+        },
+    ],
+    "DocumentType": "PLAIN_TEXT_DOCUMENT",
+    "Documents": {
+        "S3Uri": "s3://fake-bucket/documents/",
+        "TestS3Uri": "s3://fake-bucket/test-documents/",
+    },
+    "DocumentReaderConfig": {
+        "DocumentReadAction": "TEXTRACT_DETECT_DOCUMENT_TEXT",
+        "DocumentReadMode": "SERVICE_DEFAULT",
+        "FeatureTypes": [
+            "TABLES",
+            "FORMS",
+        ],
+    },
+}
+
 
 @mock_aws
 def test_list_entity_recognizers():
@@ -370,3 +409,469 @@ def test_detect_sentiment_text_too_large():
         == "Input text size exceeds limit. Max length of request text allowed is 100000 bytes while "
         f"in this request the text size is {size} bytes"
     )
+
+
+@mock_aws
+def test_create_document_classifier():
+    client = boto3.client("comprehend", region_name="ap-southeast-1")
+    resp = client.create_document_classifier(
+        DataAccessRoleArn="iam_role_with_20_chars",
+        InputDataConfig=DOCUMENT_CLASSIFIER_INPUT_DATA_CONFIG,
+        LanguageCode="en",
+        DocumentClassifierName="tf-acc-test-1726651689102157637",
+        VersionName="terraform-20221003201727469000000002",
+        OutputDataConfig=OUTPUT_DATA_CONFIG,
+        ClientRequestToken="unique-token",
+        VolumeKmsKeyId="kms-key-id",
+        VpcConfig={"SecurityGroupIds": ["sg-12345678"], "Subnets": ["subnet-12345678"]},
+        Mode="MULTI_CLASS",
+        ModelKmsKeyId="model-kms-key-id",
+        ModelPolicy="model-policy",
+    )
+
+    assert "DocumentClassifierArn" in resp
+
+
+@mock_aws
+def test_create_endpoint():
+    client = boto3.client("comprehend", region_name="ap-southeast-1")
+    resp = client.create_endpoint(
+        EndpointName="tf-acc-test-1726651689102157637",
+        ModelArn="arn:aws:comprehend:ap-southeast-1:123456789012:document-classifier/tf-acc-test-1726651689102157637",
+        DesiredInferenceUnits=123,
+        ClientRequestToken="unique-token",
+        FlywheelArn="arn:aws:comprehend:ap-southeast-1:123456789012:document-classifier/tf-acc-test-1726651689102157637",
+    )
+
+    assert "EndpointArn" in resp
+
+
+@mock_aws
+def test_create_flywheel():
+    client = boto3.client("comprehend", region_name="ap-southeast-1")
+    resp = client.create_flywheel(
+        FlywheelName="test-flywheel",
+        ActiveModelArn="arn:aws:comprehend:ap-southeast-1:123456789012:document-classifier/tf-acc-test-1726651689102157637",
+        DataAccessRoleArn="iam_role_with_20_chars",
+        TaskConfig={
+            "LanguageCode": "en",
+            "DocumentClassificationConfig": {
+                "Mode": "MULTI_CLASS",
+                "Labels": ["Label1", "Label2"],
+            },
+        },
+        ModelType="DOCUMENT_CLASSIFIER",
+        DataLakeS3Uri="s3://tf-acc-test-1726651689102157637/documents.txt",
+        DataSecurityConfig={
+            "ModelKmsKeyId": "model-kms-key-id",
+            "VolumeKmsKeyId": "volume-kms-key-id",
+            "DataLakeKmsKeyId": "data-lake-kms-key-id",
+            "VpcConfig": {
+                "SecurityGroupIds": ["sg-12345678"],
+                "Subnets": ["subnet-12345678"],
+            },
+        },
+        ClientRequestToken="unique-token",
+    )
+
+    assert "FlywheelArn" in resp
+
+
+@mock_aws
+def test_describe_document_classifier():
+    client = boto3.client("comprehend", region_name="eu-west-1")
+    resp = client.create_document_classifier(
+        DataAccessRoleArn="iam_role_with_20_chars",
+        InputDataConfig=DOCUMENT_CLASSIFIER_INPUT_DATA_CONFIG,
+        LanguageCode="en",
+        DocumentClassifierName="tf-acc-test-1726651689102157637",
+        VersionName="terraform-20221003201727469000000002",
+        OutputDataConfig=OUTPUT_DATA_CONFIG,
+        ClientRequestToken="unique-token",
+        VolumeKmsKeyId="kms-key-id",
+        VpcConfig={"SecurityGroupIds": ["sg-12345678"], "Subnets": ["subnet-12345678"]},
+        Mode="MULTI_CLASS",
+        ModelKmsKeyId="model-kms-key-id",
+        ModelPolicy="model-policy",
+    )
+    arn = resp["DocumentClassifierArn"]
+    resp = client.describe_document_classifier(DocumentClassifierArn=arn)
+    assert "DocumentClassifierProperties" in resp
+    props = resp["DocumentClassifierProperties"]
+
+    assert props["DocumentClassifierArn"] == arn
+    assert props["LanguageCode"] == "en"
+    assert props["Status"] == "TRAINING"
+    assert props["InputDataConfig"] == DOCUMENT_CLASSIFIER_INPUT_DATA_CONFIG
+    assert props["DataAccessRoleArn"] == "iam_role_with_20_chars"
+
+
+@mock_aws
+def test_describe_endpoint():
+    client = boto3.client("comprehend", region_name="eu-west-1")
+    arn = client.create_endpoint(
+        ModelArn="arn:aws:comprehend:ap-southeast-1:123456789012:document-classifier/tf-acc-test-1726651689102157637",
+        DesiredInferenceUnits=123,
+        EndpointName="tf-acc-test-1726651689102157637",
+        ClientRequestToken="unique-token",
+        FlywheelArn="arn:aws:comprehend:ap-southeast-1:123456789012:document-classifier/tf-acc-test-1726651689102157637",
+    )["EndpointArn"]
+
+    resp = client.describe_endpoint(EndpointArn=arn)
+    assert "EndpointProperties" in resp
+    props = resp["EndpointProperties"]
+
+    assert props["EndpointArn"] == arn
+    assert (
+        props["ModelArn"]
+        == "arn:aws:comprehend:ap-southeast-1:123456789012:document-classifier/tf-acc-test-1726651689102157637"
+    )
+    assert props["Status"] == "IN_SERVICE"
+    assert props["DesiredInferenceUnits"] == 123
+    assert (
+        props["FlywheelArn"]
+        == "arn:aws:comprehend:ap-southeast-1:123456789012:document-classifier/tf-acc-test-1726651689102157637"
+    )
+
+
+@mock_aws
+def test_describe_flywheel():
+    client = boto3.client("comprehend", region_name="eu-west-1")
+    arn = client.create_flywheel(
+        FlywheelName="test-flywheel",
+        ActiveModelArn="arn:aws:comprehend:ap-southeast-1:123456789012:document-classifier/tf-acc-test-1726651689102157637",
+        DataAccessRoleArn="iam_role_with_20_chars",
+        TaskConfig={
+            "LanguageCode": "en",
+            "DocumentClassificationConfig": {
+                "Mode": "MULTI_CLASS",
+                "Labels": ["Label1", "Label2"],
+            },
+        },
+        ModelType="DOCUMENT_CLASSIFIER",
+        DataLakeS3Uri="s3://tf-acc-test-1726651689102157637/documents.txt",
+        DataSecurityConfig={
+            "ModelKmsKeyId": "model-kms-key-id",
+            "VolumeKmsKeyId": "volume-kms-key-id",
+            "DataLakeKmsKeyId": "data-lake-kms-key-id",
+            "VpcConfig": {
+                "SecurityGroupIds": ["sg-12345678"],
+                "Subnets": ["subnet-12345678"],
+            },
+        },
+        ClientRequestToken="unique-token",
+    )["FlywheelArn"]
+
+    resp = client.describe_flywheel(FlywheelArn=arn)
+    assert "FlywheelProperties" in resp
+    props = resp["FlywheelProperties"]
+    assert props["FlywheelArn"] == arn
+    assert (
+        props["ActiveModelArn"]
+        == "arn:aws:comprehend:ap-southeast-1:123456789012:document-classifier/tf-acc-test-1726651689102157637"
+    )
+    assert props["DataAccessRoleArn"] == "iam_role_with_20_chars"
+    assert props["ModelType"] == "DOCUMENT_CLASSIFIER"
+    assert (
+        props["DataLakeS3Uri"] == "s3://tf-acc-test-1726651689102157637/documents.txt"
+    )
+    assert props["DataSecurityConfig"] == {
+        "ModelKmsKeyId": "model-kms-key-id",
+        "VolumeKmsKeyId": "volume-kms-key-id",
+        "DataLakeKmsKeyId": "data-lake-kms-key-id",
+        "VpcConfig": {
+            "SecurityGroupIds": ["sg-12345678"],
+            "Subnets": ["subnet-12345678"],
+        },
+    }
+
+
+@mock_aws
+def test_delete_document_classifier():
+    client = boto3.client("comprehend", region_name="ap-southeast-1")
+    arn = client.create_document_classifier(
+        DataAccessRoleArn="iam_role_with_20_chars",
+        InputDataConfig=DOCUMENT_CLASSIFIER_INPUT_DATA_CONFIG,
+        LanguageCode="en",
+        DocumentClassifierName="tf-acc-test-1726651689102157637",
+        VersionName="terraform-20221003201727469000000002",
+        OutputDataConfig=OUTPUT_DATA_CONFIG,
+        ClientRequestToken="unique-token",
+        VolumeKmsKeyId="kms-key-id",
+        VpcConfig={"SecurityGroupIds": ["sg-12345678"], "Subnets": ["subnet-12345678"]},
+        Mode="MULTI_CLASS",
+        ModelKmsKeyId="model-kms-key-id",
+        ModelPolicy="model-policy",
+    )["DocumentClassifierArn"]
+
+    client.delete_document_classifier(DocumentClassifierArn=arn)
+
+    with pytest.raises(ClientError) as exc:
+        client.describe_document_classifier(DocumentClassifierArn=arn)
+    err = exc.value.response["Error"]
+    assert err["Code"] == "ResourceNotFoundException"
+    assert err["Message"] == "RESOURCE_NOT_FOUND: Could not find specified resource."
+
+
+@mock_aws
+def test_delete_endpoint():
+    client = boto3.client("comprehend", region_name="ap-southeast-1")
+    arn = client.create_endpoint(
+        ModelArn="arn:aws:comprehend:ap-southeast-1:123456789012:document-classifier/tf-acc-test-1726651689102157637",
+        DesiredInferenceUnits=123,
+        EndpointName="tf-acc-test-1726651689102157637",
+        ClientRequestToken="unique-token",
+        FlywheelArn="arn:aws:comprehend:ap-southeast-1:123456789012:document-classifier/tf-acc-test-1726651689102157637",
+    )["EndpointArn"]
+
+    client.delete_endpoint(EndpointArn=arn)
+
+    with pytest.raises(ClientError) as exc:
+        client.describe_endpoint(EndpointArn=arn)
+    err = exc.value.response["Error"]
+    assert err["Code"] == "ResourceNotFoundException"
+    assert err["Message"] == "RESOURCE_NOT_FOUND: Could not find specified resource."
+
+
+@mock_aws
+def test_delete_flywheel():
+    client = boto3.client("comprehend", region_name="ap-southeast-1")
+    arn = client.create_flywheel(
+        FlywheelName="test-flywheel",
+        ActiveModelArn="arn:aws:comprehend:ap-southeast-1:123456789012:document-classifier/tf-acc-test-1726651689102157637",
+        DataAccessRoleArn="iam_role_with_20_chars",
+        TaskConfig={
+            "LanguageCode": "en",
+            "DocumentClassificationConfig": {
+                "Mode": "MULTI_CLASS",
+                "Labels": ["Label1", "Label2"],
+            },
+        },
+        ModelType="DOCUMENT_CLASSIFIER",
+        DataLakeS3Uri="s3://tf-acc-test-1726651689102157637/documents.txt",
+        DataSecurityConfig={
+            "ModelKmsKeyId": "model-kms-key-id",
+            "VolumeKmsKeyId": "volume-kms-key-id",
+            "DataLakeKmsKeyId": "data-lake-kms-key-id",
+            "VpcConfig": {
+                "SecurityGroupIds": ["sg-12345678"],
+                "Subnets": ["subnet-12345678"],
+            },
+        },
+        ClientRequestToken="unique-token",
+    )["FlywheelArn"]
+
+    client.delete_flywheel(FlywheelArn=arn)
+    with pytest.raises(ClientError) as exc:
+        client.describe_endpoint(EndpointArn=arn)
+    err = exc.value.response["Error"]
+    assert err["Code"] == "ResourceNotFoundException"
+
+
+@mock_aws
+def test_list_document_classifiers():
+    client = boto3.client("comprehend", region_name="us-east-2")
+    resp = client.list_document_classifiers(
+        Filter={"DocumentClassifierName": "unknown"}, MaxResults=10, NextToken="token"
+    )
+    assert resp["DocumentClassifierPropertiesList"] == []
+
+    client.create_document_classifier(
+        DataAccessRoleArn="iam_role_with_20_chars",
+        InputDataConfig=DOCUMENT_CLASSIFIER_INPUT_DATA_CONFIG,
+        LanguageCode="en",
+        DocumentClassifierName="myname",
+        VersionName="version1",
+    )
+
+    resp = client.list_document_classifiers(
+        Filter={"DocumentClassifierName": "myname"}, MaxResults=10, NextToken="token"
+    )
+    assert len(resp["DocumentClassifierPropertiesList"]) == 1
+    assert (
+        resp["DocumentClassifierPropertiesList"][0]["DocumentClassifierArn"]
+        == "arn:aws:comprehend:us-east-2:123456789012:document-classifier/myname/version1"
+    )
+
+    resp = client.list_document_classifiers(
+        Filter={"DocumentClassifierName": "unknown"}, MaxResults=10, NextToken="token"
+    )
+    assert resp["DocumentClassifierPropertiesList"] == []
+
+    resp = client.list_document_classifiers(
+        Filter={"DocumentClassifierName": "myname"}, MaxResults=10, NextToken="token"
+    )
+    assert len(resp["DocumentClassifierPropertiesList"]) == 1
+    client.create_document_classifier(
+        DataAccessRoleArn="iam_role_with_20_chars",
+        InputDataConfig=DOCUMENT_CLASSIFIER_INPUT_DATA_CONFIG,
+        LanguageCode="en",
+        DocumentClassifierName="myname",
+        VersionName="version2",
+    )
+
+    resp = client.list_document_classifiers(
+        Filter={"DocumentClassifierName": "myname"}, MaxResults=10, NextToken="token"
+    )
+
+    assert len(resp["DocumentClassifierPropertiesList"]) == 2
+
+
+@mock_aws
+def test_list_endpoints():
+    client = boto3.client("comprehend", region_name="us-east-2")
+
+    resp = client.list_endpoints(
+        Filter={"Status": "IN_SERVICE"}, MaxResults=10, NextToken="token"
+    )
+    assert resp["EndpointPropertiesList"] == []
+
+    client.create_endpoint(
+        ModelArn="arn:aws:comprehend:ap-southeast-1:123456789012:document-classifier/tf-acc-test-1726651689102157637",
+        DesiredInferenceUnits=123,
+        EndpointName="myname",
+        ClientRequestToken="unique-token",
+        FlywheelArn="arn:aws:comprehend:ap-southeast-1:123456789012:document-classifier/tf-acc-test-1726651689102157637",
+    )
+
+    resp = client.list_endpoints(
+        Filter={"Status": "FAILED"}, MaxResults=10, NextToken="token"
+    )
+    assert resp["EndpointPropertiesList"] == []
+
+    resp = client.list_endpoints(
+        Filter={"Status": "IN_SERVICE"}, MaxResults=10, NextToken="token"
+    )
+    assert len(resp["EndpointPropertiesList"]) == 1
+
+    client.create_endpoint(
+        ModelArn="arn:aws:comprehend:ap-southeast-1:123456789012:document-classifier/tf-acc-test-17266516891021576372",
+        DesiredInferenceUnits=123,
+        EndpointName="myname",
+        ClientRequestToken="unique-token",
+        FlywheelArn="arn:aws:comprehend:ap-southeast-1:123456789012:document-classifier/tf-acc-test-17266516891021576372",
+    )
+
+    resp = client.list_endpoints(
+        Filter={"Status": "IN_SERVICE"}, MaxResults=10, NextToken="token"
+    )
+    assert len(resp["EndpointPropertiesList"]) == 2
+
+
+@mock_aws
+def test_list_flywheels():
+    client = boto3.client("comprehend", region_name="us-east-2")
+
+    resp = client.list_flywheels(
+        Filter={"Status": "ACTIVE"}, MaxResults=10, NextToken="token"
+    )
+    assert resp["FlywheelSummaryList"] == []
+
+    client.create_flywheel(
+        FlywheelName="test-flywheel",
+        ActiveModelArn="arn:aws:comprehend:ap-southeast-1:123456789012:document-classifier/tf-acc-test-1726651689102157637",
+        DataAccessRoleArn="iam_role_with_20_chars",
+        TaskConfig={
+            "LanguageCode": "en",
+            "DocumentClassificationConfig": {
+                "Mode": "MULTI_CLASS",
+                "Labels": ["Label1", "Label2"],
+            },
+        },
+        ModelType="DOCUMENT_CLASSIFIER",
+        DataLakeS3Uri="s3://tf-acc-test-1726651689102157637/documents.txt",
+        DataSecurityConfig={
+            "ModelKmsKeyId": "model-kms-key-id",
+            "VolumeKmsKeyId": "volume-kms-key-id",
+            "DataLakeKmsKeyId": "data-lake-kms-key-id",
+            "VpcConfig": {
+                "SecurityGroupIds": ["sg-12345678"],
+                "Subnets": ["subnet-12345678"],
+            },
+        },
+        ClientRequestToken="unique-token",
+    )
+
+    resp = client.list_flywheels(
+        Filter={"Status": "CREATING"}, MaxResults=10, NextToken="token"
+    )
+    assert resp["FlywheelSummaryList"] == []
+
+    resp = client.list_flywheels(
+        Filter={"Status": "ACTIVE"}, MaxResults=10, NextToken="token"
+    )
+    assert len(resp["FlywheelSummaryList"]) == 1
+
+
+@mock_aws
+def test_stop_training_document_classifier():
+    client = boto3.client("comprehend", region_name="eu-west-1")
+    arn = client.create_document_classifier(
+        DataAccessRoleArn="iam_role_with_20_chars",
+        InputDataConfig=DOCUMENT_CLASSIFIER_INPUT_DATA_CONFIG,
+        LanguageCode="en",
+        DocumentClassifierName="myname",
+        VersionName="version1",
+    )["DocumentClassifierArn"]
+    client.stop_training_document_classifier(DocumentClassifierArn=arn)
+
+    props = client.describe_document_classifier(DocumentClassifierArn=arn)[
+        "DocumentClassifierProperties"
+    ]
+    assert props["Status"] == "STOP_REQUESTED"
+
+
+@mock_aws
+def test_start_flywheel_iteration():
+    client = boto3.client("comprehend", region_name="ap-southeast-1")
+    arn = client.create_flywheel(
+        FlywheelName="test-flywheel",
+        ActiveModelArn="arn:aws:comprehend:ap-southeast-1:123456789012:document-classifier/tf-acc-test-1726651689102157637",
+        DataAccessRoleArn="iam_role_with_20_chars",
+        TaskConfig={
+            "LanguageCode": "en",
+            "DocumentClassificationConfig": {
+                "Mode": "MULTI_CLASS",
+                "Labels": ["Label1", "Label2"],
+            },
+        },
+        ModelType="DOCUMENT_CLASSIFIER",
+        DataLakeS3Uri="s3://tf-acc-test-1726651689102157637/documents.txt",
+        DataSecurityConfig={
+            "ModelKmsKeyId": "model-kms-key-id",
+            "VolumeKmsKeyId": "volume-kms-key-id",
+            "DataLakeKmsKeyId": "data-lake-kms-key-id",
+            "VpcConfig": {
+                "SecurityGroupIds": ["sg-12345678"],
+                "Subnets": ["subnet-12345678"],
+            },
+        },
+        ClientRequestToken="unique-token",
+    )["FlywheelArn"]
+
+    resp = client.start_flywheel_iteration(FlywheelArn=arn)
+    assert "FlywheelIterationId" in resp
+
+
+@mock_aws
+def test_update_endpoint():
+    client = boto3.client("comprehend", region_name="ap-southeast-1")
+    arn = client.create_endpoint(
+        ModelArn="arn:aws:comprehend:ap-southeast-1:123456789012:document-classifier/tf-acc-test-1726651689102157637",
+        DesiredInferenceUnits=123,
+        EndpointName="tf-acc-test-1726651689102157637",
+        ClientRequestToken="unique-token",
+        DataAccessRoleArn="iam_role_with_20_chars",
+        FlywheelArn="arn:aws:comprehend:ap-southeast-1:123456789012:document-classifier/tf-acc-test-1726651689102157637",
+    )["EndpointArn"]
+
+    resp = client.update_endpoint(
+        EndpointArn=arn,
+        DesiredInferenceUnits=456,
+        DesiredModelArn="arn:aws:comprehend:ap-southeast-1:123456789012:document-classifier/tf-acc-test-1726651689102157637",
+    )
+
+    assert (
+        resp["DesiredModelArn"]
+        == "arn:aws:comprehend:ap-southeast-1:123456789012:document-classifier/tf-acc-test-1726651689102157637"
+    )
diff --git a/tests/test_core/test_auth.py b/tests/test_core/test_auth.py
index 7b9690dad..956c9b8e8 100644
--- a/tests/test_core/test_auth.py
+++ b/tests/test_core/test_auth.py
@@ -1166,3 +1166,60 @@ def test_perform_role_based_action() -> None:
         attached_policy_name
         in iam_client.list_role_policies(RoleName=role_name)["PolicyNames"]
     )
+
+
+@set_initial_no_auth_action_count(4)
+@mock_aws
+def test_sts_assume_role_with_external_id_unsupported_operation_should_supress_error() -> (
+    None
+):
+    user_name = "test-user"
+    role_name = "test-role"
+
+    user_policy_doc = {
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Effect": "Allow",
+                "Action": "sts:AssumeRole",
+                "Resource": f"arn:aws:iam::{ACCOUNT_ID}:role/{role_name}",
+            }
+        ],
+    }
+
+    user_access_keys = create_user_with_access_key_and_inline_policy(
+        user_name, user_policy_doc
+    )
+
+    external_id = "test-external-id"
+    trust_policy_document = {
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Effect": "Allow",
+                "Action": "sts:AssumeRole",
+                "Principal": {
+                    "AWS": f"arn:aws:iam::{ACCOUNT_ID}:user/{user_name}",
+                },
+                "Condition": {
+                    "UnsupportedOperation": {
+                        "sts:ExternalId": external_id,
+                    },
+                },
+            },
+        ],
+    }
+    create_role(role_name, trust_policy_document)
+
+    client = boto3.client(
+        "sts",
+        region_name="us-east-1",
+        aws_access_key_id=user_access_keys["AccessKeyId"],
+        aws_secret_access_key=user_access_keys["SecretAccessKey"],
+    )
+    # Not raising means success
+    client.assume_role(
+        RoleArn=f"arn:aws:iam::{ACCOUNT_ID}:role/{role_name}",
+        RoleSessionName="test-session",
+        ExternalId=external_id,
+    )
diff --git a/tests/test_dms/test_dms.py b/tests/test_dms/test_dms.py
index d11d3ff91..597c759b5 100644
--- a/tests/test_dms/test_dms.py
+++ b/tests/test_dms/test_dms.py
@@ -285,3 +285,158 @@ def test_describe_replication_instances():
     instances = response["ReplicationInstances"]
     assert len(instances) == 1
     assert instances[0]["ReplicationInstanceIdentifier"] == "test-instance-2"
+
+
+@mock_aws
+def test_create_endpoint():
+    region_name = "ap-southeast-1"
+    client = boto3.client("dms", region_name=region_name)
+
+    response = client.create_endpoint(
+        EndpointIdentifier="test-endpoint",
+        EndpointType="source",
+        EngineName="mysql",
+        Username="admin",
+        ServerName="test-server",
+        Port=3306,
+        DatabaseName="test-db",
+        ExtraConnectionAttributes="",
+        KmsKeyId="test-kms-key-id",
+        CertificateArn="test-certificate-arn",
+        SslMode="require",
+        ServiceAccessRoleArn="test-service-role-arn",
+        ExternalTableDefinition="",
+    )
+
+    endpoint = response["Endpoint"]
+
+    # Assertions for the required fields
+    assert endpoint["EndpointIdentifier"] == "test-endpoint"
+    assert endpoint["EndpointType"] == "source"
+    assert endpoint["EngineName"] == "mysql"
+    assert endpoint["Username"] == "admin"
+    assert endpoint["ServerName"] == "test-server"
+    assert endpoint["Port"] == 3306
+    assert endpoint["DatabaseName"] == "test-db"
+    assert endpoint["Status"] == "creating"
+    assert endpoint["KmsKeyId"] == "test-kms-key-id"
+    assert (
+        f"arn:aws:dms:{region_name}:123456789012:endpoint:" in endpoint["EndpointArn"]
+    )
+    assert endpoint["SslMode"] == "require"
+
+
+@mock_aws
+def test_create_endpoint_resource_identifier():
+    client = boto3.client("dms", region_name="ap-southeast-1")
+
+    resource_identifier = "test-resource-identifier"
+    response = client.create_endpoint(
+        EndpointIdentifier="test-endpoint",
+        EndpointType="source",
+        EngineName="mysql",
+        ResourceIdentifier=resource_identifier,
+    )
+
+    endpoint = response["Endpoint"]
+
+    # Assertions for the required fields
+    assert endpoint["EndpointIdentifier"] == "test-endpoint"
+    assert endpoint["EndpointType"] == "source"
+    assert endpoint["EngineName"] == "mysql"
+    assert resource_identifier == endpoint["EndpointArn"].split(":")[-1]
+
+
+@mock_aws
+def test_describe_endpoints():
+    client = boto3.client("dms", region_name="ap-southeast-1")
+    num_endpoints = 3
+    for i in range(num_endpoints):
+        client.create_endpoint(
+            EndpointIdentifier=f"test-endpoint-{i}",
+            EndpointType="source",
+            EngineName="mysql",
+        )
+
+    resp = client.describe_endpoints()
+    assert len(resp["Endpoints"]) == num_endpoints
+
+
+@mock_aws
+def test_describe_endpoints_filter():
+    client = boto3.client("dms", region_name="ap-southeast-1")
+
+    for i in range(3):
+        client.create_endpoint(
+            EndpointIdentifier=f"test-endpoint-{i}",
+            EndpointType="source",
+            EngineName="mysql",
+        )
+
+    endpoint_filter = {"Name": "endpoint-id", "Values": ["test-endpoint-1"]}
+    resp = client.describe_endpoints(Filters=[endpoint_filter])
+    assert len(resp["Endpoints"]) == 1
+
+    engine_filter = {"Name": "engine-name", "Values": ["mysql"]}
+    resp = client.describe_endpoints(Filters=[engine_filter])
+    assert len(resp["Endpoints"]) == 3
+
+
+@mock_aws
+def test_list_tags_for_resource_replication_instance():
+    client = boto3.client("dms", region_name="eu-west-1")
+
+    response = client.create_replication_instance(
+        ReplicationInstanceIdentifier="test",
+        ReplicationInstanceClass="dms.c4.large",
+        Tags=[{"Key": "Environment", "Value": "Production"}],
+    )
+
+    arn = response["ReplicationInstance"]["ReplicationInstanceArn"]
+
+    resp = client.list_tags_for_resource(ResourceArn=arn)
+    assert resp["TagList"] == [
+        {"Key": "Environment", "Value": "Production", "ResourceArn": arn}
+    ]
+
+
+@mock_aws
+def test_list_tags_for_resource_endpoint():
+    client = boto3.client("dms", region_name="eu-west-1")
+
+    response = client.create_endpoint(
+        EndpointIdentifier="test-endpoint",
+        EndpointType="source",
+        EngineName="mysql",
+        Tags=[{"Key": "Name", "Value": "Test Endpoint"}],
+    )
+    arn = response["Endpoint"]["EndpointArn"]
+
+    resp = client.list_tags_for_resource(ResourceArn=arn)
+    assert resp["TagList"] == [
+        {"Key": "Name", "Value": "Test Endpoint", "ResourceArn": arn}
+    ]
+
+
+@mock_aws
+def test_list_tags_for_resource_endpoints():
+    client = boto3.client("dms", region_name="eu-west-1")
+
+    response = client.create_endpoint(
+        EndpointIdentifier="test-endpoint",
+        EndpointType="source",
+        EngineName="mysql",
+        Tags=[{"Key": "Name", "Value": "Test Endpoint"}],
+    )
+    endpoint_arn1 = response["Endpoint"]["EndpointArn"]
+
+    response = client.create_endpoint(
+        EndpointIdentifier="test-endpoint",
+        EndpointType="source",
+        EngineName="mysql",
+        Tags=[{"Key": "Name", "Value": "Test Endpoint"}],
+    )
+    endpoint_arn2 = response["Endpoint"]["EndpointArn"]
+
+    resp = client.list_tags_for_resource(ResourceArnList=[endpoint_arn1, endpoint_arn2])
+    assert len(resp["TagList"]) == 2
diff --git a/tests/test_ec2/test_instances.py b/tests/test_ec2/test_instances.py
index 006f9a2f6..c704585d8 100644
--- a/tests/test_ec2/test_instances.py
+++ b/tests/test_ec2/test_instances.py
@@ -1979,6 +1979,56 @@ def test_describe_instance_status_with_instance_filter():
     for _id in running_instance_ids:
         assert _id not in found_instance_ids
 
+    # Filter instance using the state code and state name
+    state_code_filter = {
+        "running_and_stopped": [
+            {"Name": "instance-state-code", "Values": ["16", "80"]},
+            {"Name": "instance-state-name", "Values": ["running", "stopped"]},
+        ],
+        "running": [
+            {"Name": "instance-state-code", "Values": ["16"]},
+            {"Name": "instance-state-name", "Values": ["running"]},
+        ],
+        "stopped": [
+            {"Name": "instance-state-code", "Values": ["80"]},
+            {"Name": "instance-state-name", "Values": ["stopped"]},
+        ],
+        "contradiction": [
+            {"Name": "instance-state-code", "Values": ["80"]},
+            {"Name": "instance-state-name", "Values": ["running"]},
+        ],
+    }
+
+    found_statuses = conn.describe_instance_status(
+        IncludeAllInstances=True, Filters=state_code_filter["running_and_stopped"]
+    )["InstanceStatuses"]
+    found_instance_ids = [status["InstanceId"] for status in found_statuses]
+    for _id in all_instance_ids:
+        assert _id in found_instance_ids
+
+    found_statuses = conn.describe_instance_status(
+        IncludeAllInstances=True, Filters=state_code_filter["running"]
+    )["InstanceStatuses"]
+    found_instance_ids = [status["InstanceId"] for status in found_statuses]
+    for _id in stopped_instance_ids:
+        assert _id not in found_instance_ids
+    for _id in running_instance_ids:
+        assert _id in found_instance_ids
+
+    found_statuses = conn.describe_instance_status(
+        IncludeAllInstances=True, Filters=state_code_filter["stopped"]
+    )["InstanceStatuses"]
+    found_instance_ids = [status["InstanceId"] for status in found_statuses]
+    for _id in stopped_instance_ids:
+        assert _id in found_instance_ids
+    for _id in running_instance_ids:
+        assert _id not in found_instance_ids
+
+    found_statuses = conn.describe_instance_status(
+        IncludeAllInstances=True, Filters=state_code_filter["contradiction"]
+    )["InstanceStatuses"]
+    assert len(found_statuses) == 0
+
 
 @mock_aws
 def test_describe_instance_status_with_non_running_instances():
@@ -2836,3 +2886,75 @@ def test_instance_with_ipv6_address():
 
     assert len(instance["NetworkInterfaces"]) == 1
     assert len(instance["NetworkInterfaces"][0]["Ipv6Addresses"]) == 1
+
+
+@mock_aws
+def test_instance_without_default_subnet():
+    # Ensure that in an account without a default subnet, launching instances without specifying a subnet raises.
+    client = boto3.client("ec2", region_name="ap-south-1")
+
+    # Delete all VPCs and subnets
+    subnets = client.describe_subnets()
+    for subnet in subnets["Subnets"]:
+        client.delete_subnet(SubnetId=subnet["SubnetId"])
+    vpcs = client.describe_vpcs()
+    for vpc in vpcs["Vpcs"]:
+        client.delete_vpc(VpcId=vpc["VpcId"])
+
+    # RunInstances must raise
+    with pytest.raises(ClientError) as exc:
+        client.run_instances(ImageId=EXAMPLE_AMI_ID, MinCount=1, MaxCount=1)
+    assert exc.value.response["Error"]["Code"] == "VPCIdNotSpecified"
+    assert (
+        exc.value.response["Error"]["Message"]
+        == "No default VPC for this user. GroupName is only supported for EC2-Classic and default VPC."
+    )
+
+    # Create default VPC but without default subnets
+    default_vpc_id = client.create_default_vpc()["Vpc"]["VpcId"]
+    subnets = client.describe_subnets()
+    for subnet in subnets["Subnets"]:
+        client.delete_subnet(SubnetId=subnet["SubnetId"])
+
+    # Ensure RunInstances raises
+    with pytest.raises(ClientError) as exc:
+        client.run_instances(ImageId=EXAMPLE_AMI_ID, MinCount=1, MaxCount=1)
+    assert exc.value.response["Error"]["Code"] == "MissingInput"
+    assert (
+        exc.value.response["Error"]["Message"]
+        == f"No subnets found for the default VPC '{default_vpc_id}'. Please specify a subnet."
+    )
+
+    # Create default subnet in an AZ
+    client.create_default_subnet(AvailabilityZone="ap-south-1c")
+
+    # Ensure RunInstances without a subnet, in a different AZ raises
+    with pytest.raises(ClientError) as exc:
+        client.run_instances(
+            ImageId=EXAMPLE_AMI_ID,
+            MinCount=1,
+            MaxCount=1,
+            Placement={"AvailabilityZone": "ap-south-1b"},
+        )
+    assert exc.value.response["Error"]["Code"] == "InvalidInput"
+    assert (
+        exc.value.response["Error"]["Message"]
+        == "No default subnet for availability zone: 'ap-south-1b'."
+    )
+
+    # Ensure RunInstances without a subnet where a default subnet exists succeeds
+    client.run_instances(
+        ImageId=EXAMPLE_AMI_ID,
+        MinCount=1,
+        MaxCount=1,
+        Placement={"AvailabilityZone": "ap-south-1c"},
+    )
+
+    # Create another VPC and subnet
+    vpc = client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]
+    subnet = client.create_subnet(VpcId=vpc["VpcId"], CidrBlock="10.0.1.0/24")["Subnet"]
+
+    # Ensure RunInstances with a subnet specified succeeds
+    client.run_instances(
+        ImageId=EXAMPLE_AMI_ID, MinCount=1, MaxCount=1, SubnetId=subnet["SubnetId"]
+    )
diff --git a/tests/test_networkfirewall/__init__.py b/tests/test_networkfirewall/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/test_networkfirewall/test_networkfirewall.py b/tests/test_networkfirewall/test_networkfirewall.py
new file mode 100644
index 000000000..acebcdcc3
--- /dev/null
+++ b/tests/test_networkfirewall/test_networkfirewall.py
@@ -0,0 +1,147 @@
+"""Unit tests for networkfirewall-supported APIs."""
+
+import boto3
+
+from moto import mock_aws
+
+
+@mock_aws
+def test_create_firewall():
+    client = boto3.client("network-firewall", region_name="us-east-1")
+    firewall = client.create_firewall(
+        FirewallName="test-firewall",
+        FirewallPolicyArn="arn:aws:network-firewall:ap-southeast-1:123456789012:firewall-policy/test-policy",
+        DeleteProtection=False,
+        SubnetChangeProtection=False,
+    )["Firewall"]
+
+    assert firewall["FirewallName"] == "test-firewall"
+    assert "FirewallArn" in firewall
+    assert firewall["DeleteProtection"] is False
+    assert firewall["SubnetChangeProtection"] is False
+    assert firewall["FirewallPolicyChangeProtection"] is True
+
+
+@mock_aws
+def test_describe_logging_configuration():
+    client = boto3.client("network-firewall", region_name="eu-west-1")
+    firewall = client.create_firewall(
+        FirewallName="test-firewall",
+        FirewallPolicyArn="arn:aws:network-firewall:ap-southeast-1:123456789012:firewall-policy/test-policy",
+    )["Firewall"]
+
+    logging_config = {
+        "LogDestinationConfigs": [
+            {
+                "LogDestinationType": "S3",
+                "LogDestination": {
+                    "bucketName": "DOC-EXAMPLE-BUCKET",
+                    "prefix": "alerts",
+                },
+                "LogType": "FLOW",
+            },
+            {
+                "LogDestinationType": "CloudWatchLogs",
+                "LogDestination": {"logGroup": "alert-log-group"},
+                "LogType": "ALERT",
+            },
+        ]
+    }
+
+    # Create a logging configuration
+    client.update_logging_configuration(
+        FirewallArn=firewall["FirewallArn"], LoggingConfiguration=logging_config
+    )
+
+    # Describe the logging configuration
+    resp = client.describe_logging_configuration(FirewallArn=firewall["FirewallArn"])
+    assert resp["FirewallArn"] == firewall["FirewallArn"]
+    assert len(resp["LoggingConfiguration"]["LogDestinationConfigs"]) == 2
+    log_dest_configs = resp["LoggingConfiguration"]["LogDestinationConfigs"]
+    assert log_dest_configs[0]["LogDestinationType"] == "S3"
+    assert log_dest_configs[0]["LogType"] == "FLOW"
+    assert log_dest_configs[1]["LogDestinationType"] == "CloudWatchLogs"
+    assert log_dest_configs[1]["LogType"] == "ALERT"
+
+
+@mock_aws
+def test_update_logging_configuration():
+    client = boto3.client("network-firewall", region_name="ap-southeast-1")
+    firewall = client.create_firewall(
+        FirewallName="test-firewall",
+        FirewallPolicyArn="arn:aws:network-firewall:ap-southeast-1:123456789012:firewall-policy/test-policy",
+    )["Firewall"]
+
+    logging_config = {
+        "LogDestinationConfigs": [
+            {
+                "LogDestinationType": "S3",
+                "LogDestination": {
+                    "bucketName": "DOC-EXAMPLE-BUCKET",
+                    "prefix": "alerts",
+                },
+                "LogType": "FLOW",
+            }
+        ]
+    }
+
+    resp = client.update_logging_configuration(
+        FirewallArn=firewall["FirewallArn"], LoggingConfiguration=logging_config
+    )
+    assert resp["FirewallArn"] == firewall["FirewallArn"]
+    assert resp["FirewallName"] == "test-firewall"
+    assert len(resp["LoggingConfiguration"]["LogDestinationConfigs"]) == 1
+    assert resp["LoggingConfiguration"] == logging_config
+
+
+@mock_aws
+def test_list_firewalls():
+    client = boto3.client("network-firewall", region_name="ap-southeast-1")
+    for i in range(5):
+        client.create_firewall(
+            FirewallName=f"test-firewall-{i}",
+            FirewallPolicyArn="arn:aws:network-firewall:ap-southeast-1:123456789012:firewall-policy/test-policy",
+            VpcId=f"vpc-1234567{i}",
+        )
+
+    # List all firewalls
+    resp = client.list_firewalls()
+    assert len(resp["Firewalls"]) == 5
+    assert resp["Firewalls"][0]["FirewallName"] == "test-firewall-0"
+    assert "FirewallArn" in resp["Firewalls"][0]
+
+    # List firewalls with a specific VPC ID
+    resp = client.list_firewalls(VpcIds=["vpc-12345671"])
+    assert len(resp["Firewalls"]) == 1
+    assert resp["Firewalls"][0]["FirewallName"] == "test-firewall-1"
+
+
+@mock_aws
+def test_describe_firewall():
+    client = boto3.client("network-firewall", region_name="ap-southeast-1")
+    firewall = client.create_firewall(
+        FirewallName="test-firewall",
+        FirewallPolicyArn="arn:aws:network-firewall:ap-southeast-1:123456789012:firewall-policy/test-policy",
+        VpcId="vpc-12345678",
+        SubnetMappings=[{"SubnetId": "subnet-12345678"}],
+        DeleteProtection=False,
+        SubnetChangeProtection=False,
+        FirewallPolicyChangeProtection=False,
+        Description="Test firewall",
+        Tags=[{"Key": "Name", "Value": "test-firewall"}],
+    )["Firewall"]
+
+    # Describe the firewall using the ARN
+    resp = client.describe_firewall(FirewallArn=firewall["FirewallArn"])
+    assert resp["Firewall"]["FirewallName"] == "test-firewall"
+    assert resp["Firewall"]["VpcId"] == "vpc-12345678"
+    assert resp["Firewall"]["SubnetMappings"] == [{"SubnetId": "subnet-12345678"}]
+    assert resp["Firewall"]["DeleteProtection"] is False
+    assert resp["Firewall"]["SubnetChangeProtection"] is False
+    assert resp["Firewall"]["FirewallPolicyChangeProtection"] is False
+    assert resp["Firewall"]["Description"] == "Test firewall"
+    assert resp["Firewall"]["Tags"] == [{"Key": "Name", "Value": "test-firewall"}]
+
+    # Describe the firewall using the name
+    resp_name = client.describe_firewall(FirewallName="test-firewall")
+    assert resp_name["Firewall"]["FirewallName"] == "test-firewall"
diff --git a/tests/test_organizations/test_organizations_boto3.py b/tests/test_organizations/test_organizations_boto3.py
index a860fe722..3170e01ea 100644
--- a/tests/test_organizations/test_organizations_boto3.py
+++ b/tests/test_organizations/test_organizations_boto3.py
@@ -2369,3 +2369,52 @@ def test_aiservices_opt_out_policy():
     )
     assert len(response["Policies"]) == 1
     assert response["Policies"][0]["Id"] == policy_id
+
+
+@mock_aws
+def test_tag_policy():
+    # given
+    client = boto3.client("organizations", region_name="us-east-1")
+    org = client.create_organization(FeatureSet="ALL")["Organization"]
+    root_id = client.list_roots()["Roots"][0]["Id"]
+    client.enable_policy_type(RootId=root_id, PolicyType="TAG_POLICY")
+    tag_policy = {
+        "tags": {
+            "CostCenter": {
+                "tag_key": {"@@assign": "CostCenter"},
+                "tag_value": {"@@assign": ["100", "200", "300"]},
+                "enforced_for": {"@@assign": ["ec2:instance", "s3:bucket"]},
+            }
+        }
+    }
+
+    # when
+    response = client.create_policy(
+        Content=json.dumps(tag_policy),
+        Description="Test tag policy",
+        Name="test-tag-policy",
+        Type="TAG_POLICY",
+    )
+
+    # then
+    summary = response["Policy"]["PolicySummary"]
+    policy_id = summary["Id"]
+    assert re.match(utils.POLICY_ID_REGEX, summary["Id"])
+    assert summary["Arn"] == (
+        utils.TAG_POLICY_ARN_FORMAT.format(
+            "aws", org["MasterAccountId"], org["Id"], summary["Id"]
+        )
+    )
+    assert summary["Name"] == "test-tag-policy"
+    assert summary["Description"] == "Test tag policy"
+    assert summary["Type"] == "TAG_POLICY"
+    assert summary["AwsManaged"] is False
+    assert json.loads(response["Policy"]["Content"]) == tag_policy
+
+    # when
+    client.attach_policy(PolicyId=policy_id, TargetId=root_id)
+
+    # then
+    response = client.list_policies_for_target(TargetId=root_id, Filter="TAG_POLICY")
+    assert len(response["Policies"]) == 1
+    assert response["Policies"][0]["Id"] == policy_id
diff --git a/tests/test_resourcegroupstaggingapi/test_resourcegroupstaggingapi.py b/tests/test_resourcegroupstaggingapi/test_resourcegroupstaggingapi.py
index aff2e46d3..c6e938616 100644
--- a/tests/test_resourcegroupstaggingapi/test_resourcegroupstaggingapi.py
+++ b/tests/test_resourcegroupstaggingapi/test_resourcegroupstaggingapi.py
@@ -135,6 +135,48 @@ def test_get_resources_backup():
     assert {"Key": "Test", "Value": "1"} in resp["ResourceTagMappingList"][0]["Tags"]
 
 
+@mock_aws
+def test_get_resources_dms_endpoint():
+    client = boto3.client("dms", region_name="us-east-1")
+    endpoint = client.create_endpoint(
+        EndpointIdentifier="test-endpoint",
+        EndpointType="source",
+        EngineName="mysql",
+        ResourceIdentifier="sample_identifier",
+        Tags=[{"Key": "tag", "Value": "a tag"}],
+    )
+    endpoint_arn = endpoint["Endpoint"]["EndpointArn"]
+    rgta_client = boto3.client("resourcegroupstaggingapi", region_name="us-east-1")
+    resp = rgta_client.get_resources(
+        ResourceTypeFilters=["dms:endpoint"],
+        TagFilters=[{"Key": "tag", "Values": ["a tag"]}],
+    )
+    assert len(resp["ResourceTagMappingList"]) == 1
+    assert endpoint_arn == resp["ResourceTagMappingList"][0]["ResourceARN"]
+
+
+@mock_aws
+def test_get_resources_dms_replication_instance():
+    client = boto3.client("dms", region_name="us-east-1")
+    replication_instance = client.create_replication_instance(
+        ReplicationInstanceIdentifier="test-instance-1",
+        ReplicationInstanceClass="dms.t2.micro",
+        EngineVersion="3.4.5",
+        Tags=[{"Key": "tag", "Value": "a tag"}],
+    )
+
+    replication_instance_arn = replication_instance["ReplicationInstance"][
+        "ReplicationInstanceArn"
+    ]
+    rgta_client = boto3.client("resourcegroupstaggingapi", region_name="us-east-1")
+    resp = rgta_client.get_resources(
+        ResourceTypeFilters=["dms:replication-instance"],
+        TagFilters=[{"Key": "tag", "Values": ["a tag"]}],
+    )
+    assert len(resp["ResourceTagMappingList"]) == 1
+    assert replication_instance_arn == resp["ResourceTagMappingList"][0]["ResourceARN"]
+
+
 @mock_aws
 def test_get_resources_ecs():
     # ecs:cluster
diff --git a/tests/test_s3/test_s3.py b/tests/test_s3/test_s3.py
index 4815aa317..bd086a306 100644
--- a/tests/test_s3/test_s3.py
+++ b/tests/test_s3/test_s3.py
@@ -3561,3 +3561,132 @@ def enable_versioning(bucket_name, s3_client):
     while resp.get("Status") != "Enabled":
         sleep(0.1)
         resp = s3_client.get_bucket_versioning(Bucket=bucket_name)
+
+
+@mock_aws
+def test_put_and_get_bucket_inventory_configuration():
+    client = boto3.client("s3", region_name="us-east-1")
+    bucket = "mybucket"
+    bucket2 = "mybucket2"
+    id = "my-inventory-config"
+    inventory_configuration = {
+        "Destination": {
+            "S3BucketDestination": {
+                "AccountId": DEFAULT_ACCOUNT_ID,
+                "Bucket": f"arn:aws:s3:::{bucket2}",
+                "Format": "CSV",
+                "Prefix": "test",
+                "Encryption": {"SSEKMS": {"KeyId": "key-12345"}},
+            }
+        },
+        "IsEnabled": True,
+        "Filter": {"Prefix": "folder1/"},
+        "Id": id,
+        "IncludedObjectVersions": "All",
+        "OptionalFields": ["Size", "ETag"],
+        "Schedule": {"Frequency": "Daily"},
+    }
+
+    client.create_bucket(Bucket=bucket)
+    client.create_bucket(Bucket=bucket2)
+    resp = client.put_bucket_inventory_configuration(
+        Bucket=bucket,
+        Id=id,
+        InventoryConfiguration=inventory_configuration,
+        ExpectedBucketOwner=DEFAULT_ACCOUNT_ID,
+    )
+    assert resp["ResponseMetadata"]["HTTPStatusCode"] == 200
+
+    # Retrieve the configuration
+    resp = client.get_bucket_inventory_configuration(Bucket=bucket, Id=id)
+    assert resp["InventoryConfiguration"]["Id"] == id
+    s3_bucket_dest = resp["InventoryConfiguration"]["Destination"][
+        "S3BucketDestination"
+    ]
+    assert s3_bucket_dest["AccountId"] == DEFAULT_ACCOUNT_ID
+    assert s3_bucket_dest["Bucket"] == f"arn:aws:s3:::{bucket2}"
+    assert s3_bucket_dest["Format"] == "CSV"
+    assert s3_bucket_dest["Prefix"] == "test"
+    assert s3_bucket_dest["Encryption"]["SSEKMS"]["KeyId"] == "key-12345"
+    assert resp["InventoryConfiguration"]["IsEnabled"] is True
+    assert resp["InventoryConfiguration"]["Filter"]["Prefix"] == "folder1/"
+    assert resp["InventoryConfiguration"]["IncludedObjectVersions"] == "All"
+    assert resp["InventoryConfiguration"]["OptionalFields"] == ["Size", "ETag"]
+    assert resp["InventoryConfiguration"]["Schedule"]["Frequency"] == "Daily"
+
+
+@mock_aws
+def test_list_bucket_inventory_configurations():
+    client = boto3.client("s3", region_name="us-east-1")
+    bucket = "mybucket"
+    bucket2 = "mybucket2"
+    id = "my-inventory-config"
+
+    client.create_bucket(Bucket=bucket)
+    client.create_bucket(Bucket=bucket2)
+
+    # Create three inventory configurations for the first
+    for i in range(3):
+        inventory_configuration = {
+            "Destination": {
+                "S3BucketDestination": {
+                    "AccountId": DEFAULT_ACCOUNT_ID,
+                    "Bucket": f"arn:aws:s3:::{bucket2}",
+                    "Format": "CSV",
+                    "Prefix": "test",
+                    "Encryption": {"SSEKMS": {"KeyId": "key-12345"}},
+                }
+            },
+            "IsEnabled": True,
+            "Filter": {"Prefix": "folder1/"},
+            "Id": f"{id}-{bucket}-{i}",
+            "IncludedObjectVersions": "All",
+            "OptionalFields": ["Size", "ETag"],
+            "Schedule": {"Frequency": "Daily"},
+        }
+        client.put_bucket_inventory_configuration(
+            Bucket=bucket,
+            Id=id,
+            InventoryConfiguration=inventory_configuration,
+            ExpectedBucketOwner=DEFAULT_ACCOUNT_ID,
+        )
+
+    # Create inventory configurations for the second bucket
+    for i in range(2):
+        inventory_configuration = {
+            "Destination": {
+                "S3BucketDestination": {
+                    "AccountId": DEFAULT_ACCOUNT_ID,
+                    "Bucket": f"arn:aws:s3:::{bucket}",
+                    "Format": "CSV",
+                    "Prefix": "test",
+                    "Encryption": {"SSEKMS": {"KeyId": "key-12345"}},
+                }
+            },
+            "IsEnabled": True,
+            "Filter": {"Prefix": "folder1/"},
+            "Id": f"{id}-{bucket2}-{i}",
+            "IncludedObjectVersions": "All",
+            "OptionalFields": ["Size", "ETag"],
+            "Schedule": {"Frequency": "Daily"},
+        }
+        client.put_bucket_inventory_configuration(
+            Bucket=bucket2,
+            Id=id,
+            InventoryConfiguration=inventory_configuration,
+            ExpectedBucketOwner=DEFAULT_ACCOUNT_ID,
+        )
+
+    resp = client.list_bucket_inventory_configurations(Bucket=bucket)
+    assert len(resp["InventoryConfigurationList"]) == 3
+
+    resp = client.list_bucket_inventory_configurations(Bucket=bucket2)
+    assert len(resp["InventoryConfigurationList"]) == 2
+    assert resp["InventoryConfigurationList"][0]["Id"] == f"{id}-{bucket2}-0"
+    assert resp["InventoryConfigurationList"][1]["Id"] == f"{id}-{bucket2}-1"
+    assert (
+        resp["InventoryConfigurationList"][0]["Destination"]["S3BucketDestination"][
+            "AccountId"
+        ]
+        == DEFAULT_ACCOUNT_ID
+    )
diff --git a/tests/test_servicecatalogappregistry/__init__.py b/tests/test_servicecatalogappregistry/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/test_servicecatalogappregistry/test_servicecatalogappregistry.py b/tests/test_servicecatalogappregistry/test_servicecatalogappregistry.py
new file mode 100644
index 000000000..fc3016c3f
--- /dev/null
+++ b/tests/test_servicecatalogappregistry/test_servicecatalogappregistry.py
@@ -0,0 +1,131 @@
+"""Unit tests for servicecatalogappregistry-supported APIs."""
+
+import boto3
+import pytest
+from botocore.exceptions import ClientError
+
+from moto import mock_aws
+
+# See our Development Tips on writing tests for hints on how to write good tests:
+# http://docs.getmoto.org/en/latest/docs/contributing/development_tips/tests.html
+
+
+@mock_aws
+def test_create_application():
+    client = boto3.client("servicecatalog-appregistry", region_name="us-east-1")
+    resp = client.create_application(name="testapp", description="blah")
+
+    assert "id" in resp["application"]
+    assert resp["application"]["name"] == "testapp"
+    assert resp["application"]["description"] == "blah"
+
+
+@mock_aws
+def test_list_applications():
+    client = boto3.client("servicecatalog-appregistry", region_name="us-east-1")
+    client.create_application(name="testapp", description="blah")
+    client.create_application(name="testapp2", description="foo")
+    list_resp = client.list_applications()
+
+    assert "applications" in list_resp
+    assert list_resp["applications"][1]["name"] == "testapp2"
+    assert list_resp["applications"][1]["description"] == "foo"
+
+
+@mock_aws
+def test_associate_resource():
+    client = boto3.client("servicecatalog-appregistry", region_name="us-east-1")
+    create = client.create_application(name="testapp", description="blah")
+    resp = client.associate_resource(
+        application=create["application"]["id"],
+        resource="foo",
+        resourceType="RESOURCE_TAG_VALUE",
+        options=["APPLY_APPLICATION_TAG"],
+    )
+
+    assert "applicationArn" in resp
+    assert "resourceArn" in resp
+    assert "options" in resp
+    assert resp["applicationArn"] == create["application"]["arn"]
+
+
+@mock_aws
+def test_associate_resource_cloudformation():
+    client = boto3.client("servicecatalog-appregistry", region_name="us-east-1")
+    stack_name = "foo"
+    stack_id = boto3.client("cloudformation", region_name="us-east-1").create_stack(
+        StackName=stack_name,
+        TemplateBody="""
+---
+AWSTemplateFormatVersion: '2010-09-09'
+Resources:
+  S3:
+    Type: AWS::S3::Bucket
+    Properties:
+      BucketName: foo
+""",
+    )["StackId"]
+    create = client.create_application(name="testapp", description="blah")
+    resp = client.associate_resource(
+        application=create["application"]["id"],
+        resource=stack_id,
+        resourceType="CFN_STACK",
+        options=["APPLY_APPLICATION_TAG"],
+    )
+
+    assert "applicationArn" in resp
+    assert "resourceArn" in resp
+    assert "options" in resp
+    assert resp["applicationArn"] == create["application"]["arn"]
+
+
+@mock_aws
+def test_associate_resource_cloudformation_validation_error():
+    client = boto3.client("servicecatalog-appregistry", region_name="us-east-1")
+    create = client.create_application(name="testapp", description="blah")
+    stack_name = "foo"
+    with pytest.raises(ClientError) as exc:
+        client.associate_resource(
+            application=create["application"]["id"],
+            resource=stack_name,
+            resourceType="CFN_STACK",
+            options=["APPLY_APPLICATION_TAG"],
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "ResourceNotFoundException"
+    assert err["Message"] == f"No CloudFormation stack called '{stack_name}' found"
+
+
+@mock_aws
+def test_associate_resource_fails_resource_type():
+    client = boto3.client("servicecatalog-appregistry", region_name="us-east-1")
+    create = client.create_application(name="testapp", description="blah")
+    with pytest.raises(ClientError) as exc:
+        client.associate_resource(
+            application=create["application"]["id"],
+            resource="foo",
+            resourceType="FOO",
+            options=["APPLY_APPLICATION_TAG"],
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "ValidationException"
+    assert err["Message"] == "The request has invalid or missing parameters."
+
+
+@mock_aws
+def test_list_associated_resources():
+    client = boto3.client("servicecatalog-appregistry", region_name="us-east-1")
+    create = client.create_application(name="testapp", description="blah")
+    associate = client.associate_resource(
+        application=create["application"]["id"],
+        resource="foo",
+        resourceType="RESOURCE_TAG_VALUE",
+        options=["APPLY_APPLICATION_TAG"],
+    )
+    resp = client.list_associated_resources(application=create["application"]["arn"])
+
+    assert "resources" in resp
+    assert "options" in resp["resources"][0]
+    assert "arn" in resp["resources"][0]
+    assert "resourceDetails" in resp["resources"][0]
+    assert resp["resources"][0]["options"] == associate["options"]
diff --git a/tests/test_timestreaminfluxdb/test_timestreaminfluxdb.py b/tests/test_timestreaminfluxdb/test_timestreaminfluxdb.py
index 8721ba2b7..c99d313ad 100644
--- a/tests/test_timestreaminfluxdb/test_timestreaminfluxdb.py
+++ b/tests/test_timestreaminfluxdb/test_timestreaminfluxdb.py
@@ -278,3 +278,245 @@ def test_list_db_instances():
     response = client.list_db_instances()
     resources = response["items"]
     assert len(resources) == 1
+
+
+@mock_aws
+def test_create_db_parameter_group():
+    client = boto3.client("timestream-influxdb", region_name="us-east-1")
+
+    response = client.create_db_parameter_group(
+        name="test-parameter-group",
+        description="Test parameter group for unit tests",
+        parameters={
+            "InfluxDBv2": {
+                "fluxLogEnabled": True,
+                "logLevel": "debug",
+                "queryQueueSize": 100,
+            }
+        },
+        tags={"Environment": "Test", "Project": "Moto"},
+    )
+
+    assert response["name"] == "test-parameter-group"
+    assert response["description"] == "Test parameter group for unit tests"
+    assert response["parameters"]["InfluxDBv2"]["fluxLogEnabled"] is True
+    assert response["parameters"]["InfluxDBv2"]["logLevel"] == "debug"
+    assert response["parameters"]["InfluxDBv2"]["queryQueueSize"] == 100
+    assert "id" in response
+    assert "arn" in response
+
+
+@mock_aws
+def test_get_db_parameter_group():
+    client = boto3.client("timestream-influxdb", region_name="us-east-1")
+
+    create_response = client.create_db_parameter_group(
+        name="test-parameter-group",
+        description="Test parameter group for unit tests",
+        parameters={"InfluxDBv2": {"fluxLogEnabled": True, "logLevel": "debug"}},
+    )
+
+    param_group_id = create_response["id"]
+
+    get_response = client.get_db_parameter_group(identifier=param_group_id)
+
+    assert get_response["id"] == param_group_id
+    assert get_response["name"] == "test-parameter-group"
+    assert get_response["description"] == "Test parameter group for unit tests"
+    assert get_response["parameters"]["InfluxDBv2"]["fluxLogEnabled"] is True
+    assert get_response["parameters"]["InfluxDBv2"]["logLevel"] == "debug"
+    assert "arn" in get_response
+
+    with pytest.raises(ClientError) as exc:
+        client.get_db_parameter_group(identifier="non-existent-id")
+
+    assert exc.value.response["Error"]["Code"] == "ResourceNotFoundException"
+
+
+@mock_aws
+def test_list_db_parameter_groups():
+    client = boto3.client("timestream-influxdb", region_name="us-east-1")
+
+    list_response = client.list_db_parameter_groups()
+    assert len(list_response["items"]) == 0
+    assert "nextToken" not in list_response
+
+    param_groups = []
+    for i in range(5):
+        response = client.create_db_parameter_group(
+            name=f"test-parameter-group-{i}",
+            description=f"Test parameter group {i}",
+            parameters={
+                "InfluxDBv2": {
+                    "fluxLogEnabled": i % 2 == 0,
+                    "logLevel": "debug" if i % 2 == 0 else "info",
+                }
+            },
+        )
+        param_groups.append(response)
+
+    list_response = client.list_db_parameter_groups()
+    assert len(list_response["items"]) == 5
+
+    list_response = client.list_db_parameter_groups(maxResults=2)
+    assert len(list_response["items"]) == 2
+    assert "nextToken" in list_response
+
+    next_token = list_response["nextToken"]
+    list_response = client.list_db_parameter_groups(nextToken=next_token, maxResults=2)
+    assert len(list_response["items"]) == 2
+    assert "nextToken" in list_response
+
+    next_token = list_response["nextToken"]
+    list_response = client.list_db_parameter_groups(nextToken=next_token, maxResults=2)
+    assert len(list_response["items"]) == 1
+    assert "nextToken" not in list_response
+
+
+@mock_aws
+def test_list_db_clusters():
+    client = boto3.client("timestream-influxdb", region_name="us-east-1")
+
+    list_response = client.list_db_clusters()
+    assert len(list_response["items"]) == 0
+    assert "nextToken" not in list_response
+
+    cluster_ids = []
+    for i in range(5):
+        response = client.create_db_cluster(
+            name=f"test-cluster-{i}",
+            password="password123",
+            dbInstanceType="db.influx.medium",
+            allocatedStorage=100,
+            vpcSubnetIds=["subnet-0123456789abcdef0", "subnet-0123456789abcdef1"],
+            vpcSecurityGroupIds=["sg-0123456789abcdef0"],
+            deploymentType="MULTI_NODE_READ_REPLICAS",
+        )
+        cluster_ids.append(response["dbClusterId"])
+
+    list_response = client.list_db_clusters()
+    assert len(list_response["items"]) == 5
+
+    for item in list_response["items"]:
+        assert "id" in item
+        assert "name" in item
+        assert "arn" in item
+        assert "status" in item
+        assert "endpoint" in item
+        assert "readerEndpoint" in item
+        assert "port" in item
+        assert "deploymentType" in item
+        assert "dbInstanceType" in item
+        assert "networkType" in item
+        assert "dbStorageType" in item
+        assert "allocatedStorage" in item
+
+    list_response = client.list_db_clusters(maxResults=2)
+    assert len(list_response["items"]) == 2
+    assert "nextToken" in list_response
+
+    next_token = list_response["nextToken"]
+    list_response = client.list_db_clusters(nextToken=next_token, maxResults=2)
+    assert len(list_response["items"]) == 2
+    assert "nextToken" in list_response
+
+    next_token = list_response["nextToken"]
+    list_response = client.list_db_clusters(nextToken=next_token, maxResults=2)
+    assert len(list_response["items"]) == 1
+    assert "nextToken" not in list_response
+
+
+@mock_aws
+def test_get_db_cluster():
+    client = boto3.client("timestream-influxdb", region_name="us-east-1")
+
+    create_response = client.create_db_cluster(
+        name="test-cluster",
+        password="password123",
+        dbInstanceType="db.influx.medium",
+        allocatedStorage=100,
+        vpcSubnetIds=["subnet-0123456789abcdef0", "subnet-0123456789abcdef1"],
+        vpcSecurityGroupIds=["sg-0123456789abcdef0"],
+        deploymentType="MULTI_NODE_READ_REPLICAS",
+    )
+
+    cluster_id = create_response["dbClusterId"]
+
+    get_response = client.get_db_cluster(dbClusterId=cluster_id)
+
+    assert get_response["id"] == cluster_id
+    assert get_response["name"] == "test-cluster"
+    assert get_response["status"] == "AVAILABLE"
+    assert get_response["deploymentType"] == "MULTI_NODE_READ_REPLICAS"
+    assert get_response["dbInstanceType"] == "db.influx.medium"
+    assert get_response["allocatedStorage"] == 100
+    assert isinstance(get_response["arn"], str)
+    assert isinstance(get_response["endpoint"], str)
+    assert isinstance(get_response["readerEndpoint"], str)
+    assert isinstance(get_response["port"], int)
+    assert isinstance(get_response["vpcSubnetIds"], list)
+    assert isinstance(get_response["vpcSecurityGroupIds"], list)
+    assert len(get_response["vpcSubnetIds"]) == 2
+    assert len(get_response["vpcSecurityGroupIds"]) == 1
+
+    with pytest.raises(ClientError) as exc:
+        client.get_db_cluster(dbClusterId="non-existent-id")
+
+    assert exc.value.response["Error"]["Code"] == "ResourceNotFoundException"
+
+
+@mock_aws
+def test_create_db_cluster():
+    client = boto3.client("timestream-influxdb", region_name="us-east-1")
+
+    response = client.create_db_cluster(
+        name="test-cluster",
+        password="password123",
+        dbInstanceType="db.influx.medium",
+        allocatedStorage=100,
+        vpcSubnetIds=["subnet-0123456789abcdef0", "subnet-0123456789abcdef1"],
+        vpcSecurityGroupIds=["sg-0123456789abcdef0"],
+        deploymentType="MULTI_NODE_READ_REPLICAS",
+    )
+
+    assert "dbClusterId" in response
+    assert response["dbClusterStatus"] == "AVAILABLE"
+
+    with pytest.raises(ClientError) as exc:
+        client.create_db_cluster(
+            name="test-cluster",
+            password="password123",
+            dbInstanceType="db.influx.medium",
+            allocatedStorage=100,
+            vpcSubnetIds=["subnet-0123456789abcdef0", "subnet-0123456789abcdef1"],
+            vpcSecurityGroupIds=["sg-0123456789abcdef0"],
+            deploymentType="MULTI_NODE_READ_REPLICAS",
+        )
+
+    assert exc.value.response["Error"]["Code"] == "ConflictException"
+
+    response = client.create_db_cluster(
+        name="test-cluster-full",
+        username="admin",
+        password="password123",
+        organization="test-org",
+        bucket="test-bucket",
+        port=8088,
+        dbParameterGroupIdentifier="test-param-group",
+        dbInstanceType="db.influx.medium",
+        dbStorageType="InfluxIOIncludedT1",
+        allocatedStorage=100,
+        networkType="IPV4",
+        publiclyAccessible=True,
+        vpcSubnetIds=["subnet-0123456789abcdef0", "subnet-0123456789abcdef1"],
+        vpcSecurityGroupIds=["sg-0123456789abcdef0"],
+        deploymentType="MULTI_NODE_READ_REPLICAS",
+        failoverMode="AUTOMATIC",
+        logDeliveryConfiguration={
+            "s3Configuration": {"bucketName": "test-bucket", "enabled": True}
+        },
+        tags={"Environment": "Test", "Project": "Moto"},
+    )
+
+    assert "dbClusterId" in response
+    assert response["dbClusterStatus"] == "AVAILABLE"