Merge pull request #2539 from haxtibal/tdmg/merge_source_nodes

feat(backend/sdoc_source_code): Support merging sdoc nodes with soure nodes

Author: stanislaw

strictdoc/core/file_traceability_index.py

@@ -17,6 +17,9 @@
 from strictdoc.backend.sdoc.document_reference import DocumentReference
 from strictdoc.backend.sdoc.error_handling import StrictDocSemanticError
 from strictdoc.backend.sdoc.models.document import SDocDocument
+from strictdoc.backend.sdoc.models.document_grammar import (
+    DocumentGrammar,
+)
 from strictdoc.backend.sdoc.models.model import SDocDocumentIF, SDocNodeIF
 from strictdoc.backend.sdoc.models.node import SDocNode
 from strictdoc.backend.sdoc.models.reference import FileEntry, FileReference
@@ -37,9 +40,10 @@
     RelationMarkerType,
     SourceFileTraceabilityInfo,
 )
+from strictdoc.backend.sdoc_source_code.models.source_node import SourceNode
 from strictdoc.core.constants import GraphLinkType
 from strictdoc.core.document_iterator import SDocDocumentIterator
-from strictdoc.core.project_config import ProjectConfig
+from strictdoc.core.project_config import ProjectConfig, SourceNodesEntry
 from strictdoc.core.source_tree import SourceFile
 from strictdoc.helpers.cast import assert_cast
 from strictdoc.helpers.exception import StrictDocException
