Prevent repeating GCS referencing static CS with answers repeating for same list

app/validators/blocks/grand_calculated_summary_block_validator.py

@@ -18,6 +18,10 @@ class GrandCalculatedSummaryBlockValidator(CalculationBlockValidator):
         "Cannot have a repeating grand calculated summary reference"
         " a repeating calculated summary in a different repeating section"
     )
+    CALCULATED_SUMMARY_WITH_REPEATING_ANSWERS_FOR_SAME_LIST = (
+        "Cannot have a repeating grand calculated summary reference a static calculated summary"
+        " that has repeating answers for the same list"
+    )
 
     def __init__(self, block: Mapping, questionnaire_schema: QuestionnaireSchema):
         super().__init__(block, questionnaire_schema)
@@ -50,7 +54,7 @@ def validate(self):
             )
 
         self.validate_answer_types(answers)
-        self.validate_repeating_calculated_summaries()
+        self.validate_calculated_summaries()
 
         return self.errors
 
@@ -83,39 +87,80 @@ def validate_calculated_summary_is_before_grand_calculated_summary_block(
                     calculated_summary_id=calculated_summary_id,
                 )
 
-    def validate_repeating_calculated_summaries(self):
+    def validate_calculated_summaries(self):
         """
-        If the grand calculated summary references a repeating calculated summary, this is only valid if:
-        1) the grand calculated summary is also repeating
-        2) it is in the same repeating section as the repeating calculated summary it references
+        Run additional validation for the scenarios:
+        1) any grand calculated summary referencing a repeating calculated summary
+        2) repeating grand calculated summary referencing a static calculated summary
         """
