[feature/PI-565-questionnaire_rethink] add new spine questionnaires

pyproject.toml

@@ -27,6 +27,7 @@ urllib3 = "<3"
 orjson = "^3.9.15"
 attrs = "^24.2.0"
 locust = "^2.29.1"
+jsonschema = "^4.23.0"
 
 [tool.poetry.group.dev.dependencies]
 pre-commit = "^4.0.0"

src/layers/domain/core/questionnaire/tests/test_questionnaire_v3.py

@@ -0,0 +1,92 @@
+import json
+
+import pytest
+from domain.core.enum import Status
+from domain.core.questionnaire.v3 import Questionnaire
+from domain.core.timestamp import now
+from jsonschema import ValidationError
+
+VALID_SCHEMA = {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "type": "object",
+    "properties": {
+        "size": {
+            "type": "number",
+            "minimum": 1,
+            "maximum": 14,
+        },
+        "colour": {
+            "type": "string",
+            "enum": ["black", "white"],
+        },
+        "brand": {"type": "string"},  # not required
+    },
+    "required": ["size", "colour"],
+    "additionalProperties": False,
+}
+
+INVALID_SCHEMA = {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "type": "object",
+    "properties": {
+        "a-field": {
+            "type": "not-a-type",
+        }
+    },
+    "required": ["a-field"],
+}
+
+
+@pytest.mark.parametrize(
+    "data",
+    [
+        {"size": 1, "colour": "black"},
+        {"size": 14, "colour": "white"},
+        {"size": 7, "colour": "white", "brand": "something"},
+    ],
+)
+def test_schema_validation_pass(data):
+    questionnaire = Questionnaire(
+        name="foo", version="1", json_schema=json.dumps(VALID_SCHEMA)
+    )
+    response = questionnaire.validate(data=data)
+    assert response.name == "foo"
+    assert response.version == "1"
+    assert response.data == data
+    assert response.status is Status.ACTIVE
+    assert response.created_on.date() == now().date()
+    assert response.updated_on is None
+    assert response.deleted_on is None
+
+
+@pytest.mark.parametrize(
+    "data",
+    [
+        {"size": 1, "colour": "red"},
+        {"size": "not a number", "colour": "white"},
+        {
+            "size": 7,
+            "colour": "white",
+            "brand": "something",
+            "unknown_field": "foo",
+        },
+    ],
+)
+def test_schema_validation_fail(data):
+    questionnaire = Questionnaire(
+        name="foo", version="1", json_schema=json.dumps(VALID_SCHEMA)
+    )
+    with pytest.raises(ValidationError):
+        questionnaire.validate(data=data)
+
+
+@pytest.mark.parametrize(
+    "schema",
+    [
+        {},
+        INVALID_SCHEMA,
+        {"$schema": "invalid-path"},
+    ],
+)
+def test_invalid_schema(schema):
+    Questionnaire(name="name", version="123", schema=json.dumps(schema))

src/layers/domain/core/questionnaire/v3.py

@@ -0,0 +1,37 @@
+from datetime import datetime
+
+import jsonschema
+from domain.core.base import BaseModel
+from domain.core.enum import Status
+from domain.core.timestamp import now
+from pydantic import Field, Json, validator
+
+
+class Questionnaire(BaseModel):
+    name: str
+    version: str
+    json_schema: Json
+
+    @validator("json_schema")
+    def validate_json_schema(cls, json_schema):
+        try:
+            jsonschema.Draft7Validator.check_schema(json_schema)
+        except jsonschema.SchemaError as err:
+            raise ValueError(err.message)
+        if not json_schema:
+            raise ValueError("Cannot be empty")
+        return json_schema
+
+    def validate(self, data) -> "QuestionnaireResponse":
+        jsonschema.validate(instance=data, schema=self.json_schema)
+        return QuestionnaireResponse(name=self.name, version=self.version, data=data)
+
+
+class QuestionnaireResponse(BaseModel):
+    name: str
+    version: str
+    data: dict
+    status: Status = Status.ACTIVE
+    created_on: datetime = Field(default_factory=now)
+    updated_on: str = None
+    deleted_on: str = None

src/layers/domain/repository/questionnaire_repository/__init__.py

