[review] rewrite parse_ranges

Signed-off-by: Meret Behrens <meret.behrens@tngtech.com>
spdx · meretp · Feb 22, 2023 · Feb 1, 2023 · Feb 1, 2023 · Feb 9, 2023
commit 6b70dda4724c49a753003f40c22a344ea1a6700d
diff --git a/src/spdx/parser/rdf/snippet_parser.py b/src/spdx/parser/rdf/snippet_parser.py
@@ -8,16 +8,18 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-from typing import Tuple, Optional
+from typing import Tuple, Optional, Dict
 
 from rdflib import Graph, RDF, RDFS
+from rdflib.exceptions import UniquenessError
 from rdflib.term import URIRef, Node
+from spdx.parser.error import SPDXParsingError
 
 from spdx.model.snippet import Snippet
 from spdx.parser.logger import Logger
 from spdx.parser.parsing_functions import construct_or_raise_parsing_error, raise_parsing_error_if_logger_has_messages
 from spdx.parser.rdf.graph_parsing_functions import parse_literal, parse_spdx_id, parse_literal_or_no_assertion_or_none, \
-    get_correct_typed_value
+    get_correct_typed_value, apply_parsing_method_or_log_error
 from spdx.parser.rdf.license_expression_parser import parse_license_expression
 from spdx.rdfschema.namespace import SPDX_NAMESPACE, POINTER_NAMESPACE
 
@@ -27,8 +29,13 @@ def parse_snippet(snippet_node: URIRef, graph: Graph, doc_namespace: str) -> Sni
     spdx_id = parse_spdx_id(snippet_node, doc_namespace, graph)
     file_spdx_id_uri = graph.value(subject=snippet_node, predicate=SPDX_NAMESPACE.snippetFromFile)
     file_spdx_id = parse_spdx_id(file_spdx_id_uri, doc_namespace, graph)
-    byte_range = parse_ranges(snippet_node, graph, POINTER_NAMESPACE.ByteOffsetPointer, POINTER_NAMESPACE.offset)
-    line_range = parse_ranges(snippet_node, graph, POINTER_NAMESPACE.LineCharPointer, POINTER_NAMESPACE.lineNumber)
+    byte_range = None
+    line_range = None
+    for (_, _, start_end_pointer) in graph.triples((snippet_node, SPDX_NAMESPACE.range, None)):
+        parsed_range = apply_parsing_method_or_log_error(logger, start_end_pointer,
+                                                         parsing_method=lambda x: parse_ranges(x, graph))
+        byte_range, line_range = set_range_or_log_error(byte_range, line_range, logger, parsed_range)
+
     license_concluded = parse_literal_or_no_assertion_or_none(
         logger, graph, snippet_node, SPDX_NAMESPACE.licenseConcluded,
         parsing_method=lambda x: parse_license_expression(x, graph, doc_namespace))
@@ -57,21 +64,58 @@ def parse_snippet(snippet_node: URIRef, graph: Graph, doc_namespace: str) -> Sni
     return snippet
 
 
