Add get_indexing_configuration(), update_indexing_configuration()

IMPLEMENTATION_COVERAGE.md

@@ -4628,7 +4628,7 @@
 - [ ] get_buckets_aggregation
 - [ ] get_cardinality
 - [ ] get_effective_policies
-- [ ] get_indexing_configuration
+- [X] get_indexing_configuration
 - [X] get_job_document
 - [ ] get_logging_options
 - [ ] get_ota_update
@@ -4742,7 +4742,7 @@
 - [ ] update_dynamic_thing_group
 - [ ] update_event_configurations
 - [ ] update_fleet_metric
-- [ ] update_indexing_configuration
+- [X] update_indexing_configuration
 - [ ] update_job
 - [ ] update_mitigation_action
 - [ ] update_package

moto/iot/models.py

@@ -3,6 +3,7 @@
 import re
 import time
 from collections import OrderedDict
+from dataclasses import asdict, dataclass, field, replace
 from datetime import datetime, timedelta
 from json import JSONDecodeError
 from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Pattern, Tuple
@@ -1027,6 +1028,83 @@ def delete_from_cloudformation_json(  # type: ignore[misc]
         iot_backend.delete_role_alias(role_alias_name=role_alias_name)
 
 
+@dataclass(kw_only=True, frozen=True)
+class FakeConfigField:
+    name: str
+    type: str
+
+
+@dataclass(kw_only=True, frozen=True)
+class FakeThingGroupIndexingConfiguration:
+    customFields: List[FakeConfigField] = field(default_factory=list)
+    managedFields: List[FakeConfigField] = field(default_factory=list)
+    thingGroupIndexingMode: str = "OFF"
+
+
+@dataclass(kw_only=True, frozen=True)
+class FakeThingIndexingConfigurationFilterGeoLocations:
+    name: str
+    order: str
+
+
+@dataclass(kw_only=True, frozen=True)
+class FakeThingIndexingConfigurationFilter:
+    geoLocations: list[FakeThingIndexingConfigurationFilterGeoLocations] = field(
+        default_factory=list
+    )
+    namedShadowNames: List[str] = field(default_factory=list)
+
+
+@dataclass(kw_only=True, frozen=True)
+class FakeThingIndexingConfiguration:
+    customFields: List[FakeConfigField] = field(default_factory=list)
+    managedFields: List[FakeConfigField] = field(default_factory=list)
+    filter: FakeThingIndexingConfigurationFilter = field(
+        default_factory=FakeThingIndexingConfigurationFilter
+    )
+    deviceDefenderIndexingMode: str = "OFF"
+    namedShadowIndexingMode: str = "OFF"
+    thingConnectivityIndexingMode: str = "OFF"
+    thingIndexingMode: str = "OFF"
+
+
+@dataclass(kw_only=True, frozen=True)
+class FakeIndexingConfigurationData:
+    thingGroupIndexingConfiguration: FakeThingGroupIndexingConfiguration = field(
+        default_factory=FakeThingGroupIndexingConfiguration
+    )
+    thingIndexingConfiguration: FakeThingIndexingConfiguration = field(
+        default_factory=FakeThingIndexingConfiguration
+    )
+
+
+class FakeIndexingConfiguration(BaseModel):
+    def __init__(self, region_name: str, account_id: str) -> None:
+        self.region_name = region_name
+        self.account_id = account_id
+        self.configuration = FakeIndexingConfigurationData()
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self.configuration)
+
+    def update_configuration(
+        self,
+        thingIndexingConfiguration: Dict[str, Any],
+        thingGroupIndexingConfiguration: Dict[str, Any],
+    ) -> None:
+        self.configuration = replace(
+            self.configuration,
+            thingIndexingConfiguration=replace(
+                self.configuration.thingIndexingConfiguration,
+                **thingIndexingConfiguration,
+            ),
+            thingGroupIndexingConfiguration=replace(
+                self.configuration.thingGroupIndexingConfiguration,
+                **thingGroupIndexingConfiguration,
+            ),
+        )
+
+
 class IoTBackend(BaseBackend):
     def __init__(self, region_name: str, account_id: str):
         super().__init__(region_name, account_id)
@@ -1048,6 +1126,7 @@ def __init__(self, region_name: str, account_id: str):
         self.role_aliases: Dict[str, FakeRoleAlias] = OrderedDict()
         self.endpoint: Optional[FakeEndpoint] = None
         self.domain_configurations: Dict[str, FakeDomainConfiguration] = OrderedDict()
