feat: move polarion params and setup to class polarion worker

DSD-DBS · Dec 14, 2023 · db5d7ef · db5d7ef
1 parent b1464ba
commit db5d7ef
Show file tree

Hide file tree

Showing 3 changed files with 72 additions and 57 deletions.
diff --git a/capella2polarion/__main__.py b/capella2polarion/__main__.py
@@ -26,7 +26,7 @@
 @click.group()
 @click.option("--debug", is_flag=True, default=False)
 @click.option(
-    "--project-id",
+    "--polarion-project-id",
     type=str,
     required=False,
     default=None,
@@ -62,7 +62,7 @@
 def cli(
     ctx: click.core.Context,
     debug: bool,
-    project_id: str,
+    polarion_project_id: str,
     polarion_url: str,
     polarion_pat: str,
     polarion_delete_work_items: bool,
@@ -74,7 +74,7 @@ def cli(
     """Synchronise data from Capella to Polarion."""
     lC2PCli = C2PCli(
         debug,
-        project_id,
+        polarion_project_id,
         polarion_url,
         polarion_pat,
         polarion_delete_work_items,
@@ -107,7 +107,7 @@ def synchronize(ctx: click.core.Context) -> None:
     aC2PCli.logger.info(
         f"""
             Synchronising diagrams from diagram cache at '{str(aC2PCli.get_capella_diagram_cache_index_file_path())}'
-            to Polarion project with id {aC2PCli.ProjectId}...
+            to Polarion project with id {aC2PCli.polarion_params.project_id}...
         """
     )
     # ctx.obj["DIAGRAM_CACHE"] = None # Orignal ... aDiagramCachePath ... None damit es Crashed!
@@ -131,7 +131,7 @@ def synchronize(ctx: click.core.Context) -> None:
     from polarion import PolarionWorker
 
     lPW = PolarionWorker(
-        aC2PCli.PolarionClient, aC2PCli.logger, helpers.resolve_element_type
+        aC2PCli.polarion_params, aC2PCli.logger, helpers.resolve_element_type
     )
     lPW.load_elements_and_type_map(
         aC2PCli.SynchronizeConfigContent,
@@ -212,7 +212,6 @@ def synchronize(ctx: click.core.Context) -> None:
         aC2PCli.CapellaModel,
         lNewWorkItems,
         lDescriptionReference,
-        aC2PCli.ProjectId,
         aC2PCli.SynchronizeConfigRoles,
     )
 

diff --git a/capella2polarion/c2pcli.py b/capella2polarion/c2pcli.py
@@ -13,13 +13,10 @@
 
 import capellambse
 import click
-import polarion_rest_api_client as polarion_api
-import validators
 import yaml
 from capellambse import cli_helpers
 
-from capella2polarion import elements
-from capella2polarion.elements import serialize
+from capella2polarion.polarion import PolarionWorkerParams
 
 GLogger = logging.getLogger(__name__)
 
@@ -30,7 +27,7 @@ class C2PCli(object):
     def __init__(
         self,
         aDebug: bool,
-        aProjectId: str,
+        aPolarionProjectId: str,
         aPolarionUrl: str,
         aPolarionPat: str,
         aPolarionDeleteWorkItems: bool,
@@ -39,12 +36,11 @@ def __init__(
         synchronize_config_io: typing.TextIO,
     ) -> None:
         self.Debug = aDebug
-        self.ProjectId = aProjectId
-        self.PolarionUrl = aPolarionUrl
-        self.PolarionPat = aPolarionPat
-        self.PolarionDeleteWorkItems = aPolarionDeleteWorkItems
-        self.PolarionClient: polarion_api.OpenAPIPolarionProjectClient | None = (
-            None
+        self.polarion_params = PolarionWorkerParams(
+            aPolarionProjectId,
+            aPolarionUrl,
+            aPolarionPat,
+            aPolarionDeleteWorkItems,
         )
         self.CapellaDiagramCacheFolderPath = capella_diagram_cache_folder_path
         self.CapellaDiagramCacheIndexContent: list[
@@ -57,6 +53,9 @@ def __init__(
         self.echo = click.echo
         self.logger: logging.Logger
 
+    def _noneSaveValueString(self, aValue: str | None) -> str | None:
+        return "None" if aValue is None else aValue
+
     def printState(self) -> None:
         """Print the State of the cli tool."""
 
@@ -101,36 +100,6 @@ def _value(aValue):
             f"Synchronize Config-IO is open: {('YES' if not self.SynchronizeConfigIO.closed else 'NO')}"
         )
 
-    def setupPolarionClient(self) -> None:
-        """Instantiate the polarion client, move to PolarionWorker Class."""
-        if (self.ProjectId == None) or (len(self.ProjectId) == 0):
-            raise Exception(
-                f"""ProjectId invalid. Value '{self._noneSaveValueString(self.ProjectId)}'"""
-            )
-        if validators.url(self.PolarionUrl):
-            raise Exception(
-                f"""Polarion URL parameter is not a valid url.
-                Value {self._noneSaveValueString(self.PolarionUrl)}"""
-            )
-        if self.PolarionPat == None:
-            raise Exception(
-                f"""Polarion PAT (Personal Access Token) parameter is not a valid url. Value
-                '{self._noneSaveValueString(self.PolarionPat)}'"""
-            )
-        self.PolarionClient = polarion_api.OpenAPIPolarionProjectClient(
-            self.ProjectId,
-            self.PolarionDeleteWorkItems,
-            polarion_api_endpoint=f"{self.PolarionUrl}/rest/v1",
-            polarion_access_token=self.PolarionPat,
-            custom_work_item=serialize.CapellaWorkItem,
-            add_work_item_checksum=True,
-        )
-        # assert self.PolarionClient is not None
-        if self.PolarionClient.project_exists():
-            raise Exception(
-                f"Miss Polarion project with id {self._noneSaveValueString(self.ProjectId)}"
-            )
-
     def setupLogger(self) -> None:
         """Set the logger in the right mood."""
         lMaxLoggingLevel = logging.DEBUG if self.Debug else logging.WARNING
@@ -265,6 +234,3 @@ def load_capella_diagramm_cache_index(self) -> None:
                 )
             )
             self.CapellaDiagramCacheIndexContent = json.loads(l_text_content)
-
-    def _noneSaveValueString(self, aValue: str | None) -> str | None:
-        return "None" if aValue is None else aValue
diff --git a/capella2polarion/polarion.py b/capella2polarion/polarion.py
@@ -1,16 +1,18 @@
 # Copyright DB Netz AG and contributors
 # SPDX-License-Identifier: Apache-2.0
 """Module for polarion API client work."""
+from __future__ import annotations
+
 import logging
 import pathlib
 import typing
 from itertools import chain
 
 import capellambse
 import polarion_rest_api_client as polarion_api
+import validators
 from capellambse.model import common
 
-import capella2polarion
 from capella2polarion.elements import api_helper, element, serialize
 
 GLogger = logging.getLogger(__name__)
@@ -39,18 +41,29 @@
 }
 
 
+class PolarionWorkerParams:
+    """Container for Polarion Params."""
+
+    def __init__(
+        self, project_id: str, url: str, pat: str, delete_work_items: bool
+    ) -> None:
+        self.project_id = project_id
+        self.url = url
+        self.private_access_token = pat
+        self.delete_work_items = delete_work_items
+
+
 class PolarionWorker:
     """PolarionWorker encapsulate the Polarion API Client work."""
 
     def __init__(
         self,
-        aPolarionClient: polarion_api.OpenAPIPolarionProjectClient,
+        params: PolarionWorkerParams,
         aLogger: logging.Logger,
         aMakeTypeId: typing.Any,
     ) -> None:
-        self.client: polarion_api.OpenAPIPolarionProjectClient = (
-            aPolarionClient
-        )
+        self.PolarionParams: PolarionWorkerParams = params
+        self.client: polarion_api.OpenAPIPolarionProjectClient | None = None
         self.logger: logging.Logger = aLogger
         self.Elements: dict[str, list[common.GenericElement]]
         self.PolarionTypeMap: dict[str, str] = {}
@@ -63,6 +76,41 @@ def __init__(
         self.makeTypeId: typing.Any = aMakeTypeId
         self.Simulation: bool = False
 
+    def _noneSaveValueString(self, aValue: str | None) -> str | None:
+        return "None" if aValue is None else aValue
+
+    def setupPolarionClient(self) -> None:
+        """Instantiate the polarion client, move to PolarionWorker Class."""
+        if (self.PolarionParams.project_id == None) or (
+            len(self.PolarionParams.project_id) == 0
+        ):
+            raise Exception(
+                f"""ProjectId invalid. Value '{self._noneSaveValueString(self.PolarionParams.project_id)}'"""
+            )
+        if validators.url(self.PolarionParams.url):
+            raise Exception(
+                f"""Polarion URL parameter is not a valid url.
+                Value {self._noneSaveValueString(self.PolarionParams.url)}"""
+            )
+        if self.PolarionParams.private_access_token == None:
+            raise Exception(
+                f"""Polarion PAT (Personal Access Token) parameter is not a valid url. Value
+                '{self._noneSaveValueString(self.PolarionParams.private_access_token)}'"""
+            )
+        self.PolarionClient = polarion_api.OpenAPIPolarionProjectClient(
+            self.PolarionParams.project_id,
+            self.PolarionParams.delete_work_items,
+            polarion_api_endpoint=f"{self.PolarionParams.url}/rest/v1",
+            polarion_access_token=self.PolarionParams.private_access_token,
+            custom_work_item=serialize.CapellaWorkItem,
+            add_work_item_checksum=True,
+        )
+        # assert self.PolarionClient is not None
+        if self.PolarionClient.project_exists():
+            raise Exception(
+                f"Miss Polarion project with id {self._noneSaveValueString(self.PolarionParams.project_id)}"
+            )
+
     def load_elements_and_type_map(
         self,
         config: dict[str, typing.Any],
@@ -162,6 +210,7 @@ def load_polarion_work_item_map(self):
             work_items = []
             work_items.append(work_item)
         else:
+            assert self.client is not None
             work_items = self.client.get_all_work_items(
                 f"type:({_type})",
                 {"workitems": "id,uuid_capella,checksum,status"},
@@ -236,6 +285,7 @@ def serialize_for_delete(uuid: str) -> str:
         if work_item_ids:
             try:
                 if not self.Simulation:
+                    assert self.client is not None
                     self.client.delete_work_items(work_item_ids)
                 for uuid in uuids:
                     del self.PolarionWorkItemMap[uuid]
@@ -260,6 +310,7 @@ def post_work_items(
         if missing_work_items:
             try:
                 if not self.Simulation:
+                    assert self.client is not None
                     self.client.create_work_items(missing_work_items)
                 for work_item in missing_work_items:
                     self.PolarionIdMap[work_item.uuid_capella] = work_item.id
@@ -276,7 +327,6 @@ def patch_work_items(
         model: capellambse.MelodyModel,
         new_work_items: dict[str, serialize.CapellaWorkItem],
         descr_references,
-        project_id,
         link_roles,
     ) -> None:
         """Update work items in a Polarion project."""
@@ -299,7 +349,7 @@ def patch_work_items(
                 obj,
                 self.PolarionIdMap,
                 descr_references,
-                project_id,
+                self.PolarionParams.project_id,
                 model,
                 link_roles,
                 TYPES_POL2CAPELLA,