From 6685ac7e4db676cb302a52cc307cbc0e91ee6987 Mon Sep 17 00:00:00 2001 From: Jacek Date: Wed, 13 Nov 2024 15:02:40 +0100 Subject: [PATCH] Add get_indexing_configuration(), update_indexing_configuration() --- IMPLEMENTATION_COVERAGE.md | 4 +- moto/iot/models.py | 91 +++++++++++++++++++ moto/iot/responses.py | 10 ++ .../test_iot_indexing_configuration.py | 63 +++++++++++++ 4 files changed, 166 insertions(+), 2 deletions(-) create mode 100644 tests/test_iot/test_iot_indexing_configuration.py diff --git a/IMPLEMENTATION_COVERAGE.md b/IMPLEMENTATION_COVERAGE.md index e5ee66fffdd9..2349a0b5fc0a 100644 --- a/IMPLEMENTATION_COVERAGE.md +++ b/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 diff --git a/moto/iot/models.py b/moto/iot/models.py index a92765c957e8..d3efc19d6afc 100644 --- a/moto/iot/models.py +++ b/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(frozen=True) +class FakeConfigField: + name: str + type: str + + +@dataclass(frozen=True) +class FakeThingGroupIndexingConfiguration: + customFields: List[FakeConfigField] = field(default_factory=list) + managedFields: List[FakeConfigField] = field(default_factory=list) + thingGroupIndexingMode: str = "OFF" + + +@dataclass(frozen=True) +class FakeThingIndexingConfigurationFilterGeoLocations: + name: str + order: str + + +@dataclass(frozen=True) +class FakeThingIndexingConfigurationFilter: + geoLocations: List[FakeThingIndexingConfigurationFilterGeoLocations] = field( + default_factory=list + ) + namedShadowNames: List[str] = field(default_factory=list) + + +@dataclass(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(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") diff --git a/moto/iot/responses.py b/moto/iot/responses.py index 7d256dbb650d..55842b818989 100644 --- a/moto/iot/responses.py +++ b/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({}) diff --git a/tests/test_iot/test_iot_indexing_configuration.py b/tests/test_iot/test_iot_indexing_configuration.py new file mode 100644 index 000000000000..d2ec29a58e91 --- /dev/null +++ b/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"} + ]