fix: Fix various bugs

- Fix debug log message for empty missing types
- Merge diagram work item creator into generic work item creator
- Handle all roles from the wildcard entry in the config. Also include
`Diagram`.

capella2polarion/__main__.py

@@ -79,30 +79,67 @@ def _get_roles_from_config(ctx: dict[str, t.Any]) -> dict[str, list[str]]:
                 roles[key] = list(role_ids)
         else:
             roles[typ] = []
-    roles["Diagram"] = ["diagram_elements"]
     return roles
 
 
 def _sanitize_config(
-    config: dict[str, list[str | dict[str, t.Any]]], special: dict[str, t.Any]
+    config: dict[str, list[str | dict[str, t.Any]]],
+    special: list[str | dict[str, t.Any]],
 ) -> dict[str, t.Any]:
+    special_config: dict[str, t.Any] = {}
+    for typ in special:
+        if isinstance(typ, str):
+            special_config[typ] = None
+        else:
+            special_config.update(typ)
+
+    lookup: dict[str, dict[str, list[str]]] = {}
+    for layer, xtypes in config.items():
+        for xt in xtypes:
+            if isinstance(xt, str):
+                item: dict[str, list[str]] = {xt: []}
+            else:
+                item = xt
+
+            lookup.setdefault(layer, {}).update(item)
+
     new_config: dict[str, t.Any] = {}
     for layer, xtypes in config.items():
         new_entries: list[str | dict[str, t.Any]] = []
         for xtype in xtypes:
             if isinstance(xtype, dict):
                 for sub_key, sub_value in xtype.items():
                     new_value = (
-                        special.get("*", [])
-                        + special.get(sub_key, [])
+                        special_config.get("*", [])
+                        + special_config.get(sub_key, [])
                         + sub_value
                     )
                     new_entries.append({sub_key: new_value})
             else:
-                if new_value := special.get("*", []) + special.get(xtype, []):
+                star = special_config.get("*", [])
+                special_xtype = special_config.get(xtype, [])
+                if new_value := star + special_xtype:
                     new_entries.append({xtype: new_value})
                 else:
                     new_entries.append(xtype)
+
+        wildcard_values = special_config.get("*", [])
+        for key, value in special_config.items():
+            if key == "*":
+                continue
+
+            if isinstance(value, list):
+                new_value = (
+                    lookup.get(layer, {}).get(key, [])
+                    + wildcard_values
+                    + value
+                )
+                new_entries.append({key: new_value})
+            elif value is None and key not in [
+                entry if isinstance(entry, str) else list(entry.keys())[0]
+                for entry in new_entries
+            ]:
+                new_entries.append({key: wildcard_values})
         new_config[layer] = new_entries
 
     return new_config
