fix: Remove obsolete link serializers

capella2polarion/converters/converter_config.py

@@ -37,7 +37,7 @@ class LinkConfig:
     """
 
     capella_attr: str
-    polarion_role: str | None = None
+    polarion_role: str
     include: dict[str, str] = dataclasses.field(default_factory=dict)
 
 
@@ -286,27 +286,25 @@ def _filter_links(
 
     available_links = []
     for link in links:
-        cappela_attr = link.capella_attr.split(".")[0]
+        capella_attr = link.capella_attr.split(".")[0]
+        is_diagram_elements = capella_attr == DIAGRAM_ELEMENTS_SERIALIZER
         if (
-            cappela_attr == DESCRIPTION_REFERENCE_SERIALIZER
-            or (
-                cappela_attr == DIAGRAM_ELEMENTS_SERIALIZER
-                and c_class == diagram.Diagram
-            )
-            or hasattr(c_class, cappela_attr)
+            capella_attr == DESCRIPTION_REFERENCE_SERIALIZER
+            or (is_diagram_elements and c_class == diagram.Diagram)
+            or hasattr(c_class, capella_attr)
         ):
             available_links.append(link)
         else:
             if is_global:
                 logger.info(
                     "Global link %s is not available on Capella type %s",
-                    cappela_attr,
+                    capella_attr,
                     c_type,
                 )
             else:
                 logger.error(
                     "Link %s is not available on Capella type %s",
-                    cappela_attr,
+                    capella_attr,
                     c_type,
                 )
     return available_links

capella2polarion/converters/document_renderer.py

@@ -433,7 +433,8 @@ def _render_full_authority_documents(
                         )
                     except Exception as e:
                         logger.error(
-                            "Rendering for document %s/%s failed with the following error",
+                            "Rendering for document %s/%s failed with the "
+                            "following error",
                             instance.polarion_space,
                             instance.polarion_name,
                             exc_info=e,
@@ -456,7 +457,8 @@ def _render_full_authority_documents(
                         )
                     except Exception as e:
                         logger.error(
-                            "Rendering for document %s/%s failed with the following error",
+                            "Rendering for document %s/%s failed with the "
+                            "following error",
                             instance.polarion_space,
                             instance.polarion_name,
                             exc_info=e,

capella2polarion/converters/link_converter.py

@@ -12,7 +12,6 @@
 import polarion_rest_api_client as polarion_api
 from capellambse.model import common
 from capellambse.model import diagram as diag
-from capellambse.model.crosslayer import fa
 
 import capella2polarion.converters.polarion_html_helper
 from capella2polarion import data_models
@@ -23,7 +22,7 @@
 
 TYPE_RESOLVERS = {"Part": lambda obj: obj.type.uuid}
 _Serializer: t.TypeAlias = cabc.Callable[
-    [common.GenericElement, str, str, str, dict[str, t.Any]],
+    [common.GenericElement, str, str, dict[str, t.Any]],
     list[polarion_api.WorkItemLink],
 ]
 
@@ -48,8 +47,6 @@ def __init__(
         self.serializers: dict[str, _Serializer] = {
             converter_config.DESCRIPTION_REFERENCE_SERIALIZER: self._handle_description_reference_links,  # pylint: disable=line-too-long
             converter_config.DIAGRAM_ELEMENTS_SERIALIZER: self._handle_diagram_reference_links,  # pylint: disable=line-too-long
-            "input_exchanges": self._handle_exchanges,
-            "output_exchanges": self._handle_exchanges,
         }
 
     def create_links_for_work_item(
@@ -60,21 +57,21 @@ def create_links_for_work_item(
         obj = converter_data.capella_element
         work_item = converter_data.work_item
         assert work_item is not None
+        assert work_item.id is not None
         new_links: list[polarion_api.WorkItemLink] = []
         link_errors: list[str] = []
         for link_config in converter_data.type_config.links:
-            assert (role_id := link_config.polarion_role) is not None
-            attr_id = link_config.capella_attr or ""
-            serializer = self.serializers.get(role_id)
+            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(
-                        serializer(obj, work_item.id, role_id, attr_id, {})
+                        serializer(obj, work_item.id, role_id, {})
                     )
                 else:
-                    refs = _resolve_attribute(obj, attr_id)
+                    refs = _resolve_attribute(obj, link_config.capella_attr)
                     new: cabc.Iterable[str]
                     if isinstance(refs, common.ElementList):
                         new = refs.by_uuid  # type: ignore[assignment]
@@ -89,9 +86,14 @@ def create_links_for_work_item(
                         self._create(work_item.id, role_id, new, {})
                     )
             except Exception as error:
-                request_text = f"Requested attribute: {attr_id}"
                 error_text = f"{type(error).__name__} {str(error)}"
-                link_errors.extend([request_text, error_text, "--------"])
+                link_errors.extend(
+                    [
+                        f"Requested attribute: {link_config.capella_attr}",
+                        error_text,
+                        "--------",
+                    ]
+                )
 
         if link_errors:
             for link_error in link_errors:
@@ -137,10 +139,8 @@ def _handle_description_reference_links(
         obj: common.GenericElement,
         work_item_id: str,
         role_id: str,
-        attr_id: str,
         links: dict[str, polarion_api.WorkItemLink],
     ) -> list[polarion_api.WorkItemLink]:
-        del attr_id
         refs = self.converter_session[obj.uuid].description_references
         ref_set = set(self._get_work_item_ids(work_item_id, refs, role_id))
         return self._create(work_item_id, role_id, ref_set, links)
@@ -150,10 +150,8 @@ def _handle_diagram_reference_links(
         obj: diag.Diagram,
         work_item_id: str,
         role_id: str,
-        attr_id: str,
         links: dict[str, polarion_api.WorkItemLink],
     ) -> list[polarion_api.WorkItemLink]:
-        del attr_id
         try:
             refs = set(self._collect_uuids(obj.nodes))
             refs = set(self._get_work_item_ids(work_item_id, refs, role_id))
@@ -199,20 +197,6 @@ def _create(
         ]
         return list(filter(None, _new_links))
 
-    def _handle_exchanges(
-        self,
-        obj: fa.Function,
-        work_item_id: str,
-        role_id: str,
-        attr_id: str,
-        links: dict[str, polarion_api.WorkItemLink],
-    ) -> list[polarion_api.WorkItemLink]:
-        exchanges: list[str] = []
-        objs = _resolve_attribute(obj, attr_id)
-        exs = self._get_work_item_ids(work_item_id, objs.by_uuid, role_id)
-        exchanges.extend(set(exs))
-        return self._create(work_item_id, role_id, exchanges, links)
-
     def create_grouped_link_fields(
         self,
         data: data_session.ConverterData,
@@ -247,11 +231,9 @@ def create_grouped_link_fields(
                     config = link_config
                     break
 
+            role_id = self._remove_prefix(role)
             self._create_link_fields(
-                work_item,
-                role.removeprefix(f"{self.role_prefix}_"),
-                grouped_links,
-                config=config,
+                work_item, role_id, grouped_links, config=config
             )
 
     def _create_link_fields(
@@ -334,12 +316,13 @@ def create_grouped_back_link_fields(
             List of links referencing work_item as secondary
         """
         for role, grouped_links in _group_by("role", links).items():
