Output file location when schema check outputs errors (#98)

Signed-off-by: hoangtungdinh <[email protected]>

qc_opendrive/checks/schema/valid_schema.py

@@ -1,7 +1,9 @@
 import importlib.resources
 import logging
 
-from typing import List, Tuple
+from dataclasses import dataclass
+
+from typing import List
 
 from qc_baselib import IssueSeverity, StatusType
 
@@ -30,45 +32,16 @@
 RULE_UID = "asam.net:xodr:1.0.0:xml.valid_schema"
 
 
-def find_xpath_from_position(xml_tree, target_line, target_column):
-
-    # Function to build XPath from the element
-    def build_xpath(element):
-        path = []
-        while element is not None:
-            siblings = list(element.itersiblings(preceding=True))
-            index = len(siblings) + 1
-            path.insert(0, f"{element.tag}[{index}]")
-            element = element.getparent()
-        return "/" + "/".join(path)
-
-    # Find element by searching for position
-    def find_element_by_position(element):
-        if hasattr(element, "sourceline") and element.sourceline == target_line:
-            # Rough estimation of column position
-            # 'position' attribute might not always be available, so this is approximate
-            if hasattr(element, "position") and element.position[1] == target_column:
-                return element
-        for child in element:
-            found = find_element_by_position(child)
-            if found:
-                return found
-        return None
-
-    # Start from the root element
-    root = xml_tree.getroot()
-    element = find_element_by_position(root)
-
-    # If element is found, build its XPath
-    if element is not None:
-        return build_xpath(element)
-    else:
-        return "Element not found"
-
-
-def _is_schema_compliant(
+@dataclass
+class SchemaError:
+    message: str
+    line: int
+    column: int
+
+
+def _get_schema_errors(
     xml_file: str, schema_file: str, schema_version: str
-) -> tuple[bool, List[Tuple]]:
+) -> List[SchemaError]:
     """Check if input xml tree  is valid against the input schema file (.xsd)
 
     Args:
@@ -94,28 +67,26 @@ def _is_schema_compliant(
         schema.validate(xml_tree)
         for error in schema.error_log:
             errors.append(
-                (
-                    find_xpath_from_position(xml_tree, error.line, error.column),
-                    error.message,
+                SchemaError(
+                    message=error.message,
+                    line=error.line,
+                    column=error.column,
                 )
             )
     else:  # use xmlschema to support XSD schema 1.1 -> OpenDRIVE 1.8 and higher
         schema = xmlschema.XMLSchema11(schema_file)
         # Iterate over all validation errors
-        for error in schema.iter_errors(xml_file):
-            errors.append((error.path, error.message))
-
-    # Return True and None if there are no errors, otherwise False and the list of errors
-    if not errors:
-        logging.info("- XML is valid.")
-        return True, None
-    else:
-        logging.error("- XML is invalid!")
-        for error in errors:
-            logging.error(f"- Error: {error[1]}")
-            logging.error(f"- Path: {error[0]}")
+        xml_doc = etree.parse(xml_file)
+        for error in schema.iter_errors(xml_doc):
+            errors.append(
+                SchemaError(
+                    message=error.reason,
+                    line=error.sourceline,
+                    column=0,
+                )
+            )
 
-        return False, errors
+    return errors
 
 
 def check_rule(checker_data: models.CheckerData) -> None:
@@ -143,24 +114,24 @@ def check_rule(checker_data: models.CheckerData) -> None:
     xsd_file_path = str(
         importlib.resources.files("qc_opendrive.schema").joinpath(xsd_file)
     )
-    schema_compliant, errors = _is_schema_compliant(
+    errors = _get_schema_errors(
         checker_data.config.get_config_param("InputFile"), xsd_file_path, schema_version
     )
 
-    if not schema_compliant:
+    for error in errors:
+        issue_id = checker_data.result.register_issue(
+            checker_bundle_name=constants.BUNDLE_NAME,
+            checker_id=CHECKER_ID,
+            description="Issue flagging when input file does not follow its version schema",
+            level=IssueSeverity.ERROR,
+            rule_uid=RULE_UID,
+        )
 
-        for error in errors:
-            issue_id = checker_data.result.register_issue(
-                checker_bundle_name=constants.BUNDLE_NAME,
-                checker_id=CHECKER_ID,
-                description="Issue flagging when input file does not follow its version schema",
-                level=IssueSeverity.ERROR,
-                rule_uid=RULE_UID,
-            )
-            checker_data.result.add_xml_location(
-                checker_bundle_name=constants.BUNDLE_NAME,
-                checker_id=CHECKER_ID,
-                issue_id=issue_id,
-                xpath=error[0],
-                description=error[1],
-            )
+        checker_data.result.add_file_location(
+            checker_bundle_name=constants.BUNDLE_NAME,
+            checker_id=CHECKER_ID,
+            issue_id=issue_id,
+            row=error.line,
+            column=error.column,
+            description=error.message,
+        )