diff --git a/capella2polarion/elements/element.py b/capella2polarion/elements/element.py index fe39b2ee..46b831e6 100644 --- a/capella2polarion/elements/element.py +++ b/capella2polarion/elements/element.py @@ -4,6 +4,7 @@ from __future__ import annotations import collections.abc as cabc +import functools import logging import typing as t from itertools import chain @@ -11,6 +12,7 @@ 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"), } diff --git a/capella2polarion/elements/serialize.py b/capella2polarion/elements/serialize.py index 674c3670..5c01e6cd 100644 --- a/capella2polarion/elements/serialize.py +++ b/capella2polarion/elements/serialize.py @@ -173,8 +173,7 @@ def replace_markup( f'id="fake" data-item-id="{pid}" data-option-id="long">' "" ) - 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, diff --git a/tests/test_elements.py b/tests/test_elements.py index d83b2954..4a070740 100644 --- a/tests/test_elements.py +++ b/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