diff --git a/slp_tfplan/slp_tfplan/objects/tfplan_objects.py b/slp_tfplan/slp_tfplan/objects/tfplan_objects.py index 1e6fe37e..73f5ac03 100644 --- a/slp_tfplan/slp_tfplan/objects/tfplan_objects.py +++ b/slp_tfplan/slp_tfplan/objects/tfplan_objects.py @@ -1,5 +1,5 @@ from enum import Enum -from typing import List, Dict, Union, Optional +from typing import Union, Optional from otm.otm.entity.component import Component from otm.otm.entity.dataflow import Dataflow @@ -19,8 +19,8 @@ def __init__(self, component_type: str, parent: str, parent_type: ParentType, - tags: [str], - clones_ids: [str] = None, + tags: list[str], + clones_ids: list[str] = None, tf_resource_id: str = None, tf_type: str = None, configuration: {} = None): @@ -68,7 +68,7 @@ class SecurityGroupCIDRType(Enum): @auto_repr class SecurityGroupCIDR: - def __init__(self, cidr_blocks: List[str], description: str, type: SecurityGroupCIDRType, + def __init__(self, cidr_blocks: list[str], description: str, type: SecurityGroupCIDRType, from_port: int = None, to_port: int = None, protocol: str = None): self.cidr_blocks = cidr_blocks @@ -81,19 +81,19 @@ def __init__(self, cidr_blocks: List[str], description: str, type: SecurityGroup @auto_repr class SecurityGroup: - def __init__(self, security_group_id: str, name: str, ingress_sgs: List[str] = None, egress_sgs: List[str] = None, - ingress_cidr: List[SecurityGroupCIDR] = None, egress_cidr: List[SecurityGroupCIDR] = None): + def __init__(self, security_group_id: str, name: str, ingress_sgs: list[str] = None, egress_sgs: list[str] = None, + ingress_cidr: list[SecurityGroupCIDR] = None, egress_cidr: list[SecurityGroupCIDR] = None): self.id: str = security_group_id self.name: str = name - self.ingress_sgs: List[str] = ingress_sgs - self.egress_sgs: List[str] = egress_sgs - self.ingress_cidr: List[SecurityGroupCIDR] = ingress_cidr - self.egress_cidr: List[SecurityGroupCIDR] = egress_cidr + self.ingress_sgs: list[str] = ingress_sgs + self.egress_sgs: list[str] = egress_sgs + self.ingress_cidr: list[SecurityGroupCIDR] = ingress_cidr + self.egress_cidr: list[SecurityGroupCIDR] = egress_cidr @auto_repr class LaunchTemplate: - def __init__(self, launch_template_id: str, security_groups_ids: List[str]): + def __init__(self, launch_template_id: str, security_groups_ids: list[str]): self.id = launch_template_id self.security_groups_ids = security_groups_ids @@ -104,11 +104,11 @@ class TFPlanOTM(OTM): def __init__(self, project_id: str, project_name: str, - components: List[TFPlanComponent], - security_groups: List[SecurityGroup], - launch_templates: List[LaunchTemplate], - variables: Dict[str, Union[list, str]], - dataflows: List[Dataflow], + components: list[TFPlanComponent], + security_groups: list[SecurityGroup], + launch_templates: list[LaunchTemplate], + variables: dict[str, Union[list, str]], + dataflows: list[Dataflow], default_trustzone: Trustzone = None): super().__init__(project_name, project_id, IacType.TERRAFORM) self.default_trustzone = default_trustzone diff --git a/slp_tfplan/slp_tfplan/transformers/singleton_transformer.py b/slp_tfplan/slp_tfplan/transformers/singleton_transformer.py index 8b98af4f..b600ad0a 100644 --- a/slp_tfplan/slp_tfplan/transformers/singleton_transformer.py +++ b/slp_tfplan/slp_tfplan/transformers/singleton_transformer.py @@ -1,5 +1,4 @@ import itertools -from typing import List, Dict from otm.otm.entity.dataflow import Dataflow from sl_util.sl_util.iterations_utils import remove_from_list @@ -8,7 +7,7 @@ from slp_tfplan.slp_tfplan.transformers.transformer import Transformer -def _merge_component_configurations(otm_components: List[TFPlanComponent]) -> Dict: +def _merge_component_configurations(otm_components: list[TFPlanComponent]) -> dict: merge_configuration = {} for component in otm_components: merge_configuration = { @@ -17,7 +16,7 @@ def _merge_component_configurations(otm_components: List[TFPlanComponent]) -> Di return merge_configuration -def _find_equivalent_dataflows(dataflow: Dataflow, dataflows: List[Dataflow]) -> List[Dataflow]: +def _find_equivalent_dataflows(dataflow: Dataflow, dataflows: list[Dataflow]) -> list[Dataflow]: equivalent_dataflows = [] for df in dataflows: if _are_equivalent_dataflows(dataflow, df): @@ -27,11 +26,7 @@ def _find_equivalent_dataflows(dataflow: Dataflow, dataflows: List[Dataflow]) -> def _are_sibling(component, sibling): - if component.category and sibling.category: - return component.category == sibling.category - if not component.category and not sibling.category: - return component.type == sibling.type - return False + return component.type == sibling.type and component.category == sibling.category def _are_equivalent_dataflows(dataflow_1: Dataflow, dataflow_2: Dataflow) -> bool: @@ -45,7 +40,7 @@ def _are_equivalent_dataflows(dataflow_1: Dataflow, dataflow_2: Dataflow) -> boo return is_same_dataflow or is_reverse_bidirectional_dataflow -def _merge_dataflows(origin_dataflow: Dataflow, dataflows: List[Dataflow]) -> Dataflow: +def _merge_dataflows(origin_dataflow: Dataflow, dataflows: list[Dataflow]) -> Dataflow: for df in dataflows: if origin_dataflow.tags is None: origin_dataflow.tags = df.tags @@ -69,7 +64,7 @@ def _merge_dataflows(origin_dataflow: Dataflow, dataflows: List[Dataflow]) -> Da def __build_singleton_name(component: TFPlanComponent): return component.category or f"{component.type} (grouped)" -def _build_singleton_component(otm_components: List[TFPlanComponent]) -> TFPlanComponent: +def _build_singleton_component(otm_components: list[TFPlanComponent]) -> TFPlanComponent: tags = list(set(itertools.chain.from_iterable([c.tags or [] for c in otm_components]))) configuration = _merge_component_configurations(otm_components) component_id = otm_components[0].id @@ -91,7 +86,7 @@ def __init__(self, otm: TFPlanOTM): self.otm_components = self.otm.components self.otm_dataflows = self.otm.dataflows - self.singleton_component_relations: Dict[str, TFPlanComponent] = {} + self.singleton_component_relations: dict[str, TFPlanComponent] = {} def transform(self): self.__populate_singleton_component_relations() diff --git a/slp_tfplan/tests/unit/transformers/test_singleton_transformer.py b/slp_tfplan/tests/unit/transformers/test_singleton_transformer.py index b7cdac9a..2fe904d8 100644 --- a/slp_tfplan/tests/unit/transformers/test_singleton_transformer.py +++ b/slp_tfplan/tests/unit/transformers/test_singleton_transformer.py @@ -1,4 +1,5 @@ from random import randrange +from pytest import mark, param from otm.otm.entity.parent_type import ParentType from slp_tfplan.slp_tfplan.transformers.singleton_transformer import SingletonTransformer @@ -495,3 +496,98 @@ def test_singleton_by_group_and_type_works_mixed(self): assert len(otm.components) == 2 assert otm.components[0].id == component_a.id assert otm.components[1].id == component_c.id + + @mark.parametrize("components, expected_count", [ + param( + [ + build_mocked_component({ + 'component_name': 'component_a', + 'tf_type': 'type_a', + 'configuration': {SINGLETON_CONFIG: True, CATEGORY_CONFIG: 'Category'}, + }), + build_mocked_component({ + 'component_name': 'component_b', + 'tf_type': 'type_b', + 'configuration': {SINGLETON_CONFIG: True, CATEGORY_CONFIG: 'Category'}, + }) + ], + 2, + id="same category but different types should not singleton" + ), + param( + [ + build_mocked_component({ + 'component_name': 'component_a', + 'tf_type': 'same_type', + 'configuration': {SINGLETON_CONFIG: True, CATEGORY_CONFIG: 'Category_A'}, + }), + build_mocked_component({ + 'component_name': 'component_b', + 'tf_type': 'same_type', + 'configuration': {SINGLETON_CONFIG: True, CATEGORY_CONFIG: 'Category_B'}, + }) + ], + 2, + id="same type but different categories should not singleton" + ), + param( + [ + build_mocked_component({ + 'component_name': 'component_a', + 'tf_type': 'same_type', + 'configuration': {SINGLETON_CONFIG: True, CATEGORY_CONFIG: None}, + }), + build_mocked_component({ + 'component_name': 'component_b', + 'tf_type': 'same_type', + 'configuration': {SINGLETON_CONFIG: True, CATEGORY_CONFIG: 'Category'}, + }) + ], + 2, + id="same type but one category is none should not singleton" + ), + param( + [ + build_mocked_component({ + 'component_name': 'component_a', + 'tf_type': 'same_type', + 'configuration': {SINGLETON_CONFIG: True, CATEGORY_CONFIG: 'Category'}, + }), + build_mocked_component({ + 'component_name': 'component_b', + 'tf_type': 'same_type', + 'configuration': {SINGLETON_CONFIG: True, CATEGORY_CONFIG: 'Category'}, + }) + ], + 1, + id="same type and same category should singleton" + ), + param( + [ + build_mocked_component({ + 'component_name': 'component_a', + 'tf_type': 'same_type', + 'configuration': {SINGLETON_CONFIG: True, CATEGORY_CONFIG: None}, + }), + build_mocked_component({ + 'component_name': 'component_b', + 'tf_type': 'same_type', + 'configuration': {SINGLETON_CONFIG: True, CATEGORY_CONFIG: None}, + }) + ], + 1, + id="same type and both categories none should singleton" + ), + ]) + def test_singleton_category_and_type(self, components, expected_count): + """ + Test if singleton is correctly applied based on type and category combinations. + """ + # GIVEN an OTM with the provided components + otm = build_mocked_otm(components) + + # WHEN SingletonTransformer::transform is invoked + SingletonTransformer(otm).transform() + + # THEN verify the expected number of components remain after the transformation + assert len(otm.components) == expected_count