-            self._create_link_fields(
-                work_item,
-                role.removeprefix(f"{self.role_prefix}_"),
-                grouped_links,
-                True,
-            )
+            role_id = self._remove_prefix(role)
+            self._create_link_fields(work_item, role_id, grouped_links, True)
+
+    def _remove_prefix(self, role: str) -> str:
+        if self.role_prefix:
+            return role.removeprefix(f"{self.role_prefix}_")
+        return role
 
 
 def _group_by(

tests/test_cli.py

@@ -11,7 +11,8 @@
 from click import testing
 
 import capella2polarion.__main__ as main
-from capella2polarion.connectors.polarion_worker import CapellaPolarionWorker
+from capella2polarion.connectors import polarion_worker
+from capella2polarion.converters import model_converter
 
 # pylint: disable-next=relative-beyond-top-level, useless-suppression
 from .conftest import (  # type: ignore[import]
@@ -26,25 +27,31 @@ def test_migrate_model_elements(monkeypatch: pytest.MonkeyPatch):
     monkeypatch.setattr(polarion_api, "OpenAPIPolarionProjectClient", mock_api)
     mock_get_polarion_wi_map = mock.MagicMock()
     monkeypatch.setattr(
-        CapellaPolarionWorker,
+        polarion_worker.CapellaPolarionWorker,
         "load_polarion_work_item_map",
         mock_get_polarion_wi_map,
     )
+    mock_generate_work_items = mock.MagicMock()
+    monkeypatch.setattr(
+        model_converter.ModelConverter,
+        "generate_work_items",
+        mock_generate_work_items,
+    )
     mock_delete_work_items = mock.MagicMock()
     monkeypatch.setattr(
-        CapellaPolarionWorker,
+        polarion_worker.CapellaPolarionWorker,
         "delete_orphaned_work_items",
         mock_delete_work_items,
     )
     mock_post_work_items = mock.MagicMock()
     monkeypatch.setattr(
-        CapellaPolarionWorker,
+        polarion_worker.CapellaPolarionWorker,
         "create_missing_work_items",
         mock_post_work_items,
     )
     mock_patch_work_items = mock.MagicMock()
     monkeypatch.setattr(
-        CapellaPolarionWorker,
+        polarion_worker.CapellaPolarionWorker,
         "compare_and_update_work_items",
         mock_patch_work_items,
     )
@@ -68,6 +75,11 @@ def test_migrate_model_elements(monkeypatch: pytest.MonkeyPatch):
 
     assert result.exit_code == 0
     assert mock_get_polarion_wi_map.call_count == 1
+    assert mock_generate_work_items.call_count == 2
+    assert mock_generate_work_items.call_args_list[1][1] == {
+        "generate_links": True,
+        "generate_attachments": True,
+    }
     assert mock_delete_work_items.call_count == 1
     assert mock_patch_work_items.call_count == 1
     assert mock_post_work_items.call_count == 1
@@ -78,7 +90,7 @@ def test_render_documents(monkeypatch: pytest.MonkeyPatch):
     monkeypatch.setattr(polarion_api, "OpenAPIPolarionProjectClient", mock_api)
     mock_get_polarion_wi_map = mock.MagicMock()
     monkeypatch.setattr(
-        CapellaPolarionWorker,
+        polarion_worker.CapellaPolarionWorker,
         "load_polarion_work_item_map",
         mock_get_polarion_wi_map,
     )
@@ -89,32 +101,33 @@ def test_render_documents(monkeypatch: pytest.MonkeyPatch):
             module_name=name,
             home_page_content=polarion_api.TextContent(
                 "text/html",
-                '<h1 id="polarion_wiki macro name=module-workitem;params=id=TEST-123></h1>',
+                '<h1 id="polarion_wiki macro name=module-workitem;'
+                'params=id=TEST-123"></h1>',
             ),
         )
         if name == "id1236"
         else None
     )
     monkeypatch.setattr(
-        CapellaPolarionWorker,
+        polarion_worker.CapellaPolarionWorker,
         "get_document",
         mock_get_document,
     )
     mock_post_documents = mock.MagicMock()
     monkeypatch.setattr(
-        CapellaPolarionWorker,
+        polarion_worker.CapellaPolarionWorker,
         "post_documents",
         mock_post_documents,
     )
     mock_update_documents = mock.MagicMock()
     monkeypatch.setattr(
-        CapellaPolarionWorker,
+        polarion_worker.CapellaPolarionWorker,
         "update_documents",
         mock_update_documents,
     )
     mock_update_work_items = mock.MagicMock()
     monkeypatch.setattr(
-        CapellaPolarionWorker,
+        polarion_worker.CapellaPolarionWorker,
         "update_work_items",
         mock_update_work_items,
     )