Skip to content

Commit

Permalink
feat: A new Config class was implemented, but not yet integrated, WIP
Browse files Browse the repository at this point in the history
  • Loading branch information
micha91 committed Jan 5, 2024
1 parent 4c30c6c commit f52d8cf
Show file tree
Hide file tree
Showing 3 changed files with 134 additions and 29 deletions.
94 changes: 94 additions & 0 deletions capella2polarion/converters/converter_config.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,94 @@
# Copyright DB InfraGO AG and contributors
# SPDX-License-Identifier: Apache-2.0
"""Module providing capella2polarion config class."""
from __future__ import annotations

import dataclasses
import typing

import yaml


@dataclasses.dataclass
class CapellaTypeConfig:
"""A single Capella Type configuration."""

p_type: str | None = None
converter: str | None = None
links: list[str] = dataclasses.field(default_factory=list)


class ConverterConfig:
"""The overall Config for capella2polarion."""

def __init__(self, synchronize_config: typing.TextIO):
config_dict = yaml.safe_load(synchronize_config)
self._layer_configs: dict[str, dict[str, CapellaTypeConfig]] = {}
self._global_configs: dict[str, CapellaTypeConfig] = {}
# We handle the cross layer config separately as global_configs
global_config_dict = config_dict.pop("*", {})
all_type_config = global_config_dict.pop("*", {})
global_links = all_type_config.get("links", [])
self.__global_config = CapellaTypeConfig(links=global_links)

for c_type, type_config in global_config_dict.items():
type_config = type_config or {}
self._global_configs[c_type] = CapellaTypeConfig(
type_config.get("polarion_type"),
type_config.get("serializer"),
type_config.get("links", []) + global_links,
)

for layer, type_configs in config_dict.items():
self._layer_configs[layer] = {}
for c_type, type_config in type_configs.items():
self._layer_configs[layer][c_type] = CapellaTypeConfig(
type_config.get("polarion_type")
or self._global_configs.get(
c_type, self.__global_config
).p_type,
type_config.get("serializer")
or self._global_configs.get(
c_type, self.__global_config
).converter,
type_config.get("links", [])
+ self._global_configs.get(
c_type, self.__global_config
).links,
)

def _default_type_conversion(self, c_type: str) -> str:
return c_type[0].lower() + c_type[1:]

def _get_type_configs(
self, layer: str, c_type: str
) -> CapellaTypeConfig | None:
return self._layer_configs.get(layer, {}).get(
c_type
) or self._global_configs.get(c_type)

def get_polarion_type(self, layer: str, c_type: str) -> str:
"""Return polarion type for a given layer and Capella type."""
type_config = (
self._get_type_configs(layer, c_type) or self.__global_config
)
return type_config.p_type or self._default_type_conversion(c_type)

def get_serializer(self, layer: str, c_type: str) -> str | None:
"""Return the serializer name for a given layer and Capella type."""
type_config = (
self._get_type_configs(layer, c_type) or self.__global_config
)
return type_config.converter

def get_links(self, layer: str, c_type: str) -> list[str]:
"""Return the list of link types for a given layer and Capella type."""
type_config = (
self._get_type_configs(layer, c_type) or self.__global_config
)
return type_config.links

def __contains__(self, item: tuple[str, str]):
"""Check if there is a config for a given layer and Capella type."""
layer, c_type = item
return self._get_type_configs(layer, c_type) is not None
46 changes: 17 additions & 29 deletions capella2polarion/converters/element_converter.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@

import base64
import collections
import collections.abc as cabc
import logging
import mimetypes
import pathlib
Expand All @@ -21,10 +20,9 @@
from capellambse.model.layers import oa, pa
from lxml import etree

from capella2polarion import data_models
from capella2polarion.connectors import polarion_repo

from .. import data_models

RE_DESCR_LINK_PATTERN = re.compile(
r"<a href=\"hlink://([^\"]+)\">([^<]+)<\/a>"
)
Expand All @@ -42,12 +40,12 @@
SERIALIZERS: dict[str, str] = {
"CapabilityRealization": "include_pre_and_post_condition",
"Capability": "include_pre_and_post_condition",
"LogicalComponent": "_include_actor_in_type",
"LogicalComponent": "include_actor_in_type",
"OperationalCapability": "include_pre_and_post_condition",
"PhysicalComponent": "_include_nature_in_type",
"SystemComponent": "_include_actor_in_type",
"PhysicalComponent": "include_nature_in_type",
"SystemComponent": "include_actor_in_type",
"Scenario": "include_pre_and_post_condition",
"Constraint": "constraint",
"Constraint": "linked_text_as_description",
"SystemCapability": "include_pre_and_post_condition",
}

