Add q_code validation (#147)

app/validators/answers/__init__.py

@@ -18,6 +18,4 @@ def get_answer_validator(answer, questionnaire_schema):
     }
     validator_type = validators.get(answer["type"], AnswerValidator)
 
-    if validator_type in [OptionAnswerValidator, NumberAnswerValidator]:
-        return validator_type(answer, questionnaire_schema)
-    return validator_type(answer)
+    return validator_type(answer, questionnaire_schema)

app/validators/answers/answer_validator.py

@@ -1,11 +1,122 @@
 from app.answer_type import AnswerType
+from app.validators.questionnaire_schema import get_object_containing_key
 from app.validators.validator import Validator
 
 
 class AnswerValidator(Validator):
-    def __init__(self, schema_element):
+    OPTION_MISSING_Q_CODE = "Option q_code must be provided"
+    ANSWER_MISSING_Q_CODE = "Answer q_code must be provided"
+    NON_CHECKBOX_OPTION_HAS_Q_CODE = "Non checkbox option cannot contain q_code"
+    DETAIL_ANSWER_MISSING_Q_CODE = "Detail answer q_code must be provided"
+    CHECKBOX_DETAIL_ANSWER_HAS_Q_CODE = "Checkbox detail answer cannot contain q_code"
+    CONFIRMATION_QUESTION_HAS_Q_CODE = "Confirmation question cannot contain q_code"
+    DATA_VERSION_NOT_0_0_1_Q_CODE_PRESENT = (
+        "q_code can only be used with data_version 0.0.1"
+    )
+    CHECKBOX_ANSWER_AND_OPTIONS_Q_CODE_MUTUALLY_EXLUSIVE = (
+        "Checkbox answer and option q_code are mutually exclusive"
+    )
+    CHECKBOX_ANSWER_OR_OPTIONS_MUST_HAVE_Q_CODES = (
+        "Either checkbox answer or options must have q_codes"
+    )
+
+    def __init__(self, schema_element, questionnaire_schema):
         super().__init__(schema_element)
         self.answer = schema_element
         self.answer_id = self.answer["id"]
         self.answer_type = AnswerType(self.answer["type"])
         self.context["answer_id"] = self.answer_id
+        self.questionnaire_schema = questionnaire_schema
+
+    def validate(self):
+        self._validate_q_codes()
+
+        return self.errors
+
+    def _validate_q_codes(self):
+        is_confirmation_question = (
+            self.questionnaire_schema.get_block_by_answer_id(self.answer_id).get("type")
+            == "ConfirmationQuestion"
+        )
+
+        if (
+            self.questionnaire_schema.schema["data_version"] != "0.0.1"
+            or is_confirmation_question
+        ):
+            has_q_code = get_object_containing_key(self.answer, key_name="q_code")
+            if has_q_code:
+                self.add_error(
+                    self.CONFIRMATION_QUESTION_HAS_Q_CODE
+                    if is_confirmation_question
+                    else self.DATA_VERSION_NOT_0_0_1_Q_CODE_PRESENT,
+                    answer_id=self.answer["id"],
+                )
+
+            return None
+
+        if self.answer_type is AnswerType.CHECKBOX:
+            self._validate_checkbox_q_code()
+
+        else:
+            if not self.answer.get("q_code"):
+
+                self.add_error(self.ANSWER_MISSING_Q_CODE, answer_id=self.answer["id"])
+
+            if self.answer.get("options") and self._validate_options_q_code():
+
+                self.add_error(self.OPTION_MISSING_Q_CODE, answer_id=self.answer["id"])
+
+    def _validate_options_q_code(self):
+        any_option_missing_q_code = False
+        for option in self.answer.get("options", []):
+            option_has_q_code = option.get("q_code")
+            is_checkbox = self.answer_type is AnswerType.CHECKBOX
+
+            if is_checkbox:
+                if not option_has_q_code:
+                    any_option_missing_q_code = True
+            elif option_has_q_code:
+                self.add_error(
+                    self.NON_CHECKBOX_OPTION_HAS_Q_CODE, answer_id=self.answer["id"]
+                )
+
+            if self._validate_detail_answer_q_code(option):
+                self.add_error(
+                    self.DETAIL_ANSWER_MISSING_Q_CODE, answer_id=self.answer["id"]
+                )
+
+        return any_option_missing_q_code
+
+    def _validate_detail_answer_q_code(self, option):
+        if detail_answer := option.get("detail_answer"):
+            has_q_code = detail_answer.get("q_code")
+            is_checkbox = self.answer_type is AnswerType.CHECKBOX
+            if is_checkbox:
+                if has_q_code:
+                    self.add_error(
+                        self.CHECKBOX_DETAIL_ANSWER_HAS_Q_CODE,
+                        answer_id=self.answer["id"],
+                    )
+            elif not has_q_code:
+                self.add_error(
+                    self.DETAIL_ANSWER_MISSING_Q_CODE, answer_id=self.answer["id"]
+                )
+
+    def _validate_checkbox_q_code(self):
+        has_answer_q_code = self.answer.get("q_code")
+        any_option_missing_q_code = self._validate_options_q_code()
+
+        if has_answer_q_code:
+            any_option_has_q_code = any(
+                True for option in self.answer.get("options", []) if "q_code" in option
+            )
+            if any_option_has_q_code:
+                self.add_error(
+                    self.CHECKBOX_ANSWER_AND_OPTIONS_Q_CODE_MUTUALLY_EXLUSIVE,
+                    answer_id=self.answer["id"],
+                )
+        elif any_option_missing_q_code:
+            self.add_error(
+                self.CHECKBOX_ANSWER_OR_OPTIONS_MUST_HAVE_Q_CODES,
+                answer_id=self.answer["id"],
+            )

