merge: Merge pull request #47 from DSD-DBS/fix-interface-filters

Fix interface filters

capellambse_context_diagrams/collectors/exchanges.py

@@ -4,12 +4,10 @@
 from __future__ import annotations
 
 import abc
-import collections.abc as cabc
 import logging
 import operator
 import typing as t
 
-from capellambse import helpers
 from capellambse.model import common
 from capellambse.model.modeltypes import DiagramType as DT
 
@@ -49,10 +47,13 @@ def __init__(
         diagram: context.InterfaceContextDiagram
         | context.FunctionalContextDiagram,
         data: _elkjs.ELKInputData,
+        params: dict[str, t.Any],
     ) -> None:
         self.diagram = diagram
         self.data: _elkjs.ELKInputData = data
         self.obj = self.diagram.target
+        self.params = params
+
         src, trg, alloc_fex, fncs = self.intermap[diagram.type]
         self.get_source = operator.attrgetter(src)
         self.get_target = operator.attrgetter(trg)
@@ -123,19 +124,21 @@ def make_ports_and_update_children_size(
         data["height"] = stack_height
 
     @abc.abstractmethod
-    def collect(self) -> cabc.MutableSequence[_elkjs.ELKInputEdge]:
-        return NotImplemented
+    def collect(self) -> None:
+        """Populate the elkdata container."""
+        raise NotImplementedError
 
 
 def get_elkdata_for_exchanges(
     diagram: context.InterfaceContextDiagram
     | context.FunctionalContextDiagram,
     collector_type: type[ExchangeCollector],
+    params: dict[str, t.Any],
 ) -> _elkjs.ELKInputData:
     """Return exchange data for ELK."""
     data = makers.make_diagram(diagram)
-    collector = collector_type(diagram, data)
-    data["edges"] = collector.collect()
+    collector = collector_type(diagram, data, params)
+    collector.collect()
     for comp in data["children"]:
         collector.make_ports_and_update_children_size(comp, data["edges"])
 
@@ -159,8 +162,9 @@ def __init__(
         self,
         diagram: context.InterfaceContextDiagram,
         data: _elkjs.ELKInputData,
+        params: dict[str, t.Any],
     ) -> None:
-        super().__init__(diagram, data)
+        super().__init__(diagram, data, params)
         self.get_left_and_right()
 
     def get_left_and_right(self) -> None:
@@ -223,46 +227,42 @@ def make_boxes(
         except AttributeError:
             pass
 
-    def collect(self) -> cabc.MutableSequence[_elkjs.ELKInputEdge]:
+    def collect(self) -> None:
         """Return all allocated `FunctionalExchange`s in the context."""
-        functional_exchanges: list[_elkjs.ELKInputEdge] = []
         try:
             for ex in self.incoming_edges + self.outgoing_edges:
-                try:
-                    src, tgt = generic.collect_exchange_endpoints(ex)
-                except AttributeError:
-                    continue
-
-                width, height = helpers.extent_func(ex.name)
-                swap = ex in self.incoming_edges
-                functional_exchanges.append(
-                    _elkjs.ELKInputEdge(
-                        id=ex.uuid,
-                        sources=[tgt.uuid] if swap else [src.uuid],
-                        targets=[src.uuid] if swap else [tgt.uuid],
-                        labels=[
-                            _elkjs.ELKInputLabel(
-                                text=ex.name,
-                                width=width + 2 * makers.LABEL_HPAD,
-                                height=height + 2 * makers.LABEL_VPAD,
-                            )
-                        ],
-                    )
+                ex_data = generic.ExchangeData(
+                    ex,
+                    self.data,
+                    self.diagram.filters,
+                    self.params,
+                    is_hierarchical=False,
                 )
+                src, tgt = generic.exchange_data_collector(ex_data)
+
+                if ex in self.incoming_edges:
+                    self.data["edges"][-1]["sources"] = [tgt.uuid]
+                    self.data["edges"][-1]["targets"] = [src.uuid]
 
-            if not functional_exchanges:
+            if not self.data["edges"]:
                 logger.warning(
                     "There are no FunctionalExchanges allocated to '%s'.",
                     self.obj.name,
                 )
         except AttributeError:
             pass
 
-        return functional_exchanges
-
 
 class FunctionalContextCollector(ExchangeCollector):
-    def collect(self) -> cabc.MutableSequence[_elkjs.ELKInputEdge]:
+    def __init__(
+        self,
+        diagram: context.InterfaceContextDiagram,
+        data: _elkjs.ELKInputData,
+        params: dict[str, t.Any],
+    ) -> None:
+        super().__init__(diagram, data, params)
+
+    def collect(self) -> None:
         functional_exchanges: list[common.GenericElement] = []
         all_functions: list[common.GenericElement] = []
         made_children: set[str] = {self.obj.uuid}
@@ -295,11 +295,11 @@ def collect(self) -> cabc.MutableSequence[_elkjs.ELKInputEdge]:
 
         for ex in functional_exchanges:
             generic.exchange_data_collector(
-                generic.ExchangeData(ex, self.data, set())
+                generic.ExchangeData(
+                    ex, self.data, set(), is_hierarchical=False
+                )
             )
 
-        return self.data["edges"]
-
 
 def is_hierarchical(
     ex: common.GenericElement,

capellambse_context_diagrams/context.py

@@ -312,7 +312,7 @@ def name(self) -> str:  # type: ignore
 
     def _create_diagram(self, params: dict[str, t.Any]) -> cdiagram.Diagram:
         params["elkdata"] = exchanges.get_elkdata_for_exchanges(
-            self, exchanges.InterfaceContextCollector
+            self, exchanges.InterfaceContextCollector, params
         )
         return super()._create_diagram(params)
 
@@ -328,6 +328,6 @@ def name(self) -> str:  # type: ignore
 
     def _create_diagram(self, params: dict[str, t.Any]) -> cdiagram.Diagram:
         params["elkdata"] = exchanges.get_elkdata_for_exchanges(
-            self, exchanges.FunctionalContextCollector
+            self, exchanges.FunctionalContextCollector, params
         )
         return super()._create_diagram(params)

tests/test_context_diagrams.py

@@ -1,8 +1,6 @@
 # SPDX-FileCopyrightText: 2022 Copyright DB Netz AG and the capellambse-context-diagrams contributors
 # SPDX-License-Identifier: Apache-2.0
 
-import pathlib
-
 import capellambse
 import pytest
 
@@ -45,14 +43,10 @@
         ),
     ],
 )
-def test_context_diagrams(
-    model: capellambse.MelodyModel, uuid: str, tmp_path: pathlib.Path
-) -> None:
+def test_context_diagrams(model: capellambse.MelodyModel, uuid: str) -> None:
     obj = model.by_uuid(uuid)
-    filename = tmp_path / "tmp.svg"
 
     diag = obj.context_diagram
-    diag.render("svgdiagram").save_drawing(filename=filename)
 
     assert diag.nodes
 

tests/test_filters.py

@@ -3,9 +3,7 @@
 
 from __future__ import annotations
 
-import os
 import re
-import sys
 import typing as t
 
 import pytest
@@ -14,10 +12,17 @@
 
 from capellambse_context_diagrams import context, filters
 
-from .conftest import SYSTEM_ANALYSIS_PARAMS
+# pylint: disable-next=relative-beyond-top-level, useless-suppression
+from .conftest import SYSTEM_ANALYSIS_PARAMS  # type: ignore[import]
 
 EX_PTRN = re.compile(r"\[(.*?)\]")
 CAP_EXPLOIT = "4513c8cd-b94b-4bde-bd00-4c18aaf600ff"
+FNC_UUID = "a5642060-c9cc-4d49-af09-defaa3024bae"
+INTERF_UUID = "9cbdd233-aff5-47dd-9bef-9be1277c77c3"
+
+Types = list[
+    t.Union[crosslayer.fa.FunctionalExchange, crosslayer.fa.ComponentExchange]
+]
 
 
 @pytest.mark.parametrize(
@@ -42,16 +47,22 @@ def test_uuid_filter(model: MelodyModel, label: str, expected: str) -> None:
 
 
 def start_filter_apply_test(
-    model: MelodyModel, filter_name: str, **render_params: t.Any
-) -> tuple[list[crosslayer.fa.FunctionalExchange], diagram.Diagram]:
+    model: MelodyModel, uuid: str, filter_name: str, **render_params: t.Any
+) -> tuple[Types, diagram.Diagram]:
     """StartUp for every filter test case."""
-    obj = model.by_uuid("a5642060-c9cc-4d49-af09-defaa3024bae")
+    obj = model.by_uuid(uuid)
     diag: context.ContextDiagram = obj.context_diagram
     diag.filters.add(filter_name)
     edges = [
         elt
         for elt in diag.nodes
-        if isinstance(elt, crosslayer.fa.FunctionalExchange)
+        if isinstance(
+            elt,
+            (
+                crosslayer.fa.FunctionalExchange,
+                crosslayer.fa.ComponentExchange,
+            ),
+        )
     ]
     return edges, diag.render(None, **render_params)
 
@@ -68,8 +79,9 @@ def has_sorted_ExchangeItems(edge: diagram.Edge) -> bool:
     return exitems == sorted(exitems)
 
 
-def test_EX_ITEMS_is_applied(model: MelodyModel) -> None:
-    edges, aird_diag = start_filter_apply_test(model, filters.EX_ITEMS)
+@pytest.mark.parametrize("uuid", [FNC_UUID, INTERF_UUID])
+def test_EX_ITEMS_is_applied(model: MelodyModel, uuid: str) -> None:
+    edges, aird_diag = start_filter_apply_test(model, uuid, filters.EX_ITEMS)
 
     for edge in edges:
         aedge = aird_diag[edge.uuid]
@@ -80,12 +92,12 @@ def test_EX_ITEMS_is_applied(model: MelodyModel) -> None:
             assert get_ExchangeItems(aedge) == list(expected)
 
 
-@pytest.mark.parametrize("sort", [False, True])
+@pytest.mark.parametrize("sort,uuid", [(False, FNC_UUID), (True, INTERF_UUID)])
 def test_context_diagrams_ExchangeItems_sorting(
-    model: MelodyModel, sort: bool
+    model: MelodyModel, sort: bool, uuid: str
 ) -> None:
     edges, aird_diag = start_filter_apply_test(
-        model, filters.EX_ITEMS, sorted_exchangedItems=sort
+        model, uuid, filters.EX_ITEMS, sorted_exchangedItems=sort
     )
 
     all_sorted = True
@@ -99,10 +111,13 @@ def test_context_diagrams_ExchangeItems_sorting(
     assert all_sorted == sort
 
 
+@pytest.mark.parametrize("uuid", (FNC_UUID, INTERF_UUID))
 def test_context_diagrams_FEX_EX_ITEMS_is_applied(
-    model: MelodyModel,
+    model: MelodyModel, uuid: str
 ) -> None:
-    edges, aird_diag = start_filter_apply_test(model, filters.FEX_EX_ITEMS)
+    edges, aird_diag = start_filter_apply_test(
+        model, uuid, filters.FEX_EX_ITEMS
+    )
 
     for edge in edges:
         aedge = aird_diag[edge.uuid]
@@ -117,10 +132,13 @@ def test_context_diagrams_FEX_EX_ITEMS_is_applied(
         assert aedge.labels[0].label == expected_label
 
 
+@pytest.mark.parametrize("uuid", (FNC_UUID, INTERF_UUID))
 def test_context_diagrams_FEX_OR_EX_ITEMS_is_applied(
-    model: MelodyModel,
+    model: MelodyModel, uuid: str
 ) -> None:
-    edges, aird_diag = start_filter_apply_test(model, filters.FEX_OR_EX_ITEMS)
+    edges, aird_diag = start_filter_apply_test(
+        model, uuid, filters.FEX_OR_EX_ITEMS
+    )
 
     for edge in edges:
         aedge = aird_diag[edge.uuid]

tests/test_interface_diagrams.py

@@ -1,8 +1,6 @@
 # SPDX-FileCopyrightText: 2022 Copyright DB Netz AG and the capellambse-context-diagrams contributors
 # SPDX-License-Identifier: Apache-2.0
 
-import pathlib
-
 import capellambse
 import pytest
 
@@ -16,12 +14,10 @@
     ],
 )
 def test_interface_diagrams_get_rendered(
-    model: capellambse.MelodyModel, uuid: str, tmp_path: pathlib.Path
+    model: capellambse.MelodyModel, uuid: str
 ) -> None:
     obj = model.by_uuid(uuid)
-    filename = tmp_path / "tmp.svg"
 
     diag = obj.context_diagram
-    diag.render("svgdiagram").save_drawing(filename=filename)
 
     assert diag.nodes