Merge branch 'release_v010' of https://github.com/NHSDigital/data-validation-engine into feature/gr-ndsp-619-add_group_level_rejections

Author: georgeRobertson

src/dve/core_engine/backends/base/rules.py

@@ -396,33 +396,35 @@ def identify_and_remove_orphans(
 
         def process_node(
             node: HierarchyNode,
-            parent_entity_name: Optional[EntityName],
+            orph_messages: Messages | None = None,
             processed: bool = False,
-        ) -> bool:
-            """Recursive helper to process a node and its children."""
-            current_entity_name = node.entity_name
+        ):
+            """Identify orphans and remove in a given node"""
 
-            if parent_entity_name is not None:
-                self.logger.info(f"Identifying orphans in {current_entity_name}")
+            if orph_messages is None:
+                orph_messages = []
+
+            if node.parent_entity is not None:
+                self.logger.info(f"Identifying orphans in {node.entity_name}")
 
                 join_expr = " AND ".join(
-                    f"{parent_entity_name}.{k} = {current_entity_name}.{v}"
+                    f"{node.parent_entity}.{k} = {node.entity_name}.{v}"
                     for k, v in node.join_fields.items()
                 )
 
                 _, no_orphs = self.identify_orphans(
                     entities=entities,
                     config=OrphanIdentification(
                         id=list(node.join_fields.values())[0],
-                        entity_name=current_entity_name,
-                        target_name=parent_entity_name,
+                        entity_name=node.entity_name,
+                        target_name=node.parent_entity,
                         join_condition=join_expr,
                     ),
                 )
 
                 if no_orphs > 0:
                     self.logger.info(
-                        f"Removing records with missing parent from {current_entity_name}"
+                        f"Removing records with missing parent from {node.entity_name}"
                     )
                     processed = True
                     location = list(node.join_fields.values())[0]
