Merge branch 'main' into add-html-validation

.github/workflows/main.yml

@@ -9,7 +9,7 @@ jobs:
   docker-push:
       runs-on: ubuntu-22.04
       steps:
-        - uses: actions/checkout@v2
+        - uses: actions/checkout@v4
         - name: Build
           run: |
             docker build -t onsdigital/eq-questionnaire-validator:latest .

.github/workflows/pull_request.yml

@@ -9,13 +9,13 @@ jobs:
   lint:
     runs-on: ubuntu-22.04
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - run: |
           echo "PYTHON_VERSION=$(cat .python-version)" >> $GITHUB_ENV
-      - uses: actions/setup-python@v4
+      - uses: actions/setup-python@v5
         with:
           python-version: ${{ env.PYTHON_VERSION }}
-      - uses: actions/setup-node@v3
+      - uses: actions/setup-node@v4
         with:
           node-version-file: ".nvmrc"
           cache: "npm"
@@ -33,7 +33,7 @@ jobs:
           virtualenvs-path: ~/.virtualenvs
 
       - name: Cache Poetry virtualenv
-        uses: actions/cache@v3
+        uses: actions/cache@v4
         id: cache-virtualenv
         with:
           path: ~/.virtualenvs
@@ -47,13 +47,13 @@ jobs:
   test-unit:
     runs-on: ubuntu-22.04
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - run:
           echo "PYTHON_VERSION=$(cat .python-version)" >> $GITHUB_ENV
-      - uses: actions/setup-python@v4
+      - uses: actions/setup-python@v5
         with:
           python-version: ${{ env.PYTHON_VERSION }}
-      - uses: actions/setup-node@v3
+      - uses: actions/setup-node@v4
         with:
           node-version-file: ".nvmrc"
           cache: "npm"
@@ -73,7 +73,7 @@ jobs:
           virtualenvs-path: ~/.virtualenvs
 
       - name: Cache Poetry virtualenv
-        uses: actions/cache@v3
+        uses: actions/cache@v4
         id: cache-virtualenv
         with:
           path: ~/.virtualenvs
@@ -87,7 +87,7 @@ jobs:
   docker-push:
     runs-on: ubuntu-22.04
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - name: Tag
         run: |
           CLEAN_TAG=$(echo "${{ github.event.pull_request.head.ref  }}"  | tr / -)

.github/workflows/release.yml

@@ -10,7 +10,7 @@ jobs:
     env:
       TAG: ${{ github.event.release.tag_name }}
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v4
       - name: Build
         run: |
           docker build -t onsdigital/eq-questionnaire-validator:$TAG .

app/error_messages.py

@@ -41,3 +41,7 @@
     "Variants have more than one question type for block."
 )
 PREVIEW_WITHOUT_INTRODUCTION_BLOCK = "No introduction block found. Introduction block is mandatory when using the preview questions feature."
+ANSWER_REFERENCED_BEFORE_EXISTS = (
+    "Answer '{answer_id}' referenced as source before it has been added."
+)
+LIST_REFERENCED_BEFORE_CREATED = "List referenced as source before it has been created."

app/validators/questionnaire_schema.py

@@ -152,6 +152,7 @@ def __init__(self, schema):
             for block in list_collector.get("repeating_blocks", [])
         }
         self._answers_with_context = {}
+        self._lists_with_context = {}
 
     @lru_cache
     def get_block_ids_for_block_type(self, block_type: str) -> list[str]:
@@ -209,6 +210,41 @@ def answers_with_context(self):
         self._answers_with_context = answers_dict
         return self._answers_with_context
 
+    @property
+    def lists_with_context(self):
+        if supplementary_list := self.supplementary_lists:
+            for list_id in supplementary_list:
+                self._lists_with_context[list_id] = {
+                    "section_index": 0,
+                    "block_index": 0,
+                }
+
+        if blocks := self.list_collectors:
+            for block in blocks:
+                list_id = block["for_list"]
+                if list_id not in self._lists_with_context:
+                    section_id = self.get_section_id_for_block_id(block["id"])
+                    section_index = self.section_ids.index(section_id)
+                    self._lists_with_context[list_id] = {
+                        "section_index": section_index,
+                        "block_index": self.block_ids.index(block["id"]),
+                    }
+        if blocks := self.get_blocks(type="PrimaryPersonListCollector"):
+            for block in blocks:
+                list_id = block["for_list"]
+                if list_id not in self._lists_with_context or (
+                    self.block_ids.index(block["id"])
+                    < self._lists_with_context[list_id]["block_index"]
+                ):
+                    section_id = self.get_section_id_for_block_id(block["id"])
+                    section_index = self.section_ids.index(section_id)
+                    self._lists_with_context[list_id] = {
+                        "section_index": section_index,
+                        "block_index": self.block_ids.index(block["id"]),
+                    }
+
+        return self._lists_with_context
+
     @staticmethod
     def capture_answers(*, answers, answers_dict, context):
         for answer in answers:
@@ -584,6 +620,27 @@ def get_parent_section_for_block(self, block_id) -> dict | None:
                 if block_id == block["id"]:
                     return self.sections_by_id[section_id]
 
+    def get_parent_list_collector_for_add_block(self, block_id) -> dict | None:
+        for blocks in self.blocks_by_section_id.values():
+            for block in blocks:
+                if (
+                    block["type"] == "ListCollector"
+                    and block["add_block"]["id"] == block_id
+                ):
+                    return block["id"]
+
+    def get_parent_list_collector_for_repeating_block(self, block_id) -> dict | None:
+        for blocks in self.blocks_by_section_id.values():
+            for block in blocks:
+                if block["type"] in [
+                    "ListCollector",
+                    "ListCollectorContent",
+                ] and block.get("repeating_blocks"):
+                    for repeating_block in block["repeating_blocks"]:
+                        if repeating_block["id"] == block_id:
+                            return block["id"]
+        return None
+
     def is_block_in_repeating_section(self, block_id: str) -> bool:
         parent_section = self.get_parent_section_for_block(block_id)
         return parent_section and self.is_repeating_section(parent_section["id"])
@@ -617,3 +674,8 @@ def get_section_id_for_block(self, block: Mapping) -> str | None:
     def get_section_id_for_block_id(self, block_id: str) -> str | None:
         if block := self.get_block(block_id):
             return self.get_section_id_for_block(block)
+
+    def get_section_index_for_section_id(self, section_id: str) -> int:
+        for index, section in enumerate(self.sections):
+            if section["id"] == section_id:
+                return index

app/validators/questionnaire_validator.py

@@ -6,9 +6,14 @@
 from app.validators.answer_code_validator import AnswerCodeValidator
 from app.validators.metadata_validator import MetadataValidator
 from app.validators.placeholders.placeholder_validator import PlaceholderValidator
-from app.validators.questionnaire_schema import QuestionnaireSchema, find_duplicates
+from app.validators.questionnaire_schema import (
+    QuestionnaireSchema,
+    find_duplicates,
+    get_object_containing_key,
+)
 from app.validators.sections.section_validator import SectionValidator
 from app.validators.validator import Validator
+from app.validators.value_source_validator import ValueSourceValidator
 
 
 class QuestionnaireValidator(Validator):
@@ -31,6 +36,8 @@ def validate(self):
         self.validate_smart_quotes()
         self.validate_white_spaces()
         self.validate_html()
+        self.validate_answer_references()
+        self.validate_list_references()
 
         for section in self.questionnaire_schema.sections:
             section_validator = SectionValidator(section, self.questionnaire_schema)
@@ -176,3 +183,149 @@ def validate_introduction_block(self):
         )
         if not has_introduction_blocks:
             self.add_error(error_messages.PREVIEW_WITHOUT_INTRODUCTION_BLOCK)
