Skip to content

Commit

Permalink
fix: Remove CamelCase, fix tests, remove simulation
Browse files Browse the repository at this point in the history
  • Loading branch information
@micha91
micha91 committed Dec 21, 2023
1 parent 134949b commit c45a779
Show file tree
Hide file tree
Showing 7 changed files with 336 additions and 391 deletions.
61 changes: 31 additions & 30 deletions capella2polarion/__main__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,16 +3,20 @@
"""Main entry point into capella2polarion."""
from __future__ import annotations

import logging
import pathlib
import typing

import capellambse
import click
from capellambse import cli_helpers

from capella2polarion.c2pcli import C2PCli
from capella2polarion.c2polarion import PolarionWorker
from capella2polarion.capella2polarioncli import Capella2PolarionCli
from capella2polarion.elements import helpers, serialize

logger = logging.getLogger(__name__)


@click.group()
@click.option("--debug", is_flag=True, default=False)
Expand Down Expand Up @@ -58,11 +62,11 @@ def cli(
polarion_pat: str,
polarion_delete_work_items: bool,
capella_diagram_cache_folder_path: pathlib.Path,
capella_model: cli_helpers.ModelCLI,
capella_model: capellambse.MelodyModel,
synchronize_config: typing.TextIO,
) -> None:
"""Synchronise data from Capella to Polarion."""
lC2PCli = C2PCli(
capella2polarion_cli = Capella2PolarionCli(
debug,
polarion_project_id,
polarion_url,
Expand All @@ -72,39 +76,37 @@ def cli(
capella_model,
synchronize_config,
)
lC2PCli.setup_logger()
ctx.obj = lC2PCli
lC2PCli.echo = click.echo
lC2PCli.echo("Start")
capella2polarion_cli.setup_logger()
ctx.obj = capella2polarion_cli
capella2polarion_cli.echo = click.echo
capella2polarion_cli.echo("Start")


@cli.command()
@click.pass_obj
def printCliState(aC2PCli: C2PCli) -> None:
def print_cli_state(capella2polarion_cli: Capella2PolarionCli) -> None:
"""Print the CLI State."""
aC2PCli.setup_logger()
aC2PCli.print_state()
capella2polarion_cli.setup_logger()
capella2polarion_cli.print_state()


@cli.command()
@click.pass_context
def synchronize(ctx: click.core.Context) -> None:
"""Synchronise model elements."""
capella_to_polarion_cli: C2PCli = ctx.obj
capella_to_polarion_cli.logger.info(
f"""
Synchronising diagrams from diagram cache at
'{str(capella_to_polarion_cli.get_capella_diagram_cache_index_file_path())}'
to Polarion project with id
{capella_to_polarion_cli.polarion_params.project_id}...
"""
capella_to_polarion_cli: Capella2PolarionCli = ctx.obj
logger.info(
"Synchronising diagrams from diagram cache at %s to Polarion project with id %s...",
str(
capella_to_polarion_cli.get_capella_diagram_cache_index_file_path()
),
capella_to_polarion_cli.polarion_params.project_id,
)
capella_to_polarion_cli.load_synchronize_config()
capella_to_polarion_cli.load_roles_from_synchronize_config()
capella_to_polarion_cli.load_capella_diagramm_cache_index()
polarion_worker = PolarionWorker(
capella_to_polarion_cli.polarion_params,
capella_to_polarion_cli.logger,
helpers.resolve_element_type,
)
assert (
Expand All @@ -115,28 +117,27 @@ def synchronize(ctx: click.core.Context) -> None:
capella_to_polarion_cli.capella_model,
capella_to_polarion_cli.capella_diagram_cache_index_content,
)
# TODO - DEAKTIVIEREN - ACHTUNG!!!!!!
# polarion_worker.simulation = True

polarion_worker.fill_xtypes()
polarion_worker.load_polarion_work_item_map()
lDescriptionReference: typing.Any = {}
lNewWorkItems: dict[str, serialize.CapellaWorkItem]
lNewWorkItems = polarion_worker.create_work_items(
description_references: typing.Any = {}
new_work_items: dict[str, serialize.CapellaWorkItem]
new_work_items = polarion_worker.create_work_items(
capella_to_polarion_cli.capella_diagram_cache_folder_path,
capella_to_polarion_cli.capella_model,
lDescriptionReference,
description_references,
)
polarion_worker.delete_work_items()
polarion_worker.post_work_items(lNewWorkItems)
lNewWorkItems = polarion_worker.create_work_items(
polarion_worker.post_work_items(new_work_items)
new_work_items = polarion_worker.create_work_items(
capella_to_polarion_cli.capella_diagram_cache_folder_path,
capella_to_polarion_cli.capella_model,
lDescriptionReference,
description_references,
)
polarion_worker.patch_work_items(
capella_to_polarion_cli.capella_model,
lNewWorkItems,
lDescriptionReference,
new_work_items,
description_references,
capella_to_polarion_cli.synchronize_config_roles,
)

Expand Down
165 changes: 69 additions & 96 deletions capella2polarion/c2polarion.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@

from capella2polarion.elements import api_helper, element, serialize

GLogger = logging.getLogger(__name__)
logger = logging.getLogger(__name__)

# STATUS_DELETE = "deleted"
ACTOR_TYPES = {
Expand Down Expand Up @@ -59,28 +59,16 @@ class PolarionWorker:
def __init__(
self,
params: PolarionWorkerParams,
logger: logging.Logger,
make_type_id: typing.Any,
) -> None:
self.polarion_params: PolarionWorkerParams = params
self.client: polarion_api.OpenAPIPolarionProjectClient | None = None
self.logger: logging.Logger = logger
self.elements: dict[str, list[common.GenericElement]]
self.elements: dict[str, list[common.GenericElement]] = {}
self.polarion_type_map: dict[str, str] = {}
self.capella_uuid_s: set[str] = set()
self.x_types: set[str] = set()
self.polarion_id_map: dict[str, str] = {}
self.polarion_work_item_map: dict[
str, serialize.CapellaWorkItem
] # dict[str, typing.Any] = None
self.polarion_work_item_map: dict[str, serialize.CapellaWorkItem] = {}
self.make_type_id: typing.Any = make_type_id
self.simulation: bool = False

def _save_value_string(self, aValue: str | None) -> str | None:
return "None" if aValue is None else aValue

def setup_client(self) -> None:
"""Instantiate the polarion client, move to PolarionWorker Class."""
if (self.polarion_params.project_id is None) or (
len(self.polarion_params.project_id) == 0
):
Expand All @@ -107,7 +95,15 @@ def setup_client(self) -> None:
custom_work_item=serialize.CapellaWorkItem,
add_work_item_checksum=True,
)
if self.client.project_exists():
self.check_client()

def _save_value_string(self, value: str | None) -> str | None:
return "None" if value is None else value

def check_client(self) -> None:
"""Instantiate the polarion client, move to PolarionWorker Class."""

if not self.client.project_exists():
raise KeyError(
f"Miss Polarion project with id {self._save_value_string(self.polarion_params.project_id)}"
)
Expand All @@ -119,47 +115,6 @@ def load_elements_and_type_map(
diagram_idx: list[dict[str, typing.Any]],
) -> None:
"""Return an elements and UUID to Polarion type map."""

def _fix_components(
elements: dict[str, list[common.GenericElement]],
type_map: dict[str, str],
) -> None:
for typ, xtype in ACTOR_TYPES.items():
if typ not in elements:
continue

actors: list[common.GenericElement] = []
components: list[common.GenericElement] = []
for obj in elements[typ]:
if obj.is_actor:
actors.append(obj)
else:
components.append(obj)
type_map[obj.uuid] = xtype

elements[typ] = actors
elements[xtype] = components

nodes: list[common.GenericElement] = []
behaviors: list[common.GenericElement] = []
components = []
for obj in elements.get("PhysicalComponent", []):
if obj.nature is not None and obj.nature.name == "NODE":
nodes.append(obj)
type_map[obj.uuid] = "PhysicalComponentNode"
elif obj.nature is not None and obj.nature.name == "BEHAVIOR":
behaviors.append(obj)
type_map[obj.uuid] = "PhysicalComponentBehavior"
else:
components.append(obj)

if nodes:
elements["PhysicalComponentNode"] = nodes
if behaviors:
elements["PhysicalComponentBehavior"] = behaviors
if components:
elements["PhysicalComponent"] = components

convert_type = POL2CAPELLA_TYPES
type_map: dict[str, str] = {}
elements: dict[str, list[common.GenericElement]] = {}
Expand All @@ -178,7 +133,42 @@ def _fix_components(
for obj in objects:
type_map[obj.uuid] = typ

_fix_components(elements, type_map)
for typ, xtype in ACTOR_TYPES.items():
if typ not in elements:
continue

actors: list[common.GenericElement] = []
components: list[common.GenericElement] = []
for obj in elements[typ]:
if obj.is_actor:
actors.append(obj)
else:
components.append(obj)
type_map[obj.uuid] = xtype

elements[typ] = actors
elements[xtype] = components

nodes: list[common.GenericElement] = []
behaviors: list[common.GenericElement] = []
components = []
for obj in elements.get("PhysicalComponent", []):
if obj.nature is not None and obj.nature.name == "NODE":
nodes.append(obj)
type_map[obj.uuid] = "PhysicalComponentNode"
elif obj.nature is not None and obj.nature.name == "BEHAVIOR":
behaviors.append(obj)
type_map[obj.uuid] = "PhysicalComponentBehavior"
else:
components.append(obj)

if nodes:
elements["PhysicalComponentNode"] = nodes
if behaviors:
elements["PhysicalComponentBehavior"] = behaviors
if components:
elements["PhysicalComponent"] = components

diagrams_from_cache = {d["uuid"] for d in diagram_idx if d["success"]}
elements["Diagram"] = [
d for d in model.diagrams if d.uuid in diagrams_from_cache
Expand All @@ -201,21 +191,12 @@ def load_polarion_work_item_map(self):
"""Return a map from Capella UUIDs to Polarion work items."""
work_item_types = list(map(self.make_type_id, self.x_types))
_type = " ".join(work_item_types)
if self.simulation:
work_item = serialize.CapellaWorkItem(
"84a64a2d-3491-48af-b55b-823010a3e006", "FakeItem"
)
work_item.uuid_capella = "weck"
work_item.checksum = "doppelwegg"
work_item.status = "fake"
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"},
)

work_items = self.client.get_all_work_items(
f"type:({_type})",
{"workitems": "id,uuid_capella,checksum,status"},
)

self.polarion_work_item_map = {
wi.uuid_capella: wi
for wi in work_items
Expand Down Expand Up @@ -256,7 +237,7 @@ def create_work_items(
missing_types.add(work_item.type)

if missing_types:
self.logger.debug(
logger.debug(
"%r are missing in the capella2polarion configuration",
", ".join(missing_types),
)
Expand All @@ -270,7 +251,7 @@ def delete_work_items(self) -> None:
"""

def serialize_for_delete(uuid: str) -> str:
self.logger.info(
logger.info(
"Delete work item %r...",
workitem_id := self.polarion_id_map[uuid],
)
Expand All @@ -285,16 +266,12 @@ def serialize_for_delete(uuid: str) -> str:
work_item_ids = [serialize_for_delete(uuid) for uuid in uuids]
if work_item_ids:
try:
if not self.simulation:
assert self.client is not None
self.client.delete_work_items(work_item_ids)
self.client.delete_work_items(work_item_ids)
for uuid in uuids:
del self.polarion_work_item_map[uuid]
del self.polarion_id_map[uuid]
except polarion_api.PolarionApiException as error:
self.logger.error(
"Deleting work items failed. %s", error.args[0]
)
logger.error("Deleting work items failed. %s", error.args[0])

def post_work_items(
self, new_work_items: dict[str, serialize.CapellaWorkItem]
Expand All @@ -307,21 +284,17 @@ def post_work_items(

assert work_item is not None
missing_work_items.append(work_item)
self.logger.info("Create work item for %r...", work_item.title)
logger.info("Create work item for %r...", work_item.title)
if missing_work_items:
try:
if not self.simulation:
assert self.client is not None
self.client.create_work_items(missing_work_items)
self.client.create_work_items(missing_work_items)
for work_item in missing_work_items:
self.polarion_id_map[work_item.uuid_capella] = work_item.id
self.polarion_work_item_map[
work_item.uuid_capella
] = work_item
except polarion_api.PolarionApiException as error:
self.logger.error(
"Creating work items failed. %s", error.args[0]
)
logger.error("Creating work items failed. %s", error.args[0])

def patch_work_items(
self,
Expand Down Expand Up @@ -367,11 +340,11 @@ def patch_work_items(
element.create_grouped_back_link_fields(
new_work_item, back_links[old_work_item.id]
)
if not self.simulation:
api_helper.patch_work_item(
self.client,
new_work_item,
old_work_item,
old_work_item.title,
"element",
)

api_helper.patch_work_item(
self.client,
new_work_item,
old_work_item,
old_work_item.title,
"element",
)
Loading

0 comments on commit c45a779

Please sign in to comment.