diff --git a/capella2polarion/__main__.py b/capella2polarion/__main__.py index 06c9e2ba..3a47e75e 100644 --- a/capella2polarion/__main__.py +++ b/capella2polarion/__main__.py @@ -11,9 +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.elements import serialize +from capella2polarion.capella_polarion_conversion import element_converter logger = logging.getLogger(__name__) @@ -108,21 +109,21 @@ def synchronize(ctx: click.core.Context) -> None: capella_to_polarion_cli.load_capella_diagramm_cache_index() polarion_worker = pw.PolarionWorker( capella_to_polarion_cli.polarion_params, - serialize.resolve_element_type, + capella_to_polarion_cli.capella_model, + element_converter.resolve_element_type, ) assert ( capella_to_polarion_cli.capella_diagram_cache_index_content is not None ) polarion_worker.load_elements_and_type_map( capella_to_polarion_cli.synchronize_config_content, - capella_to_polarion_cli.capella_model, capella_to_polarion_cli.capella_diagram_cache_index_content, ) polarion_worker.fill_xtypes() polarion_worker.load_polarion_work_item_map() description_references: typing.Any = {} - new_work_items: dict[str, serialize.CapellaWorkItem] + new_work_items: dict[str, capella_work_item.CapellaWorkItem] new_work_items = polarion_worker.create_work_items( capella_to_polarion_cli.capella_diagram_cache_folder_path, capella_to_polarion_cli.capella_model, @@ -136,7 +137,6 @@ def synchronize(ctx: click.core.Context) -> None: description_references, ) polarion_worker.patch_work_items( - capella_to_polarion_cli.capella_model, new_work_items, description_references, capella_to_polarion_cli.synchronize_config_roles, diff --git a/capella2polarion/elements/__init__.py b/capella2polarion/capella_polarion_conversion/__init__.py similarity index 100% rename from capella2polarion/elements/__init__.py rename to capella2polarion/capella_polarion_conversion/__init__.py diff --git a/capella2polarion/elements/serialize.py b/capella2polarion/capella_polarion_conversion/element_converter.py similarity index 90% rename from capella2polarion/elements/serialize.py rename to capella2polarion/capella_polarion_conversion/element_converter.py index 767eeae0..78d389e0 100644 --- a/capella2polarion/elements/serialize.py +++ b/capella2polarion/capella_polarion_conversion/element_converter.py @@ -14,7 +14,6 @@ import capellambse import markupsafe -import polarion_rest_api_client as polarion_api from capellambse import helpers as chelpers from capellambse.model import common from capellambse.model import diagram as diagr @@ -22,6 +21,10 @@ from capellambse.model.layers import oa, pa from lxml import etree +from capella2polarion.polarion_connector import polarion_repo + +from .. import capella_work_item + RE_DESCR_LINK_PATTERN = re.compile(r"([^<]+)") RE_DESCR_DELETED_PATTERN = re.compile( f"" @@ -54,20 +57,6 @@ logger = logging.getLogger(__name__) -class CapellaWorkItem(polarion_api.WorkItem): - """A custom WorkItem class with additional capella related attributes.""" - - class Condition(t.TypedDict): - """A class to describe a pre or post condition.""" - - type: str - value: str - - uuid_capella: str - preCondition: Condition | None - postCondition: Condition | None - - def resolve_element_type(type_: str) -> str: """Return a valid Type ID for polarion for a given ``obj``.""" return type_[0].lower() + type_[1:] @@ -135,7 +124,9 @@ def _get_requirement_types_text( return _format_texts(type_texts) -def _condition(html: bool, value: str) -> CapellaWorkItem.Condition: +def _condition( + html: bool, value: str +) -> capella_work_item.CapellaWorkItem.Condition: _type = "text/html" if html else "text/plain" return {"type": _type, "value": value} @@ -145,11 +136,15 @@ class CapellaWorkItemSerializer: diagram_cache_path: pathlib.Path polarion_type_map: dict[str, str] + capella_polarion_mapping: polarion_repo.PolarionDataRepository model: capellambse.MelodyModel - polarion_id_map: dict[str, str] descr_references: dict[str, list[str]] + serializers: dict[ - str, cabc.Callable[[common.GenericElement], CapellaWorkItem] + str, + cabc.Callable[ + [common.GenericElement], capella_work_item.CapellaWorkItem + ], ] serializer_mapping: dict[str, str] @@ -158,14 +153,14 @@ def __init__( diagram_cache_path: pathlib.Path, polarion_type_map: dict[str, str], model: capellambse.MelodyModel, - polarion_id_map: dict[str, str], + capella_polarion_mapping: polarion_repo.PolarionDataRepository, descr_references: dict[str, list[str]], serializer_mapping: dict[str, str] | None = None, ): self.diagram_cache_path = diagram_cache_path self.polarion_type_map = polarion_type_map self.model = model - self.polarion_id_map = polarion_id_map + 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, @@ -177,7 +172,7 @@ def __init__( def serialize( self, obj: diagr.Diagram | common.GenericElement - ) -> CapellaWorkItem | None: + ) -> capella_work_item.CapellaWorkItem | None: """Return a CapellaWorkItem for the given diagram or element.""" try: if isinstance(obj, diagr.Diagram): @@ -195,7 +190,9 @@ def serialize( logger.error("Serializing model element failed. %s", error.args[0]) return None - def diagram(self, diag: diagr.Diagram) -> CapellaWorkItem: + def diagram( + self, diag: diagr.Diagram + ) -> capella_work_item.CapellaWorkItem: """Serialize a diagram for Polarion.""" diagram_path = self.diagram_cache_path / f"{diag.uuid}.svg" src = _decode_diagram(diagram_path) @@ -205,7 +202,7 @@ def diagram(self, diag: diagr.Diagram) -> CapellaWorkItem: description = ( f'

' ) - return CapellaWorkItem( + return capella_work_item.CapellaWorkItem( type="diagram", title=diag.name, description_type="text/html", @@ -216,13 +213,13 @@ def diagram(self, diag: diagr.Diagram) -> CapellaWorkItem: def _generic_work_item( self, obj: common.GenericElement - ) -> CapellaWorkItem: + ) -> capella_work_item.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 CapellaWorkItem( + return capella_work_item.CapellaWorkItem( type=resolve_element_type(xtype), title=obj.name, description_type="text/html", @@ -286,7 +283,7 @@ def replace_markup( except KeyError: logger.error("Found link to non-existing model element: %r", uuid) return strike_through(match.group(default_group)) - if pid := self.polarion_id_map.get(uuid): + if pid := self.capella_polarion_mapping.get_work_item_id(uuid): referenced_uuids.append(uuid) return POLARION_WORK_ITEM_URL.format(pid=pid) logger.warning("Found reference to non-existing work item: %r", uuid) @@ -294,7 +291,7 @@ def replace_markup( def include_pre_and_post_condition( self, obj: PrePostConditionElement - ) -> CapellaWorkItem: + ) -> capella_work_item.CapellaWorkItem: """Return generic attributes and pre- and post-condition.""" def get_condition(cap: PrePostConditionElement, name: str) -> str: @@ -328,14 +325,18 @@ def get_linked_text( self.descr_references[obj.uuid] = uuids return value - def constraint(self, obj: capellacore.Constraint) -> CapellaWorkItem: + def constraint( + self, obj: capellacore.Constraint + ) -> capella_work_item.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) return work_item - def _include_actor_in_type(self, obj: cs.Component) -> CapellaWorkItem: + def _include_actor_in_type( + self, obj: cs.Component + ) -> capella_work_item.CapellaWorkItem: """Return attributes for a ``Component``.""" work_item = self._generic_work_item(obj) if obj.is_actor: @@ -348,7 +349,7 @@ def _include_actor_in_type(self, obj: cs.Component) -> CapellaWorkItem: def _include_nature_in_type( self, obj: pa.PhysicalComponent - ) -> CapellaWorkItem: + ) -> capella_work_item.CapellaWorkItem: """Return attributes for a ``PhysicalComponent``.""" work_item = self._include_actor_in_type(obj) xtype = work_item.type diff --git a/capella2polarion/capella_polarion_conversion/link_converter.py b/capella2polarion/capella_polarion_conversion/link_converter.py new file mode 100644 index 00000000..ab73c5f5 --- /dev/null +++ b/capella2polarion/capella_polarion_conversion/link_converter.py @@ -0,0 +1,295 @@ +# Copyright DB Netz AG and contributors +# SPDX-License-Identifier: Apache-2.0 +"""Objects for synchronization of Capella model objects to Polarion.""" +from __future__ import annotations + +import collections.abc as cabc +import logging +from collections import defaultdict + +import capellambse +import polarion_rest_api_client as polarion_api +from capellambse.model import common +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 + +logger = logging.getLogger(__name__) + +TYPE_RESOLVERS = {"Part": lambda obj: obj.type.uuid} + + +class LinkSerializer: + """A converter for capella element links and description references.""" + + def __init__( + self, + capella_polarion_mapping: polarion_repo.PolarionDataRepository, + description_references: dict[str, list[str]], + project_id: str, + model: capellambse.MelodyModel, + ): + self.capella_polarion_mapping = capella_polarion_mapping + self.description_references = description_references + self.project_id = project_id + self.model = model + + def create_links_for_work_item( + self, + obj: common.GenericElement | diag.Diagram, + roles, + ) -> list[polarion_api.WorkItemLink]: + """Create work item links for a given Capella object.""" + if isinstance(obj, diag.Diagram): + repres = f"" + else: + repres = obj._short_repr_() + + work_item = ( + self.capella_polarion_mapping.get_work_item_by_capella_uuid( + obj.uuid + ) + ) + assert work_item is not None + 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", + ) + ) + else: + if (refs := getattr(obj, role_id, None)) is None: + logger.info( + "Unable to create work item link %r for [%s]. " + "There is no %r attribute on %s", + role_id, + work_item.id, + role_id, + repres, + ) + continue + + if isinstance(refs, common.ElementList): + new: cabc.Iterable[str] = refs.by_uuid # type: ignore[assignment] + else: + assert hasattr(refs, "uuid") + new = [refs.uuid] + + new = set(self._get_work_item_ids(work_item.id, new, role_id)) + new_links.extend(self._create(work_item.id, role_id, new, {})) + return new_links + + def _get_work_item_ids( + self, + primary_id: str, + uuids: cabc.Iterable[str], + role_id: str, + ) -> cabc.Iterator[str]: + for uuid in uuids: + if wid := self.capella_polarion_mapping.get_work_item_id(uuid): + yield wid + else: + obj = self.model.by_uuid(uuid) + logger.info( + "Unable to create work item link %r for [%s]. " + "Couldn't identify work item for %r", + role_id, + primary_id, + obj._short_repr_(), + ) + + def _handle_description_reference_links( + self, + obj: common.GenericElement, + work_item_id: str, + role_id: str, + links: dict[str, polarion_api.WorkItemLink], + ) -> list[polarion_api.WorkItemLink]: + refs = self.description_references.get(obj.uuid, []) + ref_set = set(self._get_work_item_ids(work_item_id, refs, role_id)) + return self._create(work_item_id, role_id, ref_set, links) + + def _handle_diagram_reference_links( + self, + obj: diag.Diagram, + work_item_id: str, + role_id: str, + links: dict[str, polarion_api.WorkItemLink], + ) -> list[polarion_api.WorkItemLink]: + try: + refs = set(self._collect_uuids(obj.nodes)) + refs = set(self._get_work_item_ids(work_item_id, refs, role_id)) + ref_links = self._create(work_item_id, role_id, refs, links) + except StopIteration: + logger.exception( + "Could not create links for diagram %r", obj._short_repr_() + ) + ref_links = [] + return ref_links + + def _collect_uuids( + self, + nodes: cabc.Iterable[common.GenericElement], + ) -> cabc.Iterator[str]: + type_resolvers = TYPE_RESOLVERS + for node in nodes: + uuid = node.uuid + if resolver := type_resolvers.get(type(node).__name__): + uuid = resolver(node) + + yield uuid + + def _create( + self, + primary_id: str, + role_id: str, + new: cabc.Iterable[str], + old: cabc.Iterable[str], + ) -> list[polarion_api.WorkItemLink]: + new = set(new) - set(old) + _new_links = [ + polarion_api.WorkItemLink( + primary_id, + id, + role_id, + secondary_work_item_project=self.project_id, + ) + for id in new + ] + return list(filter(None, _new_links)) + + def _handle_exchanges( + self, + obj: fa.Function, + work_item_id: str, + role_id: str, + links: dict[str, polarion_api.WorkItemLink], + attr: str = "inputs", + ) -> list[polarion_api.WorkItemLink]: + exchanges: list[str] = [] + for element in getattr(obj, attr): + uuids = element.exchanges.by_uuid + exs = self._get_work_item_ids(work_item_id, uuids, role_id) + exchanges.extend(set(exs)) + return self._create(work_item_id, role_id, exchanges, links) + + +def create_grouped_link_fields( + work_item: capella_work_item.CapellaWorkItem, + back_links: dict[str, list[polarion_api.WorkItemLink]] | None = None, +): + """Create the grouped link work items fields from the primary work item. + + Parameters + ---------- + work_item + WorkItem to create the fields for. + back_links + A dictionary of secondary WorkItem IDs to links to create + backlinks later. + """ + wi = f"[{work_item.id}]({work_item.type} {work_item.title})" + logger.debug("Building grouped links for work item %r...", wi) + for role, grouped_links in _group_by( + "role", work_item.linked_work_items + ).items(): + if back_links is not None: + for link in grouped_links: + key = link.secondary_work_item_id + back_links.setdefault(key, []).append(link) + + _create_link_fields(work_item, role, grouped_links) + + +def create_grouped_back_link_fields( + work_item: capella_work_item.CapellaWorkItem, + links: list[polarion_api.WorkItemLink], +): + """Create backlinks for the given WorkItem using a list of backlinks. + + Parameters + ---------- + work_item + WorkItem to create the fields for + links + List of links referencing work_item as secondary + """ + for role, grouped_links in _group_by("role", links).items(): + _create_link_fields(work_item, role, grouped_links, True) + + +def _group_by( + attr: str, + links: cabc.Iterable[polarion_api.WorkItemLink], +) -> dict[str, list[polarion_api.WorkItemLink]]: + group = defaultdict(list) + for link in links: + key = getattr(link, attr) + group[key].append(link) + return group + + +def _make_url_list( + links: cabc.Iterable[polarion_api.WorkItemLink], reverse: bool = False +) -> str: + urls: list[str] = [] + for link in links: + if reverse: + pid = link.primary_work_item_id + else: + pid = link.secondary_work_item_id + + url = element_converter.POLARION_WORK_ITEM_URL.format(pid=pid) + urls.append(f"
  • {url}
    • ") + + urls.sort() + url_list = "\n".join(urls) + return f"" + + +def _create_link_fields( + work_item: capella_work_item.CapellaWorkItem, + role: str, + links: list[polarion_api.WorkItemLink], + reverse: bool = False, +): + role = f"{role}_reverse" if reverse else role + work_item.additional_attributes[role] = { + "type": "text/html", + "value": _make_url_list(links, reverse), + } diff --git a/capella2polarion/capella_work_item.py b/capella2polarion/capella_work_item.py new file mode 100644 index 00000000..485c1f21 --- /dev/null +++ b/capella2polarion/capella_work_item.py @@ -0,0 +1,22 @@ +# Copyright DB Netz AG and contributors +# SPDX-License-Identifier: Apache-2.0 +"""Module providing the CapellaWorkItem class.""" +from __future__ import annotations + +import typing as t + +import polarion_rest_api_client as polarion_api + + +class CapellaWorkItem(polarion_api.WorkItem): + """A custom WorkItem class with additional capella related attributes.""" + + class Condition(t.TypedDict): + """A class to describe a pre or post condition.""" + + type: str + value: str + + uuid_capella: str + preCondition: Condition | None + postCondition: Condition | None diff --git a/capella2polarion/elements/capella_polarion_mapping.py b/capella2polarion/elements/capella_polarion_mapping.py deleted file mode 100644 index 38312231..00000000 --- a/capella2polarion/elements/capella_polarion_mapping.py +++ /dev/null @@ -1,115 +0,0 @@ -# Copyright DB Netz AG and contributors -# SPDX-License-Identifier: Apache-2.0 -"""Module providing a universal CapellaPolarionMapping class.""" -from __future__ import annotations - -import bidict -import capellambse - -from capella2polarion.elements import serialize - - -class CapellaPolarionMapping: - """A mapping class to access all contents by Capella and Polarion IDs.""" - - _description_references: dict[str, set[str]] - _id_mapping: bidict.bidict[str, str] - _work_items: dict[str, serialize.CapellaWorkItem] - _model: capellambse.MelodyModel - - def __init__( - self, - model: capellambse.MelodyModel, - polarion_work_items: list[serialize.CapellaWorkItem], - ): - self._description_references = {} - self._model = model - self._id_mapping = bidict.bidict( - { - work_item.uuid_capella: work_item.id - for work_item in polarion_work_items - } - ) - self._work_items = { - work_item.uuid_capella: work_item - for work_item in polarion_work_items - } - - def get_work_item_id(self, capella_uuid: str) -> str | None: - """Return a Work Item ID for a given Capella UUID.""" - return self._id_mapping.get(capella_uuid) - - def get_capella_uuid(self, work_item_id: str) -> str | None: - """Return a Capella UUID for a given Work Item ID.""" - return self._id_mapping.inverse.get(work_item_id) - - def get_work_item_by_capella_uuid( - self, capella_uuid: str - ) -> serialize.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 - ) -> serialize.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 get_model_element_by_capella_uuid( - self, capella_uuid: str - ) -> capellambse.model.GenericElement | None: - """Return a model element for a given Capella UUID.""" - try: - return self._model.by_uuid(capella_uuid) - except KeyError: - return None - - def get_model_element_by_polarion_id( - self, work_item_id: str - ) -> capellambse.model.GenericElement | None: - """Return a model element for a given Work Item ID.""" - return self.get_model_element_by_capella_uuid( - self.get_capella_uuid(work_item_id) # type: ignore - ) - - def get_description_references_by_capella_uuid( - self, capella_uuid: str - ) -> set[str] | None: - """Return the description references for a given Capella UUID.""" - return self._description_references.get(capella_uuid) - - def get_description_references_by_polarion_id( - self, work_item_id: str - ) -> set[str] | None: - """Return the description references for a given Work Item ID.""" - return self.get_description_references_by_capella_uuid( - self.get_capella_uuid(work_item_id) # type: ignore - ) - - def update_work_items( - self, - work_items: serialize.CapellaWorkItem - | list[serialize.CapellaWorkItem], - ): - """Update all mappings for the given Work Items.""" - if isinstance(work_items, serialize.CapellaWorkItem): - work_items = [work_items] - - self._id_mapping.update( - { - work_item.uuid_capella: work_item.id - for work_item in work_items - if work_item.id is not None - } - ) - self._work_items.update( - {work_item.uuid_capella: work_item for work_item in work_items} - ) - - def update_description_reference( - self, capella_uuid: str, references: list[str] - ): - """Add or replace description references for a given capella UUID.""" - self._description_references.update({capella_uuid: set(references)}) diff --git a/capella2polarion/elements/element.py b/capella2polarion/elements/element.py deleted file mode 100644 index 15932050..00000000 --- a/capella2polarion/elements/element.py +++ /dev/null @@ -1,304 +0,0 @@ -# Copyright DB Netz AG and contributors -# SPDX-License-Identifier: Apache-2.0 -"""Objects for synchronization of Capella model objects to Polarion.""" -from __future__ import annotations - -import collections.abc as cabc -import logging -from collections import defaultdict - -import polarion_rest_api_client as polarion_api -from capellambse.model import common -from capellambse.model import diagram as diag -from capellambse.model.crosslayer import fa - -from capella2polarion.elements import serialize - -logger = logging.getLogger(__name__) - -TYPE_RESOLVERS = {"Part": lambda obj: obj.type.uuid} - - -def create_links( - obj: common.GenericElement | diag.Diagram, - polarion_id_map, - work_item_map, - descr_references, - project_id, - model, - roles, -) -> list[polarion_api.WorkItemLink]: - """Create work item links for a given Capella object.""" - if isinstance(obj, diag.Diagram): - repres = f"" - else: - repres = obj._short_repr_() - - workitem = work_item_map[obj.uuid] - new_links: list[polarion_api.WorkItemLink] = [] - typ = workitem.type[0].upper() + workitem.type[1:] - for role_id in roles.get(typ, []): - if role_id == "description_reference": - new_links.extend( - _handle_description_reference_links( - polarion_id_map, - descr_references, - project_id, - model, - obj, - role_id, - {}, - ) - ) - elif role_id == "diagram_elements": - new_links.extend( - _handle_diagram_reference_links( - polarion_id_map, model, project_id, obj, role_id, {} - ) - ) - elif role_id == "input_exchanges": - new_links.extend( - _handle_exchanges( - polarion_id_map, - model, - project_id, - obj, - role_id, - {}, - "inputs", - ) - ) - elif role_id == "output_exchanges": - new_links.extend( - _handle_exchanges( - polarion_id_map, - model, - project_id, - obj, - role_id, - {}, - "outputs", - ) - ) - else: - if (refs := getattr(obj, role_id, None)) is None: - logger.info( - "Unable to create work item link %r for [%s]. " - "There is no %r attribute on %s", - role_id, - workitem.id, - role_id, - repres, - ) - continue - - if isinstance(refs, common.ElementList): - new: cabc.Iterable[str] = refs.by_uuid # type: ignore[assignment] - else: - assert hasattr(refs, "uuid") - new = [refs.uuid] - - new = set( - _get_work_item_ids( - polarion_id_map, model, workitem.id, new, role_id - ) - ) - new_links.extend( - _create(project_id, workitem.id, role_id, new, {}) - ) - return new_links - - -def _get_work_item_ids( - polarion_id_map, - model, - primary_id: str, - uuids: cabc.Iterable[str], - role_id: str, -) -> cabc.Iterator[str]: - for uuid in uuids: - if wid := polarion_id_map.get(uuid): - yield wid - else: - obj = model.by_uuid(uuid) - logger.info( - "Unable to create work item link %r for [%s]. " - "Couldn't identify work item for %r", - role_id, - primary_id, - obj._short_repr_(), - ) - - -def _handle_description_reference_links( - polarion_id_map, - descr_references, - project_id, - model, - obj: common.GenericElement, - role_id: str, - links: dict[str, polarion_api.WorkItemLink], -) -> list[polarion_api.WorkItemLink]: - refs = descr_references.get(obj.uuid, []) - wid = polarion_id_map[obj.uuid] - refs = set(_get_work_item_ids(polarion_id_map, model, wid, refs, role_id)) - return _create(project_id, wid, role_id, refs, links) - - -def _handle_diagram_reference_links( - polarion_id_map, - model, - project_id, - obj: diag.Diagram, - role_id: str, - links: dict[str, polarion_api.WorkItemLink], -) -> list[polarion_api.WorkItemLink]: - try: - refs = set(_collect_uuids(obj.nodes)) - wid = polarion_id_map[obj.uuid] - refs = set( - _get_work_item_ids(polarion_id_map, model, wid, refs, role_id) - ) - ref_links = _create(project_id, wid, role_id, refs, links) - except StopIteration: - logger.exception( - "Could not create links for diagram %r", obj._short_repr_() - ) - ref_links = [] - return ref_links - - -def _collect_uuids( - nodes: cabc.Iterable[common.GenericElement], -) -> cabc.Iterator[str]: - type_resolvers = TYPE_RESOLVERS - for node in nodes: - uuid = node.uuid - if resolver := type_resolvers.get(type(node).__name__): - uuid = resolver(node) - - yield uuid - - -def _create( - project_id, - primary_id: str, - role_id: str, - new: cabc.Iterable[str], - old: cabc.Iterable[str], -) -> list[polarion_api.WorkItemLink]: - new = set(new) - set(old) - _new_links = [ - polarion_api.WorkItemLink( - primary_id, - id, - role_id, - secondary_work_item_project=project_id, - ) - for id in new - ] - return list(filter(None, _new_links)) - - -def _handle_exchanges( - polarion_id_map, - model, - project_id, - obj: fa.Function, - role_id: str, - links: dict[str, polarion_api.WorkItemLink], - attr: str = "inputs", -) -> list[polarion_api.WorkItemLink]: - wid = polarion_id_map[obj.uuid] - exchanges: list[str] = [] - for element in getattr(obj, attr): - uuids = element.exchanges.by_uuid - exs = _get_work_item_ids(polarion_id_map, model, wid, uuids, role_id) - exchanges.extend(set(exs)) - return _create(project_id, wid, role_id, exchanges, links) - - -def create_grouped_link_fields( - work_item: serialize.CapellaWorkItem, - back_links: dict[str, list[polarion_api.WorkItemLink]] | None = None, -): - """Create the grouped link work items fields from the primary work item. - - Parameters - ---------- - work_item - WorkItem to create the fields for. - back_links - A dictionary of secondary WorkItem IDs to links to create - backlinks later. - """ - wi = f"[{work_item.id}]({work_item.type} {work_item.title})" - logger.debug("Building grouped links for work item %r...", wi) - for role, grouped_links in _group_by( - "role", work_item.linked_work_items - ).items(): - if back_links is not None: - for link in grouped_links: - key = link.secondary_work_item_id - back_links.setdefault(key, []).append(link) - - _create_link_fields(work_item, role, grouped_links) - - -def create_grouped_back_link_fields( - work_item: serialize.CapellaWorkItem, - links: list[polarion_api.WorkItemLink], -): - """Create backlinks for the given WorkItem using a list of backlinks. - - Parameters - ---------- - work_item - WorkItem to create the fields for - links - List of links referencing work_item as secondary - """ - for role, grouped_links in _group_by("role", links).items(): - _create_link_fields(work_item, role, grouped_links, True) - - -def _group_by( - attr: str, - links: cabc.Iterable[polarion_api.WorkItemLink], -) -> dict[str, list[polarion_api.WorkItemLink]]: - group = defaultdict(list) - for link in links: - key = getattr(link, attr) - group[key].append(link) - return group - - -def _make_url_list( - links: cabc.Iterable[polarion_api.WorkItemLink], reverse: bool = False -) -> str: - urls: list[str] = [] - for link in links: - if reverse: - pid = link.primary_work_item_id - else: - pid = link.secondary_work_item_id - - url = serialize.POLARION_WORK_ITEM_URL.format(pid=pid) - urls.append(f"
  • {url}
    • ") - - urls.sort() - url_list = "\n".join(urls) - return f"
      {url_list}
    " - - -def _create_link_fields( - work_item: serialize.CapellaWorkItem, - role: str, - links: list[polarion_api.WorkItemLink], - reverse: bool = False, -): - role = f"{role}_reverse" if reverse else role - work_item.additional_attributes[role] = { - "type": "text/html", - "value": _make_url_list(links, reverse), - } diff --git a/capella2polarion/polarion_connector/__init__.py b/capella2polarion/polarion_connector/__init__.py new file mode 100644 index 00000000..e42395a2 --- /dev/null +++ b/capella2polarion/polarion_connector/__init__.py @@ -0,0 +1,3 @@ +# Copyright DB Netz AG and contributors +# SPDX-License-Identifier: Apache-2.0 +"""Package for all conversion related activities.""" diff --git a/capella2polarion/polarion_connector/polarion_repo.py b/capella2polarion/polarion_connector/polarion_repo.py new file mode 100644 index 00000000..aaac0381 --- /dev/null +++ b/capella2polarion/polarion_connector/polarion_repo.py @@ -0,0 +1,118 @@ +# Copyright DB Netz AG and contributors +# SPDX-License-Identifier: Apache-2.0 +"""Module providing a universal PolarionDataRepository class.""" +from __future__ import annotations + +import typing + +import bidict + +from capella2polarion import capella_work_item + + +class PolarionDataRepository: + """A mapping class to access all contents by Capella and Polarion IDs. + + This class only holds data already present in the 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] + + def __init__( + self, + polarion_work_items: typing.Optional[ + list[capella_work_item.CapellaWorkItem] + ] = None, + ): + if polarion_work_items is None: + polarion_work_items = [] + self._id_mapping = bidict.bidict( + { + work_item.uuid_capella: work_item.id + for work_item in polarion_work_items + }, + ) + self._id_mapping.on_dup = bidict.OnDup( + key=bidict.DROP_OLD, val=bidict.DROP_OLD, kv=bidict.DROP_OLD + ) + self._work_items = { + work_item.uuid_capella: work_item + for work_item in polarion_work_items + } + + 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): + """Return the amount of registered Capella UUIDs.""" + return len(self._id_mapping) + + def __getitem__( + self, item: str + ) -> typing.Tuple[str, capella_work_item.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]: + """Iterate all Capella UUIDs.""" + return self._id_mapping.__iter__() + + def items( + self, + ): + """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] + + def get_work_item_id(self, capella_uuid: str) -> str | None: + """Return a Work Item ID for a given Capella UUID.""" + return self._id_mapping.get(capella_uuid) + + def get_capella_uuid(self, work_item_id: str) -> str | None: + """Return a Capella UUID for a given Work Item ID.""" + return self._id_mapping.inverse.get(work_item_id) + + def get_work_item_by_capella_uuid( + self, capella_uuid: str + ) -> capella_work_item.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: + """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], + ): + """Update all mappings for the given Work Items.""" + for work_item in work_items: + if uuid_capella := self._id_mapping.inverse.get(work_item.id): + del self._id_mapping[uuid_capella] + del self._work_items[uuid_capella] + + self._id_mapping.update( + { + work_item.uuid_capella: work_item.id + for work_item in work_items + if work_item.id is not None + } + ) + self._work_items.update( + {work_item.uuid_capella: work_item for work_item in work_items} + ) + + def remove_work_items_by_capella_uuid(self, uuids: typing.Iterable[str]): + """Remove entries for the given Capella UUIDs.""" + for uuid in uuids: + del self._work_items[uuid] + del self._id_mapping[uuid] diff --git a/capella2polarion/polarion_worker.py b/capella2polarion/polarion_worker.py index 0a78e09a..50f270d8 100644 --- a/capella2polarion/polarion_worker.py +++ b/capella2polarion/polarion_worker.py @@ -14,7 +14,12 @@ import polarion_rest_api_client as polarion_api from capellambse.model import common -from capella2polarion.elements import element, serialize +from capella2polarion import capella_work_item +from capella2polarion.capella_polarion_conversion import ( + element_converter, + link_converter, +) +from capella2polarion.polarion_connector import polarion_repo logger = logging.getLogger(__name__) @@ -62,15 +67,16 @@ class PolarionWorker: def __init__( self, params: PolarionWorkerParams, + model: capellambse.MelodyModel, make_type_id: typing.Any, ) -> None: self.polarion_params: PolarionWorkerParams = params self.elements: dict[str, list[common.GenericElement]] = {} - self.polarion_type_map: dict[str, str] = {} - self.capella_uuid_s: set[str] = set() + self.polarion_type_map: dict[str, str] = {} # TODO refactor + self.capella_uuid_s: set[str] = set() # TODO refactor self.x_types: set[str] = set() - self.polarion_id_map: dict[str, str] = {} - self.polarion_work_item_map: dict[str, serialize.CapellaWorkItem] = {} + self.polarion_data_repo = polarion_repo.PolarionDataRepository() + self.model = model self.make_type_id: typing.Any = make_type_id if (self.polarion_params.project_id is None) or ( len(self.polarion_params.project_id) == 0 @@ -96,7 +102,7 @@ def __init__( self.polarion_params.delete_work_items, polarion_api_endpoint=f"{self.polarion_params.url}/rest/v1", polarion_access_token=self.polarion_params.private_access_token, - custom_work_item=serialize.CapellaWorkItem, + custom_work_item=capella_work_item.CapellaWorkItem, add_work_item_checksum=True, ) self.check_client() @@ -116,7 +122,6 @@ def check_client(self) -> None: def load_elements_and_type_map( self, config: dict[str, typing.Any], - model: capellambse.MelodyModel, diagram_idx: list[dict[str, typing.Any]], ) -> None: """Return an elements and UUID to Polarion type map.""" @@ -124,7 +129,7 @@ def load_elements_and_type_map( type_map: dict[str, str] = {} elements: dict[str, list[common.GenericElement]] = {} for _below, pol_types in config.items(): - below = getattr(model, _below) + below = getattr(self.model, _below) for typ in pol_types: if isinstance(typ, dict): typ = list(typ.keys())[0] @@ -133,7 +138,7 @@ def load_elements_and_type_map( continue xtype = convert_type.get(typ, typ) - objects = model.search(xtype, below=below) + objects = self.model.search(xtype, below=below) elements.setdefault(typ, []).extend(objects) for obj in objects: type_map[obj.uuid] = typ @@ -173,7 +178,7 @@ def load_elements_and_type_map( 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 + d for d in self.model.diagrams if d.uuid in diagrams_from_cache ] for obj in elements["Diagram"]: type_map[obj.uuid] = "Diagram" @@ -199,29 +204,22 @@ def load_polarion_work_item_map(self): {"workitems": "id,uuid_capella,checksum,status"}, ) - self.polarion_work_item_map = { - wi.uuid_capella: wi - for wi in work_items - if wi.id and wi.uuid_capella - } - self.polarion_id_map = { - uuid: wi.id for uuid, wi in self.polarion_work_item_map.items() - } + self.polarion_data_repo.update_work_items(work_items) def create_work_items( self, diagram_cache_path: pathlib.Path, model, descr_references: dict[str, list[str]], - ) -> dict[str, serialize.CapellaWorkItem]: + ) -> dict[str, capella_work_item.CapellaWorkItem]: """Create a list of work items for Polarion.""" objects = chain.from_iterable(self.elements.values()) _work_items = [] - serializer = serialize.CapellaWorkItemSerializer( + serializer = element_converter.CapellaWorkItemSerializer( diagram_cache_path, self.polarion_type_map, model, - self.polarion_id_map, + self.polarion_data_repo, descr_references, ) for obj in objects: @@ -229,13 +227,15 @@ def create_work_items( _work_items = list(filter(None, _work_items)) valid_types = set(map(self.make_type_id, set(self.elements))) - work_items: list[serialize.CapellaWorkItem] = [] + work_items: list[capella_work_item.CapellaWorkItem] = [] missing_types: set[str] = set() for work_item in _work_items: assert work_item is not None assert work_item.title is not None assert work_item.type is not None - if old := self.polarion_work_item_map.get(work_item.uuid_capella): + if old := self.polarion_data_repo.get_work_item_by_capella_uuid( + work_item.uuid_capella + ): work_item.id = old.id if work_item.type in valid_types: work_items.append(work_item) @@ -257,15 +257,13 @@ def delete_work_items(self) -> None: """ def serialize_for_delete(uuid: str) -> str: - logger.info( - "Delete work item %r...", - workitem_id := self.polarion_id_map[uuid], - ) - return workitem_id + work_item_id, _ = self.polarion_data_repo[uuid] + logger.info("Delete work item %r...", work_item_id) + return work_item_id existing_work_items = { uuid - for uuid, work_item in self.polarion_work_item_map.items() + for uuid, _, work_item in self.polarion_data_repo.items() if work_item.status != "deleted" } uuids: set[str] = existing_work_items - self.capella_uuid_s @@ -273,19 +271,19 @@ def serialize_for_delete(uuid: str) -> str: if work_item_ids: try: 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] + self.polarion_data_repo.remove_work_items_by_capella_uuid( + uuids + ) except polarion_api.PolarionApiException as error: logger.error("Deleting work items failed. %s", error.args[0]) def post_work_items( - self, new_work_items: dict[str, serialize.CapellaWorkItem] + self, new_work_items: dict[str, capella_work_item.CapellaWorkItem] ) -> None: """Post work items in a Polarion project.""" - missing_work_items: list[serialize.CapellaWorkItem] = [] + missing_work_items: list[capella_work_item.CapellaWorkItem] = [] for work_item in new_work_items.values(): - if work_item.uuid_capella in self.polarion_id_map: + if work_item.uuid_capella in self.polarion_data_repo: continue assert work_item is not None @@ -294,18 +292,14 @@ def post_work_items( if missing_work_items: try: 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 + self.polarion_data_repo.update_work_items(missing_work_items) except polarion_api.PolarionApiException as error: logger.error("Creating work items failed. %s", error.args[0]) def patch_work_item( self, - new: serialize.CapellaWorkItem, - old: serialize.CapellaWorkItem, + new: capella_work_item.CapellaWorkItem, + old: capella_work_item.CapellaWorkItem, ): """Patch a given WorkItem. @@ -395,44 +389,41 @@ def _get_link_id(link: polarion_api.WorkItemLink) -> str: def patch_work_items( self, - model: capellambse.MelodyModel, - new_work_items: dict[str, serialize.CapellaWorkItem], + new_work_items: dict[str, capella_work_item.CapellaWorkItem], descr_references, link_roles, ) -> None: """Update work items in a Polarion project.""" - self.polarion_id_map = { - uuid: wi.id - for uuid, wi in self.polarion_work_item_map.items() - if wi.status == "open" and wi.uuid_capella and wi.id - } back_links: dict[str, list[polarion_api.WorkItemLink]] = {} - for uuid in self.polarion_id_map: - objects = model + link_serializer = link_converter.LinkSerializer( + self.polarion_data_repo, + descr_references, + self.polarion_params.project_id, + self.model, + ) + + for uuid in self.polarion_data_repo: + objects = self.model if uuid.startswith("_"): - objects = model.diagrams + objects = self.model.diagrams obj = objects.by_uuid(uuid) - links = element.create_links( + links = link_serializer.create_links_for_work_item( obj, - self.polarion_id_map, - self.polarion_work_item_map, - descr_references, - self.polarion_params.project_id, - model, link_roles, ) - work_item: serialize.CapellaWorkItem = new_work_items[uuid] + work_item: capella_work_item.CapellaWorkItem = new_work_items[uuid] work_item.linked_work_items = links - element.create_grouped_link_fields(work_item, back_links) + link_converter.create_grouped_link_fields(work_item, back_links) - for uuid in self.polarion_id_map: - new_work_item: serialize.CapellaWorkItem = new_work_items[uuid] - old_work_item = self.polarion_work_item_map[uuid] + for uuid, _, old_work_item in self.polarion_data_repo.items(): + new_work_item: capella_work_item.CapellaWorkItem = new_work_items[ + uuid + ] if old_work_item.id in back_links: - element.create_grouped_back_link_fields( + link_converter.create_grouped_back_link_fields( new_work_item, back_links[old_work_item.id] ) diff --git a/tests/conftest.py b/tests/conftest.py index 54580bba..a48f7d2c 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -12,7 +12,7 @@ import polarion_rest_api_client as polarion_api import pytest -from capella2polarion.elements import serialize +from capella2polarion import capella_work_item TEST_DATA_ROOT = pathlib.Path(__file__).parent / "data" TEST_DIAGRAM_CACHE = TEST_DATA_ROOT / "diagram_cache" @@ -36,9 +36,9 @@ def model() -> capellambse.MelodyModel: @pytest.fixture -def dummy_work_items() -> dict[str, serialize.CapellaWorkItem]: +def dummy_work_items() -> dict[str, capella_work_item.CapellaWorkItem]: return { - f"uuid{i}": serialize.CapellaWorkItem( + f"uuid{i}": capella_work_item.CapellaWorkItem( id=f"Obj-{i}", uuid_capella=f"uuid{i}", title=f"Fake {i}", diff --git a/tests/test_elements.py b/tests/test_elements.py index 75795b1b..3f6e56bd 100644 --- a/tests/test_elements.py +++ b/tests/test_elements.py @@ -14,8 +14,13 @@ import pytest from capellambse.model import common +from capella2polarion import capella_work_item from capella2polarion.capella2polarioncli import Capella2PolarionCli -from capella2polarion.elements import element, serialize +from capella2polarion.capella_polarion_conversion import ( + element_converter, + link_converter, +) +from capella2polarion.polarion_connector import polarion_repo from capella2polarion.polarion_worker import PolarionWorker # pylint: disable-next=relative-beyond-top-level, useless-suppression @@ -45,7 +50,6 @@ TEST_SCENARIO = "afdaa095-e2cd-4230-b5d3-6cb771a90f51" TEST_CAP_REAL = "b80b3141-a7fc-48c7-84b2-1467dcef5fce" TEST_CONSTRAINT = "95cbd4af-7224-43fe-98cb-f13dda540b8e" -TEST_POL_ID_MAP = {TEST_E_UUID: "TEST"} TEST_POL_TYPE_MAP = { TEST_ELEMENT_UUID: "LogicalComponent", TEST_OCAP_UUID: "OperationalCapability", @@ -146,7 +150,9 @@ def write(self, text: str): pass uuid = diagram_cache_index[0]["uuid"] - work_item = serialize.CapellaWorkItem(id="Diag-1", checksum="123") + work_item = capella_work_item.CapellaWorkItem( + id="Diag-1", checksum="123", uuid_capella=uuid + ) c2p_cli = Capella2PolarionCli( debug=True, polarion_project_id="project_id", @@ -166,11 +172,13 @@ def write(self, text: str): ) pw = PolarionWorker( c2p_cli.polarion_params, - serialize.resolve_element_type, + model, + element_converter.resolve_element_type, ) pw.capella_uuid_s = {d["uuid"] for d in diagram_cache_index} - pw.polarion_work_item_map = {uuid: work_item} - pw.polarion_id_map = {uuid: "Diag-1"} + pw.polarion_data_repo = polarion_repo.PolarionDataRepository( + [work_item] + ) pw.elements = {"Diagram": c2p_cli.capella_model.diagrams} return BaseObjectContainer(c2p_cli, pw) @@ -179,7 +187,7 @@ def test_create_diagrams(base_object: BaseObjectContainer): c2p_cli = base_object.c2pcli pw = base_object.pw description_reference: dict[str, list[str]] = {} - new_work_items: dict[str, serialize.CapellaWorkItem] + new_work_items: dict[str, capella_work_item.CapellaWorkItem] new_work_items = pw.create_work_items( c2p_cli.capella_diagram_cache_folder_path, c2p_cli.capella_model, @@ -187,10 +195,12 @@ def test_create_diagrams(base_object: BaseObjectContainer): ) assert len(new_work_items) == 1 work_item = new_work_items[TEST_DIAG_UUID] - assert isinstance(work_item, serialize.CapellaWorkItem) + assert isinstance(work_item, capella_work_item.CapellaWorkItem) description = work_item.description work_item.description = None - assert work_item == serialize.CapellaWorkItem(**TEST_SER_DIAGRAM) + assert work_item == capella_work_item.CapellaWorkItem( + **TEST_SER_DIAGRAM + ) assert isinstance(description, str) assert description.startswith(TEST_DIAG_DESCR) @@ -258,7 +268,7 @@ class MyIO(io.StringIO): def write(self, text: str): pass - work_item = serialize.CapellaWorkItem( + work_item = capella_work_item.CapellaWorkItem( id="Obj-1", uuid_capella="uuid1", status="open", @@ -284,10 +294,12 @@ def write(self, text: str): ) pw = PolarionWorker( c2p_cli.polarion_params, - serialize.resolve_element_type, + model, + element_converter.resolve_element_type, + ) + pw.polarion_data_repo = polarion_repo.PolarionDataRepository( + [work_item] ) - pw.polarion_work_item_map = {"uuid1": work_item} - pw.polarion_id_map = {"uuid1": "Obj-1"} pw.polarion_type_map = {"uuid1": "FakeModelObject"} fake = FakeModelObject("uuid1", name="Fake 1") pw.elements = { @@ -311,19 +323,19 @@ def test_create_work_items( base_object.pw.elements["FakeModelObject"] ) monkeypatch.setattr( - serialize.CapellaWorkItemSerializer, + element_converter.CapellaWorkItemSerializer, "serialize", mock_generic_work_item := mock.MagicMock(), ) mock_generic_work_item.side_effect = [ - expected := serialize.CapellaWorkItem( + expected := capella_work_item.CapellaWorkItem( uuid_capella="uuid1", title="Fake 1", type="fakeModelObject", description_type="text/html", description=markupsafe.Markup(""), ), - expected1 := serialize.CapellaWorkItem( + expected1 := capella_work_item.CapellaWorkItem( uuid_capella="uuid2", title="Fake 2", type="fakeModelObject", @@ -361,7 +373,7 @@ def test_create_work_items_with_special_polarion_type( base_object.pw.polarion_type_map[uuid] = _type base_object.c2pcli.capella_model = model - expected = serialize.CapellaWorkItem( + expected = capella_work_item.CapellaWorkItem( uuid_capella=uuid, type=_type[0].lower() + _type[1:], description_type="text/html", @@ -380,16 +392,17 @@ def test_create_work_items_with_special_polarion_type( @staticmethod def test_create_links_custom_resolver(base_object: BaseObjectContainer): obj = base_object.pw.elements["FakeModelObject"][1] - base_object.pw.polarion_id_map["uuid2"] = "Obj-2" - base_object.pw.polarion_work_item_map[ - "uuid2" - ] = serialize.CapellaWorkItem( - id="Obj-2", - uuid_capella="uuid2", - type="fakeModelObject", - description_type="text/html", - description=markupsafe.Markup(""), - status="open", + base_object.pw.polarion_data_repo.update_work_items( + [ + capella_work_item.CapellaWorkItem( + id="Obj-2", + uuid_capella="uuid2", + type="fakeModelObject", + description_type="text/html", + description=markupsafe.Markup(""), + status="open", + ) + ] ) base_object.c2pcli.synchronize_config_roles = { "FakeModelObject": ["description_reference"] @@ -401,13 +414,14 @@ def test_create_links_custom_resolver(base_object: BaseObjectContainer): "description_reference", secondary_work_item_project="project_id", ) - links = element.create_links( - obj, - base_object.pw.polarion_id_map, - base_object.pw.polarion_work_item_map, + link_serializer = link_converter.LinkSerializer( + base_object.pw.polarion_data_repo, description_reference, base_object.pw.polarion_params.project_id, base_object.c2pcli.capella_model, + ) + links = link_serializer.create_links_for_work_item( + obj, base_object.c2pcli.synchronize_config_roles, ) assert links == [expected] @@ -421,27 +435,29 @@ def test_create_links_custom_exchanges_resolver( obj = base_object.c2pcli.capella_model.by_uuid(function_uuid) - base_object.pw.polarion_id_map[function_uuid] = "Obj-1" - base_object.pw.polarion_work_item_map[ - function_uuid - ] = serialize.CapellaWorkItem( - id="Obj-1", - uuid_capella=function_uuid, - type=type(obj).__name__, - description_type="text/html", - description=markupsafe.Markup(""), - status="open", + base_object.pw.polarion_data_repo.update_work_items( + [ + capella_work_item.CapellaWorkItem( + id="Obj-1", + uuid_capella=function_uuid, + type=type(obj).__name__, + description_type="text/html", + description=markupsafe.Markup(""), + status="open", + ) + ] ) - base_object.pw.polarion_id_map[uuid] = "Obj-2" - base_object.pw.polarion_work_item_map[ - uuid - ] = serialize.CapellaWorkItem( - id="Obj-2", - uuid_capella=uuid, - type="functionalExchange", - description_type="text/html", - description=markupsafe.Markup(""), - status="open", + base_object.pw.polarion_data_repo.update_work_items( + [ + capella_work_item.CapellaWorkItem( + id="Obj-2", + uuid_capella=uuid, + type="functionalExchange", + description_type="text/html", + description=markupsafe.Markup(""), + status="open", + ) + ] ) base_object.c2pcli.synchronize_config_roles = { @@ -453,13 +469,14 @@ def test_create_links_custom_exchanges_resolver( "input_exchanges", secondary_work_item_project="project_id", ) - links = element.create_links( - obj, - base_object.pw.polarion_id_map, - base_object.pw.polarion_work_item_map, + link_serializer = link_converter.LinkSerializer( + base_object.pw.polarion_data_repo, {}, base_object.pw.polarion_params.project_id, base_object.c2pcli.capella_model, + ) + links = link_serializer.create_links_for_work_item( + obj, base_object.c2pcli.synchronize_config_roles, ) assert links == [expected] @@ -475,13 +492,14 @@ def test_create_links_missing_attribute( "" ) with caplog.at_level(logging.DEBUG): - links = element.create_links( - obj, - base_object.pw.polarion_id_map, - base_object.pw.polarion_work_item_map, + link_serializer = link_converter.LinkSerializer( + base_object.pw.polarion_data_repo, {}, base_object.pw.polarion_params.project_id, base_object.c2pcli.capella_model, + ) + links = link_serializer.create_links_for_work_item( + obj, base_object.c2pcli.synchronize_config_roles, ) assert not links @@ -501,20 +519,19 @@ def test_create_links_from_ElementList(base_object: BaseObjectContainer): ), ) base_object.pw.elements["FakeModelObject"].append(obj) - base_object.pw.polarion_id_map |= { - f"uuid{i}": f"Obj-{i}" for i in range(4, 7) - } - base_object.pw.polarion_work_item_map |= { - f"uuid{i}": serialize.CapellaWorkItem( - id=f"Obj-{i}", - uuid_capella=f"uuid{i}", - type="fakeModelObject", - description_type="text/html", - description=markupsafe.Markup(""), - status="open", - ) - for i in range(4, 7) - } + base_object.pw.polarion_data_repo.update_work_items( + [ + capella_work_item.CapellaWorkItem( + id=f"Obj-{i}", + uuid_capella=f"uuid{i}", + type="fakeModelObject", + description_type="text/html", + description=markupsafe.Markup(""), + status="open", + ) + for i in range(4, 7) + ] + ) expected_link = polarion_api.WorkItemLink( "Obj-6", @@ -528,13 +545,14 @@ def test_create_links_from_ElementList(base_object: BaseObjectContainer): "attribute", secondary_work_item_project="project_id", ) - links = element.create_links( - obj, - base_object.pw.polarion_id_map, - base_object.pw.polarion_work_item_map, + link_serializer = link_converter.LinkSerializer( + base_object.pw.polarion_data_repo, {}, base_object.pw.polarion_params.project_id, base_object.c2pcli.capella_model, + ) + links = link_serializer.create_links_for_work_item( + obj, base_object.c2pcli.synchronize_config_roles, ) # type: ignore[arg-type] @@ -546,16 +564,17 @@ def test_create_link_from_single_attribute( base_object: BaseObjectContainer, ): obj = base_object.pw.elements["FakeModelObject"][1] - base_object.pw.polarion_id_map["uuid2"] = "Obj-2" - base_object.pw.polarion_work_item_map[ - "uuid2" - ] = serialize.CapellaWorkItem( - id="Obj-2", - uuid_capella="uuid2", - type="fakeModelObject", - description_type="text/html", - description=markupsafe.Markup(""), - status="open", + base_object.pw.polarion_data_repo.update_work_items( + [ + capella_work_item.CapellaWorkItem( + id="Obj-2", + uuid_capella="uuid2", + type="fakeModelObject", + description_type="text/html", + description=markupsafe.Markup(""), + status="open", + ) + ] ) expected = polarion_api.WorkItemLink( @@ -564,13 +583,14 @@ def test_create_link_from_single_attribute( "attribute", secondary_work_item_project="project_id", ) - links = element.create_links( - obj, - base_object.pw.polarion_id_map, - base_object.pw.polarion_work_item_map, + link_serializer = link_converter.LinkSerializer( + base_object.pw.polarion_data_repo, {}, base_object.pw.polarion_params.project_id, base_object.c2pcli.capella_model, + ) + links = link_serializer.create_links_for_work_item( + obj, base_object.c2pcli.synchronize_config_roles, ) assert links == [expected] @@ -579,8 +599,8 @@ def test_create_link_from_single_attribute( def test_update_work_items( monkeypatch: pytest.MonkeyPatch, base_object: BaseObjectContainer ): - polarion_work_item_list: list[serialize.CapellaWorkItem] = [ - serialize.CapellaWorkItem( + polarion_work_item_list: list[capella_work_item.CapellaWorkItem] = [ + capella_work_item.CapellaWorkItem( id="Obj-1", type="type", uuid_capella="uuid1", @@ -601,7 +621,7 @@ def test_update_work_items( base_object.pw.load_polarion_work_item_map() work_items = { - "uuid1": serialize.CapellaWorkItem( + "uuid1": capella_work_item.CapellaWorkItem( id="Obj-1", uuid_capella="uuid1", title="Fake 1", @@ -610,14 +630,12 @@ def test_update_work_items( ) } base_object.c2pcli.capella_model = mock_model = mock.MagicMock() + base_object.pw.model = mock_model mock_model.by_uuid.return_value = base_object.pw.elements[ "FakeModelObject" ][0] base_object.pw.patch_work_items( - base_object.c2pcli.capella_model, - work_items, - {}, - base_object.c2pcli.synchronize_config_roles, + work_items, {}, base_object.c2pcli.synchronize_config_roles ) assert base_object.pw.client is not None assert base_object.pw.client.get_all_work_item_links.call_count == 1 @@ -625,7 +643,7 @@ def test_update_work_items( assert base_object.pw.client.create_work_item_links.call_count == 0 assert base_object.pw.client.update_work_item.call_count == 1 work_item = base_object.pw.client.update_work_item.call_args[0][0] - assert isinstance(work_item, serialize.CapellaWorkItem) + assert isinstance(work_item, capella_work_item.CapellaWorkItem) assert work_item.id == "Obj-1" assert work_item.title == "Fake 1" assert work_item.description_type == "text/html" @@ -638,17 +656,19 @@ def test_update_work_items( def test_update_work_items_filters_work_items_with_same_checksum( base_object: BaseObjectContainer, ): - base_object.pw.polarion_work_item_map[ - "uuid1" - ] = serialize.CapellaWorkItem( - id="Obj-1", - uuid_capella="uuid1", - status="open", - checksum=TEST_WI_CHECKSUM, - type="fakeModelObject", + base_object.pw.polarion_data_repo.update_work_items( + [ + capella_work_item.CapellaWorkItem( + id="Obj-1", + uuid_capella="uuid1", + status="open", + checksum=TEST_WI_CHECKSUM, + type="fakeModelObject", + ) + ] ) work_items = { - "uuid1": serialize.CapellaWorkItem( + "uuid1": capella_work_item.CapellaWorkItem( id="Obj-1", uuid_capella="uuid1", status="open", @@ -660,11 +680,10 @@ def test_update_work_items_filters_work_items_with_same_checksum( "uuid1", name="Fake 1" ) + base_object.pw.model = mock_model + base_object.pw.patch_work_items( - mock_model, - work_items, - {}, - base_object.c2pcli.synchronize_config_roles, + work_items, {}, base_object.c2pcli.synchronize_config_roles ) assert base_object.pw.client is not None @@ -672,14 +691,12 @@ def test_update_work_items_filters_work_items_with_same_checksum( @staticmethod def test_update_links_with_no_elements(base_object: BaseObjectContainer): - base_object.pw.polarion_work_item_map = {} - base_object.pw.polarion_id_map = {} - work_items: dict[str, serialize.CapellaWorkItem] = {} + base_object.pw.polarion_data_repo = ( + polarion_repo.PolarionDataRepository() + ) + work_items: dict[str, capella_work_item.CapellaWorkItem] = {} base_object.pw.patch_work_items( - base_object.c2pcli.capella_model, - work_items, - {}, - base_object.c2pcli.synchronize_config_roles, + work_items, {}, base_object.c2pcli.synchronize_config_roles ) assert base_object.pw.client.get_all_work_item_links.call_count == 0 @@ -689,25 +706,26 @@ def test_update_links(base_object: BaseObjectContainer): link = polarion_api.WorkItemLink( "Obj-1", "Obj-2", "attribute", True, "project_id" ) - base_object.pw.polarion_work_item_map["uuid1"].linked_work_items = [ - link - ] - base_object.pw.polarion_work_item_map[ - "uuid2" - ] = serialize.CapellaWorkItem( - id="Obj-2", - uuid_capella="uuid2", - status="open", - type="fakeModelObject", + _, work_item = base_object.pw.polarion_data_repo["uuid1"] + work_item.linked_work_items = [link] + base_object.pw.polarion_data_repo.update_work_items( + [ + capella_work_item.CapellaWorkItem( + id="Obj-2", + uuid_capella="uuid2", + status="open", + type="fakeModelObject", + ) + ] ) work_items = { - "uuid1": serialize.CapellaWorkItem( + "uuid1": capella_work_item.CapellaWorkItem( id="Obj-1", uuid_capella="uuid1", status="open", type="fakeModelObject", ), - "uuid2": serialize.CapellaWorkItem( + "uuid2": capella_work_item.CapellaWorkItem( id="Obj-2", uuid_capella="uuid2", status="open", @@ -726,11 +744,9 @@ def test_update_links(base_object: BaseObjectContainer): expected_new_link = polarion_api.WorkItemLink( "Obj-2", "Obj-1", "attribute", None, "project_id" ) + base_object.pw.model = mock_model base_object.pw.patch_work_items( - base_object.c2pcli.capella_model, - work_items, - {}, - base_object.c2pcli.synchronize_config_roles, + work_items, {}, base_object.c2pcli.synchronize_config_roles ) assert base_object.pw.client is not None links = base_object.pw.client.get_all_work_item_links.call_args_list @@ -750,22 +766,30 @@ def test_update_links(base_object: BaseObjectContainer): def test_patch_work_item_grouped_links( monkeypatch: pytest.MonkeyPatch, base_object: BaseObjectContainer, - dummy_work_items: dict[str, serialize.CapellaWorkItem], + dummy_work_items: dict[str, capella_work_item.CapellaWorkItem], ): work_items = dummy_work_items - base_object.pw.polarion_work_item_map = { - "uuid0": serialize.CapellaWorkItem( - id="Obj-0", uuid_capella="uuid0", status="open" - ), - "uuid1": serialize.CapellaWorkItem( - id="Obj-1", uuid_capella="uuid1", status="open" - ), - "uuid2": serialize.CapellaWorkItem( - id="Obj-2", uuid_capella="uuid2", status="open" - ), - } + base_object.pw.polarion_data_repo = ( + polarion_repo.PolarionDataRepository( + [ + capella_work_item.CapellaWorkItem( + id="Obj-0", uuid_capella="uuid0", status="open" + ), + capella_work_item.CapellaWorkItem( + id="Obj-1", uuid_capella="uuid1", status="open" + ), + capella_work_item.CapellaWorkItem( + id="Obj-2", uuid_capella="uuid2", status="open" + ), + ] + ) + ) mock_create_links = mock.MagicMock() - monkeypatch.setattr(element, "create_links", mock_create_links) + monkeypatch.setattr( + link_converter.LinkSerializer, + "create_links_for_work_item", + mock_create_links, + ) mock_create_links.side_effect = lambda obj, *args: dummy_work_items[ obj.uuid ].linked_work_items @@ -775,12 +799,12 @@ def mock_back_link(work_item, back_links): mock_grouped_links = mock.MagicMock() monkeypatch.setattr( - element, "create_grouped_link_fields", mock_grouped_links + link_converter, "create_grouped_link_fields", mock_grouped_links ) mock_grouped_links.side_effect = mock_back_link mock_grouped_links_reverse = mock.MagicMock() monkeypatch.setattr( - element, + link_converter, "create_grouped_back_link_fields", mock_grouped_links_reverse, ) @@ -788,11 +812,9 @@ def mock_back_link(work_item, back_links): mock_model.by_uuid.side_effect = [ FakeModelObject(f"uuid{i}", name=f"Fake {i}") for i in range(3) ] + base_object.pw.model = mock_model base_object.pw.patch_work_items( - base_object.c2pcli.capella_model, - work_items, - {}, - base_object.c2pcli.synchronize_config_roles, + work_items, {}, base_object.c2pcli.synchronize_config_roles ) assert base_object.pw.client is not None update_work_item_calls = ( @@ -814,10 +836,10 @@ def mock_back_link(work_item, back_links): @staticmethod def test_maintain_grouped_links_attributes( - dummy_work_items: dict[str, serialize.CapellaWorkItem] + dummy_work_items: dict[str, capella_work_item.CapellaWorkItem] ): for work_item in dummy_work_items.values(): - element.create_grouped_link_fields(work_item) + link_converter.create_grouped_link_fields(work_item) del dummy_work_items["uuid0"].additional_attributes["uuid_capella"] del dummy_work_items["uuid1"].additional_attributes["uuid_capella"] del dummy_work_items["uuid2"].additional_attributes["uuid_capella"] @@ -839,15 +861,15 @@ def test_maintain_grouped_links_attributes( @staticmethod def test_maintain_reverse_grouped_links_attributes( - dummy_work_items: dict[str, serialize.CapellaWorkItem] + dummy_work_items: dict[str, capella_work_item.CapellaWorkItem] ): reverse_polarion_id_map = {v: k for k, v in POLARION_ID_MAP.items()} back_links: dict[str, list[polarion_api.WorkItemLink]] = {} for work_item in dummy_work_items.values(): - element.create_grouped_link_fields(work_item, back_links) + link_converter.create_grouped_link_fields(work_item, back_links) for work_item_id, links in back_links.items(): work_item = dummy_work_items[reverse_polarion_id_map[work_item_id]] - element.create_grouped_back_link_fields(work_item, links) + link_converter.create_grouped_back_link_fields(work_item, links) del dummy_work_items["uuid0"].additional_attributes["uuid_capella"] del dummy_work_items["uuid1"].additional_attributes["uuid_capella"] del dummy_work_items["uuid2"].additional_attributes["uuid_capella"] @@ -877,12 +899,12 @@ def test_maintain_reverse_grouped_links_attributes( def test_grouped_linked_work_items_order_consistency(): - work_item = serialize.CapellaWorkItem("id", "Dummy") + work_item = capella_work_item.CapellaWorkItem("id", "Dummy") links = [ polarion_api.WorkItemLink("prim1", "id", "role1"), polarion_api.WorkItemLink("prim2", "id", "role1"), ] - element.create_grouped_back_link_fields(work_item, links) + link_converter.create_grouped_back_link_fields(work_item, links) check_sum = work_item.calculate_checksum() @@ -890,7 +912,7 @@ def test_grouped_linked_work_items_order_consistency(): polarion_api.WorkItemLink("prim2", "id", "role1"), polarion_api.WorkItemLink("prim1", "id", "role1"), ] - element.create_grouped_back_link_fields(work_item, links) + link_converter.create_grouped_back_link_fields(work_item, links) assert check_sum == work_item.calculate_checksum() @@ -900,7 +922,7 @@ class TestHelpers: def test_resolve_element_type(): xtype = "LogicalComponent" - type = serialize.resolve_element_type(xtype) + type = element_converter.resolve_element_type(xtype) assert type == "logicalComponent" @@ -910,15 +932,19 @@ class TestSerializers: def test_diagram(model: capellambse.MelodyModel): diag = model.diagrams.by_uuid(TEST_DIAG_UUID) - serializer = serialize.CapellaWorkItemSerializer( - TEST_DIAGRAM_CACHE, {}, model, {}, {} + serializer = element_converter.CapellaWorkItemSerializer( + TEST_DIAGRAM_CACHE, + {}, + model, + polarion_repo.PolarionDataRepository(), + {}, ) serialized_diagram = serializer.serialize(diag) if serialized_diagram is not None: serialized_diagram.description = None - assert serialized_diagram == serialize.CapellaWorkItem( + assert serialized_diagram == capella_work_item.CapellaWorkItem( type="diagram", uuid_capella=TEST_DIAG_UUID, title="[CC] Capability", @@ -931,7 +957,7 @@ def test_diagram(model: capellambse.MelodyModel): def test__decode_diagram(): diagram_path = TEST_DIAGRAM_CACHE / "_APMboAPhEeynfbzU12yy7w.svg" - diagram = serialize._decode_diagram(diagram_path) + diagram = element_converter._decode_diagram(diagram_path) assert diagram.startswith("data:image/svg+xml;base64,") @@ -1068,11 +1094,17 @@ def test_generic_work_item( ): obj = model.by_uuid(uuid) - serializer = serialize.CapellaWorkItemSerializer( + serializer = element_converter.CapellaWorkItemSerializer( pathlib.Path(""), TEST_POL_TYPE_MAP, model, - TEST_POL_ID_MAP, + polarion_repo.PolarionDataRepository( + [ + capella_work_item.CapellaWorkItem( + id="TEST", uuid_capella=TEST_E_UUID + ) + ] + ), {}, ) @@ -1081,5 +1113,5 @@ def test_generic_work_item( status = work_item.status work_item.status = None - assert work_item == serialize.CapellaWorkItem(**expected) + assert work_item == capella_work_item.CapellaWorkItem(**expected) assert status == "open"