+
+    def validate_answer_references(self):
+
+        # Handling blocks in group
+        for group in self.questionnaire_schema.groups:
+            self.validate_answer_source_group(group)
+
+        # Handling section level "enabled" rule
+        for index, section in enumerate(self.questionnaire_schema.sections):
+            self.validate_answer_source_section(section, index)
+
+    def validate_answer_source_group(self, group):
+        identifier_references = get_object_containing_key(group, "source")
+        for path, identifier_reference, parent_block in identifier_references:
+            # set up default parent_block_id for later check (group or block level)
+            parent_block_id = None
+            if (
+                "source" in identifier_reference
+                and identifier_reference["source"] == "answers"
+            ):
+
+                source_block = self.questionnaire_schema.get_block_by_answer_id(
+                    identifier_reference["identifier"]
+                )
+                # Handling non-existing blocks used as source
+                if not source_block:
+                    self.add_error(
+                        ValueSourceValidator.ANSWER_SOURCE_REFERENCE_INVALID,
+                        identifier=identifier_reference["identifier"],
+                    )
+                    return False
+                # Handling block level answer sources (skipping group level)
+                if parent_block and "blocks" in path:
+                    parent_block_id = parent_block["id"]
+                    parent_block_index = self.questionnaire_schema.block_ids.index(
+                        parent_block_id
+                    )
+                else:
+                    # Handling group level skip conditions
+                    first_block_id_in_group = group["blocks"][0]["id"]
+                    parent_block_index = self.questionnaire_schema.block_ids.index(
+                        first_block_id_in_group
+                    )
+
+                source_block_id = self.resolve_source_block_id(source_block)
+
+                source_block_index = self.questionnaire_schema.block_ids.index(
+                    source_block_id
+                )
+                if source_block_index > parent_block_index:
+                    if parent_block_id:
+                        self.add_error(
+                            error_messages.ANSWER_REFERENCED_BEFORE_EXISTS.format(
+                                answer_id=identifier_reference["identifier"]
+                            ),
+                            block_id=parent_block_id,
+                        )
+                    else:
+                        self.add_error(
+                            error_messages.ANSWER_REFERENCED_BEFORE_EXISTS.format(
+                                answer_id=identifier_reference["identifier"]
+                            ),
+                            group_id=group["id"],
+                        )
+
+    def validate_answer_source_section(self, section, section_index):
+        identifier_references = get_object_containing_key(section, "source")
+        for path, identifier_reference, _ in identifier_references:
+            if (
+                "source" in identifier_reference
+                and identifier_reference["source"] == "answers"
+                and "enabled" in path
+            ):
+                source_block = self.questionnaire_schema.get_block_by_answer_id(
+                    identifier_reference["identifier"]
+                )
+                source_block_id = self.resolve_source_block_id(source_block)
+
+                source_block_section_id = (
+                    self.questionnaire_schema.get_section_id_for_block_id(
+                        source_block_id
+                    )
+                )
+                source_block_section_index = (
+                    self.questionnaire_schema.get_section_index_for_section_id(
+                        source_block_section_id
+                    )
+                )
+                if section_index < source_block_section_index:
+                    self.add_error(
+                        error_messages.ANSWER_REFERENCED_BEFORE_EXISTS.format(
+                            answer_id=identifier_reference["identifier"]
+                        ),
+                        section_id=section["id"],
+                    )
+
+    def resolve_source_block_id(self, source_block):
+        # Handling of source block nested (list collector's add-block)
+        if source_block["type"] == "ListAddQuestion":
+            return self.questionnaire_schema.get_parent_list_collector_for_add_block(
+                source_block["id"]
+            )
+
+        # Handling of source block nested (list collector's repeating block)
+        if source_block["type"] == "ListRepeatingQuestion":
+            return (
+                self.questionnaire_schema.get_parent_list_collector_for_repeating_block(
+                    source_block["id"]
+                )
+            )
+        # Handling of standard source block
+        return source_block["id"]
+
+    def validate_list_references(self):
+        lists_with_context = self.questionnaire_schema.lists_with_context
+
+        # We need to keep track of section index for: common_definitions.json#/section_enabled
+        for section_index, section in enumerate(self.questionnaire_schema.sections):
+            identifier_references = get_object_containing_key(section, "source")
+            for _, identifier_reference, parent_block in identifier_references:
+                if identifier_reference["source"] == "list":
+                    list_identifier = identifier_reference["identifier"]
+                    if parent_block:
+                        if (
+                            self.questionnaire_schema.block_ids.index(
+                                parent_block["id"]
+                            )
+                            < lists_with_context[list_identifier]["block_index"]
+                        ):
+                            self.add_error(
+                                error_messages.LIST_REFERENCED_BEFORE_CREATED.format(),
+                                list_id=list_identifier,
+                                section_id=section["id"],
+                                block_id=parent_block["id"],
+                            )
+                    elif (
+                        section_index
+                        < lists_with_context[list_identifier]["section_index"]
+                    ):
+                        # Section level "enabled" rule that can use list source,
+                        # check: common_definitions.json#/section_enabled
+                        self.add_error(
+                            error_messages.LIST_REFERENCED_BEFORE_CREATED.format(),
+                            list_name=list_identifier,
+                            section_id=section["id"],
+                        )