feat(converter): Handle multi-serializer from config

capella2polarion/converters/converter_config.py

@@ -17,11 +17,15 @@ class CapellaTypeConfig:
     """A single Capella Type configuration."""
 
     p_type: str | None = None
-    converter: str | None = None
+    converters: str | list[str] | None = None
     links: list[str] = dataclasses.field(default_factory=list)
     is_actor: bool | None = None
     nature: str | None = None
 
+    def __post_init__(self):
+        """Post processing for the initialization."""
+        self.converters = _force_list(self.converters) or ["generic_work_item"]
+
 
 def _default_type_conversion(c_type: str) -> str:
     return c_type[0].lower() + c_type[1:]
@@ -92,7 +96,7 @@ def set_layer_config(
             self._layer_configs[layer][c_type].append(
                 CapellaTypeConfig(
                     p_type,
-                    type_config.get("serializer") or closest_config.converter,
+                    type_config.get("serializer") or closest_config.converters,
                     type_config.get("links", []) + closest_config.links,
                     type_config.get("is_actor", _C2P_DEFAULT),
                     type_config.get("nature", _C2P_DEFAULT),
@@ -184,3 +188,9 @@ def _read_capella_type_configs(
         key=lambda c: int(c.get("is_actor", _C2P_DEFAULT) != _C2P_DEFAULT)
         + 2 * int(c.get("nature", _C2P_DEFAULT) != _C2P_DEFAULT),
     )
+
+
+def _force_list(config: str | list[str] | None) -> list[str]:
+    if config is None:
+        return []
+    return [config] if not isinstance(config, list) else config

capella2polarion/converters/element_converter.py

@@ -94,7 +94,7 @@ def _get_requirement_types_text(
     obj: common.GenericElement,
 ) -> dict[str, dict[str, str]]:
     type_texts = collections.defaultdict(list)
-    for req in obj.requirements:
+    for req in getattr(obj, "requirements", []):
         if req is None:
             logger.error(
                 "RequirementsRelation with broken target found %r", obj.name
@@ -146,35 +146,34 @@ def __init__(
         self.capella_polarion_mapping = capella_polarion_mapping
         self.converter_session = converter_session
 
-    def serialize_all(self):
+    def serialize_all(self) -> list[data_models.CapellaWorkItem]:
         """Serialize all items of the converter_session."""
         work_items = [self.serialize(uuid) for uuid in self.converter_session]
         return list(filter(None, work_items))
 
-    def serialize(
-        self,
-        uuid: str,
-    ) -> data_models.CapellaWorkItem | None:
+    def serialize(self, uuid: str) -> data_models.CapellaWorkItem | None:
         """Return a CapellaWorkItem for the given diagram or element."""
         converter_data = self.converter_session[uuid]
-        try:
-            serializer: cabc.Callable[
-                [data_session.ConverterData], data_models.CapellaWorkItem
-            ] = getattr(
-                self,
-                f"_{converter_data.type_config.converter}",
-                self._generic_work_item,
-            )
-            converter_data.work_item = serializer(converter_data)
-            if old := self.capella_polarion_mapping.get_work_item_by_capella_uuid(
-                converter_data.work_item.uuid_capella
-            ):
-                converter_data.work_item.id = old.id
+        work_item = data_models.CapellaWorkItem(uuid)
+        nothing = True
+        for converter in converter_data.type_config.converters or []:
+            try:
+                serializer: cabc.Callable[
+                    [data_session.ConverterData], data_models.CapellaWorkItem
+                ] = getattr(self, f"_{converter}", self._generic_work_item)
+                converter_data.work_item = serializer(converter_data)
+                work_item += converter_data.work_item
+                nothing = False
+            except Exception as error:
+                logger.error(
+                    "Serializing model element failed. %s", error.args[0]
+                )
+                return None  # Force to not overwrite on failure
 
-            return converter_data.work_item
-        except Exception as error:
-            logger.error("Serializing model element failed. %s", error.args[0])
-            return None
+        old = self.capella_polarion_mapping.get_work_item_by_capella_uuid(uuid)
+        if not nothing and old:
+            work_item.id = old.id
+        return work_item
 
     def _diagram(
         self, converter_data: data_session.ConverterData
@@ -334,3 +333,15 @@ def _add_context_diagram(
             "value": _generate_image_html(diagram.as_datauri_svg),
         }
         return work_item
+
+    def _add_tree_diagram(
+        self, converter_data: data_session.ConverterData
+    ) -> data_models.CapellaWorkItem:
+        """Add a new custom field tree diagram."""
+        work_item = self._generic_work_item(converter_data)
+        diagram = converter_data.capella_element.tree_view
+        work_item.additional_attributes["tree_view"] = {
+            "type": "text/html",
+            "value": _generate_image_html(diagram.as_datauri_svg),
+        }
+        return work_item

capella2polarion/data_models.py

@@ -20,3 +20,25 @@ class Condition(t.TypedDict):
     uuid_capella: str
     preCondition: Condition | None
     postCondition: Condition | None
+
+    def __add__(self, other: CapellaWorkItem) -> CapellaWorkItem:
+        """Add a CapellaWorkItem to this one."""
+        if not isinstance(other, CapellaWorkItem):
+            raise TypeError("Can only merge WorkItems")
+
+        merged_data: dict[str, t.Any] = {}
+        self_dict = self.to_dict()
+        other_dict = other.to_dict()
+        for key in set(self_dict) | set(other_dict):
+            self_val: t.Any = self_dict.get(key)
+            other_val: t.Any = other_dict.get(key)
+
+            if isinstance(self_val, list) and isinstance(other_val, list):
+                merged_data[key] = self_val + other_val
+            elif isinstance(self_val, dict) and isinstance(other_val, dict):
+                merged_data[key] = {**self_val, **other_val}
+            else:
+                merged_data[key] = (
+                    other_val if other_val is not None else self_val
+                )
+        return CapellaWorkItem(**merged_data)

tests/test_elements.py

@@ -1266,3 +1266,31 @@ def test_add_context_diagram(self, model: capellambse.MelodyModel):
         assert str(
             work_item.additional_attributes["context_diagram"]["value"]
         ).startswith(TEST_DIAG_DESCR)
+
+    def test_multiple_serializers(self, model: capellambse.MelodyModel):
+        cap = model.by_uuid(TEST_OCAP_UUID)
+        type_config = converter_config.CapellaTypeConfig(
+            "test",
+            ["include_pre_and_post_condition", "add_context_diagram"],
+            [],
+        )
+        serializer = element_converter.CapellaWorkItemSerializer(
+            pathlib.Path(""),
+            model,
+            polarion_repo.PolarionDataRepository(),
+            {
+                TEST_OCAP_UUID: data_session.ConverterData(
+                    "pa", type_config, cap
+                )
+            },
+        )
+
+        work_item = serializer.serialize(TEST_OCAP_UUID)
+
+        assert work_item is not None
+        assert "preCondition" in work_item.additional_attributes
+        assert "postCondition" in work_item.additional_attributes
+        assert "context_diagram" in work_item.additional_attributes
+        assert str(
+            work_item.additional_attributes["context_diagram"]["value"]
+        ).startswith(TEST_DIAG_DESCR)