diff --git a/capella2polarion/__main__.py b/capella2polarion/__main__.py index ccbb3dcd..3cc2f83c 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__) @@ -76,8 +77,6 @@ def cli( ) capella2polarion_cli.setup_logger() ctx.obj = capella2polarion_cli - capella2polarion_cli.echo = click.echo - capella2polarion_cli.echo("Start") @cli.command() @@ -106,23 +105,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, + converter = 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, + ) + + converter.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() + + converter.generate_work_items(polarion_worker.polarion_data_repo) + + polarion_worker.delete_work_items(converter.converter_session) + polarion_worker.post_work_items(converter.converter_session) # Create missing links for new work items - polarion_worker.create_work_items() - polarion_worker.patch_work_items() + converter.generate_work_items(polarion_worker.polarion_data_repo, True) + + polarion_worker.patch_work_items(converter.converter_session) if __name__ == "__main__": diff --git a/capella2polarion/cli.py b/capella2polarion/cli.py index 6304080d..b667776d 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__) @@ -54,7 +54,6 @@ def __init__( self.synchronize_config_io: typing.TextIO = synchronize_config_io self.synchronize_config_content: dict[str, typing.Any] = {} self.synchronize_config_roles: dict[str, list[str]] | None = None - self.echo = click.echo self.config = converter_config.ConverterConfig() def _none_save_value_string(self, value: str | None) -> str | None: @@ -69,7 +68,7 @@ def _type(value): def _value(value): return value - self.echo("---------------------------------------") + click.echo("---------------------------------------") lighted_member_vars = [ attribute for attribute in dir(self) @@ -96,14 +95,14 @@ def _value(value): else: string_value = _type(member_value) string_value = self._none_save_value_string(string_value) - self.echo(f"{lighted_member_var}: '{string_value}'") + click.echo(f"{lighted_member_var}: '{string_value}'") echo = ("NO", "YES")[ self.capella_diagram_cache_index_file_path.is_file() ] - self.echo(f"""Capella Diagram Cache Index-File exists: {echo}""") + click.echo(f"""Capella Diagram Cache Index-File exists: {echo}""") echo = ("YES", "NO")[self.synchronize_config_io.closed] - self.echo(f"""Synchronize Config-IO is open: {echo}""") + click.echo(f"""Synchronize Config-IO is open: {echo}""") def setup_logger(self) -> None: """Set the logger in the right mood.""" diff --git a/capella2polarion/worker.py b/capella2polarion/connectors/polarion_worker.py similarity index 58% rename from capella2polarion/worker.py rename to capella2polarion/connectors/polarion_worker.py index 45a74d92..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,32 +34,25 @@ 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 ): raise ValueError( f"""ProjectId invalid. Value - '{self._save_value_string(self.polarion_params.project_id)}'""" + '{self.polarion_params.project_id}'""" ) result_url = parse.urlparse(self.polarion_params.url) if not all([result_url.scheme, result_url.netloc]): raise ValueError( f"""Polarion URL parameter is not a valid url. - Value {self._save_value_string(self.polarion_params.url)}""" + Value {self.polarion_params.url}""" ) if self.polarion_params.private_access_token is None: raise ValueError( @@ -84,64 +69,14 @@ def __init__( ) self.check_client() - def _save_value_string(self, value: str | None) -> str | None: - return "None" if value is None else value - def check_client(self) -> None: """Instantiate the polarion client as member.""" if not self.client.project_exists(): raise KeyError( f"Miss Polarion project with id " - f"{self._save_value_string(self.polarion_params.project_id)}" + 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) @@ -153,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 @@ -192,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: @@ -204,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( @@ -231,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: @@ -322,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/converter_config.py b/capella2polarion/converters/converter_config.py index c1c14286..5ca6bbe3 100644 --- a/capella2polarion/converters/converter_config.py +++ b/capella2polarion/converters/converter_config.py @@ -19,7 +19,7 @@ class CapellaTypeConfig: p_type: str | None = None converter: str | None = None links: list[str] = dataclasses.field(default_factory=list) - actor: bool | None = None + is_actor: bool | None = None nature: str | None = None @@ -44,7 +44,7 @@ def read_config_file(self, synchronize_config: t.TextIO): global_config_dict = config_dict.pop("*", {}) all_type_config = global_config_dict.pop("*", {}) global_links = all_type_config.get("links", []) - self.set_global_links(global_links) + self.__global_config.links = global_links if "Diagram" in global_config_dict: diagram_config = global_config_dict.pop("Diagram") or {} @@ -78,7 +78,7 @@ def set_layer_config( self.get_type_config( layer, c_type, - actor=type_config.get("actor", _C2P_DEFAULT), + actor=type_config.get("is_actor", _C2P_DEFAULT), nature=type_config.get("nature", _C2P_DEFAULT), ) or self.__global_config @@ -94,7 +94,7 @@ def set_layer_config( p_type, type_config.get("serializer") or closest_config.converter, type_config.get("links", []) + closest_config.links, - type_config.get("actor", _C2P_DEFAULT), + type_config.get("is_actor", _C2P_DEFAULT), type_config.get("nature", _C2P_DEFAULT), ) ) @@ -109,7 +109,7 @@ def set_global_config(self, c_type: str, type_config: dict[str, t.Any]): p_type, type_config.get("serializer"), type_config.get("links", []) + self.__global_config.links, - type_config.get("actor", _C2P_DEFAULT), + type_config.get("is_actor", _C2P_DEFAULT), type_config.get("nature", _C2P_DEFAULT), ) @@ -123,14 +123,6 @@ def set_diagram_config(self, diagram_config: dict[str, t.Any]): diagram_config.get("links", []) + self.__global_config.links, ) - def set_global_links(self, links: list[str]): - """Set links of the global config object. - - Must be set before adding additional configs to enable - inheritance! - """ - self.__global_config.links = links - def get_type_config( self, layer: str, c_type: str, **attributes: t.Any ) -> CapellaTypeConfig | None: @@ -186,9 +178,9 @@ def _read_capella_type_configs( return [conf] # We want to have the most generic config first followed by those - # having actor set to None + # having is_actor set to None return sorted( conf, - key=lambda c: int(c.get("actor", _C2P_DEFAULT) != _C2P_DEFAULT) + key=lambda c: int(c.get("is_actor", _C2P_DEFAULT) != _C2P_DEFAULT) + 2 * int(c.get("nature", _C2P_DEFAULT) != _C2P_DEFAULT), ) 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..773a8e80 --- /dev/null +++ b/capella2polarion/converters/model_converter.py @@ -0,0 +1,153 @@ +# 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 = {} + + 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 = { + "is_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]: + """Return a work items mapping from model elements for Polarion. + + The dictionary maps Capella UUIDs to ``CapellaWorkItem``s. 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/__init__.py b/tests/__init__.py deleted file mode 100644 index dd5d085d..00000000 --- a/tests/__init__.py +++ /dev/null @@ -1,2 +0,0 @@ -# Copyright DB InfraGO AG and contributors -# SPDX-License-Identifier: Apache-2.0 diff --git a/tests/conftest.py b/tests/conftest.py index 730b55de..cbf1ea89 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -17,7 +17,7 @@ TEST_DATA_ROOT = pathlib.Path(__file__).parent / "data" TEST_DIAGRAM_CACHE = TEST_DATA_ROOT / "diagram_cache" TEST_MODEL_ELEMENTS = TEST_DATA_ROOT / "model_elements" -TEST_MODEL_ELEMENTS_CONFIG = TEST_MODEL_ELEMENTS / "new_config.yaml" +TEST_MODEL_ELEMENTS_CONFIG = TEST_MODEL_ELEMENTS / "config.yaml" TEST_MODEL = TEST_DATA_ROOT / "model" / "Melody Model Test.aird" TEST_HOST = "https://api.example.com" diff --git a/tests/data/model_elements/config.yaml b/tests/data/model_elements/config.yaml index 84d40934..58f7fb20 100644 --- a/tests/data/model_elements/config.yaml +++ b/tests/data/model_elements/config.yaml @@ -21,7 +21,7 @@ Entity: StateMachine: -oa: +oa: # Specify below FunctionalExchange: polarion_type: operationalInteraction links: diff --git a/tests/data/model_elements/new_config.yaml b/tests/data/model_elements/new_config.yaml deleted file mode 100644 index 08cf05e7..00000000 --- a/tests/data/model_elements/new_config.yaml +++ /dev/null @@ -1,52 +0,0 @@ -# Copyright DB InfraGO AG and contributors -# SPDX-License-Identifier: Apache-2.0 - -"*": # All layers - "*": # All class types - links: - - parent - - description_reference - Class: - links: - - state_machines - Diagram: - links: - - diagram_elements - Constraint: - serializer: linked_text_as_description - Scenario: - serializer: include_pre_and_post_condition - CapabilityRealization: - serializer: include_pre_and_post_condition - Entity: - -oa: # Specify below - FunctionalExchange: - polarion_type: operationalInteraction - links: - - exchange_items - OperationalCapability: - serializer: include_pre_and_post_condition -pa: - PhysicalComponent: - - actor: false - nature: UNSET - polarion_type: physicalComponent - - actor: false - nature: NODE - polarion_type: physicalComponentNode - - actor: false - nature: BEHAVIOR - polarion_type: physicalComponentBehavior - - actor: true - nature: NODE - polarion_type: physicalActorNode - - actor: true - nature: BEHAVIOR - polarion_type: physicalActorBehavior -la: - LogicalComponent: - - actor: true - polarion_type: logicalActor - - actor: false - polarion_type: logicalComponent diff --git a/tests/test_cli.py b/tests/test_cli.py index 5a4985ea..5fbadd92 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -10,10 +10,10 @@ 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] +from .conftest import ( # type: ignore[import] TEST_DIAGRAM_CACHE, TEST_MODEL, TEST_MODEL_ELEMENTS_CONFIG, diff --git a/tests/test_elements.py b/tests/test_elements.py index 75b02c14..69e9cb88 100644 --- a/tests/test_elements.py +++ b/tests/test_elements.py @@ -17,19 +17,20 @@ 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 -from tests.conftest import ( # type: ignore[import] +from .conftest import ( # type: ignore[import] TEST_DIAGRAM_CACHE, TEST_HOST, + TEST_MODEL_ELEMENTS_CONFIG, ) # pylint: disable=redefined-outer-name @@ -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 @@ -1122,14 +1147,12 @@ def test_generic_work_item( ): obj = model.by_uuid(uuid) config = converter_config.ConverterConfig() - with open( - conftest.TEST_MODEL_ELEMENTS_CONFIG, "r", encoding="utf8" - ) as f: + with open(TEST_MODEL_ELEMENTS_CONFIG, "r", encoding="utf8") as f: config.read_config_file(f) c_type = type(obj).__name__ attributes = { - "actor": getattr(obj, "is_actor", None), + "is_actor": getattr(obj, "is_actor", None), "nature": getattr(obj, "nature", None), } type_config = config.get_type_config(layer, c_type, **attributes)