fix: Fix patch work item

Merge pull request #39 from DSD-DBS/fix-patch-work-item

capella2polarion/__main__.py

@@ -20,6 +20,7 @@
 
 @click.group()
 @click.option("--debug", is_flag=True, default=False)
+@click.option("--force-update", is_flag=True, default=False)
 @click.option(
     "--polarion-project-id",
     type=str,
@@ -56,6 +57,7 @@
 def cli(
     ctx: click.core.Context,
     debug: bool,
+    force_update: bool,
     polarion_project_id: str,
     polarion_url: str,
     polarion_pat: str,
@@ -74,6 +76,7 @@ def cli(
         capella_diagram_cache_folder_path,
         capella_model,
         synchronize_config,
+        force_update,
     )
     capella2polarion_cli.setup_logger()
     ctx.obj = capella2polarion_cli
@@ -117,7 +120,9 @@ def synchronize(ctx: click.core.Context) -> None:
     )
 
     polarion_worker = pw.CapellaPolarionWorker(
-        capella_to_polarion_cli.polarion_params, capella_to_polarion_cli.config
+        capella_to_polarion_cli.polarion_params,
+        capella_to_polarion_cli.config,
+        capella_to_polarion_cli.force_update,
     )
 
     polarion_worker.load_polarion_work_item_map()

capella2polarion/cli.py

@@ -30,6 +30,7 @@ def __init__(
         capella_diagram_cache_folder_path: pathlib.Path | None,
         capella_model: capellambse.MelodyModel,
         synchronize_config_io: typing.TextIO,
+        force_update: bool = False,
     ) -> None:
         self.debug = debug
         self.polarion_params = pw.PolarionWorkerParams(
@@ -55,6 +56,7 @@ def __init__(
         self.synchronize_config_content: dict[str, typing.Any] = {}
         self.synchronize_config_roles: dict[str, list[str]] | None = None
         self.config = converter_config.ConverterConfig()
+        self.force_update = force_update
 
     def _none_save_value_string(self, value: str | None) -> str | None:
         return "None" if value is None else value

capella2polarion/connectors/polarion_worker.py

@@ -5,6 +5,7 @@
 
 import collections.abc as cabc
 import logging
+import typing as t
 from urllib import parse
 
 import polarion_rest_api_client as polarion_api
@@ -16,6 +17,13 @@
 logger = logging.getLogger(__name__)
 
 
+DEFAULT_ATTRIBUTE_VALUES: dict[type, t.Any] = {
+    str: "",
+    int: 0,
+    bool: False,
+}
+
+
 class PolarionWorkerParams:
     """Container for Polarion Params."""
 
@@ -35,10 +43,12 @@ def __init__(
         self,
         params: PolarionWorkerParams,
         config: converter_config.ConverterConfig,
+        force_update: bool = False,
     ) -> None:
         self.polarion_params = params
         self.polarion_data_repo = polarion_repo.PolarionDataRepository()
         self.config = config
+        self.force_update = force_update
 
         if (self.polarion_params.project_id is None) or (
             len(self.polarion_params.project_id) == 0
@@ -151,20 +161,40 @@ def patch_work_item(
         """Patch a given WorkItem."""
         new = converter_session[uuid].work_item
         _, old = self.polarion_data_repo[uuid]
-        if new == old:
+        if not self.force_update and new == old:
             return
 
         assert old is not None
         assert new is not None
 
         log_args = (old.id, new.type, new.title)
         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 = self.client.get_all_work_item_links(old.id)
-        new.type = None
+        del new.additional_attributes["uuid_capella"]
+
+        old = self.client.get_work_item(old.id)
+
+        if old.linked_work_items_truncated:
+            old.linked_work_items = self.client.get_all_work_item_links(old.id)
+
+        del old.additional_attributes["uuid_capella"]
+
+        # Type will only be updated, if it is set and should be used carefully
+        if new.type == old.type:
+            new.type = None
         new.status = "open"
+
+        # If additional fields were present in the past, but aren't anymore,
+        # we have to set them to an empty value manually
+        defaults = DEFAULT_ATTRIBUTE_VALUES
+        for attribute, value in old.additional_attributes.items():
+            if attribute not in new.additional_attributes:
+                new.additional_attributes[attribute] = defaults.get(
+                    type(value)
+                )
+            elif new.additional_attributes[attribute] == value:
+                del new.additional_attributes[attribute]
+
         assert new.id is not None
         try:
             self.client.update_work_item(new)

ci-templates/gitlab/synchronise_elements.yml

@@ -15,6 +15,7 @@ capella2polarion_synchronise_elements:
       python \
         -m capella2polarion \
         $([[ $CAPELLA2POLARION_DEBUG -eq 1 ]] && echo '--debug') \
+        $([[ $CAPELLA2POLARION_FORCE_UPDATE -eq 1 ]] && echo '--force-update') \
         --polarion-project-id=${CAPELLA2POLARION_PROJECT_ID:?} \
         --capella-model="${CAPELLA2POLARION_MODEL_JSON:?}" \
         --capella-diagram-cache-folder-path=./diagram_cache \

tests/test_elements.py

@@ -672,12 +672,21 @@ def test_update_work_items(
 
         del base_object.mc.converter_session["uuid2"]
 
+        get_work_item_mock = mock.MagicMock()
+        get_work_item_mock.return_value = polarion_work_item_list[0]
+        monkeypatch.setattr(
+            base_object.pw.client,
+            "get_work_item",
+            get_work_item_mock,
+        )
+
         base_object.pw.patch_work_items(base_object.mc.converter_session)
         assert base_object.pw.client is not None
-        assert base_object.pw.client.get_all_work_item_links.call_count == 1
+        assert base_object.pw.client.get_all_work_item_links.call_count == 0
         assert base_object.pw.client.delete_work_item_links.call_count == 0
         assert base_object.pw.client.create_work_item_links.call_count == 0
         assert base_object.pw.client.update_work_item.call_count == 1
+        assert base_object.pw.client.get_work_item.call_count == 1
         work_item = base_object.pw.client.update_work_item.call_args[0][0]
         assert isinstance(work_item, data_models.CapellaWorkItem)
         assert work_item.id == "Obj-1"
@@ -719,6 +728,38 @@ def test_update_work_items_filters_work_items_with_same_checksum(
         assert base_object.pw.client is not None
         assert base_object.pw.client.update_work_item.call_count == 0
 
+    @staticmethod
+    def test_update_work_items_same_checksum_force(
+        base_object: BaseObjectContainer,
+    ):
+        base_object.pw.force_update = True
+        base_object.pw.polarion_data_repo.update_work_items(
+            [
+                data_models.CapellaWorkItem(
+                    id="Obj-1",
+                    uuid_capella="uuid1",
+                    status="open",
+                    checksum=TEST_WI_CHECKSUM,
+                    type="fakeModelObject",
+                )
+            ]
+        )
+        base_object.mc.converter_session[
+            "uuid1"
+        ].work_item = data_models.CapellaWorkItem(
+            id="Obj-1",
+            uuid_capella="uuid1",
+            status="open",
+            type="fakeModelObject",
+        )
+
+        del base_object.mc.converter_session["uuid2"]
+
+        base_object.pw.patch_work_items(base_object.mc.converter_session)
+
+        assert base_object.pw.client is not None
+        assert base_object.pw.client.update_work_item.call_count == 1
+
     @staticmethod
     def test_update_links_with_no_elements(base_object: BaseObjectContainer):
         base_object.pw.polarion_data_repo = (
@@ -774,6 +815,21 @@ def test_update_links(base_object: BaseObjectContainer):
         base_object.mc.generate_work_item_links(
             base_object.pw.polarion_data_repo
         )
+
+        work_item_1 = data_models.CapellaWorkItem(
+            **base_object.pw.polarion_data_repo["uuid1"][1].to_dict()
+        )
+        work_item_2 = data_models.CapellaWorkItem(
+            **base_object.pw.polarion_data_repo["uuid2"][1].to_dict()
+        )
+        work_item_1.linked_work_items_truncated = True
+        work_item_2.linked_work_items_truncated = True
+
+        base_object.pw.client.get_work_item.side_effect = (
+            work_item_1,
+            work_item_2,
+        )
+
         base_object.pw.patch_work_items(base_object.mc.converter_session)
         assert base_object.pw.client is not None
         links = base_object.pw.client.get_all_work_item_links.call_args_list