1487: Fix engine crash when selected rule is not applicable for version

cdisc_rules_engine/models/rule_validation_result.py

@@ -4,6 +4,7 @@
 from cdisc_rules_engine.interfaces import RepresentationInterface
 from cdisc_rules_engine.utilities.utils import get_execution_status
 from cdisc_rules_engine.models.rule import Rule
+from cdisc_rules_engine.enums.execution_status import ExecutionStatus
 
 
 @dataclass
@@ -27,6 +28,22 @@ def __init__(self, rule: Rule, results: List[dict | str]):
         self.execution_status = get_execution_status(results)
         self.results = results
 
+    @classmethod
+    def from_skipped_rule(
+        cls,
+        rule_id: str,
+        message: str | None = None,
+    ) -> "RuleValidationResult":
+        instance = cls.__new__(cls)
+        instance.id = rule_id
+        instance.cdisc_rule_id = None
+        instance.fda_rule_id = None
+        instance.executability = None
+        instance.message = message
+        instance.execution_status = ExecutionStatus.SKIPPED.value
+        instance.results = []
+        return instance
+
     def _get_rule_ids(self, rule: Rule, org: str) -> str:
         return ", ".join(
             sorted(

scripts/run_validation.py

@@ -130,9 +130,10 @@ def run_validation(args: Validation_args):
     manager.start()
     created_files = []
     try:
+        created_files = []
         shared_cache = get_cache_service(manager)
         engine_logger.info(f"Populating cache, cache path: {args.cache}")
-        rules = get_rules(args)
+        rules, skipped_rule_ids = get_rules(args)
         library_metadata: LibraryMetadataContainer = get_library_metadata_from_cache(
             args
         )
@@ -191,6 +192,12 @@ def run_validation(args: Validation_args):
             progress_handler: Callable = get_progress_displayer(args)
             results = progress_handler(rules, validation_results, results)
 
+        for skipped_rule_id, message in skipped_rule_ids or []:
+            engine_logger.info(message)
+            results.append(
+                RuleValidationResult.from_skipped_rule(skipped_rule_id, message)
+            )
+
         # build all desired reports
         end = time.time()
         elapsed_time = end - start

scripts/script_utils.py

@@ -8,7 +8,7 @@
 from cdisc_rules_engine.models.library_metadata_container import (
     LibraryMetadataContainer,
 )
-from typing import List, Iterable
+from typing import List, Iterable, Tuple
 from cdisc_rules_engine.config import config
 from cdisc_rules_engine.services import logger as engine_logger
 import os
@@ -211,10 +211,15 @@ def get_cache_service(manager):
         return manager.InMemoryCacheService()
 
 
-def get_rules(args) -> List[dict]:
-    return (
+def get_rules(args) -> Tuple[List[dict], List[Tuple[str, str]]]:
+    rules_result = (
         load_rules_from_local(args) if args.local_rules else load_rules_from_cache(args)
     )
+    if isinstance(rules_result, tuple) and len(rules_result) == 2:
+        rules, skipped_rule_ids = rules_result
+    else:
+        rules, skipped_rule_ids = rules_result, []
+    return rules, skipped_rule_ids
 
 
 def rule_cache_file(args) -> str:
@@ -256,6 +261,70 @@ def load_custom_rules(custom_data, cdisc_data, standard, version, rules, standar
     return list(rules_dict.values())
 
 
+def _determine_valid_rule_ids(
+    standard_rules: dict,
+    rule_ids: Iterable[str] | None,
+    excluded_rule_ids: Iterable[str] | None,
+) -> set:
+    include_filter = set(rule_ids) if rule_ids else None
+    exclude_filter = set(excluded_rule_ids) if excluded_rule_ids else None
+    valid_rule_ids = set()
+    for rule in standard_rules:
+        if include_filter and rule not in include_filter:
+            continue
+        if exclude_filter and rule in exclude_filter:
+            continue
+        valid_rule_ids.add(rule)
+    return valid_rule_ids
+
+
+def _collect_missing_includes(
+    rule_ids: Iterable[str] | None,
+    standard_rules: dict,
+    standard: str,
+    version: str,
+) -> List[Tuple[str, str]]:
+    if not rule_ids:
+        return []
+    available_rules = set(standard_rules)
+    skipped_rule_ids: List[Tuple[str, str]] = []
+    for rule in rule_ids:
+        if rule in available_rules:
+            continue
+        engine_logger.error(
+            f"The rule specified to include '{rule}' is not in the standard {standard} and version {version}. "
+            "It will be skipped from validation."
+        )
+        message = (
+            f"Rule '{rule}' was requested but is not available for "
+            f"standard {standard} version {version}"
+        )
+        skipped_rule_ids.append((rule, message))
+    return skipped_rule_ids
+
+
+def _log_invalid_excludes(
+    excluded_rule_ids: Iterable[str] | None,
+    standard_rules: dict,
+    standard: str,
+    version: str,
+) -> None:
+    if not excluded_rule_ids:
+        return
+    available_rules = set(standard_rules)
+    for rule in excluded_rule_ids:
+        if rule in available_rules:
+            continue
+        engine_logger.error(
+            f"The rule specified to exclude '{rule}' is not in the standard {standard} and version {version}. "
+            "It is not present and will be ignored."
+        )
+
+
+def _build_rules_from_ids(valid_rule_ids: set, rules_data) -> List[dict]:
+    return [rules_data.get(rule_id) for rule_id in valid_rule_ids]
+
+
 def load_specified_rules(
     rules_data,
     rule_ids,
@@ -267,37 +336,19 @@ def load_specified_rules(
 ):
     key = get_rules_cache_key(standard, version, substandard)
     standard_rules = standard_dict.get(key, {})
-    valid_rule_ids = set()
-
-    # Determine valid rules based on inclusion and exclusion lists
-    for rule in standard_rules:
-        if (not rule_ids or rule in rule_ids) and (
-            not excluded_rule_ids or rule not in excluded_rule_ids
-        ):
-            valid_rule_ids.add(rule)
-    # Check that all specified rules are valid
-    if rule_ids:
-        for rule in rule_ids:
-            if rule not in standard_rules:
-                raise ValueError(
-                    f"The rule specified to include '{rule}' is not in the standard {standard} and version {version}"
-                )
-    else:
-        for rule in excluded_rule_ids:
-            if rule not in standard_rules:
-                raise ValueError(
-                    f"The rule specified to exclude '{rule}' is not in the standard {standard} and version {version}"
-                )
-    rules = []
-    for rule_id in valid_rule_ids:
-        rule_data = rules_data.get(rule_id)
-        rules.append(rule_data)
-    # If no valid rules were found, raise an error
+    valid_rule_ids = _determine_valid_rule_ids(
+        standard_rules, rule_ids, excluded_rule_ids
+    )
+    skipped_rule_ids = _collect_missing_includes(
+        rule_ids, standard_rules, standard, version
+    )
+    _log_invalid_excludes(excluded_rule_ids, standard_rules, standard, version)
+    rules = _build_rules_from_ids(valid_rule_ids, rules_data)
     if not rules:
-        raise ValueError(
+        engine_logger.error(
             f"All specified rules were excluded because they are not in the standard {standard} and version {version}"
         )
-    return rules
+    return rules, skipped_rule_ids
 
 
 def load_all_rules_for_standard(
@@ -332,7 +383,9 @@ def load_all_rules(rules_data):
     return rules
 
 
-def load_rules_from_cache(args) -> List[dict]:
+def load_rules_from_cache(
+    args,
+) -> list[dict] | tuple[list[dict], List[Tuple[str, str]]]:
     rules_file, cdisc_file, standard_dict = rule_cache_file(args)
     rules_data = {}
     try:
@@ -351,6 +404,7 @@ def load_rules_from_cache(args) -> List[dict]:
     except Exception as e:
         engine_logger.error(f"Error loading rules file: {e}")
         return []
+
     if args.custom_standard:
         return load_custom_rules(
             rules_data,
@@ -361,7 +415,7 @@ def load_rules_from_cache(args) -> List[dict]:
             standard_dict,
         )
     elif args.rules or args.exclude_rules:
-        return load_specified_rules(
+        rules, skipped_rule_ids = load_specified_rules(
             rules_data,
             args.rules,
             args.exclude_rules,
@@ -370,6 +424,7 @@ def load_rules_from_cache(args) -> List[dict]:
             standard_dict,
             args.substandard,
         )
+        return rules, skipped_rule_ids
     elif args.standard and args.version:
         return load_all_rules_for_standard(
             rules_data,

tests/QARegressionTests/test_Issues/test_CoreIssue1487.py

@@ -0,0 +1,54 @@
+import os
+import subprocess
+import unittest
+import openpyxl
+import pytest
+from conftest import get_python_executable
+
+
+@pytest.mark.regression
+class TestCoreIssue1487(unittest.TestCase):
+    def test_positive_dataset(self):
+        # Run the command in the terminal
+        command = [
+            f"{get_python_executable()}",
+            "-m",
+            "core",
+            "validate",
+            "-s",
+            "sdtmig",
+            "-v",
+            "5-0",
+            "-d",
+            os.path.join("tests", "resources", "CoreIssue1487"),
+            "-r",
+            "CORE-000354",
+        ]
+        subprocess.run(command, check=True)
+
+        # Get the latest created Excel file
+        files = os.listdir()
+        excel_files = [
+            file
+            for file in files
+            if file.startswith("CORE-Report-") and file.endswith(".xlsx")
+        ]
+        excel_file_path = sorted(excel_files)[-1]
+        # # Open the Excel file
+        workbook = openpyxl.load_workbook(excel_file_path)
+
+        assert "Rules Report" in workbook.sheetnames
+        rules_sheet = workbook["Rules Report"]
+        target_row = None
+        for row in rules_sheet.iter_rows(min_row=2, values_only=True):
+            if row[0] == "CORE-000354":
+                target_row = row
+                break
+        assert target_row, "Rule CORE-000354 not present in 'Rules Report' sheet."
+        assert (
+            target_row[4] and "was requested but is not available" in target_row[4]
+        ), "Expected error message for CORE-000354 not found."
+        assert target_row[5] == "SKIPPED", "CORE-000354 status should be SKIPPED."
+
+        if os.path.exists(excel_file_path):
+            os.remove(excel_file_path)