app/validators/answers/number_answer_validator.py

@@ -22,8 +22,8 @@ class NumberAnswerValidator(AnswerValidator):
         "The referenced answer has a greater number of decimal places than answer"
     )
 
-    def __init__(self, schema_element, questionnaire_schema=None):
-        super().__init__(schema_element)
+    def __init__(self, schema_element, questionnaire_schema):
+        super().__init__(schema_element, questionnaire_schema)
         self.questionnaire_schema = questionnaire_schema
 
     def validate(self):

app/validators/answers/option_answer_validator.py

@@ -31,12 +31,12 @@ class OptionAnswerValidator(AnswerValidator):
         "option label from value answer_id do not match"
     )
 
-    def __init__(self, schema_element, questionnaire_schema=None):
-        super().__init__(schema_element)
-        if questionnaire_schema:
-            self.questionnaire_schema = questionnaire_schema
-            self.list_names = self.questionnaire_schema.list_names
-            self.block_ids = self.questionnaire_schema.block_ids
+    def __init__(self, schema_element, questionnaire_schema):
+        super().__init__(schema_element, questionnaire_schema)
+
+        self.questionnaire_schema = questionnaire_schema
+        self.list_names = self.questionnaire_schema.list_names
+        self.block_ids = self.questionnaire_schema.block_ids
 
     def validate(self):
         super().validate()

app/validators/questionnaire_schema.py

@@ -1,3 +1,4 @@
+# pylint: disable=too-many-public-methods
 import collections
 from collections import defaultdict
 from functools import cached_property, lru_cache
@@ -6,7 +7,10 @@
 from jsonpath_rw import parse
 
 from app.answer_type import AnswerType
-from app.validators.answers.number_answer_validator import MAX_NUMBER
+
+MAX_NUMBER = 9999999999
+MIN_NUMBER = -999999999
+MAX_DECIMAL_PLACES = 6
 
 
 def has_default_route(routing_rules):
@@ -262,7 +266,7 @@ def get_section(self, section_id):
 
     @lru_cache
     def get_block(self, block_id):
-        return self.blocks_by_id[block_id]
+        return self.blocks_by_id.get(block_id, None)
 
     @lru_cache
     def get_blocks(self, **filters):
@@ -339,6 +343,23 @@ def get_first_answer_in_block(self, block_id):
     def _get_path_id(self, path):
         return jp.match1(path + ".id", self.schema)
 
+    @lru_cache
+    def get_block_id_by_answer_id(self, answer_id):
+        for question, context in self.questions_with_context:
+            for answer in question.get("answers", []):
+                if answer_id == answer["id"]:
+                    return context["block"]
+                for option in answer.get("options", []):
+                    detail_answer = option.get("detail_answer")
+                    if detail_answer and answer_id == detail_answer["id"]:
+                        return context["block"]
+
+    @lru_cache
+    def get_block_by_answer_id(self, answer_id):
+        block_id = self.get_block_id_by_answer_id(answer_id)
+
+        return self.get_block(block_id)
+
     def _get_numeric_range_values(self, answer, answer_ranges):
         min_value = answer.get("minimum", {}).get("value", {})
         max_value = answer.get("maximum", {}).get("value", {})

tests/conftest.py

@@ -20,3 +20,7 @@ def get_mock_schema(questionnaire_schema=None, answers_with_context=None):
         questionnaire_schema.answers_with_context = answers_with_context
 
     return questionnaire_schema
+
+
+def get_mock_schema_with_data_version(data_version):
+    return QuestionnaireSchema({"data_version": data_version})