merge: Merge pull request #26 from DSD-DBS/grouped-links-attribute-re…

…based

feat: Grouped Links Attributes

capella2polarion/elements/__init__.py

@@ -12,7 +12,6 @@
     "STATUS_DELETE",
 ]
 
-import functools
 import logging
 import pathlib
 import typing as t
@@ -134,39 +133,41 @@ def patch_work_items(ctx: dict[str, t.Any]) -> None:
     ctx
         The context for the workitem operation to be processed.
     """
-    work_items_lookup = ctx["POLARION_WI_MAP"] | ctx["WORK_ITEMS"]
-
-    def add_content(
-        obj: common.GenericElement | diag.Diagram,
-        _: dict[str, t.Any],
-        **kwargs,
-    ) -> serialize.CapellaWorkItem:
-        work_item = work_items_lookup[obj.uuid]
-        for key, value in kwargs.items():
-            if getattr(work_item, key, None) is None:
-                continue
-
-            setattr(work_item, key, value)
-        return work_item
-
     ctx["POLARION_ID_MAP"] = uuids = {
         uuid: wi.id
         for uuid, wi in ctx["POLARION_WI_MAP"].items()
         if wi.status == "open" and wi.uuid_capella and wi.id
     }
+
+    back_links: dict[str, list[polarion_api.WorkItemLink]] = {}
     for uuid in uuids:
-        elements = ctx["MODEL"]
+        objects = ctx["MODEL"]
         if uuid.startswith("_"):
-            elements = ctx["MODEL"].diagrams
-        obj = elements.by_uuid(uuid)
+            objects = ctx["MODEL"].diagrams
+
+        obj = objects.by_uuid(uuid)
+        work_item: serialize.CapellaWorkItem = ctx["WORK_ITEMS"][uuid]
+        old_work_item: serialize.CapellaWorkItem = ctx["POLARION_WI_MAP"][uuid]
 
         links = element.create_links(obj, ctx)
+        work_item.linked_work_items = links
+        work_item.id = old_work_item.id
+
+        element.create_grouped_link_fields(work_item, back_links)
+
+    for uuid in uuids:
+        new_work_item: serialize.CapellaWorkItem = ctx["WORK_ITEMS"][uuid]
+        old_work_item = ctx["POLARION_WI_MAP"][uuid]
+        if old_work_item.id in back_links:
+            element.create_grouped_back_link_fields(
+                new_work_item, back_links[old_work_item.id]
+            )
 
         api_helper.patch_work_item(
-            ctx,
-            obj,
-            functools.partial(add_content, linked_work_items=links),
-            obj._short_repr_(),
+            ctx["API"],
+            new_work_item,
+            old_work_item,
+            old_work_item.title,
             "element",
         )
 

capella2polarion/elements/api_helper.py

@@ -6,81 +6,71 @@
 import typing as t
 
 import polarion_rest_api_client as polarion_api
-from capellambse.model import common
 
 from capella2polarion.elements import serialize
 
 logger = logging.getLogger(__name__)
 
 
 def patch_work_item(
-    ctx: dict[str, t.Any],
-    obj: common.GenericElement,
-    receiver: cabc.Callable[
-        [t.Any, dict[str, t.Any]], serialize.CapellaWorkItem
-    ],
+    api: polarion_api.OpenAPIPolarionProjectClient,
+    new: serialize.CapellaWorkItem,
+    old: serialize.CapellaWorkItem,
     name: str,
     _type: str,
 ):
     """Patch a given WorkItem.
 
     Parameters
     ----------
-    ctx
+    api
         The context to execute the patch for.
