From 860568739da38cc90126fc6e79957909271ab44a Mon Sep 17 00:00:00 2001 From: ewuerger Date: Tue, 15 Aug 2023 07:24:27 +0200 Subject: [PATCH 01/15] wip: Grouped links attribute --- capella2polarion/__main__.py | 19 +++++- capella2polarion/elements/element.py | 93 ++++++++++++++++++++++++++ capella2polarion/elements/serialize.py | 11 +-- tests/test_elements.py | 8 +++ 4 files changed, 125 insertions(+), 6 deletions(-) diff --git a/capella2polarion/__main__.py b/capella2polarion/__main__.py index 23e0f2e8..ea3dd7fd 100644 --- a/capella2polarion/__main__.py +++ b/capella2polarion/__main__.py @@ -152,7 +152,7 @@ def _sanitize_config( @click.pass_context def cli( ctx: click.core.Context, debug: bool, project_id: str, delete: bool = False -) -> None: +): """Synchronise data from Capella to Polarion. PROJECT_ID is a Polarion project id @@ -244,5 +244,22 @@ def model_elements( elements.make_model_elements_index(ctx.obj) +@cli.command() +@click.argument("types", nargs=-1, type=str) +@click.pass_context +def grouped_links_attributes( + ctx: click.core.Context, types: list[str] +) -> None: + """Synchronise model elements.""" + ctx.obj["TYPES"] = types + ctx.obj["POLARION_WI_MAP"] = get_polarion_wi_map(ctx.obj) + ctx.obj["POLARION_ID_MAP"] = { + uuid: wi.id for uuid, wi in ctx.obj["POLARION_WI_MAP"].items() + } + + elements.element.create_grouped_links_attributes(ctx.obj) + elements.element.create_reverse_grouped_links_attributes(ctx.obj) + + if __name__ == "__main__": cli(obj={}) diff --git a/capella2polarion/elements/element.py b/capella2polarion/elements/element.py index a3db417d..90d35bb1 100644 --- a/capella2polarion/elements/element.py +++ b/capella2polarion/elements/element.py @@ -7,6 +7,7 @@ import functools import logging import typing as t +from collections import defaultdict from itertools import chain import polarion_rest_api_client as polarion_api @@ -203,6 +204,98 @@ def _handle_exchanges( return _create(context, wid, role_id, exchanges, links) +def create_grouped_links_attributes(ctx: dict[str, t.Any]) -> None: + """Create list attributes for links of all work items. + + The list is updated on all primary (source) work items. + """ + ctx["LINKS"] = [] + for work_item in ctx["POLARION_WI_MAP"].values(): + wi = f"[{work_item.id}]({work_item.type} {work_item.title})" + logger.debug("Fetching links for work item %r...", wi) + links: list[polarion_api.WorkItemLink] + try: + links = ctx["API"].get_all_work_item_links(work_item.id) + except polarion_api.PolarionApiException as error: + logger.error( + "Fetching links for work item %r failed %s", wi, error.args[0] + ) + continue + + for role, links in _group_by("role", links).items(): + if len(links) < 2: + continue + + work_item.additional_attributes[role] = { + "type": "text/html", + "value": _make_url_list(links), + } + ctx["LINKS"].extend(links) + + if work_item.uuid_capella: + del work_item.additional_attributes["uuid_capella"] + work_item.type = None + try: + ctx["API"].update_work_item(work_item) + except polarion_api.PolarionApiException as error: + logger.error("Updating work item %r failed. %s", wi, error.args[0]) + + +def create_reverse_grouped_links_attributes(ctx: dict[str, t.Any]) -> None: + """Create list attributes for links of all work items. + + The list is updated on all secondary (target) work items. + """ + wid_uuid_map = {v: k for k, v in ctx["POLARION_ID_MAP"].items()} + link_groups = _group_by("secondary_work_item_id", ctx["LINKS"]) + for wid, links in link_groups.items(): + uuid = wid_uuid_map.get(wid) + work_item = ctx["POLARION_WI_MAP"].get(uuid) + if work_item is None: + logger.debug("Unknown work item ID %r found", wid) + continue + + for role, links in _group_by("role", links).items(): + if len(links) < 2: + continue + + work_item.additional_attributes[f"{role}_reverse"] = { + "type": "text/html", + "value": _make_url_list(links), + } + + if work_item.uuid_capella: + del work_item.additional_attributes["uuid_capella"] + work_item.type = None + try: + ctx["API"].update_work_item(work_item) + except polarion_api.PolarionApiException as error: + wi = f"[{work_item.id}]({work_item.type} {work_item.title})" + logger.error("Updating work item %r failed. %s", wi, error.args[0]) + + +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]) -> str: + urls: list[str] = [] + for link in links: + url = serialize.POLARION_WORK_ITEM_URL.format( + pid=link.secondary_work_item_id + ) + urls.append(f"
  • {url}
    • ") + url_list = "\n".join(urls) + return f"" + + CustomLinkMaker = cabc.Callable[ [ dict[str, t.Any], diff --git a/capella2polarion/elements/serialize.py b/capella2polarion/elements/serialize.py index 7a1ee35d..41d93e2d 100644 --- a/capella2polarion/elements/serialize.py +++ b/capella2polarion/elements/serialize.py @@ -29,6 +29,11 @@ ) RE_CAMEL_CASE_2ND_WORD_PATTERN = re.compile(r"([a-z]+)([A-Z][a-z]+)") DIAGRAM_STYLES = {"max-width": "100%"} +POLARION_WORK_ITEM_URL = ( + '' + "" +) PrePostConditionElement = t.Union[ oa.OperationalCapability, interaction.Scenario @@ -213,11 +218,7 @@ def replace_markup( uuid = match.group(1) if pid := ctx["POLARION_ID_MAP"].get(uuid): referenced_uuids.append(uuid) - return ( - '' - "" - ) + return POLARION_WORK_ITEM_URL.format(pid=pid) return non_matcher(match.group(0)) diff --git a/tests/test_elements.py b/tests/test_elements.py index 7d0f5800..f628b885 100644 --- a/tests/test_elements.py +++ b/tests/test_elements.py @@ -442,6 +442,14 @@ def test_update_links( assert context["API"].delete_work_item_links.call_count == 1 assert context["API"].delete_work_item_links.call_args[0][0] == [link] + @staticmethod + def test_create_grouped_links_attributes(context: dict[str, t.Any]): + ... + + @staticmethod + def create_reverse_grouped_links_attributes(context: dict[str, t.Any]): + ... + class TestHelpers: @staticmethod From 936969bb16bcd82bc1b2e11792180a3c0f173215 Mon Sep 17 00:00:00 2001 From: ewuerger Date: Wed, 16 Aug 2023 18:19:57 +0200 Subject: [PATCH 02/15] feat(grouped-links-attributes)!: Added new command Maintain `grouped-links-attributes` induced by Work Item Links. --- capella2polarion/__main__.py | 9 ++- capella2polarion/elements/element.py | 29 ++++++---- tests/test_elements.py | 84 +++++++++++++++++++++++++++- 3 files changed, 105 insertions(+), 17 deletions(-) diff --git a/capella2polarion/__main__.py b/capella2polarion/__main__.py index ea3dd7fd..e6a56171 100644 --- a/capella2polarion/__main__.py +++ b/capella2polarion/__main__.py @@ -250,15 +250,18 @@ def model_elements( def grouped_links_attributes( ctx: click.core.Context, types: list[str] ) -> None: - """Synchronise model elements.""" + """Maintain grouped links custom fields on work items. + + Work items are determined from the given types. + """ ctx.obj["TYPES"] = types ctx.obj["POLARION_WI_MAP"] = get_polarion_wi_map(ctx.obj) ctx.obj["POLARION_ID_MAP"] = { uuid: wi.id for uuid, wi in ctx.obj["POLARION_WI_MAP"].items() } - elements.element.create_grouped_links_attributes(ctx.obj) - elements.element.create_reverse_grouped_links_attributes(ctx.obj) + elements.element.maintain_grouped_links_attributes(ctx.obj) + elements.element.maintain_reverse_grouped_links_attributes(ctx.obj) if __name__ == "__main__": diff --git a/capella2polarion/elements/element.py b/capella2polarion/elements/element.py index 90d35bb1..f923f4ca 100644 --- a/capella2polarion/elements/element.py +++ b/capella2polarion/elements/element.py @@ -241,7 +241,7 @@ def create_grouped_links_attributes(ctx: dict[str, t.Any]) -> None: logger.error("Updating work item %r failed. %s", wi, error.args[0]) -def create_reverse_grouped_links_attributes(ctx: dict[str, t.Any]) -> None: +def maintain_reverse_grouped_links_attributes(ctx: dict[str, t.Any]) -> None: """Create list attributes for links of all work items. The list is updated on all secondary (target) work items. @@ -259,10 +259,9 @@ def create_reverse_grouped_links_attributes(ctx: dict[str, t.Any]) -> None: if len(links) < 2: continue - work_item.additional_attributes[f"{role}_reverse"] = { - "type": "text/html", - "value": _make_url_list(links), - } + work_item.additional_attributes[ + f"{role}_reverse" + ] = _make_url_list(links, reverse=True) if work_item.uuid_capella: del work_item.additional_attributes["uuid_capella"] @@ -270,8 +269,11 @@ def create_reverse_grouped_links_attributes(ctx: dict[str, t.Any]) -> None: try: ctx["API"].update_work_item(work_item) except polarion_api.PolarionApiException as error: - wi = f"[{work_item.id}]({work_item.type} {work_item.title})" - logger.error("Updating work item %r failed. %s", wi, error.args[0]) + logger.error( + "Updating work item [%r] failed. %s", + work_item.id, + error.args[0], + ) def _group_by( @@ -285,12 +287,17 @@ def _group_by( return group -def _make_url_list(links: cabc.Iterable[polarion_api.WorkItemLink]) -> str: +def _make_url_list( + links: cabc.Iterable[polarion_api.WorkItemLink], reverse: bool = False +) -> str: urls: list[str] = [] for link in links: - url = serialize.POLARION_WORK_ITEM_URL.format( - pid=link.secondary_work_item_id - ) + 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}
    • ") url_list = "\n".join(urls) return f"" diff --git a/tests/test_elements.py b/tests/test_elements.py index f628b885..8eb2e422 100644 --- a/tests/test_elements.py +++ b/tests/test_elements.py @@ -68,6 +68,28 @@ "This is a list\n\t
  • an unordered one
    • \n\n\n
      \n\t" "
    1. Ordered list
      2. \n\t
    2. Ok
      3. \n
    \n" ) +TEST_WORK_ITEM_LINKS = [ + polarion_api.WorkItemLink( + "Obj-0", "Obj-1", "attribute", True, "project_id" + ), + polarion_api.WorkItemLink( + "Obj-0", "Obj-2", "attribute", True, "project_id" + ), + polarion_api.WorkItemLink( + "Obj-0", "Obj-1", "attribute1", True, "project_id" + ), +] +TEST_WORK_ITEM_MAP = { + f"uuid{i}": serialize.CapellaWorkItem( + id=f"Obj-{i}", + uuid_capella=f"uuid{i}", + title=f"Fake {i}", + type="fakeModelObject", + description_type="text/html", + description=markupsafe.Markup(""), + ) + for i in range(3) +} class TestDiagramElements: @@ -444,11 +466,67 @@ def test_update_links( @staticmethod def test_create_grouped_links_attributes(context: dict[str, t.Any]): - ... + context["POLARION_WI_MAP"] = TEST_WORK_ITEM_MAP + mock_get_all_work_item_links = mock.MagicMock( + side_effect=(TEST_WORK_ITEM_LINKS, [], []) + ) + mock_update_work_item = mock.MagicMock() + context["API"].get_all_work_item_links = mock_get_all_work_item_links + context["API"].update_work_item = mock_update_work_item + + element.maintain_grouped_links_attributes(context) + + links = context["API"].get_all_work_item_links.call_args_list + assert context["API"].get_all_work_item_links.call_count == 3 + assert [l[0][0] for l in links] == ["Obj-0", "Obj-1", "Obj-2"] + work_items = context["API"].update_work_item.call_args_list + assert context["API"].get_all_work_item_links.call_count == 3 + assert work_items[0][0][0].additional_attributes["attribute"] == ( + "" + ) + for work_item in work_items[1:]: + assert work_item[0][0].additional_attributes == {} + + assert context["LINKS"] == TEST_WORK_ITEM_LINKS[:-1] @staticmethod - def create_reverse_grouped_links_attributes(context: dict[str, t.Any]): - ... + def test_create_reverse_grouped_links_attributes( + context: dict[str, t.Any] + ): + context["POLARION_WI_MAP"] = TEST_WORK_ITEM_MAP + context["POLARION_ID_MAP"] = {f"uuid{i}": f"Obj-{i}" for i in range(3)} + context["LINKS"] = TEST_WORK_ITEM_LINKS[:-1] + [ + polarion_api.WorkItemLink( + "Obj-1", "Obj-2", "attribute", True, "project_id" + ), + ] + mock_update_work_item = mock.MagicMock() + context["API"].update_work_item = mock_update_work_item + + element.maintain_reverse_grouped_links_attributes(context) + + work_items = context["API"].update_work_item.call_args_list + assert context["API"].update_work_item.call_count == 2 + assert work_items[0][0][0].additional_attributes == {} + assert work_items[1][0][0].additional_attributes[ + "attribute_reverse" + ] == ( + "" + ) class TestHelpers: From 676507ae18b52a54703bdc72250e2319cde780a1 Mon Sep 17 00:00:00 2001 From: ewuerger Date: Thu, 17 Aug 2023 17:34:57 +0200 Subject: [PATCH 03/15] test(grouped-links-attributes): Added CLI test and fixed naming --- tests/test_cli.py | 32 ++++++++++++++++++++++++++++++++ tests/test_elements.py | 4 ++-- 2 files changed, 34 insertions(+), 2 deletions(-) diff --git a/tests/test_cli.py b/tests/test_cli.py index 0b752e17..4de7d1b3 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -90,3 +90,35 @@ def test_migrate_model_elements(monkeypatch: pytest.MonkeyPatch): assert mock_patch_work_items.call_count == 1 assert mock_post_work_items.call_count == 1 assert ELEMENTS_IDX_PATH.exists() + + +def test_grouped_links_attributes(monkeypatch: pytest.MonkeyPatch): + mock_get_polarion_wi_map = prepare_cli_test( + monkeypatch, + {"uuid{i}": polarion_api.WorkItem("project/W-{i}") for i in range(10)}, + ) + mock_maintain_grouped_links_attributes = mock.MagicMock() + monkeypatch.setattr( + elements.element, + "maintain_grouped_links_attributes", + mock_maintain_grouped_links_attributes, + ) + mock_maintain_reverse_grouped_links_attributes = mock.MagicMock() + monkeypatch.setattr( + elements.element, + "maintain_reverse_grouped_links_attributes", + mock_maintain_reverse_grouped_links_attributes, + ) + + command = [ + "--project-id=project_id", + "grouped-links-attributes", + "OperationalCapability SystemCapability", + ] + + result = testing.CliRunner().invoke(main.cli, command) + + assert result.exit_code == 0 + assert mock_get_polarion_wi_map.call_count == 1 + assert mock_maintain_grouped_links_attributes.call_count == 1 + assert mock_maintain_reverse_grouped_links_attributes.call_count == 1 diff --git a/tests/test_elements.py b/tests/test_elements.py index 8eb2e422..ef5f766f 100644 --- a/tests/test_elements.py +++ b/tests/test_elements.py @@ -465,7 +465,7 @@ def test_update_links( assert context["API"].delete_work_item_links.call_args[0][0] == [link] @staticmethod - def test_create_grouped_links_attributes(context: dict[str, t.Any]): + def test_maintain_grouped_links_attributes(context: dict[str, t.Any]): context["POLARION_WI_MAP"] = TEST_WORK_ITEM_MAP mock_get_all_work_item_links = mock.MagicMock( side_effect=(TEST_WORK_ITEM_LINKS, [], []) @@ -497,7 +497,7 @@ def test_create_grouped_links_attributes(context: dict[str, t.Any]): assert context["LINKS"] == TEST_WORK_ITEM_LINKS[:-1] @staticmethod - def test_create_reverse_grouped_links_attributes( + def test_maintain_reverse_grouped_links_attributes( context: dict[str, t.Any] ): context["POLARION_WI_MAP"] = TEST_WORK_ITEM_MAP From 38f2865b62498d33f13c085391895982f12e5f9f Mon Sep 17 00:00:00 2001 From: Michael Harbarth Date: Tue, 14 Nov 2023 15:59:56 +0100 Subject: [PATCH 04/15] refactor: Refactoring after rebasing on current main --- capella2polarion/__main__.py | 22 +--------------------- 1 file changed, 1 insertion(+), 21 deletions(-) diff --git a/capella2polarion/__main__.py b/capella2polarion/__main__.py index e6a56171..23e0f2e8 100644 --- a/capella2polarion/__main__.py +++ b/capella2polarion/__main__.py @@ -152,7 +152,7 @@ def _sanitize_config( @click.pass_context def cli( ctx: click.core.Context, debug: bool, project_id: str, delete: bool = False -): +) -> None: """Synchronise data from Capella to Polarion. PROJECT_ID is a Polarion project id @@ -244,25 +244,5 @@ def model_elements( elements.make_model_elements_index(ctx.obj) -@cli.command() -@click.argument("types", nargs=-1, type=str) -@click.pass_context -def grouped_links_attributes( - ctx: click.core.Context, types: list[str] -) -> None: - """Maintain grouped links custom fields on work items. - - Work items are determined from the given types. - """ - ctx.obj["TYPES"] = types - ctx.obj["POLARION_WI_MAP"] = get_polarion_wi_map(ctx.obj) - ctx.obj["POLARION_ID_MAP"] = { - uuid: wi.id for uuid, wi in ctx.obj["POLARION_WI_MAP"].items() - } - - elements.element.maintain_grouped_links_attributes(ctx.obj) - elements.element.maintain_reverse_grouped_links_attributes(ctx.obj) - - if __name__ == "__main__": cli(obj={}) From 7a8e9dabab560dfe33db6512809ab906cf267b46 Mon Sep 17 00:00:00 2001 From: Michael Harbarth Date: Fri, 17 Nov 2023 15:42:41 +0100 Subject: [PATCH 05/15] fix: minor fixes and fixes for tests after rebase. wip. --- capella2polarion/elements/element.py | 11 +++++----- tests/test_cli.py | 32 ---------------------------- tests/test_elements.py | 8 ++++--- 3 files changed, 11 insertions(+), 40 deletions(-) diff --git a/capella2polarion/elements/element.py b/capella2polarion/elements/element.py index f923f4ca..99ed483d 100644 --- a/capella2polarion/elements/element.py +++ b/capella2polarion/elements/element.py @@ -46,7 +46,7 @@ def create_work_items( _work_items = list(filter(None, _work_items)) valid_types = set(map(helpers.resolve_element_type, set(ctx["ELEMENTS"]))) - work_items: list[polarion_api.CapellaWorkItem] = [] + work_items: list[serialize.CapellaWorkItem] = [] missing_types: set[str] = set() for work_item in _work_items: assert work_item is not None @@ -204,7 +204,7 @@ def _handle_exchanges( return _create(context, wid, role_id, exchanges, links) -def create_grouped_links_attributes(ctx: dict[str, t.Any]) -> None: +def maintain_grouped_links_attributes(ctx: dict[str, t.Any]) -> None: """Create list attributes for links of all work items. The list is updated on all primary (source) work items. @@ -259,9 +259,10 @@ def maintain_reverse_grouped_links_attributes(ctx: dict[str, t.Any]) -> None: if len(links) < 2: continue - work_item.additional_attributes[ - f"{role}_reverse" - ] = _make_url_list(links, reverse=True) + work_item.additional_attributes[f"{role}_reverse"] = { + "type": "text/html", + "value": _make_url_list(links, reverse=True), + } if work_item.uuid_capella: del work_item.additional_attributes["uuid_capella"] diff --git a/tests/test_cli.py b/tests/test_cli.py index 4de7d1b3..0b752e17 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -90,35 +90,3 @@ def test_migrate_model_elements(monkeypatch: pytest.MonkeyPatch): assert mock_patch_work_items.call_count == 1 assert mock_post_work_items.call_count == 1 assert ELEMENTS_IDX_PATH.exists() - - -def test_grouped_links_attributes(monkeypatch: pytest.MonkeyPatch): - mock_get_polarion_wi_map = prepare_cli_test( - monkeypatch, - {"uuid{i}": polarion_api.WorkItem("project/W-{i}") for i in range(10)}, - ) - mock_maintain_grouped_links_attributes = mock.MagicMock() - monkeypatch.setattr( - elements.element, - "maintain_grouped_links_attributes", - mock_maintain_grouped_links_attributes, - ) - mock_maintain_reverse_grouped_links_attributes = mock.MagicMock() - monkeypatch.setattr( - elements.element, - "maintain_reverse_grouped_links_attributes", - mock_maintain_reverse_grouped_links_attributes, - ) - - command = [ - "--project-id=project_id", - "grouped-links-attributes", - "OperationalCapability SystemCapability", - ] - - result = testing.CliRunner().invoke(main.cli, command) - - assert result.exit_code == 0 - assert mock_get_polarion_wi_map.call_count == 1 - assert mock_maintain_grouped_links_attributes.call_count == 1 - assert mock_maintain_reverse_grouped_links_attributes.call_count == 1 diff --git a/tests/test_elements.py b/tests/test_elements.py index ef5f766f..f338abbf 100644 --- a/tests/test_elements.py +++ b/tests/test_elements.py @@ -481,7 +481,9 @@ def test_maintain_grouped_links_attributes(context: dict[str, t.Any]): assert [l[0][0] for l in links] == ["Obj-0", "Obj-1", "Obj-2"] work_items = context["API"].update_work_item.call_args_list assert context["API"].get_all_work_item_links.call_count == 3 - assert work_items[0][0][0].additional_attributes["attribute"] == ( + assert work_items[0][0][0].additional_attributes["attribute"][ + "value" + ] == ( "
    • " '' @@ -515,8 +517,8 @@ def test_maintain_reverse_grouped_links_attributes( work_items = context["API"].update_work_item.call_args_list assert context["API"].update_work_item.call_count == 2 assert work_items[0][0][0].additional_attributes == {} - assert work_items[1][0][0].additional_attributes[ - "attribute_reverse" + assert work_items[1][0][0].additional_attributes["attribute_reverse"][ + "value" ] == ( "
      • " ' Date: Mon, 20 Nov 2023 10:57:57 +0100 Subject: [PATCH 06/15] refactor: rewrite maintain_grouped_links_attributes and tests expecting links to be present in the list of workitems --- capella2polarion/elements/element.py | 112 ++++++++++--------------- tests/conftest.py | 26 ++++++ tests/test_elements.py | 119 ++++++++++++++------------- 3 files changed, 132 insertions(+), 125 deletions(-) diff --git a/capella2polarion/elements/element.py b/capella2polarion/elements/element.py index 99ed483d..4708896e 100644 --- a/capella2polarion/elements/element.py +++ b/capella2polarion/elements/element.py @@ -204,77 +204,55 @@ def _handle_exchanges( return _create(context, wid, role_id, exchanges, links) -def maintain_grouped_links_attributes(ctx: dict[str, t.Any]) -> None: +def maintain_grouped_links_attributes( + work_item_map: dict[str, serialize.CapellaWorkItem], + polarion_id_map: dict[str, str], + include_back_links, +) -> None: """Create list attributes for links of all work items. - The list is updated on all primary (source) work items. + The list is updated on all primary work items and reverse links can + be added, too. """ - ctx["LINKS"] = [] - for work_item in ctx["POLARION_WI_MAP"].values(): + back_links: dict[str, list[polarion_api.WorkItemLink]] = {} + reverse_polarion_id_map = {v: k for k, v in polarion_id_map.items()} + + def _create_link_fields( + work_item: serialize.CapellaWorkItem, + role: str, + links: list[polarion_api.WorkItemLink], + reverse: bool = False, + ): + # TODO check why we only create links for > 2 per role + if len(links) < 2: + return + role = f"{role}_reverse" if reverse else role + work_item.additional_attributes[role] = { + "type": "text/html", + "value": _make_url_list(links, reverse), + } + + for work_item in work_item_map.values(): wi = f"[{work_item.id}]({work_item.type} {work_item.title})" - logger.debug("Fetching links for work item %r...", wi) - links: list[polarion_api.WorkItemLink] - try: - links = ctx["API"].get_all_work_item_links(work_item.id) - except polarion_api.PolarionApiException as error: - logger.error( - "Fetching links for work item %r failed %s", wi, error.args[0] - ) - continue - - for role, links in _group_by("role", links).items(): - if len(links) < 2: - continue - - work_item.additional_attributes[role] = { - "type": "text/html", - "value": _make_url_list(links), - } - ctx["LINKS"].extend(links) - - if work_item.uuid_capella: - del work_item.additional_attributes["uuid_capella"] - work_item.type = None - try: - ctx["API"].update_work_item(work_item) - except polarion_api.PolarionApiException as error: - logger.error("Updating work item %r failed. %s", wi, error.args[0]) - - -def maintain_reverse_grouped_links_attributes(ctx: dict[str, t.Any]) -> None: - """Create list attributes for links of all work items. - - The list is updated on all secondary (target) work items. - """ - wid_uuid_map = {v: k for k, v in ctx["POLARION_ID_MAP"].items()} - link_groups = _group_by("secondary_work_item_id", ctx["LINKS"]) - for wid, links in link_groups.items(): - uuid = wid_uuid_map.get(wid) - work_item = ctx["POLARION_WI_MAP"].get(uuid) - if work_item is None: - logger.debug("Unknown work item ID %r found", wid) - continue - - for role, links in _group_by("role", links).items(): - if len(links) < 2: - continue - - work_item.additional_attributes[f"{role}_reverse"] = { - "type": "text/html", - "value": _make_url_list(links, reverse=True), - } - - if work_item.uuid_capella: - del work_item.additional_attributes["uuid_capella"] - work_item.type = None - try: - ctx["API"].update_work_item(work_item) - except polarion_api.PolarionApiException as error: - logger.error( - "Updating work item [%r] failed. %s", - work_item.id, - error.args[0], - ) + 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 include_back_links: + for link in grouped_links: + uuid = reverse_polarion_id_map[link.secondary_work_item_id] + if uuid not in back_links: + back_links[uuid] = [] + back_links[uuid].append(link) + + _create_link_fields(work_item, role, grouped_links) + + if include_back_links: + for uuid, links in back_links.items(): + work_item = work_item_map[uuid] + for role, grouped_links in _group_by("role", links).items(): + _create_link_fields(work_item, role, grouped_links, True) def _group_by( diff --git a/tests/conftest.py b/tests/conftest.py index 163144cc..6bc96b58 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -8,8 +8,12 @@ import typing as t import capellambse +import markupsafe +import polarion_rest_api_client as polarion_api import pytest +from capella2polarion.elements import serialize + TEST_DATA_ROOT = pathlib.Path(__file__).parent / "data" TEST_DIAGRAM_CACHE = TEST_DATA_ROOT / "diagram_cache" TEST_MODEL_ELEMENTS = TEST_DATA_ROOT / "model_elements" @@ -29,3 +33,25 @@ def diagram_cache_index() -> list[dict[str, t.Any]]: def model() -> capellambse.MelodyModel: """Return the test model.""" return capellambse.MelodyModel(path=TEST_MODEL) + + +@pytest.fixture +def dummy_work_items() -> dict[str, serialize.CapellaWorkItem]: + return { + f"uuid{i}": serialize.CapellaWorkItem( + id=f"Obj-{i}", + uuid_capella=f"uuid{i}", + title=f"Fake {i}", + type="fakeModelObject", + description_type="text/html", + description=markupsafe.Markup(""), + linked_work_items=[ + polarion_api.WorkItemLink( + f"Obj-{i}", f"Obj-{j}", "attribute", True, "project_id" + ) + for j in range(3) + if (i not in (j, 2)) + ], + ) + for i in range(3) + } diff --git a/tests/test_elements.py b/tests/test_elements.py index f338abbf..663875b6 100644 --- a/tests/test_elements.py +++ b/tests/test_elements.py @@ -68,28 +68,7 @@ "This is a list
        • \n\t
      • an unordered one
        • \n
      \n\n
        \n\t" "
      1. Ordered list
        2. \n\t
      2. Ok
        3. \n
      \n" ) -TEST_WORK_ITEM_LINKS = [ - polarion_api.WorkItemLink( - "Obj-0", "Obj-1", "attribute", True, "project_id" - ), - polarion_api.WorkItemLink( - "Obj-0", "Obj-2", "attribute", True, "project_id" - ), - polarion_api.WorkItemLink( - "Obj-0", "Obj-1", "attribute1", True, "project_id" - ), -] -TEST_WORK_ITEM_MAP = { - f"uuid{i}": serialize.CapellaWorkItem( - id=f"Obj-{i}", - uuid_capella=f"uuid{i}", - title=f"Fake {i}", - type="fakeModelObject", - description_type="text/html", - description=markupsafe.Markup(""), - ) - for i in range(3) -} +POLARION_ID_MAP = {f"uuid{i}": f"Obj-{i}" for i in range(3)} class TestDiagramElements: @@ -465,25 +444,18 @@ def test_update_links( assert context["API"].delete_work_item_links.call_args[0][0] == [link] @staticmethod - def test_maintain_grouped_links_attributes(context: dict[str, t.Any]): - context["POLARION_WI_MAP"] = TEST_WORK_ITEM_MAP - mock_get_all_work_item_links = mock.MagicMock( - side_effect=(TEST_WORK_ITEM_LINKS, [], []) + def test_maintain_grouped_links_attributes(dummy_work_items): + element.maintain_grouped_links_attributes( + dummy_work_items, POLARION_ID_MAP, False ) - mock_update_work_item = mock.MagicMock() - context["API"].get_all_work_item_links = mock_get_all_work_item_links - context["API"].update_work_item = mock_update_work_item - element.maintain_grouped_links_attributes(context) + 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"] - links = context["API"].get_all_work_item_links.call_args_list - assert context["API"].get_all_work_item_links.call_count == 3 - assert [l[0][0] for l in links] == ["Obj-0", "Obj-1", "Obj-2"] - work_items = context["API"].update_work_item.call_args_list - assert context["API"].get_all_work_item_links.call_count == 3 - assert work_items[0][0][0].additional_attributes["attribute"][ - "value" - ] == ( + assert dummy_work_items["uuid0"].additional_attributes.pop( + "attribute" + )["value"] == ( "
      • " '' @@ -493,33 +465,60 @@ def test_maintain_grouped_links_attributes(context: dict[str, t.Any]): 'data-item-id="Obj-2" data-option-id="long">' "
      " ) - for work_item in work_items[1:]: - assert work_item[0][0].additional_attributes == {} - assert context["LINKS"] == TEST_WORK_ITEM_LINKS[:-1] + assert dummy_work_items["uuid1"].additional_attributes.pop( + "attribute" + )["value"] == ( + "
      • " + '' + "
        • \n" + "
      • " + '' + "
      " + ) + + assert dummy_work_items["uuid0"].additional_attributes == {} + assert dummy_work_items["uuid1"].additional_attributes == {} + assert dummy_work_items["uuid2"].additional_attributes == {} @staticmethod - def test_maintain_reverse_grouped_links_attributes( - context: dict[str, t.Any] - ): - context["POLARION_WI_MAP"] = TEST_WORK_ITEM_MAP - context["POLARION_ID_MAP"] = {f"uuid{i}": f"Obj-{i}" for i in range(3)} - context["LINKS"] = TEST_WORK_ITEM_LINKS[:-1] + [ - polarion_api.WorkItemLink( - "Obj-1", "Obj-2", "attribute", True, "project_id" - ), - ] - mock_update_work_item = mock.MagicMock() - context["API"].update_work_item = mock_update_work_item + def test_maintain_reverse_grouped_links_attributes(dummy_work_items): + element.maintain_grouped_links_attributes( + dummy_work_items, POLARION_ID_MAP, True + ) + + 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"] - element.maintain_reverse_grouped_links_attributes(context) + del dummy_work_items["uuid0"].additional_attributes["attribute"] + del dummy_work_items["uuid1"].additional_attributes["attribute"] - work_items = context["API"].update_work_item.call_args_list - assert context["API"].update_work_item.call_count == 2 - assert work_items[0][0][0].additional_attributes == {} - assert work_items[1][0][0].additional_attributes["attribute_reverse"][ + # TODO check if we really want to exclude groups of <2 + """ + assert dummy_work_items["uuid0"].additional_attributes.pop("attribute_reverse")[ "value" ] == ( + "
      • " + '' + "
      " + ) + + assert dummy_work_items["uuid1"].additional_attributes.pop("attribute_reverse")[ + "value" + ] == ( + "
      • " + '' + "
      " + ) + """ + assert dummy_work_items["uuid2"].additional_attributes.pop( + "attribute_reverse" + )["value"] == ( "
      • " '' @@ -530,6 +529,10 @@ def test_maintain_reverse_grouped_links_attributes( "
      " ) + assert dummy_work_items["uuid0"].additional_attributes == {} + assert dummy_work_items["uuid1"].additional_attributes == {} + assert dummy_work_items["uuid2"].additional_attributes == {} + class TestHelpers: @staticmethod From b54be4a1f721ebbd167facc0b59b9e35f0f47fc0 Mon Sep 17 00:00:00 2001 From: Michael Harbarth Date: Tue, 21 Nov 2023 08:24:28 +0100 Subject: [PATCH 07/15] feat: add new grouped links to patch_workitems method --- capella2polarion/elements/__init__.py | 43 +++++----- capella2polarion/elements/api_helper.py | 82 ++++++++---------- capella2polarion/elements/element.py | 105 +++++++++++++----------- tests/test_elements.py | 17 ++-- 4 files changed, 127 insertions(+), 120 deletions(-) diff --git a/capella2polarion/elements/__init__.py b/capella2polarion/elements/__init__.py index 73325589..f2627160 100644 --- a/capella2polarion/elements/__init__.py +++ b/capella2polarion/elements/__init__.py @@ -136,37 +136,40 @@ def patch_work_items(ctx: dict[str, t.Any]) -> None: """ work_items_lookup = ctx["POLARION_WI_MAP"] | ctx["WORK_ITEMS"] - def add_content( - obj: common.GenericElement | diag.Diagram, - _: dict[str, t.Any], - **kwargs, - ) -> serialize.CapellaWorkItem: - work_item = work_items_lookup[obj.uuid] - for key, value in kwargs.items(): - if getattr(work_item, key, None) is None: - continue - - setattr(work_item, key, value) - return work_item - ctx["POLARION_ID_MAP"] = uuids = { uuid: wi.id for uuid, wi in ctx["POLARION_WI_MAP"].items() if wi.status == "open" and wi.uuid_capella and wi.id } + + back_links: dict[str, list[polarion_api.WorkItemLink]] = {} + for uuid in uuids: - elements = ctx["MODEL"] + objects = ctx["MODEL"] if uuid.startswith("_"): - elements = ctx["MODEL"].diagrams - obj = elements.by_uuid(uuid) + objects = ctx["MODEL"].diagrams + obj = objects.by_uuid(uuid) links = element.create_links(obj, ctx) + work_item: serialize.CapellaWorkItem = work_items_lookup[obj.uuid] + work_item.linked_work_items = links + + element.create_grouped_link_fields(work_item, back_links) + + for uuid in uuids: + new_work_item: serialize.CapellaWorkItem = work_items_lookup[uuid] + old_work_item: serialize.CapellaWorkItem = ctx["POLARION_WI_MAP"][uuid] + + if old_work_item.id in back_links: + element.create_grouped_back_link_fields( + new_work_item, back_links[old_work_item.id] + ) api_helper.patch_work_item( - ctx, - obj, - functools.partial(add_content, linked_work_items=links), - obj._short_repr_(), + ctx["API"], + new_work_item, + old_work_item, + old_work_item.title, "element", ) diff --git a/capella2polarion/elements/api_helper.py b/capella2polarion/elements/api_helper.py index 966c2fc9..b3fef82b 100644 --- a/capella2polarion/elements/api_helper.py +++ b/capella2polarion/elements/api_helper.py @@ -6,7 +6,6 @@ import typing as t import polarion_rest_api_client as polarion_api -from capellambse.model import common from capella2polarion.elements import serialize @@ -14,11 +13,9 @@ def patch_work_item( - ctx: dict[str, t.Any], - obj: common.GenericElement, - receiver: cabc.Callable[ - [t.Any, dict[str, t.Any]], serialize.CapellaWorkItem - ], + api: polarion_api.OpenAPIPolarionProjectClient, + new: serialize.CapellaWorkItem, + old: serialize.CapellaWorkItem, name: str, _type: str, ): @@ -26,61 +23,54 @@ def patch_work_item( Parameters ---------- - ctx + api The context to execute the patch for. - obj - The Capella object to update the WorkItem from. - receiver - A function that receives the WorkItem from the created - instances. This function alters the WorkItem instances by adding - attributes, e.g.: `linked_work_items`. It can be useful to add - attributes which can only be computed after the work item and - its default attributes were instantiated. + new + The updated CapellaWorkItem + old + The CapellaWorkItem currently present on polarion name The name of the object, which should be displayed in log messages. _type The type of element, which should be shown in log messages. """ - if new := receiver(obj, ctx): - wid = ctx["POLARION_ID_MAP"][obj.uuid] - old: serialize.CapellaWorkItem = ctx["POLARION_WI_MAP"][obj.uuid] - if new == old: - return + if new == old: + return - log_args = (wid, _type, name) - logger.info("Update work item %r for model %s %r...", *log_args) - if "uuid_capella" in new.additional_attributes: - del new.additional_attributes["uuid_capella"] + log_args = (old.id, _type, name) + logger.info("Update work item %r for model %s %r...", *log_args) + if "uuid_capella" in new.additional_attributes: + del new.additional_attributes["uuid_capella"] - old.linked_work_items = ctx["API"].get_all_work_item_links(old.id) - new.type = None - new.status = "open" - new.id = wid - try: - ctx["API"].update_work_item(new) - handle_links( - old.linked_work_items, - new.linked_work_items, - ("Delete", _type, name), - ctx["API"].delete_work_item_links, - ) - handle_links( - new.linked_work_items, - old.linked_work_items, - ("Create", _type, name), - ctx["API"].create_work_item_links, - ) - except polarion_api.PolarionApiException as error: - wi = f"{wid}({_type} {name})" - logger.error("Updating work item %r failed. %s", wi, error.args[0]) + old.linked_work_items = api.get_all_work_item_links(old.id) + new.type = None + new.status = "open" + new.id = old.id + try: + api.update_work_item(new) + handle_links( + old.linked_work_items, + new.linked_work_items, + ("Delete", _type, name), + api.delete_work_item_links, + ) + handle_links( + new.linked_work_items, + old.linked_work_items, + ("Create", _type, name), + api.create_work_item_links, + ) + except polarion_api.PolarionApiException as error: + wi = f"{old.id}({_type} {name})" + logger.error("Updating work item %r failed. %s", wi, error.args[0]) def handle_links( left: cabc.Iterable[polarion_api.WorkItemLink], right: cabc.Iterable[polarion_api.WorkItemLink], log_args: tuple[str, ...], - handler: cabc.Callable[[cabc.Iterable[polarion_api.WorkItemLink]], None], + handler: cabc.Callable[[cabc.Iterable[polarion_api.WorkItemLink]], t.Any], ): """Handle work item links on Polarion.""" for link in (links := get_links(left, right)): diff --git a/capella2polarion/elements/element.py b/capella2polarion/elements/element.py index 4708896e..16234d2d 100644 --- a/capella2polarion/elements/element.py +++ b/capella2polarion/elements/element.py @@ -204,55 +204,48 @@ def _handle_exchanges( return _create(context, wid, role_id, exchanges, links) -def maintain_grouped_links_attributes( - work_item_map: dict[str, serialize.CapellaWorkItem], - polarion_id_map: dict[str, str], - include_back_links, -) -> None: - """Create list attributes for links of all work items. - - The list is updated on all primary work items and reverse links can - be added, too. +def create_grouped_link_fields( + work_item: serialize.CapellaWorkItem, + back_links: dict[str, list[polarion_api.WorkItemLink]] | None = None, +): + """Create the grouped link workitems fields from of the primary work items. + + Parameters + ---------- + work_item + WorkItem to create the fields for + back_links + A dictionary of secondary WorkItem IDs to links to create backlinks afterwards """ - back_links: dict[str, list[polarion_api.WorkItemLink]] = {} - reverse_polarion_id_map = {v: k for k, v in polarion_id_map.items()} - - def _create_link_fields( - work_item: serialize.CapellaWorkItem, - role: str, - links: list[polarion_api.WorkItemLink], - reverse: bool = False, - ): - # TODO check why we only create links for > 2 per role - if len(links) < 2: - return - role = f"{role}_reverse" if reverse else role - work_item.additional_attributes[role] = { - "type": "text/html", - "value": _make_url_list(links, reverse), - } - - for work_item in work_item_map.values(): - 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 include_back_links: - for link in grouped_links: - uuid = reverse_polarion_id_map[link.secondary_work_item_id] - if uuid not in back_links: - back_links[uuid] = [] - back_links[uuid].append(link) - - _create_link_fields(work_item, role, grouped_links) - - if include_back_links: - for uuid, links in back_links.items(): - work_item = work_item_map[uuid] - for role, grouped_links in _group_by("role", links).items(): - _create_link_fields(work_item, role, grouped_links, True) + 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: + if link.secondary_work_item_id not in back_links: + back_links[link.secondary_work_item_id] = [] + back_links[link.secondary_work_item_id].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( @@ -282,6 +275,22 @@ def _make_url_list( return f"
        {url_list}
      " +def _create_link_fields( + work_item: serialize.CapellaWorkItem, + role: str, + links: list[polarion_api.WorkItemLink], + reverse: bool = False, +): + # TODO check why we only create links for > 2 per role + if len(links) < 2: + return + role = f"{role}_reverse" if reverse else role + work_item.additional_attributes[role] = { + "type": "text/html", + "value": _make_url_list(links, reverse), + } + + CustomLinkMaker = cabc.Callable[ [ dict[str, t.Any], diff --git a/tests/test_elements.py b/tests/test_elements.py index 663875b6..a43207e5 100644 --- a/tests/test_elements.py +++ b/tests/test_elements.py @@ -445,9 +445,8 @@ def test_update_links( @staticmethod def test_maintain_grouped_links_attributes(dummy_work_items): - element.maintain_grouped_links_attributes( - dummy_work_items, POLARION_ID_MAP, False - ) + for work_item in dummy_work_items.values(): + element.create_grouped_link_fields(work_item) del dummy_work_items["uuid0"].additional_attributes["uuid_capella"] del dummy_work_items["uuid1"].additional_attributes["uuid_capella"] @@ -485,9 +484,15 @@ def test_maintain_grouped_links_attributes(dummy_work_items): @staticmethod def test_maintain_reverse_grouped_links_attributes(dummy_work_items): - element.maintain_grouped_links_attributes( - dummy_work_items, POLARION_ID_MAP, True - ) + 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) + + 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) del dummy_work_items["uuid0"].additional_attributes["uuid_capella"] del dummy_work_items["uuid1"].additional_attributes["uuid_capella"] From 55170e15ac40a9425d8b8b6294c4a42cfbb219c2 Mon Sep 17 00:00:00 2001 From: Michael Harbarth Date: Wed, 22 Nov 2023 09:28:52 +0100 Subject: [PATCH 08/15] test: Add test for linkedgroupedworkitems in patch_work_items --- capella2polarion/elements/__init__.py | 5 +- tests/conftest.py | 1 + tests/test_elements.py | 170 +++++++++++++++++++------- 3 files changed, 129 insertions(+), 47 deletions(-) diff --git a/capella2polarion/elements/__init__.py b/capella2polarion/elements/__init__.py index f2627160..de0ec677 100644 --- a/capella2polarion/elements/__init__.py +++ b/capella2polarion/elements/__init__.py @@ -134,6 +134,7 @@ def patch_work_items(ctx: dict[str, t.Any]) -> None: ctx The context for the workitem operation to be processed. """ + # TODO Why did we have this lookup? work_items_lookup = ctx["POLARION_WI_MAP"] | ctx["WORK_ITEMS"] ctx["POLARION_ID_MAP"] = uuids = { @@ -151,13 +152,13 @@ def patch_work_items(ctx: dict[str, t.Any]) -> None: obj = objects.by_uuid(uuid) links = element.create_links(obj, ctx) - work_item: serialize.CapellaWorkItem = work_items_lookup[obj.uuid] + work_item: serialize.CapellaWorkItem = ctx["WORK_ITEMS"][uuid] work_item.linked_work_items = links element.create_grouped_link_fields(work_item, back_links) for uuid in uuids: - new_work_item: serialize.CapellaWorkItem = work_items_lookup[uuid] + new_work_item: serialize.CapellaWorkItem = ctx["WORK_ITEMS"][uuid] old_work_item: serialize.CapellaWorkItem = ctx["POLARION_WI_MAP"][uuid] if old_work_item.id in back_links: diff --git a/tests/conftest.py b/tests/conftest.py index 6bc96b58..54580bba 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -52,6 +52,7 @@ def dummy_work_items() -> dict[str, serialize.CapellaWorkItem]: for j in range(3) if (i not in (j, 2)) ], + status="open", ) for i in range(3) } diff --git a/tests/test_elements.py b/tests/test_elements.py index a43207e5..4f225ac6 100644 --- a/tests/test_elements.py +++ b/tests/test_elements.py @@ -70,6 +70,55 @@ ) POLARION_ID_MAP = {f"uuid{i}": f"Obj-{i}" for i in range(3)} +HTML_LINK_0 = { + "attribute": ( + "
      • " + '' + "
        • \n" + "
      • " + '' + "
      " + ), + "attribute_reverse": ( + "
      • " + '' + "
      " + ), +} +HTML_LINK_1 = { + "attribute": ( + "
      • " + '' + "
        • \n" + "
      • " + '' + "
      " + ), + "attribute_reverse": ( + "
      • " + '' + "
      " + ), +} +HTML_LINK_2 = { + "attribute_reverse": ( + "
      • " + '' + "
        • \n" + "
      • " + '' + "
      " + ) +} + class TestDiagramElements: @staticmethod @@ -443,6 +492,65 @@ def test_update_links( assert context["API"].delete_work_item_links.call_count == 1 assert context["API"].delete_work_item_links.call_args[0][0] == [link] + @staticmethod + def test_patch_work_item_grouped_links( + monkeypatch: pytest.MonkeyPatch, + context: dict[str, t.Any], + dummy_work_items, + ): + context["WORK_ITEMS"] = dummy_work_items + + context["POLARION_WI_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" + ), + } + mock_create_links = mock.MagicMock() + monkeypatch.setattr(element, "create_links", mock_create_links) + mock_create_links.side_effect = lambda obj, ctx: dummy_work_items[ + obj.uuid + ].linked_work_items + + context["MODEL"] = mock_model = mock.MagicMock() + mock_model.by_uuid.side_effect = [ + FakeModelObject(f"uuid{i}", name=f"Fake {i}") for i in range(3) + ] + + elements.patch_work_items(context) + + update_work_item_calls = context["API"].update_work_item.call_args_list + + assert len(update_work_item_calls) == 3 + + work_item_0 = update_work_item_calls[0][0][0] + work_item_1 = update_work_item_calls[1][0][0] + work_item_2 = update_work_item_calls[2][0][0] + + assert ( + work_item_0.additional_attributes.pop("attribute")["value"] + == HTML_LINK_0["attribute"] + ) + + assert ( + work_item_1.additional_attributes.pop("attribute")["value"] + == HTML_LINK_1["attribute"] + ) + + assert ( + work_item_2.additional_attributes.pop("attribute_reverse")["value"] + == HTML_LINK_2["attribute_reverse"] + ) + + assert work_item_0.additional_attributes == {} + assert work_item_1.additional_attributes == {} + assert work_item_2.additional_attributes == {} + @staticmethod def test_maintain_grouped_links_attributes(dummy_work_items): for work_item in dummy_work_items.values(): @@ -452,30 +560,18 @@ def test_maintain_grouped_links_attributes(dummy_work_items): del dummy_work_items["uuid1"].additional_attributes["uuid_capella"] del dummy_work_items["uuid2"].additional_attributes["uuid_capella"] - assert dummy_work_items["uuid0"].additional_attributes.pop( - "attribute" - )["value"] == ( - "
      • " - '' - "
        • \n" - "
      • " - '' - "
      " + assert ( + dummy_work_items["uuid0"].additional_attributes.pop("attribute")[ + "value" + ] + == HTML_LINK_0["attribute"] ) - assert dummy_work_items["uuid1"].additional_attributes.pop( - "attribute" - )["value"] == ( - "
      • " - '' - "
        • \n" - "
      • " - '' - "
      " + assert ( + dummy_work_items["uuid1"].additional_attributes.pop("attribute")[ + "value" + ] + == HTML_LINK_1["attribute"] ) assert dummy_work_items["uuid0"].additional_attributes == {} @@ -505,33 +601,17 @@ def test_maintain_reverse_grouped_links_attributes(dummy_work_items): """ assert dummy_work_items["uuid0"].additional_attributes.pop("attribute_reverse")[ "value" - ] == ( - "
      • " - '' - "
      " - ) + ] == HTML_LINK_0["attribute_reverse"] assert dummy_work_items["uuid1"].additional_attributes.pop("attribute_reverse")[ "value" - ] == ( - "
      • " - '' - "
      " - ) + ] == HTML_LINK_1["attribute_reverse"] """ - assert dummy_work_items["uuid2"].additional_attributes.pop( - "attribute_reverse" - )["value"] == ( - "
      • " - '' - "
        • \n" - "
      • " - '' - "
      " + assert ( + dummy_work_items["uuid2"].additional_attributes.pop( + "attribute_reverse" + )["value"] + == HTML_LINK_2["attribute_reverse"] ) assert dummy_work_items["uuid0"].additional_attributes == {} From 47e4af15210213249850a8624886ec2207cca022 Mon Sep 17 00:00:00 2001 From: Michael Harbarth Date: Wed, 22 Nov 2023 10:06:38 +0100 Subject: [PATCH 09/15] refactor: pylint --- capella2polarion/elements/element.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/capella2polarion/elements/element.py b/capella2polarion/elements/element.py index 16234d2d..b20481fe 100644 --- a/capella2polarion/elements/element.py +++ b/capella2polarion/elements/element.py @@ -215,7 +215,7 @@ def create_grouped_link_fields( work_item WorkItem to create the fields for back_links - A dictionary of secondary WorkItem IDs to links to create backlinks afterwards + 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) From 37577c4fb1c767ff49da6cf9d65725c8c8aebbd4 Mon Sep 17 00:00:00 2001 From: ewuerger Date: Wed, 29 Nov 2023 11:45:49 +0100 Subject: [PATCH 10/15] fix: Missing work item IDs in patch --- capella2polarion/elements/__init__.py | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/capella2polarion/elements/__init__.py b/capella2polarion/elements/__init__.py index de0ec677..35980484 100644 --- a/capella2polarion/elements/__init__.py +++ b/capella2polarion/elements/__init__.py @@ -12,7 +12,6 @@ "STATUS_DELETE", ] -import functools import logging import pathlib import typing as t @@ -134,9 +133,6 @@ def patch_work_items(ctx: dict[str, t.Any]) -> None: ctx The context for the workitem operation to be processed. """ - # TODO Why did we have this lookup? - work_items_lookup = ctx["POLARION_WI_MAP"] | ctx["WORK_ITEMS"] - ctx["POLARION_ID_MAP"] = uuids = { uuid: wi.id for uuid, wi in ctx["POLARION_WI_MAP"].items() @@ -144,23 +140,24 @@ def patch_work_items(ctx: dict[str, t.Any]) -> None: } back_links: dict[str, list[polarion_api.WorkItemLink]] = {} - for uuid in uuids: objects = ctx["MODEL"] if uuid.startswith("_"): objects = ctx["MODEL"].diagrams + obj = objects.by_uuid(uuid) + work_item: serialize.CapellaWorkItem = ctx["WORK_ITEMS"][uuid] + old_work_item: serialize.CapellaWorkItem = ctx["POLARION_WI_MAP"][uuid] links = element.create_links(obj, ctx) - work_item: serialize.CapellaWorkItem = ctx["WORK_ITEMS"][uuid] work_item.linked_work_items = links + work_item.id = old_work_item.id element.create_grouped_link_fields(work_item, back_links) for uuid in uuids: new_work_item: serialize.CapellaWorkItem = ctx["WORK_ITEMS"][uuid] - old_work_item: serialize.CapellaWorkItem = ctx["POLARION_WI_MAP"][uuid] - + old_work_item = ctx["POLARION_WI_MAP"][uuid] if old_work_item.id in back_links: element.create_grouped_back_link_fields( new_work_item, back_links[old_work_item.id] From f97401f9645b5f2f3511ac6b205bb2b8d27f105c Mon Sep 17 00:00:00 2001 From: ewuerger Date: Wed, 29 Nov 2023 11:46:23 +0100 Subject: [PATCH 11/15] docs: Fix `create_grouped_link_fields` docstring --- capella2polarion/elements/element.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/capella2polarion/elements/element.py b/capella2polarion/elements/element.py index b20481fe..7b6241f8 100644 --- a/capella2polarion/elements/element.py +++ b/capella2polarion/elements/element.py @@ -208,14 +208,15 @@ def create_grouped_link_fields( work_item: serialize.CapellaWorkItem, back_links: dict[str, list[polarion_api.WorkItemLink]] | None = None, ): - """Create the grouped link workitems fields from of the primary work items. + """Create the grouped link work items fields from the primary work item. Parameters ---------- work_item - WorkItem to create the fields for + WorkItem to create the fields for. back_links - A dictionary of secondary WorkItem IDs to links to create backlinks later + 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) From 3366d4603f89b354372bbdd1515a7a3c55a26576 Mon Sep 17 00:00:00 2001 From: micha91 Date: Wed, 29 Nov 2023 11:55:45 +0100 Subject: [PATCH 12/15] refactor: apply changes from review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Ernst Würger <50786483+ewuerger@users.noreply.github.com> --- capella2polarion/elements/element.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/capella2polarion/elements/element.py b/capella2polarion/elements/element.py index 7b6241f8..4175ecb5 100644 --- a/capella2polarion/elements/element.py +++ b/capella2polarion/elements/element.py @@ -225,9 +225,8 @@ def create_grouped_link_fields( ).items(): if back_links is not None: for link in grouped_links: - if link.secondary_work_item_id not in back_links: - back_links[link.secondary_work_item_id] = [] - back_links[link.secondary_work_item_id].append(link) + key = link.secondary_work_item_id + back_links.setdefault(key, []).append(link) _create_link_fields(work_item, role, grouped_links) From 125c3ea78d15e072c2abd224c2da414ae84fbcd0 Mon Sep 17 00:00:00 2001 From: Michael Harbarth Date: Wed, 29 Nov 2023 13:48:35 +0100 Subject: [PATCH 13/15] feat: Also add fields, if there are less than 2 links of a type --- capella2polarion/elements/element.py | 3 -- tests/test_elements.py | 58 ++++++++++++++++------------ 2 files changed, 33 insertions(+), 28 deletions(-) diff --git a/capella2polarion/elements/element.py b/capella2polarion/elements/element.py index 4175ecb5..58646feb 100644 --- a/capella2polarion/elements/element.py +++ b/capella2polarion/elements/element.py @@ -281,9 +281,6 @@ def _create_link_fields( links: list[polarion_api.WorkItemLink], reverse: bool = False, ): - # TODO check why we only create links for > 2 per role - if len(links) < 2: - return role = f"{role}_reverse" if reverse else role work_item.additional_attributes[role] = { "type": "text/html", diff --git a/tests/test_elements.py b/tests/test_elements.py index 4f225ac6..e27fc770 100644 --- a/tests/test_elements.py +++ b/tests/test_elements.py @@ -517,6 +517,18 @@ def test_patch_work_item_grouped_links( obj.uuid ].linked_work_items + mock_grouped_links = mock.MagicMock() + monkeypatch.setattr( + element, "create_grouped_link_fields", mock_grouped_links + ) + + mock_grouped_links_reverse = mock.MagicMock() + monkeypatch.setattr( + element, + "create_grouped_back_link_fields", + mock_grouped_links_reverse, + ) + context["MODEL"] = mock_model = mock.MagicMock() mock_model.by_uuid.side_effect = [ FakeModelObject(f"uuid{i}", name=f"Fake {i}") for i in range(3) @@ -525,27 +537,19 @@ def test_patch_work_item_grouped_links( elements.patch_work_items(context) update_work_item_calls = context["API"].update_work_item.call_args_list - assert len(update_work_item_calls) == 3 - work_item_0 = update_work_item_calls[0][0][0] - work_item_1 = update_work_item_calls[1][0][0] - work_item_2 = update_work_item_calls[2][0][0] + mock_grouped_links_calls = mock_grouped_links.call_args_list - assert ( - work_item_0.additional_attributes.pop("attribute")["value"] - == HTML_LINK_0["attribute"] - ) + assert len(mock_grouped_links_calls) == 3 - assert ( - work_item_1.additional_attributes.pop("attribute")["value"] - == HTML_LINK_1["attribute"] - ) + assert mock_grouped_links_calls[0][0][0] == dummy_work_items["uuid0"] + assert mock_grouped_links_calls[1][0][0] == dummy_work_items["uuid1"] + assert mock_grouped_links_calls[2][0][0] == dummy_work_items["uuid2"] - assert ( - work_item_2.additional_attributes.pop("attribute_reverse")["value"] - == HTML_LINK_2["attribute_reverse"] - ) + work_item_0 = update_work_item_calls[0][0][0] + work_item_1 = update_work_item_calls[1][0][0] + work_item_2 = update_work_item_calls[2][0][0] assert work_item_0.additional_attributes == {} assert work_item_1.additional_attributes == {} @@ -597,16 +601,20 @@ def test_maintain_reverse_grouped_links_attributes(dummy_work_items): del dummy_work_items["uuid0"].additional_attributes["attribute"] del dummy_work_items["uuid1"].additional_attributes["attribute"] - # TODO check if we really want to exclude groups of <2 - """ - assert dummy_work_items["uuid0"].additional_attributes.pop("attribute_reverse")[ - "value" - ] == HTML_LINK_0["attribute_reverse"] + assert ( + dummy_work_items["uuid0"].additional_attributes.pop( + "attribute_reverse" + )["value"] + == HTML_LINK_0["attribute_reverse"] + ) + + assert ( + dummy_work_items["uuid1"].additional_attributes.pop( + "attribute_reverse" + )["value"] + == HTML_LINK_1["attribute_reverse"] + ) - assert dummy_work_items["uuid1"].additional_attributes.pop("attribute_reverse")[ - "value" - ] == HTML_LINK_1["attribute_reverse"] - """ assert ( dummy_work_items["uuid2"].additional_attributes.pop( "attribute_reverse" From b9e24fd5d3c92f68145797d283f96810d498a989 Mon Sep 17 00:00:00 2001 From: Michael Harbarth Date: Wed, 29 Nov 2023 14:06:51 +0100 Subject: [PATCH 14/15] test: update test_patch_work_item_grouped_links test --- tests/test_elements.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/tests/test_elements.py b/tests/test_elements.py index e27fc770..5d969dec 100644 --- a/tests/test_elements.py +++ b/tests/test_elements.py @@ -517,10 +517,14 @@ def test_patch_work_item_grouped_links( obj.uuid ].linked_work_items + def mock_back_link(work_item, back_links): + back_links[work_item.id] = [] + mock_grouped_links = mock.MagicMock() monkeypatch.setattr( element, "create_grouped_link_fields", mock_grouped_links ) + mock_grouped_links.side_effect = mock_back_link mock_grouped_links_reverse = mock.MagicMock() monkeypatch.setattr( @@ -542,6 +546,7 @@ def test_patch_work_item_grouped_links( mock_grouped_links_calls = mock_grouped_links.call_args_list assert len(mock_grouped_links_calls) == 3 + assert mock_grouped_links_reverse.call_count == 3 assert mock_grouped_links_calls[0][0][0] == dummy_work_items["uuid0"] assert mock_grouped_links_calls[1][0][0] == dummy_work_items["uuid1"] From 996d17fc58248a48380e54a488d79ef2eb1ac44c Mon Sep 17 00:00:00 2001 From: Michael Harbarth Date: Wed, 29 Nov 2023 15:09:25 +0100 Subject: [PATCH 15/15] fix: the order of the links in grouped linked workitems should be stable --- capella2polarion/elements/element.py | 2 ++ tests/test_elements.py | 19 +++++++++++++++++++ 2 files changed, 21 insertions(+) diff --git a/capella2polarion/elements/element.py b/capella2polarion/elements/element.py index 58646feb..e533f8f1 100644 --- a/capella2polarion/elements/element.py +++ b/capella2polarion/elements/element.py @@ -271,6 +271,8 @@ def _make_url_list( url = serialize.POLARION_WORK_ITEM_URL.format(pid=pid) urls.append(f"
    • {url}
      • ") + + urls.sort() url_list = "\n".join(urls) return f"
        {url_list}
      " diff --git a/tests/test_elements.py b/tests/test_elements.py index 5d969dec..c9ef64e8 100644 --- a/tests/test_elements.py +++ b/tests/test_elements.py @@ -632,6 +632,25 @@ def test_maintain_reverse_grouped_links_attributes(dummy_work_items): assert dummy_work_items["uuid2"].additional_attributes == {} +def test_grouped_linked_work_items_order_consistency(): + work_item = serialize.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) + + check_sum = work_item.calculate_checksum() + + links = [ + polarion_api.WorkItemLink("prim2", "id", "role1"), + polarion_api.WorkItemLink("prim1", "id", "role1"), + ] + element.create_grouped_back_link_fields(work_item, links) + + assert check_sum == work_item.calculate_checksum() + + class TestHelpers: @staticmethod def test_resolve_element_type():