feat(model-elements): Add input|output_exchanges link handler

capella2polarion/elements/element.py

@@ -4,13 +4,15 @@
 from __future__ import annotations
 
 import collections.abc as cabc
+import functools
 import logging
 import typing as t
 from itertools import chain
 
 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
 
 from capella2polarion import elements
 from capella2polarion.elements import helpers, serialize
@@ -80,7 +82,7 @@ def create_links(
             continue
 
         if isinstance(refs, common.ElementList):
-            new = refs.by_uuid
+            new: cabc.Iterable[str] = refs.by_uuid  # type: ignore[assignment]
         else:
             assert hasattr(refs, "uuid")
             new = [refs.uuid]
@@ -141,7 +143,9 @@ def _handle_diagram_reference_links(
     return ref_links
 
 
-def _collect_uuids(nodes: list[common.GenericElement]) -> cabc.Iterator[str]:
+def _collect_uuids(
+    nodes: cabc.Iterable[common.GenericElement],
+) -> cabc.Iterator[str]:
     type_resolvers = TYPE_RESOLVERS
     for node in nodes:
         uuid = node.uuid
@@ -171,7 +175,34 @@ def _create(
     return list(filter(None.__ne__, _new_links))
 
 
-CUSTOM_LINKS = {
-    "description_reference": _handle_description_reference_links,
-    "diagram_elements": _handle_diagram_reference_links,
+def _handle_exchanges(
+    context: dict[str, t.Any],
+    obj: fa.Function,
+    role_id: str,
+    links: dict[str, polarion_api.WorkItemLink],
+    attr: str = "inputs",
+) -> list[polarion_api.WorkItemLink]:
+    wid = context["POLARION_ID_MAP"][obj.uuid]
+    exchanges: list[str] = []
+    for element in getattr(obj, attr):
+        uuids = element.exchanges.by_uuid
+        exs = _get_work_item_ids(context, wid, uuids, role_id)
+        exchanges.extend(set(exs))
+    return _create(context, wid, role_id, exchanges, links)
+
+
+CustomLinkMaker = cabc.Callable[
+    [
+        dict[str, t.Any],
+        diag.Diagram | common.GenericElement,
+        str,
+        dict[str, t.Any],
+    ],
+    list[polarion_api.WorkItemLink],
+]
+CUSTOM_LINKS: dict[str, CustomLinkMaker] = {
+    "description_reference": _handle_description_reference_links,  # type: ignore[dict-item]
+    "diagram_elements": _handle_diagram_reference_links,  # type: ignore[dict-item]
+    "input_exchanges": functools.partial(_handle_exchanges, attr="inputs"),
+    "output_exchanges": functools.partial(_handle_exchanges, attr="outputs"),
 }

capella2polarion/elements/serialize.py

@@ -173,8 +173,7 @@ def replace_markup(
             f'id="fake" data-item-id="{pid}" data-option-id="long">'
             "</span>"
         )
-    else:
-        return non_matcher(match.group(0))
+    return non_matcher(match.group(0))
 
 
 def include_pre_and_post_condition(
@@ -262,7 +261,10 @@ def physical_component(
     return work_item
 
 
-SERIALIZERS = {
+Serializer = cabc.Callable[
+    [common.GenericElement, dict[str, t.Any]], CapellaWorkItem
+]
+SERIALIZERS: dict[str, Serializer] = {
     "CapabilityRealization": include_pre_and_post_condition,
     "LogicalComponent": component_or_actor,
     "OperationalCapability": include_pre_and_post_condition,

tests/test_elements.py

@@ -137,7 +137,7 @@ def __init__(
         self,
         uuid: str,
         name: str = "",
-        attribute: FakeModelObject | None = None,
+        attribute: t.Any | None = None,
     ):
         self.uuid = uuid
         self.name = name
@@ -237,6 +237,26 @@ def test_create_links_custom_resolver(context: dict[str, t.Any]):
 
         assert links == [expected]
 
+    @staticmethod
+    def test_create_links_custom_exchanges_resolver(context: dict[str, t.Any]):
+        function_uuid = "ceffa011-7b66-4b3c-9885-8e075e312ffa"
+        obj = context["MODEL"].by_uuid(function_uuid)
+        context["POLARION_ID_MAP"][function_uuid] = "Obj-1"
+        context["POLARION_ID_MAP"][
+            "1a414995-f4cd-488c-8152-486e459fb9de"
+        ] = "Obj-2"
+        context["ROLES"] = {"SystemFunction": ["input_exchanges"]}
+        expected = polarion_api.WorkItemLink(
+            "Obj-1",
+            "Obj-2",
+            "input_exchanges",
+            secondary_work_item_project="project_id",
+        )
+
+        links = element.create_links(obj, context)
+
+        assert links == [expected]
+
     @staticmethod
     def test_create_links_missing_attribute(
         context: dict[str, t.Any], caplog: pytest.LogCaptureFixture
@@ -282,7 +302,7 @@ def test_create_links_from_ElementList(context: dict[str, t.Any]):
             secondary_work_item_project="project_id",
         )
 
-        links = element.create_links(obj, context)
+        links = element.create_links(obj, context)  # type: ignore[arg-type]
 
         assert expected_link in links
         assert expected_link1 in links