-def parse_ranges(snippet_node: URIRef, graph: Graph, pointer: Node, member: Node) -> Tuple[int, int]:
-    start_range = None
-    end_range = None
-    for (_, _, start_end_pointer) in graph.triples((snippet_node, SPDX_NAMESPACE.range, None)):
-        for (_, _, pointer_node) in graph.triples((start_end_pointer, POINTER_NAMESPACE.startPointer, None)):
-            for (typed_pointer_node, _, _) in graph.triples((pointer_node, RDF.type, pointer)):
-                start_range = parse_range_value(graph, typed_pointer_node, member)
-        for (_, _, pointer_node) in graph.triples((start_end_pointer, POINTER_NAMESPACE.endPointer, None)):
-            for (typed_pointer_node, _, _) in graph.triples((pointer_node, RDF.type, pointer)):
-                end_range = parse_range_value(graph, typed_pointer_node, member)
-    return start_range, end_range
+def set_range_or_log_error(
+    byte_range: Optional[Tuple[int, int]], line_range: Optional[Tuple[int, int]], logger: Logger,
+    parsed_range: Dict[str, Tuple[int, int]]) -> Tuple[Optional[Tuple[int, int]], Optional[Tuple[int, int]]]:
+    if not parsed_range:
+        return byte_range, line_range
+    if "ByteOffsetPointer" in parsed_range.keys() and not byte_range:
+        byte_range = parsed_range["ByteOffsetPointer"]
+    elif "ByteOffsetPointer" in parsed_range.keys() and byte_range:
+        logger.append("Multiple ByteOffsetPointer found.")
+    elif "LineCharPointer" in parsed_range.keys() and not line_range:
+        line_range = parsed_range["LineCharPointer"]
+    elif "LineCharPointer" in parsed_range.keys() and line_range:
+        logger.append("Multiple LineCharPointer found.")
+    return byte_range, line_range
+
+
+def parse_ranges(start_end_pointer: URIRef, graph: Graph) -> Dict[str, Tuple[int, int]]:
+    range_values = dict()
+    start_pointer_type, start_pointer_node = get_pointer_type(graph, POINTER_NAMESPACE.startPointer, start_end_pointer)
+    end_pointer_type, end_pointer_node = get_pointer_type(graph, POINTER_NAMESPACE.endPointer, start_end_pointer)
+
+    if start_pointer_type != end_pointer_type:
+        raise SPDXParsingError([f"Types of startPointer and endPointer don't match"])
+
+    range_values["startPointer"] = parse_range_value(graph, start_pointer_node, POINTER_MATCHING[start_pointer_type])
+    range_values["endPointer"] = parse_range_value(graph, end_pointer_node, POINTER_MATCHING[end_pointer_type])
+
+    return {str(start_pointer_type.fragment): (
+        range_values["startPointer"], range_values["endPointer"])}
+
+
+def get_pointer_type(graph: Graph, pointer: URIRef, start_end_pointer: URIRef) -> Tuple[URIRef, URIRef]:
+    try:
+        pointer_node = graph.value(start_end_pointer, pointer, any=False)
+    except UniquenessError:
+        raise SPDXParsingError([f"Multiple values for {pointer.fragment}"])
+    if not pointer_node:
+        raise SPDXParsingError([f"Couldn't find pointer of type {pointer.fragment}."])
+    pointer_type = graph.value(pointer_node, RDF.type)
+    return pointer_type, pointer_node
+
+
+POINTER_MATCHING = {
+    POINTER_NAMESPACE.ByteOffsetPointer: POINTER_NAMESPACE.offset,
+    POINTER_NAMESPACE.LineCharPointer: POINTER_NAMESPACE.lineNumber}
 
 
-def parse_range_value(graph: Graph, pointer_node: Node, predicate: Node) -> Optional[int]:
-    value = graph.value(pointer_node, predicate)
+def parse_range_value(graph: Graph, pointer_node: Node, predicate: URIRef) -> Optional[int]:
+    try:
+        value = graph.value(pointer_node, predicate, any=False)
+    except UniquenessError:
+        raise SPDXParsingError([f"Multiple values for {predicate.fragment} found."])
     if value:
         value = int(value)
     return value
diff --git a/tests/spdx/parser/rdf/test_snippet_parser.py b/tests/spdx/parser/rdf/test_snippet_parser.py
@@ -11,11 +11,13 @@
 import os
 from unittest import TestCase
 
+import pytest
 from license_expression import get_spdx_licensing
-from rdflib import Graph, RDF
+from rdflib import Graph, RDF, BNode, Literal
 
-from spdx.parser.rdf.snippet_parser import parse_snippet
-from spdx.rdfschema.namespace import SPDX_NAMESPACE
+from spdx.parser.error import SPDXParsingError
+from spdx.parser.rdf.snippet_parser import parse_snippet, parse_ranges
+from spdx.rdfschema.namespace import SPDX_NAMESPACE, POINTER_NAMESPACE
 
 
 def test_parse_snippet():
@@ -37,3 +39,86 @@ def test_parse_snippet():
     assert snippet.comment == "snippetComment"
     assert snippet.name == "snippetName"
     assert snippet.attribution_texts == ["snippetAttributionText"]