Expand Down Expand Up @@ -141,11 +139,6 @@ class CapellaWorkItemSerializer:
capella_polarion_mapping: polarion_repo.PolarionDataRepository
model: capellambse.MelodyModel
descr_references: dict[str, list[str]]

serializers: dict[
str,
cabc.Callable[[common.GenericElement], data_models.CapellaWorkItem],
]
serializer_mapping: dict[str, str]

def __init__(
Expand All @@ -162,12 +155,6 @@ def __init__(
self.model = model
self.capella_polarion_mapping = capella_polarion_mapping
self.descr_references = descr_references
self.serializers = {
"include_pre_and_post_condition": self.include_pre_and_post_condition,
"_include_actor_in_type": self._include_actor_in_type,
"_include_nature_in_type": self._include_nature_in_type,
"constraint": self.constraint,
}
self.serializer_mapping = serializer_mapping or SERIALIZERS

def serialize(
Expand All @@ -176,21 +163,22 @@ def serialize(
"""Return a CapellaWorkItem for the given diagram or element."""
try:
if isinstance(obj, diagr.Diagram):
return self.diagram(obj)
return self._diagram(obj)
else:
xtype = self.polarion_type_map.get(
obj.uuid, type(obj).__name__
)
serializer = self.serializers.get(
self.serializer_mapping.get(xtype, ""),
serializer = getattr(
self,
f"_{self.serializer_mapping.get(xtype)}",
self._generic_work_item,
)
return serializer(obj)
except Exception as error:
logger.error("Serializing model element failed. %s", error.args[0])
return None

def diagram(self, diag: diagr.Diagram) -> data_models.CapellaWorkItem:
def _diagram(self, diag: diagr.Diagram) -> data_models.CapellaWorkItem:
"""Serialize a diagram for Polarion."""
diagram_path = self.diagram_cache_path / f"{diag.uuid}.svg"
src = _decode_diagram(diagram_path)
Expand Down Expand Up @@ -232,7 +220,7 @@ def _sanitize_description(
) -> tuple[list[str], markupsafe.Markup]:
referenced_uuids: list[str] = []
replaced_markup = RE_DESCR_LINK_PATTERN.sub(
lambda match: self.replace_markup(match, referenced_uuids, 2),
lambda match: self._replace_markup(match, referenced_uuids, 2),
descr,
)

Expand Down Expand Up @@ -264,7 +252,7 @@ def repair_images(node: etree._Element) -> None:
)
return referenced_uuids, repaired_markup

def replace_markup(
def _replace_markup(
self,
match: re.Match,
referenced_uuids: list[str],
Expand All @@ -287,7 +275,7 @@ def replace_markup(
logger.warning("Found reference to non-existing work item: %r", uuid)
return match.group(default_group)

def include_pre_and_post_condition(
def _include_pre_and_post_condition(
self, obj: PrePostConditionElement
) -> data_models.CapellaWorkItem:
"""Return generic attributes and pre- and post-condition."""
Expand All @@ -298,7 +286,7 @@ def get_condition(cap: PrePostConditionElement, name: str) -> str:
return condition.specification["capella:linkedText"].striptags()

def matcher(match: re.Match) -> str:
return strike_through(self.replace_markup(match, []))
return strike_through(self._replace_markup(match, []))

work_item = self._generic_work_item(obj)
pre_condition = RE_DESCR_DELETED_PATTERN.sub(
Expand All @@ -313,7 +301,7 @@ def matcher(match: re.Match) -> str:

return work_item

def get_linked_text(
def _get_linked_text(
self, obj: capellacore.Constraint
) -> markupsafe.Markup:
"""Return sanitized markup of the given ``obj`` linked text."""
Expand All @@ -323,13 +311,13 @@ def get_linked_text(
self.descr_references[obj.uuid] = uuids
return value

def constraint(
def _linked_text_as_description(
self, obj: capellacore.Constraint
) -> data_models.CapellaWorkItem:
"""Return attributes for a ``Constraint``."""
work_item = self._generic_work_item(obj)
# pylint: disable-next=attribute-defined-outside-init
work_item.description = self.get_linked_text(obj)
work_item.description = self._get_linked_text(obj)
return work_item

def _include_actor_in_type(
Expand Down
23 changes: 23 additions & 0 deletions tests/data/model_elements/new_config.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,23 @@
# Copyright DB InfraGO AG and contributors
# SPDX-License-Identifier: Apache-2.0

"*": # All layers
"*": # All class types
links:
- parent # Specify workitem links
- description_reference # Custom attribute
Class:
links:
- state_machines
Diagram:
links:
- diagram_elements
Constraint:

oa: # Specify below
FunctionalExchange:
polarion_type: operationalInteraction
links:
- exchange_items
OperationalCapability:
serializer: include_pre_and_post_condition

0 comments on commit f52d8cf

Please sign in to comment.