-    obj
-        The Capella object to update the WorkItem from.
-    receiver
-        A function that receives the WorkItem from the created
-        instances. This function alters the WorkItem instances by adding
-        attributes, e.g.: `linked_work_items`. It can be useful to add
-        attributes which can only be computed after the work item and
-        its default attributes were instantiated.
+    new
+        The updated CapellaWorkItem
+    old
+        The CapellaWorkItem currently present on polarion
     name
         The name of the object, which should be displayed in log
         messages.
     _type
         The type of element, which should be shown in log messages.
     """
-    if new := receiver(obj, ctx):
-        wid = ctx["POLARION_ID_MAP"][obj.uuid]
-        old: serialize.CapellaWorkItem = ctx["POLARION_WI_MAP"][obj.uuid]
-        if new == old:
-            return
+    if new == old:
+        return
 
-        log_args = (wid, _type, name)
-        logger.info("Update work item %r for model %s %r...", *log_args)
-        if "uuid_capella" in new.additional_attributes:
-            del new.additional_attributes["uuid_capella"]
+    log_args = (old.id, _type, name)
+    logger.info("Update work item %r for model %s %r...", *log_args)
+    if "uuid_capella" in new.additional_attributes:
+        del new.additional_attributes["uuid_capella"]
 
-        old.linked_work_items = ctx["API"].get_all_work_item_links(old.id)
-        new.type = None
-        new.status = "open"
-        new.id = wid
-        try:
-            ctx["API"].update_work_item(new)
-            handle_links(
-                old.linked_work_items,
-                new.linked_work_items,
-                ("Delete", _type, name),
-                ctx["API"].delete_work_item_links,
-            )
-            handle_links(
-                new.linked_work_items,
-                old.linked_work_items,
-                ("Create", _type, name),
-                ctx["API"].create_work_item_links,
-            )
-        except polarion_api.PolarionApiException as error:
-            wi = f"{wid}({_type} {name})"
-            logger.error("Updating work item %r failed. %s", wi, error.args[0])
+    old.linked_work_items = api.get_all_work_item_links(old.id)
+    new.type = None
+    new.status = "open"
+    new.id = old.id
+    try:
+        api.update_work_item(new)
+        handle_links(
+            old.linked_work_items,
+            new.linked_work_items,
+            ("Delete", _type, name),
+            api.delete_work_item_links,
+        )
+        handle_links(
+            new.linked_work_items,
+            old.linked_work_items,
+            ("Create", _type, name),
+            api.create_work_item_links,
+        )
+    except polarion_api.PolarionApiException as error:
+        wi = f"{old.id}({_type} {name})"
+        logger.error("Updating work item %r failed. %s", wi, error.args[0])
 
 
 def handle_links(
     left: cabc.Iterable[polarion_api.WorkItemLink],
     right: cabc.Iterable[polarion_api.WorkItemLink],
     log_args: tuple[str, ...],
-    handler: cabc.Callable[[cabc.Iterable[polarion_api.WorkItemLink]], None],
+    handler: cabc.Callable[[cabc.Iterable[polarion_api.WorkItemLink]], t.Any],
 ):
     """Handle work item links on Polarion."""
     for link in (links := get_links(left, right)):

capella2polarion/elements/element.py

@@ -7,6 +7,7 @@
 import functools
 import logging
 import typing as t
+from collections import defaultdict
 from itertools import chain
 
 import polarion_rest_api_client as polarion_api
@@ -45,7 +46,7 @@ def create_work_items(
 
     _work_items = list(filter(None, _work_items))
     valid_types = set(map(helpers.resolve_element_type, set(ctx["ELEMENTS"])))
-    work_items: list[polarion_api.CapellaWorkItem] = []
+    work_items: list[serialize.CapellaWorkItem] = []
     missing_types: set[str] = set()
     for work_item in _work_items:
         assert work_item is not None
@@ -203,6 +204,92 @@ def _handle_exchanges(
     return _create(context, wid, role_id, exchanges, links)
 
 
+def create_grouped_link_fields(
+    work_item: serialize.CapellaWorkItem,
+    back_links: dict[str, list[polarion_api.WorkItemLink]] | None = None,
+):
+    """Create the grouped link work items fields from the primary work item.
+
+    Parameters
+    ----------
+    work_item
+        WorkItem to create the fields for.
+    back_links
+        A dictionary of secondary WorkItem IDs to links to create
+        backlinks later.
+    """
+    wi = f"[{work_item.id}]({work_item.type} {work_item.title})"
+    logger.debug("Building grouped links for work item %r...", wi)
+    for role, grouped_links in _group_by(
+        "role", work_item.linked_work_items
+    ).items():
+        if back_links is not None:
+            for link in grouped_links:
+                key = link.secondary_work_item_id
+                back_links.setdefault(key, []).append(link)
+
+        _create_link_fields(work_item, role, grouped_links)
+
+
+def create_grouped_back_link_fields(
+    work_item: serialize.CapellaWorkItem,
+    links: list[polarion_api.WorkItemLink],
+):
+    """Create backlinks for the given WorkItem using a list of backlinks.
+
+    Parameters
+    ----------
+    work_item
+        WorkItem to create the fields for
+    links
+        List of links referencing work_item as secondary
+    """
+    for role, grouped_links in _group_by("role", links).items():
+        _create_link_fields(work_item, role, grouped_links, True)
+
+
+def _group_by(
+    attr: str,
+    links: cabc.Iterable[polarion_api.WorkItemLink],
+) -> dict[str, list[polarion_api.WorkItemLink]]:
+    group = defaultdict(list)
+    for link in links:
+        key = getattr(link, attr)
+        group[key].append(link)
+    return group
+
+
+def _make_url_list(
+    links: cabc.Iterable[polarion_api.WorkItemLink], reverse: bool = False
+) -> str:
+    urls: list[str] = []
+    for link in links:
+        if reverse:
+            pid = link.primary_work_item_id
+        else:
+            pid = link.secondary_work_item_id
+
+        url = serialize.POLARION_WORK_ITEM_URL.format(pid=pid)
+        urls.append(f"<li>{url}</li>")
+
+    urls.sort()
+    url_list = "\n".join(urls)
+    return f"<ul>{url_list}</ul>"
+
+
+def _create_link_fields(
+    work_item: serialize.CapellaWorkItem,
+    role: str,
+    links: list[polarion_api.WorkItemLink],
+    reverse: bool = False,
+):
+    role = f"{role}_reverse" if reverse else role
+    work_item.additional_attributes[role] = {
+        "type": "text/html",
+        "value": _make_url_list(links, reverse),
+    }
+
+
 CustomLinkMaker = cabc.Callable[
     [
         dict[str, t.Any],

capella2polarion/elements/serialize.py

@@ -29,6 +29,11 @@
 )
 RE_CAMEL_CASE_2ND_WORD_PATTERN = re.compile(r"([a-z]+)([A-Z][a-z]+)")
 DIAGRAM_STYLES = {"max-width": "100%"}
+POLARION_WORK_ITEM_URL = (
+    '<span class="polarion-rte-link" data-type="workItem" '
+    'id="fake" data-item-id="{pid}" data-option-id="long">'
+    "</span>"
+)
 
 PrePostConditionElement = t.Union[
     oa.OperationalCapability, interaction.Scenario
@@ -213,11 +218,7 @@ def replace_markup(
     uuid = match.group(1)
     if pid := ctx["POLARION_ID_MAP"].get(uuid):
         referenced_uuids.append(uuid)
-        return (
-            '<span class="polarion-rte-link" data-type="workItem" '
-            f'id="fake" data-item-id="{pid}" data-option-id="long">'
-            "</span>"
-        )
+        return POLARION_WORK_ITEM_URL.format(pid=pid)
     return non_matcher(match.group(0))
 
 

tests/conftest.py

@@ -8,8 +8,12 @@
 import typing as t
 
 import capellambse
+import markupsafe
+import polarion_rest_api_client as polarion_api
 import pytest
 
+from capella2polarion.elements import serialize
+
 TEST_DATA_ROOT = pathlib.Path(__file__).parent / "data"
 TEST_DIAGRAM_CACHE = TEST_DATA_ROOT / "diagram_cache"
 TEST_MODEL_ELEMENTS = TEST_DATA_ROOT / "model_elements"
@@ -29,3 +33,26 @@ def diagram_cache_index() -> list[dict[str, t.Any]]:
 def model() -> capellambse.MelodyModel:
     """Return the test model."""
     return capellambse.MelodyModel(path=TEST_MODEL)
+
+
+@pytest.fixture
+def dummy_work_items() -> dict[str, serialize.CapellaWorkItem]:
+    return {
+        f"uuid{i}": serialize.CapellaWorkItem(
+            id=f"Obj-{i}",
+            uuid_capella=f"uuid{i}",
+            title=f"Fake {i}",
+            type="fakeModelObject",
+            description_type="text/html",
+            description=markupsafe.Markup(""),
+            linked_work_items=[
+                polarion_api.WorkItemLink(
+                    f"Obj-{i}", f"Obj-{j}", "attribute", True, "project_id"
+                )
+                for j in range(3)
+                if (i not in (j, 2))
+            ],
+            status="open",
+        )
+        for i in range(3)
+    }