@@ -435,7 +437,7 @@ def process_node(
                         _orph_records = self.remove_orphans(
                             entities=entities,
                             config=OrphanRemoval(
-                                entity_name=current_entity_name,
+                                entity_name=node.entity_name,
                                 reporting=ReportingConfig(
                                     emit="record_failure",
                                     code=node.missing_parent_id_error_code,
@@ -448,7 +450,7 @@ def process_node(
                         msg_writer.write_queue.put(
                             [
                                 FeedbackMessage(
-                                    entity=current_entity_name,
+                                    entity=node.entity_name,
                                     record=record,  # type: ignore
                                     error_location=location,
                                     error_message=node.missing_parent_id_error_message,
@@ -462,16 +464,13 @@ def process_node(
                             ]
                         )
 
-            if node.children:
-                for child_node in node.children:
-                    processed = process_node(child_node, current_entity_name, processed)
-
             return processed
 
         processed = False
 
-        for root_node in entity_hierarchy.entity_trees.values():
-            processed = process_node(root_node, parent_entity_name=None, processed=processed)
+        for tree in entity_hierarchy.entity_trees.values():
+            for node in tree.iterate_root_down():
+                processed = process_node(node)
 
         _orph_rel = entities.get(ORPHANED_RECORD_ENTITY_NAME)
         if _orph_rel is not None:

src/dve/core_engine/configuration/v1/hierarchy.py

@@ -16,6 +16,7 @@ class HierarchyNode(BaseModel):
     """Stores entity hierarchy information"""
 
     entity_name: str
+    parent_entity: Optional[str] = None
     children: list["HierarchyNode"] = Field(default_factory=list)
     mandatory: bool = False
     join_fields: dict[str, str] = Field(default_factory=dict)
@@ -26,14 +27,18 @@ class HierarchyNode(BaseModel):
         "Records removed due to no valid parent record"
     )
 
-    def get_descendents(self) -> list[str]:
+    def get_descendents(self) -> list["HierarchyNode"]:
         """Recursively list all descendents of the node"""
         descendents = []
         for node in self.children:  # type: ignore
-            descendents.append(node.entity_name)
+            descendents.append(node)
             descendents.extend(node.get_descendents())
         return descendents
 
+    def get_descendent_names(self) -> list[str]:
+        """Recursively list all names of descendents of the node"""
+        return [node.entity_name for node in self.get_descendents()]
+
     def get_node(self, entity_name: str) -> Union["HierarchyNode", None]:
         """Recursively search for node and return if found"""
         node = None
@@ -65,6 +70,20 @@ def as_dict(self) -> dict[str, dict[str, Any]]:
 
         return {self.entity_name: ret_dict}
 
+    def _get_full_tree(self):
+        """Get all nodes in tree, including the root"""
+        desc = self.get_descendents()
+        desc.insert(0, self)
+        return desc
+
+    def iterate_root_down(self):
+        """Iterate through nodes from root to lowest descendent"""
+        yield from self._get_full_tree()
+
+    def iterate_lowest_descendent_up(self):
+        """Iterate through nodes from lowest descendent to root"""
+        yield from self._get_full_tree()[::-1]
+
 
 class EntityHierarchy:
     """Determines and stores entity hierarchy information from config"""
@@ -83,6 +102,7 @@ def determine_trees(
         top_level_parents: dict[EntityName, HierarchyNode] = {
             entity_name: HierarchyNode(
                 entity_name=entity_name,
+                parent_entity=None,
                 **config.model_dump(
                     exclude={
                         "parent_entity",
@@ -102,6 +122,7 @@ def determine_trees(
             for entity_name in default_roots:
                 top_level_parents[entity_name] = HierarchyNode(
                     entity_name=entity_name,
+                    parent_entity=None,
                     missing_parent_id_error_code=None,
                     missing_parent_id_error_message=None,
                 )
@@ -110,13 +131,11 @@ def determine_trees(
             for main_entity, parent_node in top_level_parents.items():
                 if (
                     linkage_detail.parent_entity == main_entity
-                    or linkage_detail.parent_entity in parent_node.get_descendents()
+                    or linkage_detail.parent_entity in parent_node.get_descendent_names()
                 ):
                     parent_node.add_child_node(
                         linkage_detail.parent_entity,
-                        HierarchyNode(
-                            entity_name=name, **linkage_detail.model_dump(exclude={"parent_entity"})
-                        ),
+                        HierarchyNode(entity_name=name, **linkage_detail.model_dump()),
                     )
                     break
             else:

src/dve/pipeline/pipeline.py

@@ -886,14 +886,18 @@ def error_report(
                 .agg(pl.col("Count").sum())  # type: ignore
                 .iter_rows(named=True)
             }
+            submission_rejections = err_types.get(
+                ErrorReportCategories.FILE_REJECTION.reporting_name, 0
+            )
             sub_stats = SubmissionStatisticsRecord(
                 submission_id=submission_info.submission_id,
                 record_count=submission_status.number_of_records,
-                number_submission_rejections=err_types.get(
-                    ErrorReportCategories.FILE_REJECTION.reporting_name, 0
-                ),
-                number_record_rejections=err_types.get(
-                    ErrorReportCategories.RECORD_REJECTION.reporting_name, 0
+                number_submission_rejections=submission_rejections,
+                number_record_rejections=(
+                    submission_status.number_of_records
+                    if submission_rejections > 0 else err_types.get(  # type: ignore
+                        ErrorReportCategories.RECORD_REJECTION.reporting_name, 0
+                    )
                 ),
                 number_warnings=err_types.get(ErrorReportCategories.WARNING.reporting_name, 0),
             )

src/dve/pipeline/utils.py

@@ -60,7 +60,7 @@ def load_reader(
         if file_extension:
             err_msg = (
                 f"The supplied file extension `{file_extension}`"
-                +f" is not a supported file format for {model_name}."
+                + f" is not a supported file format for {model_name}."
             )
         else:
             err_msg = "No supplied file extension. Unable to parse file without a file extension."

src/dve/reporting/excel_report.py

@@ -153,17 +153,21 @@ def _add_submission_info(self, status: str, summary: Worksheet):
                 ),  # pylint: disable=C0301
             ]
         )
-        if status not in (
+        if status in (
             ErrorReportStatus.PROCESSING_FAILED,
             ErrorReportStatus.FILE_REJECTION,
         ):
-            summary.append(
-                [
-                    "",
-                    "Total Number of Records Rejected",
-                    self.submission_status.number_of_records_rejected,
-                ]
-            )
+            _records_rejected = self.submission_status.number_of_records
+        else:
+            _records_rejected = self.submission_status.number_of_records_rejected
+        summary.append(
+            [
+                "",
+                "Total Number of Records Rejected",
+                _records_rejected,
+            ]
+        )
+
         summary.append(["", ""])
 
 

tests/features/animals.feature

@@ -55,5 +55,5 @@ Feature: Pipeline tests using the animal dataset
             | parameter                    | value |
             | record_count                 | 7     |
             | number_submission_rejections | 1     |
-            | number_record_rejections     | 2     |
+            | number_record_rejections     | 7     |
             | number_warnings              | 1     |

tests/features/flights.feature

@@ -10,6 +10,7 @@ Feature: Pipeline tests using the flights dataset
         When I run the file transformation phase
         Then the country entity is stored as a parquet after the file_transformation phase
         And the airport entity is stored as a parquet after the file_transformation phase
+        And the staff entity is stored as a parquet after the file_transformation phase
         And the flights entity is stored as a parquet after the file_transformation phase
         And the passengers entity is stored as a parquet after the file_transformation phase
         And the latest audit record for the submission is marked with processing status data_contract
@@ -36,19 +37,20 @@ Feature: Pipeline tests using the flights dataset
         When I run the file transformation phase
         Then the country entity is stored as a parquet after the file_transformation phase
         And the airport entity is stored as a parquet after the file_transformation phase
+        And the staff entity is stored as a parquet after the file_transformation phase
         And the flights entity is stored as a parquet after the file_transformation phase
         And the passengers entity is stored as a parquet after the file_transformation phase
         And the latest audit record for the submission is marked with processing status data_contract
         When I run the data contract phase
         Then there are no file rejections from the data_contract phase
-        And there are no record rejections from the data_contract phase
+        And there is 1 record rejection from the data_contract phase
         When I run the business rules phase
         Then there are errors with the following details and associated error_count from the business_rules phase
             | ErrorType | ErrorCode | error_count |
-            | record    | C1        | 1           |
-            | record    | AG1       | 1           |
-            | record    | FG1       | 2           |
-            | record    | PG1       | 4           |
+            | record    | AG1       | 3           |
+            | record    | SG1       | 15          |
+            | record    | FG1       | 10          |
+            | record    | PG1       | 25          |
         When I run the error report phase
         Then An error report is produced
     # TODO - fix the stats calculations as they're currently incorrect for hiearchical datasets
@@ -66,6 +68,7 @@ Feature: Pipeline tests using the flights dataset
         When I run the file transformation phase
         Then the country entity is stored as a parquet after the file_transformation phase
         And the airport entity is stored as a parquet after the file_transformation phase
+        And the staff entity is stored as a parquet after the file_transformation phase
         And the flights entity is stored as a parquet after the file_transformation phase
         And the passengers entity is stored as a parquet after the file_transformation phase
         And the latest audit record for the submission is marked with processing status data_contract
@@ -76,10 +79,10 @@ Feature: Pipeline tests using the flights dataset
         Then there are errors with the following details and associated error_count from the business_rules phase
             | ErrorType | ErrorCode | error_count |
             | record    | F1        | 1           |
-            | record    | PG1       | 2           |
+            | record    | PG1       | 3           |
         When I run the error report phase
         Then An error report is produced
-    # TODO - fix the stats calculations as they're currently incorrect for hiearchical datasets
+    #TODO - fix the stats calculations as they're currently incorrect for hiearchical datasets
     # And The statistics entry for the submission shows the following information
     #     | parameter                | value |
     #     | record_count             | 1     |
@@ -95,6 +98,7 @@ Feature: Pipeline tests using the flights dataset
         Then the country entity is stored as a parquet after the file_transformation phase
         And the airport entity is stored as a parquet after the file_transformation phase
         And the flights entity is stored as a parquet after the file_transformation phase
+        And the staff entity is stored as a parquet after the file_transformation phase
         And the passengers entity is stored as a parquet after the file_transformation phase
         And the latest audit record for the submission is marked with processing status data_contract
         When I run the data contract phase
@@ -104,8 +108,9 @@ Feature: Pipeline tests using the flights dataset
         Then there are errors with the following details and associated error_count from the business_rules phase
             | ErrorType | ErrorCode | error_count |
             | record    | F1        | 1           |
-            | record    | PG1       | 2           |
+            | record    | PG1       | 3           |
             | record    | P1        | 1           |
+            | record    | S1        | 7           |
         When I run the error report phase
         Then An error report is produced
     # TODO - fix the stats calculations as they're currently incorrect for hiearchical datasets
@@ -161,7 +166,7 @@ Feature: Pipeline tests using the flights dataset
             | ErrorType | Status        | ErrorCode | error_count |
             | record    | error         | F2        | 2           |
             | record    | error         | PG1       | 4           |
-            | record    | informational | A1        | 1           |
+            # | record    | informational | A1        | 1           |
         When I run the error report phase
         Then An error report is produced
 # TODO - fix the stats calculations as they're currently incorrect for hiearchical datasets

tests/features/movies.feature

@@ -42,7 +42,7 @@ Feature: Pipeline tests using the movies dataset
             | parameter                    | value |
             | record_count                 | 5     |
             | number_submission_rejections | 1     |
-            | number_record_rejections     | 3     |
+            | number_record_rejections     | 5     |
             | number_warnings              | 2     |
         And the error aggregates are persisted
 
@@ -81,7 +81,7 @@ Feature: Pipeline tests using the movies dataset
             | parameter                    | value |
             | record_count                 | 5     |
             | number_submission_rejections | 1     |
-            | number_record_rejections     | 3     |
+            | number_record_rejections     | 5     |
             | number_warnings              | 2     |
         And the error aggregates are persisted
 

tests/test_core_engine/test_backends/test_implementations/test_duckdb/test_rules.py

@@ -656,9 +656,11 @@ def test_identify_and_remove_orphans(self):
                     children=[
                         HierarchyNode(
                             entity_name="passengers",
+                            parent_entity="flights",
                             children=[
                                 HierarchyNode(
                                     entity_name="food",
+                                    parent_entity="passengers",
                                     children=[],
                                     join_fields={"passenger_id": "passenger_id"},
                                     mandatory=False