-        gcs_section_id = self.questionnaire_schema.get_section_id_for_block_id(
+        gcs_section = self.questionnaire_schema.get_parent_section_for_block(
             self.block["id"]
         )
         is_gcs_repeating = self.questionnaire_schema.is_repeating_section(
-            gcs_section_id
+            gcs_section["id"]
         )
         for calculated_summary_id in self.calculated_summaries_to_calculate:
-            if not self.questionnaire_schema.is_block_in_repeating_section(
+            if is_cs_repeating := self.questionnaire_schema.is_block_in_repeating_section(
                 calculated_summary_id
             ):
-                # validation below only required for repeating calculated summaries
-                continue
-
-            if not is_gcs_repeating:
-                self.add_error(
-                    self.REPEATING_CALCULATED_SUMMARY_OUTSIDE_REPEAT,
-                    block_id=self.block["id"],
+                self._validate_repeating_calculated_summary_in_grand_calculated_summary(
                     calculated_summary_id=calculated_summary_id,
+                    is_gcs_repeating=is_gcs_repeating,
+                    grand_calculated_summary_section_id=gcs_section["id"],
                 )
-            elif (
-                gcs_section_id
-                != self.questionnaire_schema.get_section_id_for_block_id(
-                    calculated_summary_id
+
+            if is_gcs_repeating and not is_cs_repeating:
+                list_name = gcs_section["repeat"]["for_list"]
+                self._validate_static_calculated_summary_in_repeating_grand_calculated_summary(
+                    list_name=list_name, calculated_summary_id=calculated_summary_id
                 )
-            ):
+
+    def _validate_static_calculated_summary_in_repeating_grand_calculated_summary(
+        self, *, list_name: str, calculated_summary_id: str
+    ):
+        """
+        If the grand calculated summary is repeating, and references a static calculated summary with repeating answers,
+        this is only valid if the repeating answers are for a different list to the grand calculated summary.
+        """
+        for answer_id in self.calculated_summary_answers[calculated_summary_id]:
+            if (
+                answer_list := self.questionnaire_schema.get_list_name_for_answer_id(
+                    answer_id
+                )
+            ) and answer_list == list_name:
                 self.add_error(
-                    self.CALCULATED_SUMMARY_IN_DIFFERENT_REPEATING_SECTION,
+                    self.CALCULATED_SUMMARY_WITH_REPEATING_ANSWERS_FOR_SAME_LIST,
                     block_id=self.block["id"],
                     calculated_summary_id=calculated_summary_id,
+                    list_name=list_name,
                 )
+
+    def _validate_repeating_calculated_summary_in_grand_calculated_summary(
+        self,
+        *,
+        calculated_summary_id: str,
+        is_gcs_repeating: bool,
+        grand_calculated_summary_section_id: str,
+    ):
+        """
+        If the grand calculated summary references a repeating calculated summary, this is only valid if:
+        1) the grand calculated summary is also repeating
+        2) it is in the same repeating section as the repeating calculated summary it references
+        """
+        if not is_gcs_repeating:
+            self.add_error(
+                self.REPEATING_CALCULATED_SUMMARY_OUTSIDE_REPEAT,
+                block_id=self.block["id"],
+                calculated_summary_id=calculated_summary_id,
+            )
+        elif (
+            grand_calculated_summary_section_id
+            != self.questionnaire_schema.get_section_id_for_block_id(
+                calculated_summary_id
+            )
+        ):
+            self.add_error(
+                self.CALCULATED_SUMMARY_IN_DIFFERENT_REPEATING_SECTION,
+                block_id=self.block["id"],
+                calculated_summary_id=calculated_summary_id,
+            )

app/validators/questionnaire_schema.py

@@ -139,17 +139,35 @@ def __init__(self, schema):
         self.supplementary_lists = jp.match(
             "$..supplementary_data.lists[*]", self.schema
         )
-        self.list_collector_names = jp.match(
-            '$..blocks[?(@.type=="ListCollector")].for_list', self.schema
+        self.list_collectors = jp.match(
+            '$..blocks[?(@.type=="ListCollector")]', self.schema
         )
+        self.list_collector_names = [
+            list_collector["for_list"] for list_collector in self.list_collectors
+        ]
         self.list_names = self.list_collector_names + self.supplementary_lists
-
+        self.list_names_by_repeating_block_id = {
+            block["id"]: list_collector["for_list"]
+            for list_collector in self.list_collectors
+            for block in list_collector.get("repeating_blocks", [])
+        }
         self._answers_with_context = {}
 
     @lru_cache
     def get_block_ids_for_block_type(self, block_type: str) -> list[str]:
         return [block["id"] for block in self.blocks if block["type"] == block_type]
 
+    @cached_property
+    def list_names_by_dynamic_answer_id(self) -> dict[str, str]:
+        answer_id_to_list: dict[str, str] = {}
+        for dynamic_answer in jp.match("$..dynamic_answers[*]", self.schema):
+            if dynamic_answer["values"]["source"] == "list":
+                list_name = dynamic_answer["values"]["identifier"]
+                answer_id_to_list.update(
+                    {answer["id"]: list_name for answer in dynamic_answer["answers"]}
+                )
+        return answer_id_to_list
+
     @cached_property
     def numeric_answer_ranges(self):
         numeric_answer_ranges = {}
@@ -396,6 +414,22 @@ def get_all_dynamic_answer_ids(self, block_id):
             for answer in question.get("dynamic_answers", {}).get("answers", [])
         }
 
+    def get_list_name_for_answer_id(self, answer_id: str) -> str | None:
+        """
+        If the answer is dynamic or in a repeating block or section, return the name of the list it repeats over
+        otherwise None
+        """
+        if list_name := self.list_names_by_dynamic_answer_id.get(answer_id):
+            return list_name
+        block = self.get_block_by_answer_id(answer_id)
+        if list_name := self.list_names_by_repeating_block_id.get(block["id"]):
+            return list_name
+        if block["type"] == "ListCollector":
+            return block["for_list"]
+        section = self.get_parent_section_for_block(block["id"])
+        if section.get("repeat"):
+            return section["repeat"]["for_list"]
+
     @lru_cache
     def get_first_answer_in_block(self, block_id):
         questions = self.get_all_questions_for_block(self.blocks_by_id[block_id])