+
+
+@pytest.mark.parametrize("predicate_value_class_member",
+                         [([(POINTER_NAMESPACE.startPointer, 1, POINTER_NAMESPACE.ByteOffsetPointer,
+                             POINTER_NAMESPACE.offset),
+                            (POINTER_NAMESPACE.endPointer, 2, POINTER_NAMESPACE.ByteOffsetPointer,
+                             POINTER_NAMESPACE.offset)]),
+                          ([(POINTER_NAMESPACE.startPointer, 100, POINTER_NAMESPACE.LineCharPointer,
+                             POINTER_NAMESPACE.lineNumber),
+                            (POINTER_NAMESPACE.endPointer, 200, POINTER_NAMESPACE.LineCharPointer,
+                             POINTER_NAMESPACE.lineNumber)])
+                          ])
+def test_parse_ranges(predicate_value_class_member):
+    graph = Graph()
+    pointer_class = predicate_value_class_member[0][2]
+
+    add_range_to_graph_helper(graph, predicate_value_class_member)
+
+    range_dict = parse_ranges(graph.value(predicate=RDF.type, object=POINTER_NAMESPACE.StartEndPointer), graph)
+
+    assert pointer_class.fragment in range_dict.keys()
+    assert range_dict[pointer_class.fragment][0] == predicate_value_class_member[0][1]
+    assert range_dict[pointer_class.fragment][1] == predicate_value_class_member[1][1]
+
+
+@pytest.mark.parametrize("predicate_value_class_member",
+                         [([(POINTER_NAMESPACE.startPointer, 1, POINTER_NAMESPACE.ByteOffsetPointer,
+                             POINTER_NAMESPACE.lineNumber),
+                            (POINTER_NAMESPACE.endPointer, 2, POINTER_NAMESPACE.ByteOffsetPointer,
+                             POINTER_NAMESPACE.lineNumber)]),
+                          ([(POINTER_NAMESPACE.startPointer, 100, POINTER_NAMESPACE.LineCharPointer,
+                             POINTER_NAMESPACE.offset),
+                            (POINTER_NAMESPACE.endPointer, 200, POINTER_NAMESPACE.LineCharPointer,
+                             POINTER_NAMESPACE.offset)])
+                          ])
+def test_parse_ranges_wrong_pair_of_pointer_classes(predicate_value_class_member):
+    graph = Graph()
+    pointer_class = predicate_value_class_member[0][2]
+
+    add_range_to_graph_helper(graph, predicate_value_class_member)
+
+    range_dict = parse_ranges(graph.value(predicate=RDF.type, object=POINTER_NAMESPACE.StartEndPointer), graph)
+
+    assert pointer_class.fragment in range_dict.keys()
+    assert range_dict[pointer_class.fragment][0] is None
+    assert range_dict[pointer_class.fragment][1] is None
+
+
+@pytest.mark.parametrize(
+    "predicate_value_class_member,expected_message",
+    [([(POINTER_NAMESPACE.endPointer, 1, POINTER_NAMESPACE.ByteOffsetPointer, POINTER_NAMESPACE.offset),
+       (POINTER_NAMESPACE.endPointer, 2, POINTER_NAMESPACE.ByteOffsetPointer, POINTER_NAMESPACE.offset)],
+      "Couldn't find pointer of type startPointer."),
+     ([(POINTER_NAMESPACE.startPointer, 1, POINTER_NAMESPACE.ByteOffsetPointer, POINTER_NAMESPACE.offset)],
+      "Couldn't find pointer of type endPointer."),
+     ([(POINTER_NAMESPACE.startPointer, 1, POINTER_NAMESPACE.ByteOffsetPointer, POINTER_NAMESPACE.offset),
+       (POINTER_NAMESPACE.endPointer, 2, POINTER_NAMESPACE.ByteOffsetPointer, POINTER_NAMESPACE.offset),
+       (POINTER_NAMESPACE.endPointer, 3, POINTER_NAMESPACE.ByteOffsetPointer, POINTER_NAMESPACE.offset)],
+      "Multiple values for endPointer."),
+     ([(POINTER_NAMESPACE.startPointer, 100, POINTER_NAMESPACE.LineCharPointer, POINTER_NAMESPACE.lineNumber),
+       (POINTER_NAMESPACE.startPointer, 200, POINTER_NAMESPACE.LineCharPointer, POINTER_NAMESPACE.lineNumber)],
+      "Multiple values for startPointer"),
+     ([(POINTER_NAMESPACE.startPointer, 100, POINTER_NAMESPACE.LineCharPointer, POINTER_NAMESPACE.lineNumber),
+       (POINTER_NAMESPACE.endPointer, 200, POINTER_NAMESPACE.ByteOffsetPointer, POINTER_NAMESPACE.offset)],
+      f"Types of startPointer and endPointer don't match")
+     ])
+def test_parse_ranges_error(predicate_value_class_member, expected_message):
+    graph = Graph()
+
+    add_range_to_graph_helper(graph, predicate_value_class_member)
+
+    with pytest.raises(SPDXParsingError, match=expected_message):
+        parse_ranges(graph.value(predicate=RDF.type, object=POINTER_NAMESPACE.StartEndPointer), graph)
+
+
+def add_range_to_graph_helper(graph, predicate_value_class_member):
+    start_end_pointer = BNode()
+    graph.add((start_end_pointer, RDF.type, POINTER_NAMESPACE.StartEndPointer))
+    for (predicate, value, pointer_class, pointer_member) in predicate_value_class_member:
+        pointer_node = BNode()
+        graph.add((pointer_node, RDF.type, pointer_class))
+        graph.add((start_end_pointer, predicate, pointer_node))
+        graph.add((pointer_node, pointer_member, Literal(value)))