merge: Merge pull request #99 from DSD-DBS/refactor-prefixes

capella2polarion/__main__.py

@@ -84,8 +84,8 @@ def print_cli_state(capella2polarion_cli: Capella2PolarionCli) -> None:
 @click.pass_context
 def synchronize(
     ctx: click.core.Context,
-    force_update: bool,
     synchronize_config: typing.TextIO,
+    force_update: bool,
     type_prefix: str,
     role_prefix: str,
 ) -> None:
@@ -95,16 +95,14 @@ def synchronize(
         "Synchronising model elements to Polarion project with id %s...",
         capella_to_polarion_cli.polarion_params.project_id,
     )
-    capella_to_polarion_cli.load_synchronize_config(synchronize_config)
+    capella_to_polarion_cli.load_synchronize_config(
+        synchronize_config, type_prefix, role_prefix
+    )
     capella_to_polarion_cli.force_update = force_update
-    capella_to_polarion_cli.type_prefix = type_prefix
-    capella_to_polarion_cli.role_prefix = role_prefix
 
     converter = model_converter.ModelConverter(
         capella_to_polarion_cli.capella_model,
         capella_to_polarion_cli.polarion_params.project_id,
-        type_prefix=capella_to_polarion_cli.type_prefix,
-        role_prefix=capella_to_polarion_cli.role_prefix,
     )
 
     converter.read_model(capella_to_polarion_cli.config)

capella2polarion/cli.py

