feat: Generate WorkItems only once (#123)

capella2polarion/__main__.py

@@ -150,8 +150,6 @@ def synchronize(
 
     polarion_worker.load_polarion_work_item_map()
 
-    converter.generate_work_items(polarion_worker.polarion_data_repo)
-
     polarion_worker.delete_orphaned_work_items(converter.converter_session)
     polarion_worker.create_missing_work_items(converter.converter_session)
 

capella2polarion/connectors/polarion_repo.py

@@ -48,7 +48,7 @@ def __contains__(self, item: str) -> bool:
         """Return True, if the given capella UUID is in the repository."""
         return item in self._id_mapping
 
-    def __sizeof__(self) -> int:
+    def __len__(self) -> int:
         """Return the amount of registered Capella UUIDs."""
         return len(self._id_mapping)
 

capella2polarion/connectors/polarion_worker.py

@@ -159,22 +159,15 @@ def create_missing_work_items(
         self, converter_session: data_session.ConverterSession
     ) -> None:
         """Post work items in a Polarion project."""
-        missing_work_items: list[data_models.CapellaWorkItem] = []
-        for uuid, converter_data in converter_session.items():
-            if not (work_item := converter_data.work_item):
-                logger.warning(
-                    "Expected to find a WorkItem for %s, but there is none",
-                    uuid,
-                )
-                continue
-
-            assert work_item is not None
-            if work_item.uuid_capella in self.polarion_data_repo:
-                continue
-
-            work_item.calculate_checksum()
-            missing_work_items.append(work_item)
-            logger.info("Create work item for %r...", work_item.title)
+        missing_work_items = [
+            data_models.CapellaWorkItem(
+                title=converter_data.capella_element.name,
+                type=converter_data.type_config.p_type,
+                uuid_capella=uuid,
+            )
+            for uuid, converter_data in converter_session.items()
+            if uuid not in self.polarion_data_repo
+        ]
         if missing_work_items:
             try:
                 self.project_client.work_items.create(missing_work_items)

pyproject.toml

@@ -29,10 +29,10 @@ classifiers = [
 ]
 dependencies = [
   "capellambse>=0.6.6,<0.7",
-  "capellambse_context_diagrams>=0.4.0",
+  "capellambse_context_diagrams>=0.5.0,<0.6",
   "click",
   "PyYAML",
-  "polarion-rest-api-client==1.2.0",
+  "polarion-rest-api-client==1.2.1",
   "bidict",
   "cairosvg",
   "jinja2",

tests/test_cli.py

@@ -137,8 +137,8 @@ def test_migrate_model_elements(cli_mocks: CLIMocks):
 
     assert result.exit_code == 0
     assert cli_mocks.get_polarion_wi_map.call_count == 1
-    assert cli_mocks.generate_work_items.call_count == 2
-    assert cli_mocks.generate_work_items.call_args_list[1][1] == {
+    assert cli_mocks.generate_work_items.call_count == 1
+    assert cli_mocks.generate_work_items.call_args_list[0][1] == {
         "generate_links": True,
         "generate_attachments": True,
         "generate_grouped_links_custom_fields": True,

tests/test_elements.py

@@ -161,6 +161,11 @@
 DIAGRAM_CONFIG = converter_config.CapellaTypeConfig("diagram", "diagram")
 
 
+def _set_work_item_id(work_items: list[polarion_api.WorkItem]):
+    for index, work_item in enumerate(work_items):
+        work_item.id = f"AUTO-{index}"
+
+
 class GroupedLinksBaseObject(t.TypedDict):
     link_serializer: link_converter.LinkSerializer
     work_items: dict[str, data_models.CapellaWorkItem]
@@ -880,6 +885,44 @@ def test_update_deleted_work_item(
         assert isinstance(work_item, data_models.CapellaWorkItem)
         assert work_item.status == "open"
 
+    @staticmethod
+    def test_create_new_work_item(base_object: BaseObjectContainer):
+        polarion_api_get_all_work_items = mock.MagicMock()
+        polarion_api_get_all_work_items.return_value = [
+            data_models.CapellaWorkItem(
+                id="Obj-1",
+                type="type",
+                uuid_capella="uuid1",
+                status="open",
+            )
+        ]
+        base_object.pw.project_client.work_items.get_all = (
+            polarion_api_get_all_work_items
+        )
+        polarion_api_create_work_items = mock.MagicMock()
+        polarion_api_create_work_items.side_effect = _set_work_item_id
+        base_object.pw.project_client.work_items.create = (
+            polarion_api_create_work_items
+        )
+
+        base_object.pw.load_polarion_work_item_map()
+        base_object.pw.create_missing_work_items(
+            base_object.mc.converter_session
+        )
+
+        assert polarion_api_create_work_items.call_count == 1
+        assert len(polarion_api_create_work_items.call_args[0][0]) == 1
+        work_item = polarion_api_create_work_items.call_args[0][0][0]
+        assert isinstance(work_item, data_models.CapellaWorkItem)
+        assert work_item.id == "AUTO-0"
+        assert len(base_object.pw.polarion_data_repo) == 2
+        assert (
+            base_object.pw.polarion_data_repo.get_work_item_by_capella_uuid(
+                "uuid2"
+            ).id
+            == "AUTO-0"
+        )
+
     @staticmethod
     def test_update_work_items_filters_work_items_with_same_checksum(
         base_object: BaseObjectContainer,