@@ -592,9 +596,11 @@ def create_folder_section(
         documents_with_generated_content = set()
 
         section_cache = {}
-        source_nodes_config: List[Dict[str, str]] = project_config.source_nodes
+        source_nodes_config: List[SourceNodesEntry] = (
+            project_config.source_nodes
+        )
         unused_source_node_paths = {
-            config_entry_["path"] for config_entry_ in source_nodes_config
+            config_entry_.path for config_entry_ in source_nodes_config
         }
         for (
             path_to_source_file_,
@@ -606,16 +612,19 @@ def create_folder_section(
             if len(source_nodes_config) == 0:
                 continue
 
-            for config_entry_ in source_nodes_config:
-                config_entry_path = config_entry_["path"]
-                if path_to_source_file_.startswith(config_entry_path):
-                    relevant_source_node_entry = config_entry_
-                    unused_source_node_paths.discard(config_entry_path)
-                    break
+            relevant_source_node_entry = (
+                project_config.get_relevant_source_nodes_entry(
+                    path_to_source_file_
+                )
+            )
+            if relevant_source_node_entry is not None:
+                unused_source_node_paths.discard(
+                    relevant_source_node_entry.path
+                )
             else:
                 continue
 
-            document_uid = relevant_source_node_entry["uid"]
+            document_uid = relevant_source_node_entry.uid
             document = traceability_index.get_node_by_uid(document_uid)
             documents_with_generated_content.add(document)
 
@@ -638,95 +647,43 @@ def create_folder_section(
                 current_top_node = section_cache[path_component_]
 
             for source_node_ in traceability_info_.source_nodes:
-                source_sdoc_node = SDocNode(
-                    parent=document,
-                    node_type=relevant_source_node_entry["node_type"],
-                    fields=[],
-                    relations=[],
-                    # It is important that this autogenerated node is marked as such.
-                    autogen=True,
-                )
                 assert source_node_.entity_name is not None
                 source_sdoc_node_uid = f"{document_uid}/{path_to_source_file_}/{source_node_.entity_name}"
-                source_sdoc_node.ng_document_reference = DocumentReference()
-                source_sdoc_node.ng_document_reference.set_document(document)
-                source_sdoc_node.ng_including_document_reference = (
-                    DocumentReference()
-                )
-                source_sdoc_node.set_field_value(
-                    field_name="UID",
-                    form_field_index=0,
-                    value=source_sdoc_node_uid,
-                )
-                source_sdoc_node.set_field_value(
-                    field_name="TITLE",
-                    form_field_index=0,
-                    value=source_node_.entity_name,
-                )
 
-                for node_name_, node_value_ in source_node_.fields.items():
-                    source_sdoc_node.set_field_value(
-                        field_name=node_name_,
-                        form_field_index=0,
-                        value=node_value_,
+                source_sdoc_node = traceability_index.get_node_by_uid_weak(
+                    source_sdoc_node_uid
+                )
+                if source_sdoc_node is not None:
+                    source_sdoc_node = assert_cast(source_sdoc_node, SDocNode)
+                    self.merge_sdoc_node_with_source_node(
+                        source_sdoc_node,
+                        source_node_,
+                        document,
+                        relevant_source_node_entry,
+                    )
+                else:
+                    source_sdoc_node = self.create_sdoc_node_from_source_node(
+                        source_node_,
+                        source_sdoc_node_uid,
+                        document,
+                        relevant_source_node_entry,
+                    )
+                    current_top_node.section_contents.append(source_sdoc_node)
+                    traceability_index.graph_database.create_link(
+                        link_type=GraphLinkType.UID_TO_NODE,
+                        lhs_node=source_sdoc_node_uid,
+                        rhs_node=source_sdoc_node,
                     )
-                current_top_node.section_contents.append(source_sdoc_node)
 
-                traceability_index.graph_database.create_link(
-                    link_type=GraphLinkType.UID_TO_NODE,
-                    lhs_node=source_sdoc_node_uid,
-                    rhs_node=source_sdoc_node,
+                self.connect_source_node_function(
+                    source_node_, source_sdoc_node_uid, traceability_info_
                 )
-
-                source_node_function = source_node_.function
-                assert source_node_function is not None
-
-                function_marker = self.forward_function_marker_from_function(
-                    function=source_node_function,
-                    marker_type=RangeMarkerType.FUNCTION,
-                    reqs=[Req(None, source_sdoc_node_uid)],
-                    role=None,
-                    description=f"function {source_node_function.display_name}()",
+                self.connect_sdoc_node_with_file_path(
+                    source_sdoc_node, path_to_source_file_
+                )
+                self.connect_source_node_requirements(
+                    source_node_, source_sdoc_node, traceability_index
                 )
-
-                traceability_info_.ng_map_reqs_to_markers.setdefault(
-                    source_sdoc_node_uid, []
-                ).append(function_marker)
-
-                self.map_reqs_uids_to_paths.setdefault(
-                    source_sdoc_node_uid, OrderedSet()
-                ).add(path_to_source_file_)
-
-                self.map_paths_to_reqs.setdefault(
-                    path_to_source_file_, OrderedSet()
-                ).add(source_sdoc_node)
-
-                function_marker_copy = function_marker.create_end_marker()
-
-                traceability_info_.markers.append(function_marker)
-                traceability_info_.markers.append(function_marker_copy)
-
-                #
-                # This connects:
-                # - Source nodes and auto-generated requirements.
-                # - Source nodes-related requirements and auto-generated requirements.
-                #
-                for marker_ in source_node_.markers:
-                    if not isinstance(marker_, FunctionRangeMarker):
-                        continue
-
-                    for req_ in marker_.reqs:
-                        node = traceability_index.get_node_by_uid_weak2(req_)
-                        traceability_index.graph_database.create_link(
-                            link_type=GraphLinkType.NODE_TO_PARENT_NODES,
-                            lhs_node=source_sdoc_node,
-                            rhs_node=node,
-                        )
-                        traceability_index.graph_database.create_link(
-                            link_type=GraphLinkType.NODE_TO_CHILD_NODES,
-                            lhs_node=node,
-                            rhs_node=source_sdoc_node,
-                        )
 
         # Warn if source_node was not matched by any include_source_paths, it indicates misconfiguration
         for unused_source_node_path in unused_source_node_paths:
@@ -990,3 +947,160 @@ def compare_sdocnode_by_uid(node_: SDocNode) -> str:
                 return assert_cast(node_.reserved_uid, str)
 
             path_requirements_.sort(key=compare_sdocnode_by_uid)
+
+    def connect_source_node_function(
+        self,
+        source_node: SourceNode,
+        source_sdoc_node_uid: str,
+        traceability_info: SourceFileTraceabilityInfo,
+    ) -> None:
+        source_node_function = source_node.function
+        assert source_node_function is not None
+
+        function_marker = self.forward_function_marker_from_function(
+            function=source_node_function,
+            marker_type=RangeMarkerType.FUNCTION,
+            reqs=[Req(None, source_sdoc_node_uid)],
+            role=None,
+            description=f"function {source_node_function.display_name}()",
+        )
+
+        traceability_info.ng_map_reqs_to_markers.setdefault(
+            source_sdoc_node_uid, []
+        ).append(function_marker)
+        function_marker_copy = function_marker.create_end_marker()
+        traceability_info.markers.append(function_marker)
+        traceability_info.markers.append(function_marker_copy)
+
+    @staticmethod
+    def create_sdoc_node_from_source_node(
+        source_node: SourceNode,
+        uid: str,
+        parent_document: SDocDocumentIF,
+        relevant_source_node_entry: SourceNodesEntry,
+    ) -> SDocNode:
+        source_sdoc_node = SDocNode(
+            parent=parent_document,
+            node_type=relevant_source_node_entry.node_type,
+            fields=[],
+            relations=[],
+            # It is important that this autogenerated node is marked as such.
+            autogen=True,
+        )
+        source_sdoc_node.ng_document_reference = DocumentReference()
+        source_sdoc_node.ng_document_reference.set_document(parent_document)
+        source_sdoc_node.ng_including_document_reference = DocumentReference()
+        source_sdoc_node.set_field_value(
+            field_name="UID",
+            form_field_index=0,
+            value=uid,
+        )
+        source_sdoc_node.set_field_value(
+            field_name="TITLE",
+            form_field_index=0,
+            value=source_node.entity_name,
+        )
+        for node_name_, node_value_ in source_node.fields.items():
+            source_sdoc_node.set_field_value(
+                field_name=node_name_,
+                form_field_index=0,
+                value=node_value_,
+            )
+        return source_sdoc_node
+
+    @staticmethod
+    def merge_sdoc_node_with_source_node(
+        sdoc_node: SDocNode,
+        source_node: SourceNode,
+        parent_document: SDocDocumentIF,
+        source_node_config_entry: SourceNodesEntry,
+    ) -> None:
+        # First check if grammar element definitions are compatible.
+        source_node_type = source_node_config_entry.node_type
+        source_node_grammar = assert_cast(
+            parent_document.grammar, DocumentGrammar
+        )
+        source_node_grammar_element = source_node_grammar.elements_by_type[
+            source_node_type
+        ]
+        sdoc_node_document = assert_cast(
+            sdoc_node.get_document(), SDocDocumentIF
+        )
+        sdoc_node_grammar = assert_cast(
+            sdoc_node_document.grammar, DocumentGrammar
+        )
+        sdoc_node_grammar_element = sdoc_node_grammar.elements_by_type[
+            source_node_type
+        ]
+        if source_node_grammar_element != sdoc_node_grammar_element:
+            raise StrictDocException(
+                f"Can't merge node {sdoc_node.reserved_uid} with source portion: "
+                f"Grammar element {sdoc_node_document.reserved_uid}::{source_node_type} "
+                f"incompatible with {parent_document.reserved_uid}::{source_node_type}"
+            )
+        # Merge strategy: overwrite title if there's a TITLE from custom tags.
+        if "TITLE" in source_node.fields:
+            sdoc_node.set_field_value(
+                field_name="TITLE",
+                form_field_index=0,
+                value=source_node.fields["TITLE"],
+            )
+        # Merge strategy: overwrite any field if there's a field with same name from custom tags.
+        for node_name_, node_value_ in source_node.fields.items():
+            sdoc_node.set_field_value(
+                field_name=node_name_,
+                form_field_index=0,
+                value=node_value_,
+            )
+
+    def connect_sdoc_node_with_file_path(
+        self, sdoc_node: SDocNode, path_to_source_file_: str
+    ) -> None:
+        uid = sdoc_node.reserved_uid
+        assert uid is not None
+        self.map_reqs_uids_to_paths.setdefault(uid, OrderedSet()).add(
+            path_to_source_file_
+        )
+        self.map_paths_to_reqs.setdefault(
+            path_to_source_file_, OrderedSet()
+        ).add(sdoc_node)
+
+    @staticmethod
+    def connect_source_node_requirements(
+        source_node: SourceNode,
+        sdoc_node: SDocNode,
+        traceability_index: "TraceabilityIndex",
+    ) -> None:
+        """
+        Connect auto-generated requirement with function marker and with marker target requirement.
+
+        If function comment has @relation(REQ, scope=function), connections shall become
+        [REQ] <-parent- [auto-generated/merged sdoc_node] -file-> [function marker]
+
+        Here we link REQ and sdoc_node bidirectional.
+        """
+        for marker_ in source_node.markers:
+            if not isinstance(marker_, FunctionRangeMarker):
+                continue
+            for req_ in marker_.reqs:
+                node = traceability_index.get_node_by_uid_weak2(req_)
+                if not traceability_index.graph_database.has_link(
+                    link_type=GraphLinkType.NODE_TO_PARENT_NODES,
+                    lhs_node=sdoc_node,
+                    rhs_node=node,
+                ):
+                    traceability_index.graph_database.create_link(
+                        link_type=GraphLinkType.NODE_TO_PARENT_NODES,
+                        lhs_node=sdoc_node,
+                        rhs_node=node,
+                    )
+                if not traceability_index.graph_database.has_link(
+                    link_type=GraphLinkType.NODE_TO_CHILD_NODES,
+                    lhs_node=node,
+                    rhs_node=sdoc_node,
+                ):
+                    traceability_index.graph_database.create_link(
+                        link_type=GraphLinkType.NODE_TO_CHILD_NODES,
+                        lhs_node=node,
+                        rhs_node=sdoc_node,
+                    )

strictdoc/core/graph/abstract_bucket.py

@@ -8,7 +8,13 @@
 
 class AbstractBucket(ABC):
     @abstractmethod
-    def has_link(self, *, lhs_node: Any) -> bool:
+    def has_any_link(self, *, lhs_node: Any) -> bool:
+        raise NotImplementedError
+
+    @abstractmethod
+    def has_link(
+        self, *, lhs_node: Any, rhs_node: Any, edge: Optional[str] = None
+    ) -> bool:
         raise NotImplementedError
 
     @abstractmethod

strictdoc/core/graph/many_to_many_set.py

@@ -14,10 +14,16 @@ def __init__(self, lhs_type: Type[Any], rhs_type: Type[Any]) -> None:
         self._lhs_type: type = lhs_type
         self._rhs_type: type = rhs_type
 
-    def has_link(self, *, lhs_node: Any) -> bool:
+    def has_any_link(self, *, lhs_node: Any) -> bool:
         assert isinstance(lhs_node, self._lhs_type), lhs_node
         return lhs_node in self._links and len(self._links[lhs_node]) > 0
 
+    def has_link(
+        self, *, lhs_node: Any, rhs_node: Any, edge: Optional[str] = None
+    ) -> bool:
+        assert isinstance(rhs_node, self._rhs_type), rhs_node
+        return rhs_node in self.get_link_values(lhs_node=lhs_node, edge=edge)
+
     def get_count(self, edge: Optional[str] = None) -> int:
         total_count = 0
         for _, lhs_node_links in self._links.items():