refactor: Changes from code review

DSD-DBS · Jan 5, 2024 · 80185e7 · 80185e7
1 parent efb825c
commit 80185e7
Show file tree

Hide file tree

Showing 13 changed files with 124 additions and 163 deletions.
diff --git a/capella2polarion/__main__.py b/capella2polarion/__main__.py
@@ -11,10 +11,10 @@
 import click
 from capellambse import cli_helpers
 
-from capella2polarion import capella_work_item
-from capella2polarion import polarion_worker as pw
-from capella2polarion.capella2polarioncli import Capella2PolarionCli
-from capella2polarion.capella_polarion_conversion import element_converter
+from capella2polarion import data
+from capella2polarion import worker as pw
+from capella2polarion.cli import Capella2PolarionCli
+from capella2polarion.converters import element_converter
 
 logger = logging.getLogger(__name__)
 
@@ -122,7 +122,7 @@ def synchronize(ctx: click.core.Context) -> None:
     polarion_worker.fill_xtypes()
     polarion_worker.load_polarion_work_item_map()
     description_references: typing.Any = {}
-    new_work_items: dict[str, capella_work_item.CapellaWorkItem]
+    new_work_items: dict[str, data.CapellaWorkItem]
     new_work_items = polarion_worker.create_work_items(
         capella_to_polarion_cli.capella_diagram_cache_folder_path,
         capella_to_polarion_cli.capella_model,

diff --git a/capella2polarion/capella2polarioncli.py → capella2polarion/cli.py b/capella2polarion/capella2polarioncli.py → capella2polarion/cli.py
@@ -13,7 +13,7 @@
 import click
 import yaml
 
-from capella2polarion import polarion_worker as pw
+from capella2polarion import worker as pw
 
 logger = logging.getLogger(__name__)
 

diff --git a/...a2polarion/polarion_connector/__init__.py → capella2polarion/connectors/__init__.py b/...a2polarion/polarion_connector/__init__.py → capella2polarion/connectors/__init__.py
diff --git a/...arion/polarion_connector/polarion_repo.py → capella2polarion/connectors/polarion_repo.py b/...arion/polarion_connector/polarion_repo.py → capella2polarion/connectors/polarion_repo.py
@@ -3,29 +3,27 @@
 """Module providing a universal PolarionDataRepository class."""
 from __future__ import annotations
 
-import typing
+import collections.abc as cabc
 
 import bidict
 
-from capella2polarion import capella_work_item
+from capella2polarion import data
 
 
 class PolarionDataRepository:
-    """A mapping class to access all contents by Capella and Polarion IDs.
+    """A mapping to access all contents by Capella and Polarion IDs.
 
-    This class only holds data already present in the Polarion. It only
+    This class only holds data already present in Polarion. It only
     receives updates if data were written to Polarion. There shall be no
     intermediate data stored here during serialization.
     """
 
     _id_mapping: bidict.bidict[str, str]
-    _work_items: dict[str, capella_work_item.CapellaWorkItem]
+    _work_items: dict[str, data.CapellaWorkItem]
 
     def __init__(
         self,
-        polarion_work_items: typing.Optional[
-            list[capella_work_item.CapellaWorkItem]
-        ] = None,
+        polarion_work_items: list[data.CapellaWorkItem] | None = None,
     ):
         if polarion_work_items is None:
             polarion_work_items = []
@@ -47,23 +45,21 @@ 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):
+    def __sizeof__(self) -> int:
         """Return the amount of registered Capella UUIDs."""
         return len(self._id_mapping)
 
-    def __getitem__(
-        self, item: str
-    ) -> typing.Tuple[str, capella_work_item.CapellaWorkItem]:
+    def __getitem__(self, item: str) -> tuple[str, data.CapellaWorkItem]:
         """Return the polarion ID and work_item for a given Capella UUID."""
         return self._id_mapping[item], self._work_items[item]
 
-    def __iter__(self) -> typing.Iterator[str]:
+    def __iter__(self) -> cabc.Iterator[str]:
         """Iterate all Capella UUIDs."""
         return self._id_mapping.__iter__()
 
     def items(
         self,
-    ):
+    ) -> cabc.Iterator[tuple[str, str, data.CapellaWorkItem]]:
         """Yield all Capella UUIDs, Work Item IDs and Work Items."""
         for uuid, polarion_id in self._id_mapping.items():
             yield uuid, polarion_id, self._work_items[uuid]
@@ -78,21 +74,21 @@ def get_capella_uuid(self, work_item_id: str) -> str | None:
 
     def get_work_item_by_capella_uuid(
         self, capella_uuid: str
-    ) -> capella_work_item.CapellaWorkItem | None:
+    ) -> data.CapellaWorkItem | None:
         """Return a Work Item for a provided Capella UUID."""
         return self._work_items.get(capella_uuid)
 
     def get_work_item_by_polarion_id(
         self, work_item_id: str
-    ) -> capella_work_item.CapellaWorkItem | None:
+    ) -> data.CapellaWorkItem | None:
         """Return a Work Item for a provided Work Item ID."""
         return self.get_work_item_by_capella_uuid(
             self.get_capella_uuid(work_item_id)  # type: ignore
         )
 
     def update_work_items(
         self,
-        work_items: list[capella_work_item.CapellaWorkItem],
+        work_items: list[data.CapellaWorkItem],
     ):
         """Update all mappings for the given Work Items."""
         for work_item in work_items:
@@ -111,7 +107,7 @@ def update_work_items(
             {work_item.uuid_capella: work_item for work_item in work_items}
         )
 
-    def remove_work_items_by_capella_uuid(self, uuids: typing.Iterable[str]):
+    def remove_work_items_by_capella_uuid(self, uuids: cabc.Iterable[str]):
         """Remove entries for the given Capella UUIDs."""
         for uuid in uuids:
             del self._work_items[uuid]

diff --git a/...n/capella_polarion_conversion/__init__.py → capella2polarion/converters/__init__.py b/...n/capella_polarion_conversion/__init__.py → capella2polarion/converters/__init__.py
diff --git a/..._polarion_conversion/element_converter.py → ...2polarion/converters/element_converter.py b/..._polarion_conversion/element_converter.py → ...2polarion/converters/element_converter.py
@@ -21,9 +21,9 @@
 from capellambse.model.layers import oa, pa
 from lxml import etree
 
-from capella2polarion.polarion_connector import polarion_repo
+from capella2polarion.connectors import polarion_repo
 
-from .. import capella_work_item
+from .. import data
 
 RE_DESCR_LINK_PATTERN = re.compile(
     r"<a href=\"hlink://([^\"]+)\">([^<]+)<\/a>"
@@ -126,9 +126,7 @@ def _get_requirement_types_text(
     return _format_texts(type_texts)
 
 
-def _condition(
-    html: bool, value: str
-) -> capella_work_item.CapellaWorkItem.Condition:
+def _condition(html: bool, value: str) -> data.CapellaWorkItem.Condition:
     _type = "text/html" if html else "text/plain"
     return {"type": _type, "value": value}
 
@@ -144,9 +142,7 @@ class CapellaWorkItemSerializer:
 
     serializers: dict[
         str,
-        cabc.Callable[
-            [common.GenericElement], capella_work_item.CapellaWorkItem
-        ],
+        cabc.Callable[[common.GenericElement], data.CapellaWorkItem],
     ]
     serializer_mapping: dict[str, str]
 
@@ -174,7 +170,7 @@ def __init__(
 
     def serialize(
         self, obj: diagr.Diagram | common.GenericElement
-    ) -> capella_work_item.CapellaWorkItem | None:
+    ) -> data.CapellaWorkItem | None:
         """Return a CapellaWorkItem for the given diagram or element."""
         try:
             if isinstance(obj, diagr.Diagram):
@@ -192,9 +188,7 @@ def serialize(
             logger.error("Serializing model element failed. %s", error.args[0])
             return None
 
-    def diagram(
-        self, diag: diagr.Diagram
-    ) -> capella_work_item.CapellaWorkItem:
+    def diagram(self, diag: diagr.Diagram) -> data.CapellaWorkItem:
         """Serialize a diagram for Polarion."""
         diagram_path = self.diagram_cache_path / f"{diag.uuid}.svg"
         src = _decode_diagram(diagram_path)
@@ -204,7 +198,7 @@ def diagram(
         description = (
             f'<html><p><img style="{style}" src="{src}" /></p></html>'
         )
-        return capella_work_item.CapellaWorkItem(
+        return data.CapellaWorkItem(
             type="diagram",
             title=diag.name,
             description_type="text/html",
@@ -215,13 +209,13 @@ def diagram(
 
     def _generic_work_item(
         self, obj: common.GenericElement
-    ) -> capella_work_item.CapellaWorkItem:
+    ) -> data.CapellaWorkItem:
         xtype = self.polarion_type_map.get(obj.uuid, type(obj).__name__)
         raw_description = getattr(obj, "description", markupsafe.Markup(""))
         uuids, value = self._sanitize_description(obj, raw_description)
         self.descr_references[obj.uuid] = uuids
         requirement_types = _get_requirement_types_text(obj)
-        return capella_work_item.CapellaWorkItem(
+        return data.CapellaWorkItem(
             type=resolve_element_type(xtype),
             title=obj.name,
             description_type="text/html",
@@ -293,7 +287,7 @@ def replace_markup(
 
     def include_pre_and_post_condition(
         self, obj: PrePostConditionElement
-    ) -> capella_work_item.CapellaWorkItem:
+    ) -> data.CapellaWorkItem:
         """Return generic attributes and pre- and post-condition."""
 
         def get_condition(cap: PrePostConditionElement, name: str) -> str:
@@ -327,9 +321,7 @@ def get_linked_text(
             self.descr_references[obj.uuid] = uuids
         return value
 
-    def constraint(
-        self, obj: capellacore.Constraint
-    ) -> capella_work_item.CapellaWorkItem:
+    def constraint(self, obj: capellacore.Constraint) -> data.CapellaWorkItem:
         """Return attributes for a ``Constraint``."""
         work_item = self._generic_work_item(obj)
         if work_item.uuid_capella == "b24a9e54-4386-4d38-aea3-7c5172d73bb6":
@@ -340,7 +332,7 @@ def constraint(
 
     def _include_actor_in_type(
         self, obj: cs.Component
-    ) -> capella_work_item.CapellaWorkItem:
+    ) -> data.CapellaWorkItem:
         """Return attributes for a ``Component``."""
         work_item = self._generic_work_item(obj)
         if obj.is_actor:
@@ -353,7 +345,7 @@ def _include_actor_in_type(
 
     def _include_nature_in_type(
         self, obj: pa.PhysicalComponent
-    ) -> capella_work_item.CapellaWorkItem:
+    ) -> data.CapellaWorkItem:
         """Return attributes for a ``PhysicalComponent``."""
         work_item = self._include_actor_in_type(obj)
         xtype = work_item.type

diff --git a/...lla_polarion_conversion/link_converter.py → ...lla2polarion/converters/link_converter.py b/...lla_polarion_conversion/link_converter.py → ...lla2polarion/converters/link_converter.py
@@ -4,7 +4,9 @@
 from __future__ import annotations
 
 import collections.abc as cabc
+import functools
 import logging
+import typing as t
 from collections import defaultdict
 
 import capellambse
@@ -13,9 +15,9 @@
 from capellambse.model import diagram as diag
 from capellambse.model.crosslayer import fa
 
-from capella2polarion import capella_work_item
-from capella2polarion.capella_polarion_conversion import element_converter
-from capella2polarion.polarion_connector import polarion_repo
+from capella2polarion import data
+from capella2polarion.connectors import polarion_repo
+from capella2polarion.converters import element_converter
 
 logger = logging.getLogger(__name__)
 
@@ -28,7 +30,7 @@ class LinkSerializer:
     def __init__(
         self,
         capella_polarion_mapping: polarion_repo.PolarionDataRepository,
-        new_work_items: dict[str, capella_work_item.CapellaWorkItem],
+        new_work_items: dict[str, data.CapellaWorkItem],
         description_references: dict[str, list[str]],
         project_id: str,
         model: capellambse.MelodyModel,
@@ -39,6 +41,23 @@ def __init__(
         self.project_id = project_id
         self.model = model
 
+        self.serializers: dict[
+            str,
+            cabc.Callable[
+                [common.GenericElement, str, str, dict[str, t.Any]],
+                list[polarion_api.WorkItemLink],
+            ],
+        ] = {
+            "description_reference": self._handle_description_reference_links,
+            "diagram_elements": self._handle_diagram_reference_links,
+            "input_exchanges": functools.partial(
+                self._handle_exchanges, attr="inputs"
+            ),
+            "output_exchanges": functools.partial(
+                self._handle_exchanges, attr="outputs"
+            ),
+        }
+
     def create_links_for_work_item(
         self,
         obj: common.GenericElement | diag.Diagram,
@@ -54,41 +73,8 @@ def create_links_for_work_item(
         new_links: list[polarion_api.WorkItemLink] = []
         typ = work_item.type[0].upper() + work_item.type[1:]
         for role_id in roles.get(typ, []):
-            if role_id == "description_reference":
-                new_links.extend(
-                    self._handle_description_reference_links(
-                        obj,
-                        work_item.id,
-                        role_id,
-                        {},
-                    )
-                )
-            elif role_id == "diagram_elements":
-                new_links.extend(
-                    self._handle_diagram_reference_links(
-                        obj, work_item.id, role_id, {}
-                    )
-                )
-            elif role_id == "input_exchanges":
-                new_links.extend(
-                    self._handle_exchanges(
-                        obj,
-                        work_item.id,
-                        role_id,
-                        {},
-                        "inputs",
-                    )
-                )
-            elif role_id == "output_exchanges":
-                new_links.extend(
-                    self._handle_exchanges(
-                        obj,
-                        work_item.id,
-                        role_id,
-                        {},
-                        "outputs",
-                    )
-                )
+            if serializer := self.serializers.get(role_id):
+                new_links.extend(serializer(obj, work_item.id, role_id, {}))
             else:
                 if (refs := getattr(obj, role_id, None)) is None:
                     logger.info(
@@ -207,10 +193,10 @@ def _handle_exchanges(
 
 
 def create_grouped_link_fields(
-    work_item: capella_work_item.CapellaWorkItem,
+    work_item: data.CapellaWorkItem,
     back_links: dict[str, list[polarion_api.WorkItemLink]] | None = None,
 ):
-    """Create the grouped link work items fields from the primary work item.
+    """Create the grouped link fields from the primary work item.
 
     Parameters
     ----------
@@ -234,10 +220,10 @@ def create_grouped_link_fields(
 
 
 def create_grouped_back_link_fields(
-    work_item: capella_work_item.CapellaWorkItem,
+    work_item: data.CapellaWorkItem,
     links: list[polarion_api.WorkItemLink],
 ):
-    """Create backlinks for the given WorkItem using a list of backlinks.
+    """Create fields for the given WorkItem using a list of backlinks.
 
     Parameters
     ----------
@@ -280,7 +266,7 @@ def _make_url_list(
 
 
 def _create_link_fields(
-    work_item: capella_work_item.CapellaWorkItem,
+    work_item: data.CapellaWorkItem,
     role: str,
     links: list[polarion_api.WorkItemLink],
     reverse: bool = False,

diff --git a/capella2polarion/capella_work_item.py → capella2polarion/data.py b/capella2polarion/capella_work_item.py → capella2polarion/data.py
diff --git a/capella2polarion/py.typed b/capella2polarion/py.typed