diff --git a/capella2polarion/converters/converter_config.py b/capella2polarion/converters/converter_config.py new file mode 100644 index 00000000..dc89c121 --- /dev/null +++ b/capella2polarion/converters/converter_config.py @@ -0,0 +1,94 @@ +# Copyright DB InfraGO AG and contributors +# SPDX-License-Identifier: Apache-2.0 +"""Module providing capella2polarion config class.""" +from __future__ import annotations + +import dataclasses +import typing + +import yaml + + +@dataclasses.dataclass +class CapellaTypeConfig: + """A single Capella Type configuration.""" + + p_type: str | None = None + converter: str | None = None + links: list[str] = dataclasses.field(default_factory=list) + + +class ConverterConfig: + """The overall Config for capella2polarion.""" + + def __init__(self, synchronize_config: typing.TextIO): + config_dict = yaml.safe_load(synchronize_config) + self._layer_configs: dict[str, dict[str, CapellaTypeConfig]] = {} + self._global_configs: dict[str, CapellaTypeConfig] = {} + # We handle the cross layer config separately as global_configs + global_config_dict = config_dict.pop("*", {}) + all_type_config = global_config_dict.pop("*", {}) + global_links = all_type_config.get("links", []) + self.__global_config = CapellaTypeConfig(links=global_links) + + for c_type, type_config in global_config_dict.items(): + type_config = type_config or {} + self._global_configs[c_type] = CapellaTypeConfig( + type_config.get("polarion_type"), + type_config.get("serializer"), + type_config.get("links", []) + global_links, + ) + + for layer, type_configs in config_dict.items(): + self._layer_configs[layer] = {} + for c_type, type_config in type_configs.items(): + self._layer_configs[layer][c_type] = CapellaTypeConfig( + type_config.get("polarion_type") + or self._global_configs.get( + c_type, self.__global_config + ).p_type, + type_config.get("serializer") + or self._global_configs.get( + c_type, self.__global_config + ).converter, + type_config.get("links", []) + + self._global_configs.get( + c_type, self.__global_config + ).links, + ) + + def _default_type_conversion(self, c_type: str) -> str: + return c_type[0].lower() + c_type[1:] + + def _get_type_configs( + self, layer: str, c_type: str + ) -> CapellaTypeConfig | None: + return self._layer_configs.get(layer, {}).get( + c_type + ) or self._global_configs.get(c_type) + + def get_polarion_type(self, layer: str, c_type: str) -> str: + """Return polarion type for a given layer and Capella type.""" + type_config = ( + self._get_type_configs(layer, c_type) or self.__global_config + ) + return type_config.p_type or self._default_type_conversion(c_type) + + def get_serializer(self, layer: str, c_type: str) -> str | None: + """Return the serializer name for a given layer and Capella type.""" + type_config = ( + self._get_type_configs(layer, c_type) or self.__global_config + ) + return type_config.converter + + def get_links(self, layer: str, c_type: str) -> list[str]: + """Return the list of link types for a given layer and Capella type.""" + type_config = ( + self._get_type_configs(layer, c_type) or self.__global_config + ) + return type_config.links + + def __contains__(self, item: tuple[str, str]): + """Check if there is a config for a given layer and Capella type.""" + layer, c_type = item + return self._get_type_configs(layer, c_type) is not None diff --git a/capella2polarion/converters/element_converter.py b/capella2polarion/converters/element_converter.py index 67957aa2..874d48a7 100644 --- a/capella2polarion/converters/element_converter.py +++ b/capella2polarion/converters/element_converter.py @@ -5,7 +5,6 @@ import base64 import collections -import collections.abc as cabc import logging import mimetypes import pathlib @@ -21,10 +20,9 @@ from capellambse.model.layers import oa, pa from lxml import etree +from capella2polarion import data_models from capella2polarion.connectors import polarion_repo -from .. import data_models - RE_DESCR_LINK_PATTERN = re.compile( r"([^<]+)<\/a>" ) @@ -42,12 +40,12 @@ SERIALIZERS: dict[str, str] = { "CapabilityRealization": "include_pre_and_post_condition", "Capability": "include_pre_and_post_condition", - "LogicalComponent": "_include_actor_in_type", + "LogicalComponent": "include_actor_in_type", "OperationalCapability": "include_pre_and_post_condition", - "PhysicalComponent": "_include_nature_in_type", - "SystemComponent": "_include_actor_in_type", + "PhysicalComponent": "include_nature_in_type", + "SystemComponent": "include_actor_in_type", "Scenario": "include_pre_and_post_condition", - "Constraint": "constraint", + "Constraint": "linked_text_as_description", "SystemCapability": "include_pre_and_post_condition", } @@ -141,11 +139,6 @@ class CapellaWorkItemSerializer: capella_polarion_mapping: polarion_repo.PolarionDataRepository model: capellambse.MelodyModel descr_references: dict[str, list[str]] - - serializers: dict[ - str, - cabc.Callable[[common.GenericElement], data_models.CapellaWorkItem], - ] serializer_mapping: dict[str, str] def __init__( @@ -162,12 +155,6 @@ def __init__( self.model = model self.capella_polarion_mapping = capella_polarion_mapping self.descr_references = descr_references - self.serializers = { - "include_pre_and_post_condition": self.include_pre_and_post_condition, - "_include_actor_in_type": self._include_actor_in_type, - "_include_nature_in_type": self._include_nature_in_type, - "constraint": self.constraint, - } self.serializer_mapping = serializer_mapping or SERIALIZERS def serialize( @@ -176,13 +163,14 @@ def serialize( """Return a CapellaWorkItem for the given diagram or element.""" try: if isinstance(obj, diagr.Diagram): - return self.diagram(obj) + return self._diagram(obj) else: xtype = self.polarion_type_map.get( obj.uuid, type(obj).__name__ ) - serializer = self.serializers.get( - self.serializer_mapping.get(xtype, ""), + serializer = getattr( + self, + f"_{self.serializer_mapping.get(xtype)}", self._generic_work_item, ) return serializer(obj) @@ -190,7 +178,7 @@ def serialize( logger.error("Serializing model element failed. %s", error.args[0]) return None - def diagram(self, diag: diagr.Diagram) -> data_models.CapellaWorkItem: + def _diagram(self, diag: diagr.Diagram) -> data_models.CapellaWorkItem: """Serialize a diagram for Polarion.""" diagram_path = self.diagram_cache_path / f"{diag.uuid}.svg" src = _decode_diagram(diagram_path) @@ -232,7 +220,7 @@ def _sanitize_description( ) -> tuple[list[str], markupsafe.Markup]: referenced_uuids: list[str] = [] replaced_markup = RE_DESCR_LINK_PATTERN.sub( - lambda match: self.replace_markup(match, referenced_uuids, 2), + lambda match: self._replace_markup(match, referenced_uuids, 2), descr, ) @@ -264,7 +252,7 @@ def repair_images(node: etree._Element) -> None: ) return referenced_uuids, repaired_markup - def replace_markup( + def _replace_markup( self, match: re.Match, referenced_uuids: list[str], @@ -287,7 +275,7 @@ def replace_markup( logger.warning("Found reference to non-existing work item: %r", uuid) return match.group(default_group) - def include_pre_and_post_condition( + def _include_pre_and_post_condition( self, obj: PrePostConditionElement ) -> data_models.CapellaWorkItem: """Return generic attributes and pre- and post-condition.""" @@ -298,7 +286,7 @@ def get_condition(cap: PrePostConditionElement, name: str) -> str: return condition.specification["capella:linkedText"].striptags() def matcher(match: re.Match) -> str: - return strike_through(self.replace_markup(match, [])) + return strike_through(self._replace_markup(match, [])) work_item = self._generic_work_item(obj) pre_condition = RE_DESCR_DELETED_PATTERN.sub( @@ -313,7 +301,7 @@ def matcher(match: re.Match) -> str: return work_item - def get_linked_text( + def _get_linked_text( self, obj: capellacore.Constraint ) -> markupsafe.Markup: """Return sanitized markup of the given ``obj`` linked text.""" @@ -323,13 +311,13 @@ def get_linked_text( self.descr_references[obj.uuid] = uuids return value - def constraint( + def _linked_text_as_description( self, obj: capellacore.Constraint ) -> data_models.CapellaWorkItem: """Return attributes for a ``Constraint``.""" work_item = self._generic_work_item(obj) # pylint: disable-next=attribute-defined-outside-init - work_item.description = self.get_linked_text(obj) + work_item.description = self._get_linked_text(obj) return work_item def _include_actor_in_type( diff --git a/tests/data/model_elements/new_config.yaml b/tests/data/model_elements/new_config.yaml new file mode 100644 index 00000000..bb3dfb70 --- /dev/null +++ b/tests/data/model_elements/new_config.yaml @@ -0,0 +1,23 @@ +# Copyright DB InfraGO AG and contributors +# SPDX-License-Identifier: Apache-2.0 + +"*": # All layers + "*": # All class types + links: + - parent # Specify workitem links + - description_reference # Custom attribute + Class: + links: + - state_machines + Diagram: + links: + - diagram_elements + Constraint: + +oa: # Specify below + FunctionalExchange: + polarion_type: operationalInteraction + links: + - exchange_items + OperationalCapability: + serializer: include_pre_and_post_condition