+        self.indexing_configuration = FakeIndexingConfiguration(region_name, account_id)
 
     @staticmethod
     def default_vpc_endpoint_service(
@@ -2384,5 +2463,17 @@ def delete_role_alias(self, role_alias_name: str) -> None:
         self.describe_role_alias(role_alias_name=role_alias_name)
         del self.role_aliases[role_alias_name]
 
+    def get_index_configuration(self) -> Dict[str, Any]:
+        return self.indexing_configuration.to_dict()
+
+    def update_indexing_configuration(
+        self,
+        thingIndexingConfiguration: Dict[str, Any],
+        thingGroupIndexingConfiguration: Dict[str, Any],
+    ) -> None:
+        self.indexing_configuration.update_configuration(
+            thingIndexingConfiguration, thingGroupIndexingConfiguration
+        )
+
 
 iot_backends = BackendDict(IoTBackend, "iot")

moto/iot/responses.py

@@ -826,3 +826,13 @@ def delete_role_alias(self) -> str:
         role_alias_name = self._get_param("roleAlias")
         self.iot_backend.delete_role_alias(role_alias_name=role_alias_name)
         return json.dumps({})
+
+    def get_indexing_configuration(self) -> str:
+        return json.dumps(self.iot_backend.get_index_configuration())
+
+    def update_indexing_configuration(self) -> str:
+        self.iot_backend.update_indexing_configuration(
+            self._get_param("thingIndexingConfiguration", {}),
+            self._get_param("thingGroupIndexingConfiguration", {}),
+        )
+        return json.dumps({})

tests/test_iot/test_iot_indexing_configuration.py

@@ -0,0 +1,63 @@
+import boto3
+
+from moto import mock_aws
+
+
+@mock_aws
+def test_validate_default_indexing_configuration():
+    client = boto3.client("iot", region_name="us-east-1")
+    indexing_config = client.get_indexing_configuration()
+
+    thingIndexingConfiguration = indexing_config["thingIndexingConfiguration"]
+    assert thingIndexingConfiguration["thingIndexingMode"] in [
+        "OFF",
+        "REGISTRY",
+        "REGISTRY_AND_SHADOW",
+    ]
+
+    thingGroupIndexingConfiguration = indexing_config["thingGroupIndexingConfiguration"]
+    assert thingGroupIndexingConfiguration["thingGroupIndexingMode"] in ["ON", "OFF"]
+
+
+@mock_aws
+def test_update_indexing_mode():
+    client = boto3.client("iot", region_name="us-east-1")
+    client.update_indexing_configuration(
+        thingIndexingConfiguration={
+            "thingIndexingMode": "REGISTRY",
+            "thingConnectivityIndexingMode": "STATUS",
+            "deviceDefenderIndexingMode": "VIOLATIONS",
+            "namedShadowIndexingMode": "ON",
+            "managedFields": [{"name": "field1", "type": "String"}],
+            "customFields": [{"name": "custom_field1", "type": "Boolean"}],
+            "filter": {"namedShadowNames": ["shadow_1", "shadow_2"]},
+        },
+        thingGroupIndexingConfiguration={
+            "thingGroupIndexingMode": "ON",
+            "managedFields": [{"name": "thing_field_1", "type": "String"}],
+            "customFields": [{"name": "thing_custom_field", "type": "Number"}],
+        },
+    )
+
+    indexing_config = client.get_indexing_configuration()
+
+    thingIndexingConfiguration = indexing_config["thingIndexingConfiguration"]
+    assert thingIndexingConfiguration["thingIndexingMode"] == "REGISTRY"
+    assert thingIndexingConfiguration["thingConnectivityIndexingMode"] == "STATUS"
+    assert thingIndexingConfiguration["deviceDefenderIndexingMode"] == "VIOLATIONS"
+    assert thingIndexingConfiguration["namedShadowIndexingMode"] == "ON"
+    assert thingIndexingConfiguration["managedFields"] == [
+        {"name": "field1", "type": "String"}
+    ]
+    assert thingIndexingConfiguration["customFields"] == [
+        {"name": "custom_field1", "type": "Boolean"}
+    ]
+
+    thingGroupIndexingConfiguration = indexing_config["thingGroupIndexingConfiguration"]
+    assert thingGroupIndexingConfiguration["thingGroupIndexingMode"] == "ON"
+    assert thingGroupIndexingConfiguration["managedFields"] == [
+        {"name": "thing_field_1", "type": "String"}
+    ]
+    assert thingGroupIndexingConfiguration["customFields"] == [
+        {"name": "thing_custom_field", "type": "Number"}
+    ]