@@ -1,53 +1 @@
-from pathlib import Path
-
-from domain.core.questionnaire.v2 import Questionnaire
-from domain.repository.errors import ItemNotFound
-from domain.repository.questionnaire_repository.deserialisers import (
-    QUESTION_DESERIALISERS,
-)
-from event.json import json_load
-
-PATH_TO_QUESTIONNAIRES = Path(__file__).parent / "questionnaires"
-
-
-def deserialise_question(question: dict) -> dict:
-    for field, deserialiser in QUESTION_DESERIALISERS.items():
-        value = question.get(field)
-        if value:
-            question[field] = deserialiser(value)
-    return question
-
-
-def version_from_file_path(file_path: Path) -> int:
-    return int(file_path.stem.lstrip("v"))
-
-
-def get_latest_questions_by_name(name: str) -> Path | None:
-    possible_paths = PATH_TO_QUESTIONNAIRES.glob(f"{name}/v*.json")
-    paths_sorted_by_version = sorted(possible_paths, key=version_from_file_path)
-    try:
-        path = paths_sorted_by_version[-1]
-    except IndexError:
-        path = None
-    return path
-
-
-def read_questions(path: Path):
-    with open(path, "r") as fp:
-        raw_questions = json_load(fp)
-    return list(map(deserialise_question, raw_questions))
-
-
-class QuestionnaireRepository:
-
-    def read(self, name: str) -> Questionnaire:
-        path = get_latest_questions_by_name(name=name)
-        if not path:
-            raise ItemNotFound(name, item_type=Questionnaire)
-
-        version = version_from_file_path(path)
-        questions = read_questions(path=path)
-        questionnaire = Questionnaire(name=name, version=version)
-        for question in questions:
-            questionnaire.add_question(**question)
-        return questionnaire
+from .v1 import *  # noqa

src/layers/domain/repository/questionnaire_repository/v1/__init__.py

@@ -0,0 +1,52 @@
+from pathlib import Path
+
+from domain.core.questionnaire.v2 import Questionnaire
+from domain.repository.errors import ItemNotFound
+from event.json import json_load
+
+from .deserialisers import QUESTION_DESERIALISERS
+
+PATH_TO_QUESTIONNAIRES = Path(__file__).parent / "questionnaires"
+
+
+def deserialise_question(question: dict) -> dict:
+    for field, deserialiser in QUESTION_DESERIALISERS.items():
+        value = question.get(field)
+        if value:
+            question[field] = deserialiser(value)
+    return question
+
+
+def version_from_file_path(file_path: Path) -> int:
+    return int(file_path.stem.lstrip("v"))
+
+
+def get_latest_questions_by_name(name: str) -> Path | None:
+    possible_paths = PATH_TO_QUESTIONNAIRES.glob(f"{name}/v*.json")
+    paths_sorted_by_version = sorted(possible_paths, key=version_from_file_path)
+    try:
+        path = paths_sorted_by_version[-1]
+    except IndexError:
+        path = None
+    return path
+
+
+def read_questions(path: Path):
+    with open(path, "r") as fp:
+        raw_questions = json_load(fp)
+    return list(map(deserialise_question, raw_questions))
+
+
+class QuestionnaireRepository:
+
+    def read(self, name: str) -> Questionnaire:
+        path = get_latest_questions_by_name(name=name)
+        if not path:
+            raise ItemNotFound(name, item_type=Questionnaire)
+
+        version = version_from_file_path(path)
+        questions = read_questions(path=path)
+        questionnaire = Questionnaire(name=name, version=version)
+        for question in questions:
+            questionnaire.add_question(**question)
+        return questionnaire

src/layers/domain/repository/questionnaire_repository/questionnaires/tests/test_spine_device_questionnaire.py → src/layers/domain/repository/questionnaire_repository/v1/questionnaires/tests/test_spine_device_questionnaire.py

@@ -1,7 +1,7 @@
 import pytest
 from domain.core.questionnaire.v2 import Questionnaire