@@ -28,8 +28,6 @@ def __init__(
         polarion_delete_work_items: bool,
         capella_model: capellambse.MelodyModel,
         force_update: bool = False,
-        type_prefix: str = "",
-        role_prefix: str = "",
     ) -> None:
         self.debug = debug
         self.polarion_params = pw.PolarionWorkerParams(
@@ -42,8 +40,6 @@ def __init__(
         self.capella_model = capella_model
         self.config = converter_config.ConverterConfig()
         self.force_update = force_update
-        self.type_prefix = type_prefix
-        self.role_prefix = role_prefix
 
     def _none_save_value_string(self, value: str | None) -> str | None:
         return "None" if value is None else value
@@ -97,7 +93,10 @@ def setup_logger(self) -> None:
         logging.getLogger("httpcore").setLevel("WARNING")
 
     def load_synchronize_config(
-        self, synchronize_config_io: typing.TextIO
+        self,
+        synchronize_config_io: typing.TextIO,
+        type_prefix: str = "",
+        role_prefix: str = "",
     ) -> None:
         """Read the sync config into SynchronizeConfigContent.
 
@@ -107,4 +106,6 @@ def load_synchronize_config(
             raise RuntimeError("synchronize config io stream is closed ")
         if not synchronize_config_io.readable():
             raise RuntimeError("synchronize config io stream is not readable")
-        self.config.read_config_file(synchronize_config_io)
+        self.config.read_config_file(
+            synchronize_config_io, type_prefix, role_prefix
+        )

capella2polarion/converters/converter_config.py

@@ -39,6 +39,8 @@ class LinkConfig:
     capella_attr: str
     polarion_role: str
     include: dict[str, str] = dataclasses.field(default_factory=dict)
+    link_field: str = ""
+    reverse_field: str = ""
 
 
 @dataclasses.dataclass
@@ -70,18 +72,25 @@ def __init__(self):
         self.diagram_config: CapellaTypeConfig | None = None
         self.__global_config = CapellaTypeConfig()
 
-    def read_config_file(self, synchronize_config: t.TextIO):
+    def read_config_file(
+        self,
+        synchronize_config: t.TextIO,
+        type_prefix: str = "",
+        role_prefix: str = "",
+    ):
         """Read a given yaml file as config."""
         config_dict = yaml.safe_load(synchronize_config)
         # 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.links = _force_link_config(global_links)
+        self.__global_config.links = self._force_link_config(
+            global_links, role_prefix
+        )
 
         if "Diagram" in global_config_dict:
             diagram_config = global_config_dict.pop("Diagram") or {}
-            self.set_diagram_config(diagram_config)
+            self.set_diagram_config(diagram_config, type_prefix)
 
         for c_type, type_config in global_config_dict.items():
             type_config = type_config or {}
@@ -91,7 +100,9 @@ def read_config_file(self, synchronize_config: t.TextIO):
             type_configs = type_configs or {}
             self.add_layer(layer)
             for c_type, c_type_config in type_configs.items():
-                self.set_layer_config(c_type, c_type_config, layer)
+                self.set_layer_config(
+                    c_type, c_type_config, layer, type_prefix, role_prefix
+                )
 
     def add_layer(self, layer: str):
         """Add a new layer without configuring any types."""
@@ -105,6 +116,8 @@ def set_layer_config(
         c_type: str,
         c_type_config: dict[str, t.Any] | list[dict[str, t.Any]] | None,
         layer: str,
+        type_prefix: str = "",
+        role_prefix: str = "",
     ):
         """Set one or multiple configs for a type to an existing layer."""
         type_configs = _read_capella_type_configs(c_type_config)
@@ -122,21 +135,23 @@ def set_layer_config(
             # As we set up all types this way, we can expect that all
             # non-compliant links are coming from global context here
             closest_links = _filter_links(c_type, closest_config.links, True)
-            p_type = (
-                type_config.get("polarion_type")
-                or closest_config.p_type
-                or _default_type_conversion(c_type)
+            p_type = add_prefix(
+                (
+                    type_config.get("polarion_type")
+                    or closest_config.p_type
+                    or _default_type_conversion(c_type)
+                ),
+                type_prefix,
             )
             self.polarion_types.add(p_type)
+            links = self._force_link_config(
+                type_config.get("links", []), role_prefix
+            )
             self._layer_configs[layer][c_type].append(
                 CapellaTypeConfig(
                     p_type,
                     type_config.get("serializer") or closest_config.converters,
-                    _filter_links(
-                        c_type,
-                        _force_link_config(type_config.get("links", [])),
-                    )
-                    + closest_links,
+                    _filter_links(c_type, links) + closest_links,
                     type_config.get("is_actor", _C2P_DEFAULT),
                     type_config.get("nature", _C2P_DEFAULT),
                 )
@@ -152,27 +167,61 @@ def set_global_config(self, c_type: str, type_config: dict[str, t.Any]):
             p_type,
             type_config.get("serializer"),
             _filter_links(
-                c_type, _force_link_config(type_config.get("links", []))
+                c_type, self._force_link_config(type_config.get("links", []))
             )
             + self._get_global_links(c_type),
             type_config.get("is_actor", _C2P_DEFAULT),
             type_config.get("nature", _C2P_DEFAULT),
         )
 
-    def set_diagram_config(self, diagram_config: dict[str, t.Any]):
+    def set_diagram_config(
+        self, diagram_config: dict[str, t.Any], type_prefix: str = ""
+    ):
         """Set the diagram config."""
         c_type = "diagram"
         p_type = diagram_config.get("polarion_type") or "diagram"
         self.polarion_types.add(p_type)
         links = _filter_links(
-            c_type, _force_link_config(diagram_config.get("links", []))
+            c_type, self._force_link_config(diagram_config.get("links", []))
         )
         self.diagram_config = CapellaTypeConfig(
-            p_type,
+            add_prefix(p_type, type_prefix),
             diagram_config.get("serializer") or "diagram",
             links + self._get_global_links(c_type),
         )
 
+    def _force_link_config(
+        self, links: t.Any, role_prefix: str = ""
+    ) -> list[LinkConfig]:
+        result: list[LinkConfig] = []
+        for link in links:
+            if isinstance(link, str):
+                config = LinkConfig(
+                    capella_attr=link,
+                    polarion_role=add_prefix(link, role_prefix),
+                    link_field=link,
+                    reverse_field=f"{link}_reverse",
+                )
+            elif isinstance(link, dict):
+                config = LinkConfig(
+                    capella_attr=(lid := link["capella_attr"]),
+                    polarion_role=add_prefix(
+                        (pid := link.get("polarion_role", lid)),
+                        role_prefix,
+                    ),
+                    include=link.get("include", {}),
+                    link_field=(lf := link.get("link_field", pid)),
+                    reverse_field=link.get("reverse_field", f"{lf}_reverse"),
+                )
+            else:
+                logger.error(
+                    "Link not configured correctly: %r",
+                    link,
+                )
+                continue
+            result.append(config)
+        return result
+
     def get_type_config(
         self, layer: str, c_type: str, **attributes: t.Any
     ) -> CapellaTypeConfig | None:
@@ -252,25 +301,11 @@ def _force_dict(
             raise TypeError("Unsupported Type")
 
 
-def _force_link_config(links: t.Any) -> list[LinkConfig]:
-    result: list[LinkConfig] = []
-    for link in links:
-        if isinstance(link, str):
-            config = LinkConfig(capella_attr=link, polarion_role=link)
-        elif isinstance(link, dict):
-            config = LinkConfig(
-                capella_attr=(lid := link["capella_attr"]),
-                polarion_role=link.get("polarion_role", lid),
-                include=link.get("include", {}),
-            )
-        else:
-            logger.error(
-                "Link not configured correctly: %r",
-                link,
-            )
-            continue
-        result.append(config)
-    return result
+def add_prefix(polarion_type: str, prefix: str) -> str:
+    """Add a prefix to the given ``polarion_type``."""
+    if prefix:
+        return f"{prefix}_{polarion_type}"
+    return polarion_type
 
 
 def _filter_links(

capella2polarion/converters/element_converter.py

@@ -85,13 +85,11 @@ def __init__(
         capella_polarion_mapping: polarion_repo.PolarionDataRepository,
         converter_session: data_session.ConverterSession,
         generate_attachments: bool,
-        type_prefix: str = "",
     ):
         self.model = model
         self.capella_polarion_mapping = capella_polarion_mapping
         self.converter_session = converter_session
         self.generate_attachments = generate_attachments
-        self.type_prefix = type_prefix
         self.jinja_envs: dict[str, jinja2.Environment] = {}
 
     def serialize_all(self) -> list[data_models.CapellaWorkItem]:
@@ -125,11 +123,6 @@ def serialize(self, uuid: str) -> data_models.CapellaWorkItem | None:
                 )
                 converter_data.work_item = None
 
-        if self.type_prefix and converter_data.work_item is not None:
-            converter_data.work_item.type = (
-                f"{self.type_prefix}_{converter_data.work_item.type}"
-            )
-
         if converter_data.errors:
             log_args = (
                 converter_data.capella_element._short_repr_(),

capella2polarion/converters/link_converter.py

@@ -36,13 +36,11 @@ def __init__(
         converter_session: data_session.ConverterSession,
         project_id: str,
         model: capellambse.MelodyModel,
-        role_prefix: str = "",
     ):
         self.capella_polarion_mapping = capella_polarion_mapping
         self.converter_session = converter_session
         self.project_id = project_id
         self.model = model
-        self.role_prefix = role_prefix
 
         self.serializers: dict[str, _Serializer] = {
             converter_config.DESCRIPTION_REFERENCE_SERIALIZER: self._handle_description_reference_links,  # pylint: disable=line-too-long
@@ -63,8 +61,6 @@ def create_links_for_work_item(
         for link_config in converter_data.type_config.links:
             serializer = self.serializers.get(link_config.capella_attr)
             role_id = link_config.polarion_role
-            if self.role_prefix:
-                role_id = f"{self.role_prefix}_{role_id}"
             try:
                 if serializer:
                     new_links.extend(
@@ -225,16 +221,11 @@ def create_grouped_link_fields(
                     key = link.secondary_work_item_id
                     back_links.setdefault(key, []).append(link)
 
-            role_id = self._remove_prefix(role)
-            config: converter_config.LinkConfig | None = None
-            for link_config in data.type_config.links:
-                if link_config.polarion_role == role_id:
-                    config = link_config
-                    break
-
-            self._create_link_fields(
-                work_item, role_id, grouped_links, config=config
-            )
+            config = find_link_config(data, role)
+            if config is not None and config.link_field:
+                self._create_link_fields(
+                    work_item, config.link_field, grouped_links, config=config
+                )
 
     def _create_link_fields(
         self,
@@ -246,7 +237,6 @@ def _create_link_fields(
     ):
         link_map: dict[str, dict[str, list[str]]]
         if reverse:
-            role = f"{role}_reverse"
             link_map = {link.primary_work_item_id: {} for link in links}
         else:
             link_map = {link.secondary_work_item_id: {} for link in links}
@@ -303,26 +293,41 @@ def _create_link_fields(
 
     def create_grouped_back_link_fields(
         self,
-        work_item: data_models.CapellaWorkItem,
+        data: data_session.ConverterData,
         links: list[polarion_api.WorkItemLink],
     ):
         """Create fields for the given WorkItem using a list of backlinks.
 
         Parameters
         ----------
-        work_item
-            WorkItem to create the fields for
+        data
+            The ConverterData that stores the WorkItem to create the
+            fields for.
         links
-            List of links referencing work_item as secondary
+            List of links referencing work_item as secondary.
         """
+        work_item = data.work_item
+        assert work_item is not None
+        wi = f"[{work_item.id}]({work_item.type} {work_item.title})"
+        logger.debug("Building grouped back links for work item %r...", wi)
         for role, grouped_links in _group_by("role", links).items():
-            role_id = self._remove_prefix(role)
-            self._create_link_fields(work_item, role_id, grouped_links, True)
+            config = find_link_config(data, role)
+            if config is not None and config.reverse_field:
+                self._create_link_fields(
+                    work_item, config.reverse_field, grouped_links, True
+                )
+
+
+def find_link_config(
+    data: data_session.ConverterData, role: str
+) -> converter_config.LinkConfig | None:
+    """Search for LinkConfig with matching polarion_role in ``data``."""
+    for link_config in data.type_config.links:
+        if link_config.polarion_role == role:
+            return link_config
 
-    def _remove_prefix(self, role: str) -> str:
-        if self.role_prefix:
-            return role.removeprefix(f"{self.role_prefix}_")
-        return role
+    logger.error("No LinkConfig found for %r", role)
+    return None
 
 
 def _group_by(