@@ -189,16 +226,20 @@ def model_elements(
     ctx.obj["POLARION_ID_MAP"] = {
         uuid: wi.id for uuid, wi in ctx.obj["POLARION_WI_MAP"].items()
     }
-    diagrams = ctx.obj["ELEMENTS"].pop("Diagram", [])
-    work_items = elements.element.create_work_items(ctx.obj)
-    ctx.obj["ELEMENTS"]["Diagram"] = diagrams
-    pdiagrams = elements.diagram.create_diagrams(ctx.obj)
-    ctx.obj["WORK_ITEMS"] = {
-        wi.uuid_capella: wi for wi in work_items + pdiagrams
+    duuids = {
+        diag["uuid"] for diag in ctx.obj["DIAGRAM_IDX"] if diag["success"]
     }
+    ctx.obj["ELEMENTS"]["Diagram"] = [
+        diag for diag in ctx.obj["ELEMENTS"]["Diagram"] if diag.uuid in duuids
+    ]
+    work_items = elements.element.create_work_items(ctx.obj)
+    ctx.obj["WORK_ITEMS"] = {wi.uuid_capella: wi for wi in work_items}
 
     elements.delete_work_items(ctx.obj)
     elements.post_work_items(ctx.obj)
+
+    work_items = elements.element.create_work_items(ctx.obj)
+    ctx.obj["WORK_ITEMS"] = {wi.uuid_capella: wi for wi in work_items}
     elements.patch_work_items(ctx.obj)
 
     elements.make_model_elements_index(ctx.obj)

capella2polarion/elements/__init__.py

@@ -114,6 +114,11 @@ def post_work_items(ctx: dict[str, t.Any]) -> None:
     if work_items:
         try:
             ctx["API"].create_work_items(work_items)
+            workitems = {wi.uuid_capella: wi for wi in work_items if wi.id}
+            ctx["POLARION_WI_MAP"].update(workitems)
+            ctx["POLARION_ID_MAP"] = {
+                uuid: wi.id for uuid, wi in ctx["POLARION_WI_MAP"].items()
+            }
         except polarion_api.PolarionApiException as error:
             logger.error("Creating work items failed. %s", error.args[0])
 
@@ -140,21 +145,16 @@ def add_content(
             setattr(work_item, key, value)
         return work_item
 
-    ctx["POLARION_WI_MAP"] = get_polarion_wi_map(ctx)
     ctx["POLARION_ID_MAP"] = uuids = {
         uuid: wi.id
         for uuid, wi in ctx["POLARION_WI_MAP"].items()
-        if wi.status == "open" and uuid in ctx["WORK_ITEMS"]
+        if wi.status == "open" and wi.uuid_capella and wi.id
     }
     for uuid in uuids:
         elements = ctx["MODEL"]
         if uuid.startswith("_"):
             elements = ctx["MODEL"].diagrams
-        try:
-            obj = elements.by_uuid(uuid)
-        except KeyError:
-            logger.error("Weird %r", uuid)
-            continue
+        obj = elements.by_uuid(uuid)
 
         links = element.create_links(obj, ctx)
 
@@ -189,6 +189,9 @@ def get_elements_and_type_map(
             if isinstance(typ, dict):
                 typ = list(typ.keys())[0]
 
+            if typ == "Diagram":
+                continue
+
             xtype = convert_type.get(typ, typ)
             objects = ctx["MODEL"].search(xtype, below=below)
             elements.setdefault(typ, []).extend(objects)
@@ -277,7 +280,6 @@ def make_model_elements_index(ctx: dict[str, t.Any]) -> None:
 
 from . import (  # pylint: disable=cyclic-import
     api_helper,
-    diagram,
     element,
     helpers,
     serialize,

capella2polarion/elements/element.py

@@ -28,12 +28,21 @@
 def create_work_items(
     ctx: dict[str, t.Any]
 ) -> list[serialize.CapellaWorkItem]:
-    """Create a set of work items in Polarion."""
+    """Create a list of work items for Polarion."""
     objects = chain.from_iterable(ctx["ELEMENTS"].values())
-    _work_items = [
-        serialize.element(obj, ctx, serialize.generic_work_item)
-        for obj in objects
+    _work_items = []
+    serializer: cabc.Callable[
+        [diag.Diagram | common.GenericElement, dict[str, t.Any]],
+        serialize.CapellaWorkItem,
     ]
+    for obj in objects:
+        if isinstance(obj, diag.Diagram):
+            serializer = serialize.diagram
+        else:
+            serializer = serialize.generic_work_item
+
+        _work_items.append(serialize.element(obj, ctx, serializer))
+
     _work_items = list(filter(None.__ne__, _work_items))
     valid_types = set(map(helpers.resolve_element_type, set(ctx["ELEMENTS"])))
     work_items: list[polarion_api.CapellaWorkItem] = []
@@ -44,10 +53,12 @@ def create_work_items(
             work_items.append(work_item)
         else:
             missing_types.add(work_item.type)
-    logger.debug(
-        "%r are missing in the capella2polarion configuration",
-        ", ".join(missing_types),
-    )
+
+    if missing_types:
+        logger.debug(
+            "%r are missing in the capella2polarion configuration",
+            ", ".join(missing_types),
+        )
     return work_items
 
 
@@ -118,7 +129,7 @@ def _handle_description_reference_links(
     role_id: str,
     links: dict[str, polarion_api.WorkItemLink],
 ) -> list[polarion_api.WorkItemLink]:
-    refs = context["DESCR_REFERENCES"].get(obj.uuid)
+    refs = context["DESCR_REFERENCES"].get(obj.uuid, [])
     wid = context["POLARION_ID_MAP"][obj.uuid]
     refs = set(_get_work_item_ids(context, wid, refs, role_id))
     return _create(context, wid, role_id, refs, links)

tests/data/model_elements/config.yaml

@@ -2,11 +2,14 @@
 # SPDX-License-Identifier: Apache-2.0
 
 "*": # All layers
-  "*": # All class types
-    - parent # Specify workitem links
-    - description_reference # Custom attribute
-  Class:
-    - state_machines
+  - "*": # All class types
+      - parent # Specify workitem links
+      - description_reference # Custom attribute
+  - Class:
+      - state_machines
+  - Diagram:
+      - diagram_elements
+  - Constraint
 
 oa: # Specify below
   - OperationalCapability: # Capella Type with references
@@ -19,7 +22,6 @@ oa: # Specify below
   - CommunicationMean
   - Class
   - StateMachine
-  - Constraint
 
 sa:
   - SystemComponent:
@@ -34,7 +36,6 @@ sa:
       - exchanged_items
   - ExchangeItem
   - Class
-  - Constraint
 
 pa:
   - PhysicalComponent:

tests/test_cli.py

@@ -67,10 +67,6 @@ def test_migrate_model_elements(monkeypatch: pytest.MonkeyPatch):
             {},
         ),
     )
-    mock_create_diagrams = mock.MagicMock()
-    monkeypatch.setattr(
-        elements.diagram, "create_diagrams", mock_create_diagrams
-    )
     mock_delete_work_items = mock.MagicMock()
     monkeypatch.setattr(elements, "delete_work_items", mock_delete_work_items)
     mock_post_work_items = mock.MagicMock()
@@ -93,5 +89,4 @@ def test_migrate_model_elements(monkeypatch: pytest.MonkeyPatch):
     assert mock_delete_work_items.call_count == 1
     assert mock_patch_work_items.call_count == 1
     assert mock_post_work_items.call_count == 1
-    assert mock_create_diagrams.call_count == 1
     assert ELEMENTS_IDX_PATH.exists()

tests/test_elements.py

@@ -14,7 +14,7 @@
 from capellambse.model import common
 
 from capella2polarion import elements
-from capella2polarion.elements import diagram, element, helpers, serialize
+from capella2polarion.elements import element, helpers, serialize
 
 # pylint: disable-next=relative-beyond-top-level, useless-suppression
 from .conftest import TEST_DIAGRAM_CACHE, TEST_HOST  # type: ignore[import]
@@ -90,7 +90,7 @@ def context(
     def test_create_diagrams(context: dict[str, t.Any]):
         context["ELEMENTS"] = {"Diagram": context["MODEL"].diagrams}
 
-        diagrams = diagram.create_diagrams(context)
+        diagrams = element.create_work_items(context)
 
         assert len(diagrams) == 1
         work_item = diagrams[0]
@@ -116,7 +116,7 @@ def test_create_diagrams_filters_non_diagram_elements(
         attributes.return_value = None
         monkeypatch.setattr(serialize, "element", attributes)
 
-        diagram.create_diagrams(context)
+        element.create_work_items(context)
 
         assert context["API"].create_work_items.call_count == 0
 
@@ -385,7 +385,7 @@ def test_update_work_items_filters_work_items_with_same_checksum(
 
     @staticmethod
     def test_update_links_with_no_elements(context: dict[str, t.Any]):
-        context["POLARION_ID_MAP"] = {}
+        context["POLARION_WI_MAP"] = {}
 
         elements.patch_work_items(context)