-from domain.repository.questionnaire_repository import QuestionnaireRepository
-from domain.repository.questionnaire_repository.questionnaires import (
+from domain.repository.questionnaire_repository.v1 import QuestionnaireRepository
+from domain.repository.questionnaire_repository.v1.questionnaires import (
     QuestionnaireInstance,
 )
 from event.json import json_load

src/layers/domain/repository/questionnaire_repository/questionnaires/tests/test_spine_endpoint_questionnaire.py → src/layers/domain/repository/questionnaire_repository/v1/questionnaires/tests/test_spine_endpoint_questionnaire.py

@@ -1,7 +1,7 @@
 import pytest
 from domain.core.questionnaire.v2 import Questionnaire
 from domain.repository.questionnaire_repository import QuestionnaireRepository
-from domain.repository.questionnaire_repository.questionnaires import (
+from domain.repository.questionnaire_repository.v1.questionnaires import (
     QuestionnaireInstance,
 )
 from event.json import json_load

src/layers/domain/repository/questionnaire_repository/tests/test_deserialisers.py → src/layers/domain/repository/questionnaire_repository/v1/tests/test_deserialisers.py

@@ -1,6 +1,6 @@
 import pytest
 from domain.core.questionnaire.v1 import NoSuchQuestionType
-from domain.repository.questionnaire_repository.deserialisers import (
+from domain.repository.questionnaire_repository.v1.deserialisers import (
     _deserialise_answer_type,
     _deserialise_rule,
 )

src/layers/domain/repository/questionnaire_repository/tests/test_questionnaire_repository.py → src/layers/domain/repository/questionnaire_repository/v1/tests/test_questionnaire_repository.py

@@ -2,11 +2,11 @@
 
 import pytest
 from domain.repository.errors import ItemNotFound
-from domain.repository.questionnaire_repository import (
+from domain.repository.questionnaire_repository.v1 import (
     PATH_TO_QUESTIONNAIRES,
     QuestionnaireRepository,
 )
-from domain.repository.questionnaire_repository.questionnaires import (
+from domain.repository.questionnaire_repository.v1.questionnaires import (
     QuestionnaireInstance,
 )
 

src/layers/domain/repository/questionnaire_repository/v2/__init__.py

@@ -0,0 +1,36 @@
+from pathlib import Path
+
+from domain.core.questionnaire.v3 import Questionnaire
+from domain.repository.errors import ItemNotFound
+
+PATH_TO_QUESTIONNAIRES = Path(__file__).parent / "questionnaires"
+
+
+def version_from_file_path(file_path: Path) -> int:
+    return int(file_path.stem.lstrip("v"))
+
+
+def get_latest_schema_path_by_name(name: str) -> Path | None:
+    possible_paths = PATH_TO_QUESTIONNAIRES.glob(f"{name}/v*.json")
+    paths_sorted_by_version = sorted(possible_paths, key=version_from_file_path)
+    try:
+        path = paths_sorted_by_version[-1]
+    except IndexError:
+        path = None
+    return path
+
+
+def read_schema(path: Path) -> str:
+    with open(path, "r") as fp:
+        return fp.read()
+
+
+class QuestionnaireRepository:
+
+    def read(self, name: str) -> Questionnaire:
+        path = get_latest_schema_path_by_name(name=name)
+        if not path:
+            raise ItemNotFound(name, item_type=Questionnaire)
+        version = version_from_file_path(path)
+        schema = read_schema(path=path)
+        return Questionnaire(name=name, version=version, json_schema=schema)

src/layers/domain/repository/questionnaire_repository/v2/questionnaires/__init__.py

@@ -0,0 +1,8 @@
+from enum import StrEnum, auto
+
+
+class QuestionnaireInstance(StrEnum):
+    SPINE_AS = auto()
+    SPINE_MHS = auto()
+    SPINE_AS_INTERACTIONS = auto()
+    SPINE_MHS_INTERACTIONS = auto()

src/layers/domain/repository/questionnaire_repository/v2/questionnaires/spine_as/field_mapping.json

@@ -0,0 +1,18 @@
+{
+  "nhs_mhs_manufacturer_org": "MHS Manufacturer Organisation",
+  "nhs_mhs_party_key": "Party Key",
+  "nhs_id_code": "ODS Code",
+  "nhs_product_name": "Product Name",
+  "unique_identifier": "ASID",
+  "nhs_as_client": "Client ODS Codes",
+  "nhs_approver_urp": "Approver URP",
+  "nhs_date_approved": "Date Approved",
+  "nhs_requestor_urp": "Requestor URP",
+  "nhs_date_requested": "Date Requested",
+  "nhs_product_key": "Product Key",
+  "nhs_product_version": "Product Version",
+  "nhs_as_acf": "AS ACF",
+  "nhs_temp_uid": "Temp UID",
+  "description": "Description",
+  "nhs_as_category_bag": "AS Category Bag"
+}