From 4ecc9869e5064f68f70a71cc5b3f899578abdf00 Mon Sep 17 00:00:00 2001 From: Michael Harbarth Date: Thu, 11 Jan 2024 12:30:08 +0100 Subject: [PATCH] refactor: Clear split between polarion and capella specific work introduced --- capella2polarion/__main__.py | 30 ++-- capella2polarion/cli.py | 2 +- .../polarion_worker.py} | 163 ++---------------- .../converters/element_converter.py | 5 + .../converters/model_converter.py | 154 +++++++++++++++++ tests/test_cli.py | 2 +- tests/test_elements.py | 141 ++++++++------- 7 files changed, 281 insertions(+), 216 deletions(-) rename capella2polarion/{worker.py => connectors/polarion_worker.py} (60%) create mode 100644 capella2polarion/converters/model_converter.py diff --git a/capella2polarion/__main__.py b/capella2polarion/__main__.py index ccbb3dcd..c852a1b7 100644 --- a/capella2polarion/__main__.py +++ b/capella2polarion/__main__.py @@ -11,8 +11,9 @@ import click from capellambse import cli_helpers -from capella2polarion import worker as pw from capella2polarion.cli import Capella2PolarionCli +from capella2polarion.connectors import polarion_worker as pw +from capella2polarion.converters import model_converter logger = logging.getLogger(__name__) @@ -106,23 +107,32 @@ def synchronize(ctx: click.core.Context) -> None: capella_to_polarion_cli.capella_diagram_cache_index_content is not None ) - polarion_worker = pw.CapellaPolarionWorker( - capella_to_polarion_cli.polarion_params, + mc = model_converter.ModelConverter( capella_to_polarion_cli.capella_model, + capella_to_polarion_cli.capella_diagram_cache_folder_path, + capella_to_polarion_cli.polarion_params.project_id, + ) + + mc.read_model( capella_to_polarion_cli.config, capella_to_polarion_cli.capella_diagram_cache_index_content, - capella_to_polarion_cli.capella_diagram_cache_folder_path, ) - polarion_worker.generate_converter_session() + + polarion_worker = pw.CapellaPolarionWorker( + capella_to_polarion_cli.polarion_params, capella_to_polarion_cli.config + ) polarion_worker.load_polarion_work_item_map() - polarion_worker.create_work_items() - polarion_worker.delete_work_items() - polarion_worker.post_work_items() + + mc.generate_work_items(polarion_worker.polarion_data_repo) + + polarion_worker.delete_work_items(mc.converter_session) + polarion_worker.post_work_items(mc.converter_session) # Create missing links for new work items - polarion_worker.create_work_items() - polarion_worker.patch_work_items() + mc.generate_work_items(polarion_worker.polarion_data_repo, True) + + polarion_worker.patch_work_items(mc.converter_session) if __name__ == "__main__": diff --git a/capella2polarion/cli.py b/capella2polarion/cli.py index 6304080d..18a3f947 100644 --- a/capella2polarion/cli.py +++ b/capella2polarion/cli.py @@ -11,7 +11,7 @@ import capellambse import click -from capella2polarion import worker as pw +from capella2polarion.connectors import polarion_worker as pw from capella2polarion.converters import converter_config logger = logging.getLogger(__name__) diff --git a/capella2polarion/worker.py b/capella2polarion/connectors/polarion_worker.py similarity index 60% rename from capella2polarion/worker.py rename to capella2polarion/connectors/polarion_worker.py index cc0f4087..bbf244f6 100644 --- a/capella2polarion/worker.py +++ b/capella2polarion/connectors/polarion_worker.py @@ -5,21 +5,13 @@ import collections.abc as cabc import logging -import pathlib -import typing as t from urllib import parse -import capellambse import polarion_rest_api_client as polarion_api from capella2polarion import data_models from capella2polarion.connectors import polarion_repo -from capella2polarion.converters import ( - converter_config, - data_session, - element_converter, - link_converter, -) +from capella2polarion.converters import converter_config, data_session logger = logging.getLogger(__name__) @@ -42,18 +34,11 @@ class CapellaPolarionWorker: def __init__( self, params: PolarionWorkerParams, - model: capellambse.MelodyModel, config: converter_config.ConverterConfig, - diagram_idx: list[dict[str, t.Any]], - diagram_cache_path: pathlib.Path, ) -> None: self.polarion_params = params self.polarion_data_repo = polarion_repo.PolarionDataRepository() - self.converter_session: data_session.ConverterSession = {} - self.model = model self.config = config - self.diagram_idx = diagram_idx - self.diagram_cache_path = diagram_cache_path if (self.polarion_params.project_id is None) or ( len(self.polarion_params.project_id) == 0 @@ -92,53 +77,6 @@ def check_client(self) -> None: f"{self.polarion_params.project_id}" ) - def generate_converter_session( - self, - ) -> None: - """Return an elements and UUID to Polarion type map.""" - missing_types: set[tuple[str, str, dict[str, t.Any]]] = set() - for layer, c_type in self.config.layers_and_types(): - below = getattr(self.model, layer) - if c_type == "Diagram": - continue - - objects = self.model.search(c_type, below=below) - for obj in objects: - attributes = { - "actor": getattr(obj, "is_actor", None), - "nature": getattr(obj, "nature", None), - } - if config := self.config.get_type_config( - layer, c_type, **attributes - ): - self.converter_session[ - obj.uuid - ] = data_session.ConverterData(layer, config, obj) - else: - missing_types.add((layer, c_type, attributes)) - - if self.config.diagram_config: - diagrams_from_cache = { - d["uuid"] for d in self.diagram_idx if d["success"] - } - for d in self.model.diagrams: - if d.uuid in diagrams_from_cache: - self.converter_session[ - d.uuid - ] = data_session.ConverterData( - "", self.config.diagram_config, d - ) - - if missing_types: - for missing_type in missing_types: - layer, c_type, attributes = missing_type - logger.warning( - "Capella type %r is configured in layer %r, but not for %s.", - layer, - c_type, - ", ".join(f"{k!r}={v!r}" for k, v in attributes.items()), - ) - def load_polarion_work_item_map(self): """Return a map from Capella UUIDs to Polarion work items.""" _type = " ".join(self.config.polarion_types) @@ -150,29 +88,9 @@ def load_polarion_work_item_map(self): self.polarion_data_repo.update_work_items(work_items) - def create_work_items( - self, - ) -> dict[str, data_models.CapellaWorkItem]: - """Create a list of work items for Polarion.""" - serializer = element_converter.CapellaWorkItemSerializer( - self.diagram_cache_path, - self.model, - self.polarion_data_repo, - self.converter_session, - ) - work_items = serializer.serialize_all() - 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_data_repo.get_work_item_by_capella_uuid( - work_item.uuid_capella - ): - work_item.id = old.id - - return {wi.uuid_capella: wi for wi in work_items} - - def delete_work_items(self) -> None: + def delete_work_items( + self, converter_session: data_session.ConverterSession + ) -> None: """Delete work items in a Polarion project. If the delete flag is set to ``False`` in the context work items are @@ -189,7 +107,7 @@ def serialize_for_delete(uuid: str) -> str: for uuid, _, work_item in self.polarion_data_repo.items() if work_item.status != "deleted" } - uuids: set[str] = existing_work_items - set(self.converter_session) + uuids: set[str] = existing_work_items - set(converter_session) work_item_ids = [serialize_for_delete(uuid) for uuid in uuids] if work_item_ids: try: @@ -201,11 +119,11 @@ def serialize_for_delete(uuid: str) -> str: logger.error("Deleting work items failed. %s", error.args[0]) def post_work_items( - self, + self, converter_session: data_session.ConverterSession ) -> None: """Post work items in a Polarion project.""" missing_work_items: list[data_models.CapellaWorkItem] = [] - for uuid, converter_data in self.converter_session.items(): + for uuid, converter_data in converter_session.items(): work_item = converter_data.work_item if work_item is None: logger.warning( @@ -228,24 +146,17 @@ def post_work_items( logger.error("Creating work items failed. %s", error.args[0]) def patch_work_item( - self, - new: data_models.CapellaWorkItem, - old: data_models.CapellaWorkItem, + self, uuid: str, converter_session: data_session.ConverterSession ): - """Patch a given WorkItem. - - Parameters - ---------- - api - The context to execute the patch for. - new - The updated CapellaWorkItem - old - The CapellaWorkItem currently present on polarion - """ + """Patch a given WorkItem.""" + new = converter_session[uuid].work_item + _, old = self.polarion_data_repo[uuid] if new == old: return + assert old is not None + assert new is not None + log_args = (old.id, new.type, new.title) logger.info("Update work item %r for model %s %r...", *log_args) if "uuid_capella" in new.additional_attributes: @@ -319,48 +230,8 @@ def _get_link_id(link: polarion_api.WorkItemLink) -> str: ) def patch_work_items( - self, + self, converter_session: data_session.ConverterSession ) -> None: """Update work items in a Polarion project.""" - - back_links: dict[str, list[polarion_api.WorkItemLink]] = {} - link_serializer = link_converter.LinkSerializer( - self.polarion_data_repo, - self.converter_session, - self.polarion_params.project_id, - self.model, - ) - - for uuid, converter_data in self.converter_session.items(): - if converter_data.work_item is None: - logger.warning( - "Expected to find a WorkItem for %s, but there is none", - uuid, - ) - continue - - links = link_serializer.create_links_for_work_item(uuid) - converter_data.work_item.linked_work_items = links - - link_converter.create_grouped_link_fields( - converter_data.work_item, back_links - ) - - for uuid, converter_data in self.converter_session.items(): - if converter_data.work_item is None: - logger.warning( - "Expected to find a WorkItem for %s, but there is none", - uuid, - ) - continue - - _, old_work_item = self.polarion_data_repo[uuid] - if old_work_item.id in back_links: - link_converter.create_grouped_back_link_fields( - converter_data.work_item, back_links[old_work_item.id] - ) - - self.patch_work_item( - converter_data.work_item, - old_work_item, - ) + for uuid in converter_session: + self.patch_work_item(uuid, converter_session) diff --git a/capella2polarion/converters/element_converter.py b/capella2polarion/converters/element_converter.py index 327cac43..309c34d8 100644 --- a/capella2polarion/converters/element_converter.py +++ b/capella2polarion/converters/element_converter.py @@ -157,6 +157,11 @@ def serialize( self._generic_work_item, ) converter_data.work_item = serializer(converter_data) + if old := self.capella_polarion_mapping.get_work_item_by_capella_uuid( + converter_data.work_item.uuid_capella + ): + converter_data.work_item.id = old.id + return converter_data.work_item except Exception as error: logger.error("Serializing model element failed. %s", error.args[0]) diff --git a/capella2polarion/converters/model_converter.py b/capella2polarion/converters/model_converter.py new file mode 100644 index 00000000..249579ad --- /dev/null +++ b/capella2polarion/converters/model_converter.py @@ -0,0 +1,154 @@ +# Copyright DB InfraGO AG and contributors +# SPDX-License-Identifier: Apache-2.0 +"""A module containing the overall model conversion class.""" + +from __future__ import annotations + +import logging +import pathlib +import typing as t + +import capellambse +import polarion_rest_api_client as polarion_api + +from capella2polarion import data_models +from capella2polarion.connectors import polarion_repo +from capella2polarion.converters import ( + converter_config, + data_session, + element_converter, + link_converter, +) + +logger = logging.getLogger(__name__) + + +class ModelConverter: + """Class to convert elements of a model and store related data.""" + + def __init__( + self, + model: capellambse.MelodyModel, + diagram_cache_path: pathlib.Path, + project_id: str, + ): + self.model = model + self.diagram_cache_path = diagram_cache_path + self.project_id = project_id + self.converter_session: data_session.ConverterSession = ( + data_session.ConverterSession() + ) + + def read_model( + self, + config: converter_config.ConverterConfig, + diagram_idx: list[dict[str, t.Any]], + ): + """Read the model using a given config and diagram_idx.""" + missing_types: set[tuple[str, str, dict[str, t.Any]]] = set() + for layer, c_type in config.layers_and_types(): + below = getattr(self.model, layer) + if c_type == "Diagram": + continue + + objects = self.model.search(c_type, below=below) + for obj in objects: + attributes = { + "actor": getattr(obj, "is_actor", None), + "nature": getattr(obj, "nature", None), + } + if type_config := config.get_type_config( + layer, c_type, **attributes + ): + self.converter_session[ + obj.uuid + ] = data_session.ConverterData(layer, type_config, obj) + else: + missing_types.add((layer, c_type, attributes)) + + if config.diagram_config: + diagrams_from_cache = { + d["uuid"] for d in diagram_idx if d["success"] + } + for d in self.model.diagrams: + if d.uuid in diagrams_from_cache: + self.converter_session[ + d.uuid + ] = data_session.ConverterData( + "", config.diagram_config, d + ) + + if missing_types: + for missing_type in missing_types: + layer, c_type, attributes = missing_type + logger.warning( + "Capella type %r is configured in layer %r, but not for %s.", + layer, + c_type, + ", ".join(f"{k!r}={v!r}" for k, v in attributes.items()), + ) + + def generate_work_items( + self, + polarion_data_repo: polarion_repo.PolarionDataRepository, + generate_links: bool = False, + ) -> dict[str, data_models.CapellaWorkItem]: + """Generate a list of work items from known elements for Polarion. + + In addition, it is ensured that neither title nor type are None, + Links are not created in this step by default. + """ + serializer = element_converter.CapellaWorkItemSerializer( + self.diagram_cache_path, + self.model, + polarion_data_repo, + self.converter_session, + ) + work_items = serializer.serialize_all() + for work_item in work_items: + assert work_item.title is not None + assert work_item.type is not None + + if generate_links: + self.generate_work_item_links(polarion_data_repo) + + return {wi.uuid_capella: wi for wi in work_items} + + def generate_work_item_links( + self, polarion_data_repo: polarion_repo.PolarionDataRepository + ): + """Generate links for all work items and add custom fields for them.""" + back_links: dict[str, list[polarion_api.WorkItemLink]] = {} + link_serializer = link_converter.LinkSerializer( + polarion_data_repo, + self.converter_session, + self.project_id, + self.model, + ) + for uuid, converter_data in self.converter_session.items(): + if converter_data.work_item is None: + logger.warning( + "Expected to find a WorkItem for %s, but there is none", + uuid, + ) + continue + + links = link_serializer.create_links_for_work_item(uuid) + converter_data.work_item.linked_work_items = links + + link_converter.create_grouped_link_fields( + converter_data.work_item, back_links + ) + + for uuid, converter_data in self.converter_session.items(): + if converter_data.work_item is None: + logger.warning( + "Expected to find a WorkItem for %s, but there is none", + uuid, + ) + continue + + if local_back_links := back_links.get(converter_data.work_item.id): + link_converter.create_grouped_back_link_fields( + converter_data.work_item, local_back_links + ) diff --git a/tests/test_cli.py b/tests/test_cli.py index 5a4985ea..2e853713 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -10,7 +10,7 @@ from click import testing import capella2polarion.__main__ as main -from capella2polarion.worker import CapellaPolarionWorker +from capella2polarion.connectors.polarion_worker import CapellaPolarionWorker # pylint: disable-next=relative-beyond-top-level, useless-suppression from tests.conftest import ( # type: ignore[import] diff --git a/tests/test_elements.py b/tests/test_elements.py index 75b02c14..c9348515 100644 --- a/tests/test_elements.py +++ b/tests/test_elements.py @@ -17,13 +17,14 @@ from capella2polarion import data_models from capella2polarion.cli import Capella2PolarionCli from capella2polarion.connectors import polarion_repo +from capella2polarion.connectors.polarion_worker import CapellaPolarionWorker from capella2polarion.converters import ( converter_config, data_session, element_converter, link_converter, + model_converter, ) -from capella2polarion.worker import CapellaPolarionWorker from tests import conftest # pylint: disable-next=relative-beyond-top-level, useless-suppression @@ -133,9 +134,11 @@ def __init__( self, cli: Capella2PolarionCli, pw: CapellaPolarionWorker, + mc: model_converter.ModelConverter, ) -> None: self.c2pcli: Capella2PolarionCli = cli self.pw: CapellaPolarionWorker = pw + self.mc = mc class TestDiagramElements: @@ -174,28 +177,33 @@ def write(self, text: str): polarion_api, "OpenAPIPolarionProjectClient", mock_api ) c2p_cli.config = mock.Mock(converter_config.ConverterConfig) - pw = CapellaPolarionWorker( - c2p_cli.polarion_params, + + mc = model_converter.ModelConverter( model, - c2p_cli.config, - c2p_cli.capella_diagram_cache_index_content, c2p_cli.capella_diagram_cache_folder_path, + c2p_cli.polarion_params.project_id, ) - pw.converter_session = { + + mc.converter_session = { TEST_DIAG_UUID: data_session.ConverterData( "", DIAGRAM_CONFIG, model.diagrams.by_uuid(TEST_DIAG_UUID) ) } + + pw = CapellaPolarionWorker(c2p_cli.polarion_params, c2p_cli.config) + pw.polarion_data_repo = polarion_repo.PolarionDataRepository( [work_item] ) - return BaseObjectContainer(c2p_cli, pw) + return BaseObjectContainer(c2p_cli, pw, mc) @staticmethod def test_create_diagrams(base_object: BaseObjectContainer): pw = base_object.pw new_work_items: dict[str, data_models.CapellaWorkItem] - new_work_items = pw.create_work_items() + new_work_items = base_object.mc.generate_work_items( + pw.polarion_data_repo + ) assert len(new_work_items) == 1 work_item = new_work_items[TEST_DIAG_UUID] assert isinstance(work_item, data_models.CapellaWorkItem) @@ -211,20 +219,20 @@ def test_create_diagrams_filters_non_diagram_elements( ): # This test does not make any sense, but it also didn't before pw = base_object.pw - pw.create_work_items() - assert pw.client.create_work_items.call_count == 0 + base_object.mc.generate_work_items(pw.polarion_data_repo) + assert pw.client.generate_work_items.call_count == 0 @staticmethod def test_delete_diagrams(base_object: BaseObjectContainer): pw = base_object.pw - pw.converter_session = {} - pw.create_work_items() - pw.post_work_items() - pw.delete_work_items() + base_object.mc.converter_session = {} + base_object.mc.generate_work_items(pw.polarion_data_repo) + pw.post_work_items(base_object.mc.converter_session) + pw.delete_work_items(base_object.mc.converter_session) assert pw.client is not None assert pw.client.delete_work_items.call_count == 1 assert pw.client.delete_work_items.call_args[0][0] == ["Diag-1"] - assert pw.client.create_work_items.call_count == 0 + assert pw.client.generate_work_items.call_count == 0 class FakeModelObject: @@ -288,21 +296,19 @@ def write(self, text: str): polarion_api, "OpenAPIPolarionProjectClient", mock_api ) c2p_cli.config = mock.Mock(converter_config.ConverterConfig) - pw = CapellaPolarionWorker( - c2p_cli.polarion_params, - model, - c2p_cli.config, - c2p_cli.capella_diagram_cache_index_content, - c2p_cli.capella_diagram_cache_folder_path, - ) - pw.polarion_data_repo = polarion_repo.PolarionDataRepository( - [work_item] - ) + fake = FakeModelObject("uuid1", name="Fake 1") fake_model_type_config = converter_config.CapellaTypeConfig( "fakeModelObject", links=["attribute"] ) - pw.converter_session = { + + mc = model_converter.ModelConverter( + model, + c2p_cli.capella_diagram_cache_folder_path, + c2p_cli.polarion_params.project_id, + ) + + mc.converter_session = { "uuid1": data_session.ConverterData( "oa", fake_model_type_config, @@ -321,7 +327,12 @@ def write(self, text: str): FakeModelObject("uuid2", name="Fake 2", attribute=fake), ), } - return BaseObjectContainer(c2p_cli, pw) + + pw = CapellaPolarionWorker(c2p_cli.polarion_params, c2p_cli.config) + pw.polarion_data_repo = polarion_repo.PolarionDataRepository( + [work_item] + ) + return BaseObjectContainer(c2p_cli, pw, mc) @staticmethod def test_create_work_items( @@ -349,7 +360,9 @@ def test_create_work_items( description=markupsafe.Markup(""), ), ] - work_items = base_object.pw.create_work_items() + work_items = base_object.mc.generate_work_items( + base_object.pw.polarion_data_repo + ) assert list(work_items.values()) == [expected, expected1] @staticmethod @@ -371,7 +384,7 @@ def test_create_work_items_with_special_polarion_type( _type: str, attrs: dict[str, t.Any], ): - base_object.pw.converter_session = { + base_object.mc.converter_session = { uuid: data_session.ConverterData( "oa", converter_config.CapellaTypeConfig( @@ -391,7 +404,9 @@ def test_create_work_items_with_special_polarion_type( **attrs, ) - work_items = base_object.pw.create_work_items() + work_items = base_object.mc.generate_work_items( + base_object.pw.polarion_data_repo + ) assert len(work_items) == 1 assert work_items[uuid] == expected @@ -407,11 +422,11 @@ def test_create_links_custom_resolver(base_object: BaseObjectContainer): status="open", ) base_object.pw.polarion_data_repo.update_work_items([work_item_obj_2]) - base_object.pw.converter_session["uuid2"].work_item = work_item_obj_2 - base_object.pw.converter_session["uuid2"].type_config.links = [ + base_object.mc.converter_session["uuid2"].work_item = work_item_obj_2 + base_object.mc.converter_session["uuid2"].type_config.links = [ "description_reference" ] - base_object.pw.converter_session["uuid2"].description_references = [ + base_object.mc.converter_session["uuid2"].description_references = [ "uuid1" ] expected = polarion_api.WorkItemLink( @@ -422,7 +437,7 @@ def test_create_links_custom_resolver(base_object: BaseObjectContainer): ) link_serializer = link_converter.LinkSerializer( base_object.pw.polarion_data_repo, - base_object.pw.converter_session, + base_object.mc.converter_session, base_object.pw.polarion_params.project_id, base_object.c2pcli.capella_model, ) @@ -461,7 +476,7 @@ def test_create_links_custom_exchanges_resolver( base_object.pw.polarion_data_repo.update_work_items( [work_item_obj_1, work_item_obj_2] ) - base_object.pw.converter_session[ + base_object.mc.converter_session[ function_uuid ] = data_session.ConverterData( "fa", @@ -471,7 +486,7 @@ def test_create_links_custom_exchanges_resolver( funtion_obj, work_item_obj_1, ) - base_object.pw.converter_session[uuid] = data_session.ConverterData( + base_object.mc.converter_session[uuid] = data_session.ConverterData( "fa", converter_config.CapellaTypeConfig( "functionalExchange", @@ -491,7 +506,7 @@ def test_create_links_custom_exchanges_resolver( ) link_serializer = link_converter.LinkSerializer( base_object.pw.polarion_data_repo, - base_object.pw.converter_session, + base_object.mc.converter_session, base_object.pw.polarion_params.project_id, base_object.c2pcli.capella_model, ) @@ -510,7 +525,7 @@ def test_create_links_missing_attribute( with caplog.at_level(logging.DEBUG): link_serializer = link_converter.LinkSerializer( base_object.pw.polarion_data_repo, - base_object.pw.converter_session, + base_object.mc.converter_session, base_object.pw.polarion_params.project_id, base_object.c2pcli.capella_model, ) @@ -548,11 +563,11 @@ def test_create_links_from_ElementList(base_object: BaseObjectContainer): ] base_object.pw.polarion_data_repo.update_work_items(work_items) for work_item in work_items: - base_object.pw.converter_session[ + base_object.mc.converter_session[ work_item.uuid_capella ] = data_session.ConverterData( "", - base_object.pw.converter_session["uuid1"].type_config, + base_object.mc.converter_session["uuid1"].type_config, fake_objects[work_item.uuid_capella], work_item, ) @@ -571,7 +586,7 @@ def test_create_links_from_ElementList(base_object: BaseObjectContainer): ) link_serializer = link_converter.LinkSerializer( base_object.pw.polarion_data_repo, - base_object.pw.converter_session, + base_object.mc.converter_session, base_object.pw.polarion_params.project_id, base_object.c2pcli.capella_model, ) @@ -596,7 +611,7 @@ def test_create_link_from_single_attribute( ) base_object.pw.polarion_data_repo.update_work_items([work_item_2]) - base_object.pw.converter_session["uuid2"].work_item = work_item_2 + base_object.mc.converter_session["uuid2"].work_item = work_item_2 expected = polarion_api.WorkItemLink( "Obj-2", @@ -606,7 +621,7 @@ def test_create_link_from_single_attribute( ) link_serializer = link_converter.LinkSerializer( base_object.pw.polarion_data_repo, - base_object.pw.converter_session, + base_object.mc.converter_session, base_object.pw.polarion_params.project_id, base_object.c2pcli.capella_model, ) @@ -644,7 +659,7 @@ def test_update_work_items( base_object.pw.load_polarion_work_item_map() - base_object.pw.converter_session[ + base_object.mc.converter_session[ "uuid1" ].work_item = data_models.CapellaWorkItem( id="Obj-1", @@ -655,9 +670,9 @@ def test_update_work_items( description=markupsafe.Markup(""), ) - del base_object.pw.converter_session["uuid2"] + del base_object.mc.converter_session["uuid2"] - base_object.pw.patch_work_items() + base_object.pw.patch_work_items(base_object.mc.converter_session) assert base_object.pw.client is not None assert base_object.pw.client.get_all_work_item_links.call_count == 1 assert base_object.pw.client.delete_work_item_links.call_count == 0 @@ -688,7 +703,7 @@ def test_update_work_items_filters_work_items_with_same_checksum( ) ] ) - base_object.pw.converter_session[ + base_object.mc.converter_session[ "uuid1" ].work_item = data_models.CapellaWorkItem( id="Obj-1", @@ -697,9 +712,9 @@ def test_update_work_items_filters_work_items_with_same_checksum( type="fakeModelObject", ) - del base_object.pw.converter_session["uuid2"] + del base_object.mc.converter_session["uuid2"] - base_object.pw.patch_work_items() + base_object.pw.patch_work_items(base_object.mc.converter_session) assert base_object.pw.client is not None assert base_object.pw.client.update_work_item.call_count == 0 @@ -709,8 +724,8 @@ def test_update_links_with_no_elements(base_object: BaseObjectContainer): base_object.pw.polarion_data_repo = ( polarion_repo.PolarionDataRepository() ) - base_object.pw.converter_session = {} - base_object.pw.patch_work_items() + base_object.mc.converter_session = {} + base_object.pw.patch_work_items(base_object.mc.converter_session) assert base_object.pw.client.get_all_work_item_links.call_count == 0 @@ -731,7 +746,7 @@ def test_update_links(base_object: BaseObjectContainer): ) ] ) - base_object.pw.converter_session[ + base_object.mc.converter_session[ "uuid1" ].work_item = data_models.CapellaWorkItem( id="Obj-1", @@ -739,7 +754,7 @@ def test_update_links(base_object: BaseObjectContainer): status="open", type="fakeModelObject", ) - base_object.pw.converter_session[ + base_object.mc.converter_session[ "uuid2" ].work_item = data_models.CapellaWorkItem( id="Obj-2", @@ -756,7 +771,10 @@ def test_update_links(base_object: BaseObjectContainer): expected_new_link = polarion_api.WorkItemLink( "Obj-2", "Obj-1", "attribute", None, "project_id" ) - base_object.pw.patch_work_items() + base_object.mc.generate_work_item_links( + base_object.pw.polarion_data_repo + ) + base_object.pw.patch_work_items(base_object.mc.converter_session) assert base_object.pw.client is not None links = base_object.pw.client.get_all_work_item_links.call_args_list assert base_object.pw.client.get_all_work_item_links.call_count == 2 @@ -777,7 +795,7 @@ def test_patch_work_item_grouped_links( base_object: BaseObjectContainer, dummy_work_items: dict[str, data_models.CapellaWorkItem], ): - base_object.pw.converter_session = { + base_object.mc.converter_session = { work_item.uuid_capella: data_session.ConverterData( "", converter_config.CapellaTypeConfig("fakeModelObject"), @@ -812,7 +830,11 @@ def test_patch_work_item_grouped_links( ].linked_work_items def mock_back_link(work_item, back_links): - back_links[work_item.id] = [] + back_links[work_item.id] = [ + polarion_api.WorkItemLink( + "Obj-0", work_item.id, "attribute", True, "project_id" + ) + ] mock_grouped_links = mock.MagicMock() monkeypatch.setattr( @@ -829,8 +851,11 @@ 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.mc.model = mock_model + base_object.mc.generate_work_item_links( + base_object.pw.polarion_data_repo + ) + base_object.pw.patch_work_items(base_object.mc.converter_session) assert base_object.pw.client is not None update_work_item_calls = ( base_object.pw.client.update_work_item.call_args_list