984: define subversion workaround

cdisc_rules_engine/services/define_xml/base_define_xml_reader.py

@@ -474,6 +474,9 @@ def validate_schema(self) -> bool:
     def get_define_version(self) -> Optional[str]:
         """Use to extract DefineVersion from file"""
         self.read()
+        if hasattr(self, "_original_define_version") and self._original_define_version:
+            return self._original_define_version
+
         mdv_attrib: dict = self._odm_loader.loader.parser.mdv[0].attrib
         for key, val in mdv_attrib.items():
             if key.endswith("DefineVersion"):

cdisc_rules_engine/services/define_xml/define_xml_reader_factory.py

@@ -1,7 +1,9 @@
+import re
+from pathlib import Path
 from os.path import dirname, join
-from xml.etree import ElementTree
 from re import compile
 from typing import Union
+from xml.etree import ElementTree
 
 from cdisc_rules_engine.constants.define_xml_constants import (
     DEFINE_XML_FILE_NAME,
@@ -41,16 +43,8 @@ class DefineXMLReaderFactory:
 
     @classmethod
     def from_filename(cls, filename: str):
-        """
-        Inits a DefineXMLReader object from file.
-        """
-        logger.info(f"Reading Define-XML from file name. filename={filename}")
-        define_xml_reader_class: type = cls._get_define_xml_reader(
-            ElementTree.parse(filename).getroot()
-        )
-        reader: BaseDefineXMLReader = define_xml_reader_class()
-        reader._odm_loader.open_odm_document(filename)
-        return reader
+        logger.info(f"Reading Define-XML from file. filename={filename}")
+        return cls._build_reader(Path(filename).read_bytes())
 
     @classmethod
     def from_file_contents(
@@ -60,35 +54,50 @@ def from_file_contents(
         study_id=None,
         data_bundle_id=None,
     ):
-        """
-        Inits a DefineXMLReader object from file contents.
-        """
         logger.info("Reading Define-XML from file contents")
-        define_xml_reader_class: type = cls._get_define_xml_reader(
-            ElementTree.fromstring(file_contents)
-        )
-        reader: BaseDefineXMLReader = define_xml_reader_class(
-            cache_service_obj, study_id, data_bundle_id
+        return cls._build_reader(
+            file_contents, cache_service_obj, study_id, data_bundle_id
         )
-        reader._odm_loader.load_odm_string(file_contents)
+
+    @classmethod
+    def _build_reader(
+        cls,
+        content: Union[str, bytes],
+        cache_service_obj=None,
+        study_id=None,
+        data_bundle_id=None,
+    ) -> "BaseDefineXMLReader":
+        root = ElementTree.fromstring(content)
+        reader_class, original_version = cls._get_define_xml_reader(root)
+
+        if isinstance(content, bytes):
+            content = ElementTree.tostring(root, encoding="unicode")
+
+        if original_version:
+            content = content.replace(
+                f'DefineVersion="{original_version}"',
+                'DefineVersion="2.1"',
+            )
+
+        reader = reader_class(cache_service_obj, study_id, data_bundle_id)
+        reader._original_define_version = original_version
+        reader._odm_loader.load_odm_string(content)
         return reader
 
     @classmethod
-    def _get_define_xml_reader(cls, root: ElementTree.Element) -> BaseDefineXMLReader:
-        elt = root.find(
-            "Study/MetaDataVersion",
-            namespaces={"": ODM_NAMESPACE},
-        )
+    def _get_define_xml_reader(
+        cls, root: ElementTree.Element
+    ) -> tuple[type[BaseDefineXMLReader], str]:
+        elt = root.find("Study/MetaDataVersion", namespaces={"": ODM_NAMESPACE})
         pattern = compile(r"(\{(.*)\})?DefineVersion")
-        define_version = next(
-            iter(
-                cls._from_namespace(match.group(2))
-                for match in [pattern.fullmatch(name) for name, _ in elt.items()]
-                if match
-            ),
-            None,
-        )
-        return define_version
+        for name, value in elt.items():
+            match = pattern.fullmatch(name)
+            if match:
+                reader_class = cls._from_namespace(match.group(2))
+                # only look after 2.1.* versions
+                original_version = value if re.fullmatch(r"2\.1\.\d+", value) else None
+                return reader_class, original_version
+        return None, None
 
     @classmethod
     def _from_namespace(cls, namespace: str) -> BaseDefineXMLReader:

tests/QARegressionTests/test_Issues/test_CoreIssue984.py

@@ -0,0 +1,67 @@
+import os
+import subprocess
+
+import pytest
+import json
+from conftest import get_python_executable
+
+
+@pytest.mark.regression
+class TestCoreIssue984:
+    def test_define_subversion_ignored(self):
+        # Run the command in the terminal
+        command = [
+            f"{get_python_executable()}",
+            "-m",
+            "core",
+            "validate",
+            "-s",
+            "sdtmig",
+            "-r",
+            "CORE-000007",
+            "-v",
+            "3.4",
+            "-dp",
+            os.path.join(
+                "tests",
+                "resources",
+                "test_dataset.json",
+            ),
+            "-dxp",
+            os.path.join("tests", "resources", "CoreIssue984", "define.xml"),
+            "--output-format",
+            "json",
+            "-ps",
+            "1",
+        ]
+        subprocess.run(command, check=True)
+
+        files = os.listdir()
+        json_files = [
+            file
+            for file in files
+            if file.startswith("CORE-Report-") and file.endswith(".json")
+        ]
+        json_report_path = sorted(json_files)[-1]
+        # Open the JSON report file
+        json_report = json.load(open(json_report_path))
+        assert {
+            "Conformance_Details",
+            "Dataset_Details",
+            "Issue_Summary",
+            "Issue_Details",
+            "Rules_Report",
+        }.issubset(json_report.keys())
+
+        assert {
+            "Conformance_Details",
+            "Dataset_Details",
+            "Issue_Summary",
+            "Issue_Details",
+            "Rules_Report",
+        }.issubset(json_report.keys())
+        assert json_report["Conformance_Details"]["Standard"] == "SDTMIG"
+        assert json_report["Conformance_Details"]["Define_XML_Version"] == "2.1.5"
+
+        if os.path.exists(json_report_path):
+            os.remove(json_report_path)