diff --git a/care/emr/tests/test_questionnaire_api.py b/care/emr/tests/test_questionnaire_api.py index 390c50a281..6b67abe3ef 100644 --- a/care/emr/tests/test_questionnaire_api.py +++ b/care/emr/tests/test_questionnaire_api.py @@ -41,6 +41,24 @@ def setUp(self): self.questionnaire_data = self._create_questionnaire() self.questions = self.questionnaire_data.get("questions", []) + @classmethod + def _prepare_questionnaire_definition(cls, definition): + """ + Fills the fields QuestionnaireCreateSpec requires but the historical + test payloads predate: auth_context (required, no default) and a + unique id on every question. + """ + definition.setdefault("auth_context", "instance") + cls._assign_question_ids(definition.get("questions") or []) + return definition + + @classmethod + def _assign_question_ids(cls, questions): + for question in questions: + question.setdefault("id", str(uuid.uuid4())) + if question.get("questions"): + cls._assign_question_ids(question["questions"]) + def _submit_questionnaire(self, payload): """ Submits a questionnaire response and returns the submission results. @@ -52,7 +70,7 @@ def _submit_questionnaire(self, payload): tuple: A pair of (status_code, response_data) from the submission """ submit_url = reverse( - "questionnaire-submit", kwargs={"slug": self.questionnaire_data["slug"]} + "questionnaire-submit", kwargs={"external_id": self.questionnaire_data["id"]} ) response = self.client.post(submit_url, payload, format="json") return response.status_code, response.json() @@ -156,7 +174,9 @@ def _create_questionnaire(self): } response = self.client.post( - self.base_url, questionnaire_definition, format="json" + self.base_url, + self._prepare_questionnaire_definition(questionnaire_definition), + format="json", ) self.assertEqual( response.status_code, @@ -300,12 +320,14 @@ def test_submit_inactive_questionnaire(self): ], } response = self.client.post( - self.base_url, questionnaire_definition, format="json" + self.base_url, + self._prepare_questionnaire_definition(questionnaire_definition), + format="json", ) self.assertEqual(response.status_code, 200) submit_url = reverse( - "questionnaire-submit", kwargs={"slug": response.json()["slug"]} + "questionnaire-submit", kwargs={"external_id": response.json()["id"]} ) payload = { @@ -343,7 +365,9 @@ def test_false_choice_values_validations(self): ], } response = self.client.post( - self.base_url, questionnaire_definition, format="json" + self.base_url, + self._prepare_questionnaire_definition(questionnaire_definition), + format="json", ) data = response.json() status_code = response.status_code @@ -392,7 +416,9 @@ def _create_questionnaire(self, questions): "questions": questions, } response = self.client.post( - self.base_url, questionnaire_definition, format="json" + self.base_url, + self._prepare_questionnaire_definition(questionnaire_definition), + format="json", ) self.assertEqual( response.status_code, @@ -1734,7 +1760,9 @@ def _create_questionnaire(self): } response = self.client.post( - self.base_url, questionnaire_definition, format="json" + self.base_url, + self._prepare_questionnaire_definition(questionnaire_definition), + format="json", ) self.assertEqual( response.status_code, @@ -1804,7 +1832,9 @@ def _create_questionnaire(self, questions=None): "questions": questions, } response = self.client.post( - self.base_url, questionnaire_definition, format="json" + self.base_url, + self._prepare_questionnaire_definition(questionnaire_definition), + format="json", ) self.assertEqual( response.status_code, @@ -1987,7 +2017,7 @@ def test_repeatable_group_responses_validation(self): ] ) submit_url = reverse( - "questionnaire-submit", kwargs={"slug": self.questionnaire_data["slug"]} + "questionnaire-submit", kwargs={"external_id": self.questionnaire_data["id"]} ) response = self.client.post(submit_url, payload, format="json") self.assertEqual( @@ -2050,7 +2080,9 @@ def _create_questionnaire(self): } response = self.client.post( - self.base_url, questionnaire_definition, format="json" + self.base_url, + self._prepare_questionnaire_definition(questionnaire_definition), + format="json", ) self.assertEqual( response.status_code, @@ -2101,7 +2133,7 @@ def _create_questionnaire(self): Returns: dict: Basic questionnaire definition for permission testing """ - return { + return self._prepare_questionnaire_definition({ "title": "Permission Test Assessment", "slug": "permission-test", "description": "Questionnaire for testing access controls", @@ -2121,13 +2153,15 @@ def _create_questionnaire(self): }, } ], - } + }) def create_questionnaire_instance(self): """ Helper method to create a questionnaire instance for testing permissions. - Temporarily authenticates as super user to ensure creation, then reverts - to regular user authentication. + Temporarily authenticates as super user to create the questionnaire and + tag it with the test organization (creation no longer accepts + organizations — visibility is granted via set_organizations), then + reverts to regular user authentication. Returns: dict: The created questionnaire instance data @@ -2136,16 +2170,28 @@ def create_questionnaire_instance(self): response = self.client.post( self.base_url, self._create_questionnaire(), format="json" ) + questionnaire = response.json() + set_organizations_url = reverse( + "questionnaire-set-organizations", + kwargs={"external_id": questionnaire["id"]}, + ) + self.client.post( + set_organizations_url, + {"organizations": [str(self.organization.external_id)]}, + format="json", + ) self.client.force_authenticate(self.user) - return response.json() + return questionnaire - def test_questionnaire_list_access_denied(self): + def test_questionnaire_list_scoped_to_user_grants(self): """ - Verifies that users without proper permissions cannot list questionnaires. - Tests the basic access control for questionnaire listing functionality. + Verifies that the questionnaire list is scoped: a user without any + grants gets an empty list even when questionnaires exist. """ + self.create_questionnaire_instance() response = self.client.get(self.base_url) - self.assertEqual(response.status_code, 403) + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json()["count"], 0) def test_questionnaire_list_access_granted(self): """ @@ -2169,10 +2215,10 @@ def test_questionnaire_creation_access_denied(self): ) self.assertEqual(response.status_code, 403) - def test_questionnaire_creation_access_granted(self): + def test_instance_questionnaire_creation_restricted_to_super_users(self): """ - Verifies that users with write permissions can successfully create questionnaires. - Tests proper access grant for users with explicit write permissions. + Instance level questionnaires can only be created by super users — + write permission alone is not enough. """ permissions = [QuestionnairePermissions.can_write_questionnaire.name] role = self.create_role_with_permissions(permissions) @@ -2185,19 +2231,41 @@ def test_questionnaire_creation_access_granted(self): questionnaire_data["title"] = self.fake.text(max_nb_chars=255) response = self.client.post(self.base_url, questionnaire_data, format="json") + self.assertEqual(response.status_code, 403) + + def test_questionnaire_creation_access_granted(self): + """ + Verifies that users with write permissions in a facility organization + can create a user scoped questionnaire in that facility. + """ + facility = self.create_facility(self.super_user) + facility_organization = self.create_facility_organization(facility) + permissions = [QuestionnairePermissions.can_write_questionnaire.name] + role = self.create_role_with_permissions(permissions) + self.attach_role_facility_organization_user( + facility_organization, self.user, role + ) + + questionnaire_data = self._create_questionnaire() + questionnaire_data["auth_context"] = "user" + questionnaire_data["facility"] = str(facility.external_id) + # patient subject_type is only allowed at the instance level + questionnaire_data["subject_type"] = "encounter" + response = self.client.post(self.base_url, questionnaire_data, format="json") self.assertEqual(response.status_code, 200) def test_questionnaire_retrieval_access_denied(self): """ - Verifies that users without proper permissions cannot retrieve individual questionnaires. - Tests access control for detailed questionnaire viewing. + Verifies that users without proper permissions cannot retrieve individual + questionnaires. Unauthorized questionnaires are hidden from the scoped + queryset, so the API responds with 404. """ questionnaire = self.create_questionnaire_instance() detail_url = reverse( - "questionnaire-detail", kwargs={"slug": questionnaire["slug"]} + "questionnaire-detail", kwargs={"external_id": questionnaire["id"]} ) response = self.client.get(detail_url) - self.assertEqual(response.status_code, 403) + self.assertEqual(response.status_code, 404) def test_questionnaire_retrieval_access_granted(self): """ @@ -2210,7 +2278,7 @@ def test_questionnaire_retrieval_access_granted(self): questionnaire = self.create_questionnaire_instance() detail_url = reverse( - "questionnaire-detail", kwargs={"slug": questionnaire["slug"]} + "questionnaire-detail", kwargs={"external_id": questionnaire["id"]} ) response = self.client.get(detail_url) self.assertEqual(response.status_code, 200) @@ -2230,7 +2298,7 @@ def test_questionnaire_deletion_access_denied(self): questionnaire = self.create_questionnaire_instance() detail_url = reverse( - "questionnaire-detail", kwargs={"slug": questionnaire["slug"]} + "questionnaire-detail", kwargs={"external_id": questionnaire["id"]} ) response = self.client.delete(detail_url) self.assertEqual(response.status_code, 403) @@ -2242,7 +2310,7 @@ def test_questionnaire_deletion_super_user_allowed(self): """ questionnaire = self.create_questionnaire_instance() detail_url = reverse( - "questionnaire-detail", kwargs={"slug": questionnaire["slug"]} + "questionnaire-detail", kwargs={"external_id": questionnaire["id"]} ) self.client.force_authenticate(user=self.super_user) @@ -2263,12 +2331,17 @@ def test_questionnaire_update_access_denied(self): questionnaire = self.create_questionnaire_instance() detail_url = reverse( - "questionnaire-detail", kwargs={"slug": questionnaire["slug"]} + "questionnaire-detail", kwargs={"external_id": questionnaire["id"]} ) updated_data = self._create_questionnaire() updated_data["questions"] = [ - {"link_id": "1", "type": "boolean", "text": "Modified question text"} + { + "id": str(uuid.uuid4()), + "link_id": "1", + "type": "boolean", + "text": "Modified question text", + } ] response = self.client.put(detail_url, updated_data, format="json") @@ -2282,7 +2355,7 @@ def test_questionnaire_update_super_user_allowed(self): """ questionnaire = self.create_questionnaire_instance() detail_url = reverse( - "questionnaire-detail", kwargs={"slug": questionnaire["slug"]} + "questionnaire-detail", kwargs={"external_id": questionnaire["id"]} ) self.client.force_authenticate(user=self.super_user) @@ -2290,6 +2363,7 @@ def test_questionnaire_update_super_user_allowed(self): updated_data["description"] = "" updated_data["questions"] = [ { + "id": str(uuid.uuid4()), "link_id": "1", "type": "boolean", "text": "Modified question text", @@ -2318,7 +2392,7 @@ def test_questionnaire_update_super_user_allowed(self): # questionnaire = self.create_questionnaire_instance() # self.questionnaire_data = questionnaire # detail_url = reverse( - # "questionnaire-detail", kwargs={"slug": questionnaire["slug"]} + # "questionnaire-detail", kwargs={"external_id": questionnaire["id"]} # ) # self.client.force_authenticate(user=self.super_user) # @@ -2345,21 +2419,21 @@ def test_questionnaire_update_super_user_allowed(self): def test_questionnaire_organization_list_access_denied(self): """ Verifies that users without proper permissions cannot view the organizations - associated with a questionnaire. - + associated with a questionnaire. The questionnaire itself is hidden from + the scoped queryset, so the API responds with 404. """ questionnaire = self.create_questionnaire_instance() organization_list_url = reverse( - "questionnaire-get-organizations", kwargs={"slug": questionnaire["slug"]} + "questionnaire-get-organizations", kwargs={"external_id": questionnaire["id"]} ) response = self.client.get(organization_list_url) - self.assertEqual(response.status_code, 403) + self.assertEqual(response.status_code, 404) - def test_questionnaire_organization_list_access_granted(self): + def test_questionnaire_organization_list_restricted_to_super_users(self): """ - Verifies that users with read permissions can successfully view the organizations - associated with a questionnaire. - + Verifies that organization management on instance questionnaires is + restricted to super users: a user with read permissions can see the + questionnaire but not its organizations. """ permissions = [QuestionnairePermissions.can_read_questionnaire.name] role = self.create_role_with_permissions(permissions) @@ -2367,44 +2441,47 @@ def test_questionnaire_organization_list_access_granted(self): questionnaire = self.create_questionnaire_instance() organization_list_url = reverse( - "questionnaire-get-organizations", kwargs={"slug": questionnaire["slug"]} + "questionnaire-get-organizations", kwargs={"external_id": questionnaire["id"]} ) response = self.client.get(organization_list_url) + self.assertEqual(response.status_code, 403) + + self.client.force_authenticate(user=self.super_user) + response = self.client.get(organization_list_url) self.assertEqual(response.status_code, 200) + self.assertEqual(response.json()["count"], 1) - def test_set_organizations_without_authentication(self): - """Tests that setting organizations without authentication returns 403 forbidden.""" + def test_set_organizations_without_permissions(self): + """Setting organizations on a questionnaire the user cannot even see returns 404.""" questionnaire = self.create_questionnaire_instance() url = reverse( - "questionnaire-set-organizations", kwargs={"slug": questionnaire["slug"]} + "questionnaire-set-organizations", kwargs={"external_id": questionnaire["id"]} ) payload = {"organizations": [self.create_organization().external_id]} response = self.client.post(url, payload, format="json") - self.assertEqual(response.status_code, 403) + self.assertEqual(response.status_code, 404) def test_set_organizations_with_read_only_access(self): """Tests that setting organizations with read-only permissions returns 403 forbidden.""" - questionnaire = self.create_questionnaire_instance() - url = reverse( - "questionnaire-set-organizations", kwargs={"slug": questionnaire["slug"]} - ) - permissions = [QuestionnairePermissions.can_read_questionnaire.name] role = self.create_role_with_permissions(permissions) self.attach_role_organization_user(self.organization, self.user, role) - payload = {"organizations": [self.create_organization().external_id]} - response = self.client.post(url, payload, format="json") - self.assertEqual(response.status_code, 403) - - def test_set_organizations_with_invalid_organization_id(self): - """Tests that setting organizations with non-existent organization ID returns 404 not found.""" questionnaire = self.create_questionnaire_instance() url = reverse( - "questionnaire-set-organizations", kwargs={"slug": questionnaire["slug"]} + "questionnaire-set-organizations", kwargs={"external_id": questionnaire["id"]} ) + payload = {"organizations": [self.create_organization().external_id]} + response = self.client.post(url, payload, format="json") + self.assertEqual(response.status_code, 403) + + def test_set_organizations_restricted_to_super_users(self): + """ + Organization management on instance questionnaires is restricted to + super users — write permission alone is rejected. + """ permissions = [ QuestionnairePermissions.can_read_questionnaire.name, QuestionnairePermissions.can_write_questionnaire.name, @@ -2412,42 +2489,35 @@ def test_set_organizations_with_invalid_organization_id(self): role = self.create_role_with_permissions(permissions) self.attach_role_organization_user(self.organization, self.user, role) - payload = {"organizations": [uuid.uuid4()]} + questionnaire = self.create_questionnaire_instance() + url = reverse( + "questionnaire-set-organizations", kwargs={"external_id": questionnaire["id"]} + ) + + payload = {"organizations": [self.organization.external_id]} response = self.client.post(url, payload, format="json") - self.assertEqual(response.status_code, 404) + self.assertEqual(response.status_code, 403) - def test_set_organizations_without_organization_access(self): - """Tests that setting organizations without access to target organization returns 403 forbidden.""" + def test_set_organizations_with_invalid_organization_id(self): + """Tests that setting organizations with non-existent organization ID returns 404 not found.""" questionnaire = self.create_questionnaire_instance() url = reverse( - "questionnaire-set-organizations", kwargs={"slug": questionnaire["slug"]} + "questionnaire-set-organizations", kwargs={"external_id": questionnaire["id"]} ) - permissions = [ - QuestionnairePermissions.can_read_questionnaire.name, - QuestionnairePermissions.can_write_questionnaire.name, - ] - role = self.create_role_with_permissions(permissions) - self.attach_role_organization_user(self.organization, self.user, role) - - payload = {"organizations": [self.create_organization().external_id]} + self.client.force_authenticate(user=self.super_user) + payload = {"organizations": [uuid.uuid4()]} response = self.client.post(url, payload, format="json") - self.assertEqual(response.status_code, 403) + self.assertEqual(response.status_code, 404) def test_set_organizations_with_valid_access(self): - """Tests that setting organizations succeeds with proper permissions and organization access.""" + """Tests that a super user can set questionnaire organizations.""" questionnaire = self.create_questionnaire_instance() url = reverse( - "questionnaire-set-organizations", kwargs={"slug": questionnaire["slug"]} + "questionnaire-set-organizations", kwargs={"external_id": questionnaire["id"]} ) - permissions = [ - QuestionnairePermissions.can_read_questionnaire.name, - QuestionnairePermissions.can_write_questionnaire.name, - ] - role = self.create_role_with_permissions(permissions) - self.attach_role_organization_user(self.organization, self.user, role) - + self.client.force_authenticate(user=self.super_user) payload = {"organizations": [self.organization.external_id]} response = self.client.post(url, payload, format="json") self.assertEqual(response.status_code, 200) @@ -2480,6 +2550,7 @@ def _create_questionnaire(self): "slug": "appointment", "status": "active", "subject_type": "encounter", + "auth_context": "instance", "organizations": [str(self.organization.external_id)], "questions": [ { @@ -2515,7 +2586,7 @@ def _submit(self, results): "results": results, } url = reverse( - "questionnaire-submit", kwargs={"slug": self.questionnaire["slug"]} + "questionnaire-submit", kwargs={"external_id": self.questionnaire["id"]} ) resp = self.client.post(url, payload, format="json") return resp.status_code, resp.json() @@ -2636,6 +2707,7 @@ def _create_questionnaire(self): "slug": "appointment-any", "status": "active", "subject_type": "encounter", + "auth_context": "instance", "organizations": [str(self.organization.external_id)], "questions": [ { @@ -2671,7 +2743,7 @@ def _submit(self, results): "results": results, } url = reverse( - "questionnaire-submit", kwargs={"slug": self.questionnaire["slug"]} + "questionnaire-submit", kwargs={"external_id": self.questionnaire["id"]} ) resp = self.client.post(url, payload, format="json") return resp.status_code, resp.json() diff --git a/care/emr/tests/test_questionnaire_scoping_api.py b/care/emr/tests/test_questionnaire_scoping_api.py new file mode 100644 index 0000000000..f41b1d382e --- /dev/null +++ b/care/emr/tests/test_questionnaire_scoping_api.py @@ -0,0 +1,396 @@ +import uuid + +from django.urls import reverse + +from care.emr.models import Questionnaire +from care.security.permissions.questionnaire import QuestionnairePermissions +from care.utils.tests.base import CareAPITestBase + + +def questionnaire_definition(slug, **overrides): + """Minimal valid questionnaire create payload.""" + definition = { + "title": f"Questionnaire {slug}", + "slug": slug, + "version": "1.0", + "description": "Questionnaire scoping test", + "status": "active", + "subject_type": "encounter", + "auth_context": "instance", + "questions": [ + { + "id": str(uuid.uuid4()), + "link_id": "1", + "type": "string", + "text": "Note", + } + ], + } + definition.update(overrides) + return definition + + +class QuestionnaireScopingTestBase(CareAPITestBase): + def setUp(self): + super().setUp() + self.super_user = self.create_super_user() + self.organization = self.create_organization(org_type="govt") + self.facility = self.create_facility(self.super_user) + self.facility_organization = self.create_facility_organization(self.facility) + self.base_url = reverse("questionnaire-list") + self.client.force_authenticate(user=self.super_user) + + def create_questionnaire(self, slug, **overrides): + response = self.client.post( + self.base_url, questionnaire_definition(slug, **overrides), format="json" + ) + self.assertEqual( + response.status_code, + 200, + f"Questionnaire creation failed: {response.json()}", + ) + return response.json() + + def create_facility_questionnaire(self, slug, facility=None, **overrides): + facility = facility or self.facility + return self.create_questionnaire( + slug, + auth_context="facility", + facility=str(facility.external_id), + **overrides, + ) + + def detail_url(self, questionnaire_id): + return reverse( + "questionnaire-detail", kwargs={"external_id": questionnaire_id} + ) + + def list_slugs(self, params=None): + response = self.client.get(self.base_url, params or {}) + self.assertEqual(response.status_code, 200) + return {entry["slug"] for entry in response.json()["results"]} + + +class QuestionnaireAuthContextCreateTests(QuestionnaireScopingTestBase): + """Validation of the auth_context / facility / subject_type create rules.""" + + def test_create_requires_auth_context(self): + definition = questionnaire_definition("missing-auth-context") + del definition["auth_context"] + response = self.client.post(self.base_url, definition, format="json") + self.assertEqual(response.status_code, 400) + errors = response.json()["errors"] + self.assertTrue( + any(error["loc"] == ["auth_context"] for error in errors), + f"Expected a missing auth_context error, got {errors}", + ) + + def test_facility_context_requires_facility(self): + response = self.client.post( + self.base_url, + questionnaire_definition("facility-no-facility", auth_context="facility"), + format="json", + ) + self.assertEqual(response.status_code, 400) + self.assertIn("Facility is required", str(response.json()["errors"])) + + def test_user_context_requires_facility(self): + response = self.client.post( + self.base_url, + questionnaire_definition("user-no-facility", auth_context="user"), + format="json", + ) + self.assertEqual(response.status_code, 400) + self.assertIn("Facility is required", str(response.json()["errors"])) + + def test_facility_organization_context_requires_facility_organization(self): + response = self.client.post( + self.base_url, + questionnaire_definition( + "org-no-org", auth_context="facility_organization" + ), + format="json", + ) + self.assertEqual(response.status_code, 400) + self.assertIn( + "Facility organization is required", str(response.json()["errors"]) + ) + + def test_patient_subject_type_rejected_outside_instance(self): + response = self.client.post( + self.base_url, + questionnaire_definition( + "facility-patient-subject", + auth_context="facility", + facility=str(self.facility.external_id), + subject_type="patient", + ), + format="json", + ) + self.assertEqual(response.status_code, 400) + self.assertIn( + "Patient questionnaires are only supported at the instance level", + str(response.json()["errors"]), + ) + + def test_patient_subject_type_allowed_at_instance(self): + questionnaire = self.create_questionnaire( + "instance-patient-subject", subject_type="patient" + ) + self.assertEqual(questionnaire["subject_type"], "patient") + + def test_facility_create_persists_facility_scope(self): + questionnaire = self.create_facility_questionnaire("facility-scoped") + obj = Questionnaire.objects.get(external_id=questionnaire["id"]) + self.assertEqual(obj.auth_context, "facility") + self.assertEqual(obj.facility.external_id, self.facility.external_id) + + def test_update_ignores_auth_context_and_subject_type(self): + questionnaire = self.create_facility_questionnaire("immutable-scope") + payload = questionnaire_definition( + "immutable-scope", + auth_context="instance", + subject_type="patient", + questions=questionnaire["questions"], + ) + response = self.client.put( + self.detail_url(questionnaire["id"]), payload, format="json" + ) + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json()["subject_type"], "encounter") + obj = Questionnaire.objects.get(external_id=questionnaire["id"]) + self.assertEqual(obj.auth_context, "facility") + self.assertEqual(obj.subject_type, "encounter") + self.assertEqual(obj.facility.external_id, self.facility.external_id) + + +class QuestionnaireVisibilityTests(QuestionnaireScopingTestBase): + """Scoped queryset behaviour for facility and user questionnaires.""" + + def setUp(self): + super().setUp() + self.reader = self.create_user() + read_role = self.create_role_with_permissions( + [QuestionnairePermissions.can_read_questionnaire.name] + ) + self.attach_role_facility_organization_user( + self.facility_organization, self.reader, read_role + ) + + def set_facility_organizations(self, questionnaire_id, organizations): + url = reverse( + "questionnaire-set-facility-organizations", + kwargs={"external_id": questionnaire_id}, + ) + response = self.client.post( + url, + {"facility_organizations": [str(o.external_id) for o in organizations]}, + format="json", + ) + self.assertEqual(response.status_code, 200) + + def test_facility_questionnaire_visible_to_facility_org_member(self): + questionnaire = self.create_facility_questionnaire("facility-visible") + self.set_facility_organizations( + questionnaire["id"], [self.facility_organization] + ) + + self.client.force_authenticate(user=self.reader) + self.assertIn("facility-visible", self.list_slugs()) + response = self.client.get(self.detail_url(questionnaire["id"])) + self.assertEqual(response.status_code, 200) + + def test_facility_questionnaire_hidden_without_organization_tagging(self): + questionnaire = self.create_facility_questionnaire("facility-untagged") + + self.client.force_authenticate(user=self.reader) + self.assertNotIn("facility-untagged", self.list_slugs()) + response = self.client.get(self.detail_url(questionnaire["id"])) + self.assertEqual(response.status_code, 404) + + def test_facility_questionnaire_hidden_from_other_facility_members(self): + questionnaire = self.create_facility_questionnaire("facility-private") + self.set_facility_organizations( + questionnaire["id"], [self.facility_organization] + ) + + other_facility = self.create_facility(self.super_user) + other_org = self.create_facility_organization(other_facility) + outsider = self.create_user() + read_role = self.create_role_with_permissions( + [QuestionnairePermissions.can_read_questionnaire.name] + ) + self.attach_role_facility_organization_user(other_org, outsider, read_role) + + self.client.force_authenticate(user=outsider) + self.assertNotIn("facility-private", self.list_slugs()) + response = self.client.get(self.detail_url(questionnaire["id"])) + self.assertEqual(response.status_code, 404) + + def test_user_questionnaire_visible_only_to_creator(self): + author = self.create_user() + write_role = self.create_role_with_permissions( + [QuestionnairePermissions.can_write_questionnaire.name] + ) + self.attach_role_facility_organization_user( + self.facility_organization, author, write_role + ) + + self.client.force_authenticate(user=author) + response = self.client.post( + self.base_url, + questionnaire_definition( + "user-private", + auth_context="user", + facility=str(self.facility.external_id), + ), + format="json", + ) + self.assertEqual(response.status_code, 200, response.json()) + questionnaire = response.json() + + self.assertIn("user-private", self.list_slugs()) + + self.client.force_authenticate(user=self.reader) + self.assertNotIn("user-private", self.list_slugs()) + response = self.client.get(self.detail_url(questionnaire["id"])) + self.assertEqual(response.status_code, 404) + + +class QuestionnaireRevisionTests(QuestionnaireScopingTestBase): + """Revision snapshots created by question edits.""" + + def updated_payload(self, slug, questions): + return questionnaire_definition(slug, questions=questions) + + def test_question_change_creates_revision_snapshot(self): + questionnaire = self.create_questionnaire("versioned") + self.assertEqual(questionnaire["internal_revision"], 1) + + questions = questionnaire["questions"] + questions[0]["text"] = "Updated note" + response = self.client.put( + self.detail_url(questionnaire["id"]), + self.updated_payload("versioned", questions), + format="json", + ) + self.assertEqual(response.status_code, 200, response.json()) + self.assertEqual(response.json()["internal_revision"], 2) + + head = Questionnaire.objects.get(external_id=questionnaire["id"]) + archived = Questionnaire.objects.get(latest_revision=head) + self.assertEqual(archived.internal_revision, 1) + self.assertEqual(archived.slug, "versioned") + self.assertNotEqual(archived.external_id, head.external_id) + + # The archived revision is listed through the parent_revision filter + listed = self.client.get( + self.base_url, {"parent_revision": questionnaire["id"]} + ).json() + self.assertEqual(listed["count"], 1) + self.assertEqual(listed["results"][0]["id"], str(archived.external_id)) + self.assertEqual(listed["results"][0]["internal_revision"], 1) + + # ... and hidden from the default listing + self.assertNotIn( + str(archived.external_id), + {entry["id"] for entry in self.client.get(self.base_url).json()["results"]}, + ) + + def test_metadata_only_update_does_not_create_revision(self): + questionnaire = self.create_questionnaire("metadata-only") + payload = self.updated_payload("metadata-only", questionnaire["questions"]) + payload["title"] = "Renamed questionnaire" + response = self.client.put( + self.detail_url(questionnaire["id"]), payload, format="json" + ) + self.assertEqual(response.status_code, 200, response.json()) + self.assertEqual(response.json()["internal_revision"], 1) + self.assertEqual(response.json()["title"], "Renamed questionnaire") + head = Questionnaire.objects.get(external_id=questionnaire["id"]) + self.assertFalse( + Questionnaire.objects.filter(latest_revision=head).exists() + ) + + def test_past_revision_cannot_be_updated(self): + questionnaire = self.create_questionnaire("no-editing-history") + questions = questionnaire["questions"] + questions[0]["text"] = "Second revision" + self.client.put( + self.detail_url(questionnaire["id"]), + self.updated_payload("no-editing-history", questions), + format="json", + ) + head = Questionnaire.objects.get(external_id=questionnaire["id"]) + archived = Questionnaire.objects.get(latest_revision=head) + + response = self.client.put( + self.detail_url(str(archived.external_id)), + self.updated_payload("no-editing-history", questions), + format="json", + ) + self.assertEqual(response.status_code, 403) + self.assertIn("past revision", str(response.json())) + + +class QuestionnaireListFilterTests(QuestionnaireScopingTestBase): + """auth_context and facility list filters.""" + + def setUp(self): + super().setUp() + self.other_facility = self.create_facility(self.super_user) + self.create_questionnaire("filter-instance") + self.create_facility_questionnaire("filter-facility-a") + self.create_facility_questionnaire( + "filter-facility-b", facility=self.other_facility + ) + + def test_auth_context_filter(self): + self.assertEqual( + self.list_slugs({"auth_context": "facility"}), + {"filter-facility-a", "filter-facility-b"}, + ) + self.assertEqual( + self.list_slugs({"auth_context": "instance"}), {"filter-instance"} + ) + + def test_facility_filter(self): + self.assertEqual( + self.list_slugs({"facility": str(self.facility.external_id)}), + {"filter-facility-a"}, + ) + + def test_auth_context_and_facility_filters_combined(self): + self.assertEqual( + self.list_slugs( + { + "auth_context": "facility", + "facility": str(self.other_facility.external_id), + } + ), + {"filter-facility-b"}, + ) + + +class QuestionnaireSlugScopingTests(QuestionnaireScopingTestBase): + """Slug uniqueness is scoped per auth context.""" + + def test_same_slug_in_different_auth_contexts_coexists(self): + instance_questionnaire = self.create_questionnaire("shared-slug") + facility_questionnaire = self.create_facility_questionnaire("shared-slug") + + self.assertNotEqual( + instance_questionnaire["id"], facility_questionnaire["id"] + ) + for questionnaire in (instance_questionnaire, facility_questionnaire): + response = self.client.get(self.detail_url(questionnaire["id"])) + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json()["slug"], "shared-slug") + + def test_same_slug_in_different_facilities_coexists(self): + other_facility = self.create_facility(self.super_user) + first = self.create_facility_questionnaire("shared-facility-slug") + second = self.create_facility_questionnaire( + "shared-facility-slug", facility=other_facility + ) + self.assertNotEqual(first["id"], second["id"]) diff --git a/care/fixtures/base.py b/care/fixtures/base.py index 6aa596ea72..d512cb84cf 100644 --- a/care/fixtures/base.py +++ b/care/fixtures/base.py @@ -1,4 +1,5 @@ import json +import logging import re import secrets import string @@ -33,6 +34,8 @@ from care.facility.models.facility import REVERSE_FACILITY_TYPES, FacilityFeature from care.fixtures.constants import build_price_components +logger = logging.getLogger(__name__) + class FixtureError(Exception): pass @@ -97,6 +100,16 @@ def post(self, url, data): raise FixtureError(msg) return to_attr_dict(response.data) + def put(self, url, data): + response = self.client.put(url, data, format="json") + if response.status_code not in ( + http_status.HTTP_200_OK, + http_status.HTTP_201_CREATED, + ): + msg = f"PUT {url} failed ({response.status_code}): {response.data}" + raise FixtureError(msg) + return to_attr_dict(response.data) + def patch(self, url, data): response = self.client.patch(url, data, format="json") if response.status_code not in ( @@ -261,8 +274,19 @@ def create_encounter(self, patient_id, facility_id, organizations=None, **kwargs return self.post(reverse("encounter-list"), data) def create_questionnaire(self, organizations, data): - questionnaire_data = {**data, "organizations": organizations} - return self.post(reverse("questionnaire-list"), questionnaire_data) + # QuestionnaireCreateSpec does not accept "organizations"; visibility is + # granted through the set_organizations action after creation. + questionnaire_data = {k: v for k, v in data.items() if k != "organizations"} + questionnaire = self.post(reverse("questionnaire-list"), questionnaire_data) + if organizations: + self.post( + reverse( + "questionnaire-set-organizations", + kwargs={"external_id": questionnaire.id}, + ), + {"organizations": organizations}, + ) + return questionnaire def load_questionnaires_from_file( self, organizations, path="data/questionnaire_fixtures.json" @@ -277,8 +301,12 @@ def load_questionnaires_from_file( try: result = self.create_questionnaire(organizations, questionnaire_data) results.append(result) - except FixtureError: - pass + except FixtureError as exc: + logger.warning( + "Skipped questionnaire fixture %r: %s", + questionnaire_data.get("slug", ""), + exc, + ) return results def create_resource_category(self, facility_id, title, resource_type, **kwargs): @@ -801,6 +829,10 @@ def load_templates_from_file( for entry in templates: try: results.append(self.create_template(facility=facility, **entry)) - except FixtureError: - pass + except FixtureError as exc: + logger.warning( + "Skipped template fixture %r: %s", + entry.get("slug_value", ""), + exc, + ) return results diff --git a/care/fixtures/context.py b/care/fixtures/context.py index 3a270e8861..40c20c0e91 100644 --- a/care/fixtures/context.py +++ b/care/fixtures/context.py @@ -18,7 +18,15 @@ class _NoOpLock: - """Bypass PatientCreateLock inside an outer transaction.""" + """Bypass redis locks inside an outer transaction. + + Lock release is deferred with transaction.on_commit, which never fires + inside the fixture context's wrapping transaction — a second operation + on the same object would deadlock on the still-held lock. + """ + + def __init__(self, *args, **kwargs): + pass def acquire(self): pass @@ -44,6 +52,10 @@ def care_fixture_context(base_cls: type[CareFixtureBase] = CareFixtureBase): with ( transaction.atomic(), patch("care.emr.api.viewsets.patient.PatientCreateLock", _NoOpLock), + patch( + "care.emr.api.viewsets.questionnaire.questionnaire.QuestionnaireLock", + _NoOpLock, + ), warnings.catch_warnings(), ): warnings.filterwarnings( diff --git a/care/fixtures/fixtures.md b/care/fixtures/fixtures.md index da8a474c30..0bdebeec45 100644 --- a/care/fixtures/fixtures.md +++ b/care/fixtures/fixtures.md @@ -16,7 +16,8 @@ care/fixtures/ ├── context.py # care_fixture_context() — setup/teardown ├── scripts/ │ ├── __init__.py -│ └── default_fixtures.py # Default seed file (loaded by manage.py) +│ ├── default_fixtures.py # Default seed file (loaded by manage.py) +│ └── questionnaire_e2e_fixtures.py # Questionnaire E2E seed file (see below) └── fixtures.md # This file ``` @@ -81,6 +82,36 @@ python manage.py load_fixtures --path care/fixtures/scripts/minimal.py > executes the file with `__name__ == "__main__"`. You can omit the > guard and put the `with` block at the top level — both are fine. +### Questionnaire E2E fixtures + +`care/fixtures/scripts/questionnaire_e2e_fixtures.py` seeds a deterministic +set of questionnaires for frontend E2E tests, all slugged with an `e2e-` +prefix: + +- `e2e-kitchen-sink-instance` / `e2e-kitchen-sink-facility` — every simple + question type, groups with layout presets, enable_when coverage, + repeating questions and an observation-bound question. +- `e2e-subject-location` / `e2e-subject-device` / `e2e-subject-facility` — + one questionnaire per non-encounter subject type. +- `e2e-org-scope` (facility-organization scoped), `e2e-user-scope` + (user scoped, created as `care-fac-admin`). +- `e2e-versioned` — carries two archived revisions (`internal_revision` 3). +- `e2e-pagination-001` … `e2e-pagination-018` — enough active + questionnaires to page past the default page size. +- An E2E patient with one `planned` and one `in_progress` encounter in + "FACILITY WITH PATIENTS". + +Run it **after** the default fixtures: + +```bash +python manage.py load_fixtures --path care/fixtures/scripts/questionnaire_e2e_fixtures.py +``` + +Unlike `default_fixtures.py`, this script is **idempotent and additive**: +every create is preceded by an API lookup (slug / phone number / status) +and skipped when the record already exists, so it is safe to run +repeatedly against a populated development database. + ### CI-injected fixtures Because `--path` accepts an absolute path and the file doesn't need to @@ -182,7 +213,7 @@ The `load_billing(base, facility_id, patients, encounters=None)` orchestrator in - `load_questionnaires_from_file(...)` — bulk-load from JSON - `get_roles()` - `get_facility_organizations(facility_id)` -- `get(...)/post(...)` - if utility unavailable make one can use these to load data +- `get(...)/post(...)/put(...)/patch(...)` - if utility unavailable make one can use these to load data Every `create_*` method accepts `**kwargs` for any additional fields the API supports. diff --git a/care/fixtures/scripts/questionnaire_e2e_fixtures.py b/care/fixtures/scripts/questionnaire_e2e_fixtures.py new file mode 100644 index 0000000000..3ec484cef8 --- /dev/null +++ b/care/fixtures/scripts/questionnaire_e2e_fixtures.py @@ -0,0 +1,887 @@ +""" +Questionnaire E2E fixtures. + +Seeds a deterministic set of questionnaires (plus one patient with fresh +encounters) so the frontend can exercise the entire questionnaire feature +in E2E tests. Everything created here is prefixed with ``e2e-`` and uses +deterministic slugs / question ids, so runs are stable across machines. + +Run it with:: + + python manage.py load_fixtures --path care/fixtures/scripts/questionnaire_e2e_fixtures.py + +Requirements: the default fixtures must already be loaded (the script +resolves "FACILITY WITH PATIENTS", the "General Medicine" facility +organization and the default users by name via the API). + +Idempotency: the script is additive and safe to run repeatedly against a +populated development database. Before each create it resolves the record +via the API (questionnaires by slug, the patient by phone number, +encounters by patient + status) and skips anything that already exists. +A second run is a no-op, with two deliberate exceptions: + +- ``e2e-versioned`` is topped up with follow-up PUTs until it reaches + ``internal_revision`` 3 (two archived revisions). +- The E2E encounters are re-created when the existing ones are older than + ~60 days, because the frontend's encounter setup only looks at a 90 day + ``created_date`` window. + +What it seeds: + +- ``e2e-kitchen-sink-instance`` / ``e2e-kitchen-sink-facility`` — every + simple question type, choice + quantity with custom options, groups with + ``containerClasses`` layout presets, enable_when coverage (boolean + Yes/No, numeric greater/less, string equals, 2-condition "any" + behavior, a protected ``disabled_display``), repeating questions and a + LOINC-bound observation question. +- ``e2e-units`` — unit-semantics coverage: integer/decimal questions with + a question-level ``unit`` and a quantity question whose + ``answer_value_set`` is the ``e2e-dose-units`` instance valueset (three + enumerated UCUM units: mg/g/kg — small enough to render as inline unit + chips in the frontend). +- ``e2e-subject-location`` / ``e2e-subject-device`` / + ``e2e-subject-facility`` — minimal facility questionnaires covering the + remaining subject types. +- ``e2e-org-scope`` — scoped to the "General Medicine" facility + organization. +- ``e2e-user-scope`` — a user-scoped questionnaire created as + ``care-fac-admin`` through a second API client. (``care-doctor`` cannot + be used: the Doctor role does not carry ``can_write_questionnaire``.) +- ``e2e-versioned`` — instance questionnaire with two archived revisions + (``internal_revision`` 3). +- ``e2e-pagination-001`` … ``e2e-pagination-018`` — tiny active facility + questionnaires to page past the default page size of 14. +- ``e2e-structured-`` for all 11 core structured types + (``allergy_intolerance``, ``symptom``, ``diagnosis``, + ``medication_request``, ``medication_statement``, ``encounter``, + ``appointment``, ``files``, ``time_of_death``, ``service_request``, + ``charge_item``) — encounter-subject, one structured question of that + type plus one plain string question. ``-required`` variants + (``e2e-structured--required``) exist for the five types the + structured-rearchitecture test plan takes from zero to full coverage + (``medication_statement``, ``files``, ``appointment``, + ``time_of_death``, ``encounter``) so the required-section-blocks-submit + path has a fixture to exercise; the other six types already have + row-level validation coverage and don't need a dedicated variant. +- ``e2e-structured-unknown`` / ``e2e-structured-unknown-optional`` — a + required (resp. non-required) structured question whose + ``structured_type`` (``x_e2e.missing``) is a namespaced plugin id no + build registers, plus a plain string question, pinning both halves of + hard-block behavior: required blocks Save with a named error; + non-required is skipped and Save succeeds with only the plain answer. +- ``e2e-structured-kitchen-sink`` — several structured types (allergy, + diagnosis, medication request, encounter, files) plus plain questions + in one questionnaire, for session/draft/merge specs. +- ``e2e-subject-patient`` — instance-scoped, ``subject_type: "patient"`` + (the only scope patient questionnaires are allowed in), a couple of + plain questions. +- One patient (phone ``+919999888777``) with one ``planned`` and one + ``in_progress`` encounter in "FACILITY WITH PATIENTS" / "General + Medicine", mirroring the default fixture wiring. +""" + +import uuid +from datetime import UTC, datetime, timedelta + +from django.contrib.auth import get_user_model +from django.urls import reverse +from rest_framework.test import APIClient + +from care.emr.resources.encounter.constants import ( + ClassChoices, + EncounterPriorityChoices, +) +from care.emr.resources.encounter.constants import ( + StatusChoices as EncounterStatusChoices, +) +from care.emr.models.valueset import ValueSet +from care.fixtures.base import CareFixtureBase, FixtureError +from care.fixtures.context import care_fixture_context + +FACILITY_NAME = "FACILITY WITH PATIENTS" +FACILITY_ORG_NAME = "General Medicine" +ADMIN_FACILITY_ORG_NAME = "Administration" +GEO_ORG_NAME = "Kerala" +USER_SCOPE_USERNAME = "care-fac-admin" + +E2E_PATIENT_NAME = "E2E QUESTIONNAIRE PATIENT" +E2E_PATIENT_PHONE = "+919999888777" +ENCOUNTER_FRESHNESS_DAYS = 60 + +PAGINATION_QUESTIONNAIRE_COUNT = 18 +VERSIONED_TARGET_REVISION = 3 + +UCUM_MG = { + "system": "http://unitsofmeasure.org", + "code": "mg", + "display": "milligram", +} +UCUM_G = { + "system": "http://unitsofmeasure.org", + "code": "g", + "display": "gram", +} +UCUM_KG = { + "system": "http://unitsofmeasure.org", + "code": "kg", + "display": "kilogram", +} +UCUM_PER_MIN = { + "system": "http://unitsofmeasure.org", + "code": "/min", + "display": "per minute", +} +UCUM_CEL = { + "system": "http://unitsofmeasure.org", + "code": "Cel", + "display": "degree Celsius", +} + +UNITS_VALUESET_SLUG = "e2e-dose-units" +UNITS_QUESTIONNAIRE_SLUG = "e2e-units" +LOINC_HEART_RATE = { + "system": "http://loinc.org", + "code": "8867-4", + "display": "Heart rate", +} + +# The 11 core structured question types (registry parity with the +# frontend's QuestionnaireV2/structured/registry.ts). +STRUCTURED_TYPES = [ + ("allergy_intolerance", "Allergy Intolerance"), + ("symptom", "Symptom"), + ("diagnosis", "Diagnosis"), + ("medication_request", "Medication Request"), + ("medication_statement", "Medication Statement"), + ("encounter", "Encounter"), + ("appointment", "Appointment"), + ("files", "Files"), + ("time_of_death", "Time of Death"), + ("service_request", "Service Request"), + ("charge_item", "Charge Item"), +] + +# Types that get an additional `-required` questionnaire variant: per the +# structured-rearchitecture test plan these five go from zero test +# coverage to the full per-type matrix (add/edit/remove/validation/ +# draft/submit), which includes proving a required, unanswered structured +# section blocks submit. The other six types already have row-level +# validation specs (e.g. medicationRequest.spec.ts's "Dosage* This field +# is required") that exercise validation on the plain, optional fixture — +# they don't need a dedicated required questionnaire. +STRUCTURED_REQUIRED_VARIANT_TYPES = { + "medication_statement", + "files", + "appointment", + "time_of_death", + "encounter", +} + +# Namespaced plugin id (`{slug}.{name}`) that no build registers a +# component for, so the frontend always resolves it as `unknown_type`. +UNKNOWN_STRUCTURED_TYPE = "x_e2e.missing" + + +def log(message): + print(message) # noqa: T201 + + +def question_id(slug, link_id): + """Deterministic question id so re-seeded data stays stable.""" + return str(uuid.uuid5(uuid.NAMESPACE_URL, f"care:e2e:{slug}:{link_id}")) + + +def simple_question(slug, link_id, question_type, text, **kwargs): + return { + "id": question_id(slug, link_id), + "link_id": link_id, + "type": question_type, + "text": text, + **kwargs, + } + + +def kitchen_sink_questions(slug): + """Every simple question type + groups, enable_when, repeats, observation.""" + q = lambda *args, **kwargs: simple_question(slug, *args, **kwargs) # noqa: E731 + return [ + q("q-string", "string", "Primary symptom"), + q("q-text", "text", "Detailed history"), + q("q-url", "url", "Reference document URL"), + q("q-decimal", "decimal", "Body temperature (C)"), + q("q-integer", "integer", "Pain score (0-10)"), + q("q-date", "date", "Symptom onset date"), + q("q-datetime", "dateTime", "Admission timestamp"), + q("q-time", "time", "Last medication time"), + q("q-boolean", "boolean", "Is the patient stable?"), + q( + "q-choice", + "choice", + "Severity assessment", + answer_option=[ + {"value": "Mild"}, + {"value": "Moderate", "initial_selected": True}, + {"value": "Severe"}, + ], + ), + q( + "q-quantity", + "quantity", + "Dose administered", + answer_option=[ + {"value": "250"}, + {"value": "500"}, + {"value": "1000"}, + ], + unit=UCUM_MG, + ), + q( + "grp-main", + "group", + "Examination findings", + styling_metadata={"containerClasses": "grid grid-cols-2"}, + questions=[ + q("q-grp-string", "string", "General appearance"), + q( + "grp-nested", + "group", + "Cardiovascular", + styling_metadata={"containerClasses": "grid grid-cols-1"}, + questions=[ + q("q-nested-string", "string", "Heart sounds"), + ], + ), + ], + ), + q( + "q-ew-yes", + "string", + "Stability notes", + enable_when=[ + {"question": "q-boolean", "operator": "equals", "answer": "Yes"} + ], + ), + q( + "q-ew-no", + "text", + "Escalation plan", + enable_when=[ + {"question": "q-boolean", "operator": "equals", "answer": "No"} + ], + ), + q( + "q-ew-greater", + "string", + "Severe pain follow-up", + enable_when=[ + {"question": "q-integer", "operator": "greater", "answer": 7} + ], + ), + q( + "q-ew-less", + "string", + "Low pain follow-up", + enable_when=[{"question": "q-integer", "operator": "less", "answer": 3}], + ), + q( + "q-ew-equals", + "string", + "Fever details", + enable_when=[ + {"question": "q-string", "operator": "equals", "answer": "fever"} + ], + ), + q( + "q-ew-any", + "string", + "Any-behavior follow-up", + enable_behavior="any", + enable_when=[ + {"question": "q-boolean", "operator": "equals", "answer": "Yes"}, + {"question": "q-integer", "operator": "greater", "answer": 5}, + ], + ), + q( + "q-ew-protected", + "string", + "Protected note (visible but locked when disabled)", + disabled_display="protected", + enable_when=[ + {"question": "q-boolean", "operator": "equals", "answer": "Yes"} + ], + ), + q( + "q-repeat-choice", + "choice", + "Symptoms observed (repeats)", + repeats=True, + answer_option=[ + {"value": "Cough"}, + {"value": "Fever"}, + {"value": "Fatigue"}, + ], + ), + q("q-repeat-string", "string", "Medications taken (repeats)", repeats=True), + q( + "q-obs-heart-rate", + "integer", + "Heart rate (bpm)", + code=LOINC_HEART_RATE, + is_observation=True, + ), + ] + + +def units_questions(slug): + """Unit-semantics coverage: integer/decimal with a question-level unit + (label display) and a quantity whose ``answer_value_set`` is a small, + bounded unit valueset (renders as inline unit chips in the frontend).""" + q = lambda *args, **kwargs: simple_question(slug, *args, **kwargs) # noqa: E731 + return [ + q("q-int-unit", "integer", "Resting heart rate", unit=UCUM_PER_MIN), + q("q-dec-unit", "decimal", "Body temperature", unit=UCUM_CEL), + q( + "q-qty-vs", + "quantity", + "Dose given", + answer_value_set={"slug": UNITS_VALUESET_SLUG}, + unit=UCUM_MG, + ), + ] + + +def structured_question(slug, link_id, structured_type, text, **kwargs): + return simple_question( + slug, link_id, "structured", text, structured_type=structured_type, **kwargs + ) + + +def structured_type_questions(slug, structured_type, label, *, required): + """One structured question of ``structured_type`` plus one plain string + question, so specs can assert the plain answer submits while the + structured section is exercised.""" + q = lambda *args, **kwargs: simple_question(slug, *args, **kwargs) # noqa: E731 + return [ + structured_question( + slug, + "q-structured", + structured_type, + f"{label} section", + required=required, + ), + q("q-note", "string", "Plain note"), + ] + + +def unknown_structured_questions(slug, *, required): + """A structured question of an unregistered plugin type, plus a plain + string question — for hard-block / unknown-type specs. ``required`` + picks which half of the hard-block behavior the fixture pins: required + blocks Save with a named error, non-required is skipped and Save + succeeds with only the plain answer.""" + q = lambda *args, **kwargs: simple_question(slug, *args, **kwargs) # noqa: E731 + return [ + structured_question( + slug, + "q-structured", + UNKNOWN_STRUCTURED_TYPE, + "Missing Plugin Section", + required=required, + ), + q("q-note", "string", "Clinical Note"), + ] + + +def structured_kitchen_sink_questions(slug): + """Several structured types plus plain questions in one questionnaire, + for session/draft/merge specs.""" + q = lambda *args, **kwargs: simple_question(slug, *args, **kwargs) # noqa: E731 + return [ + q("q-note", "string", "Chief complaint"), + structured_question(slug, "q-allergy", "allergy_intolerance", "Allergies"), + structured_question(slug, "q-diagnosis", "diagnosis", "Diagnoses"), + structured_question( + slug, "q-medication-request", "medication_request", "Medications" + ), + structured_question(slug, "q-encounter", "encounter", "Encounter details"), + structured_question(slug, "q-files", "files", "Attachments"), + q("q-boolean", "boolean", "Ready for discharge?"), + ] + + +def patient_subject_questions(slug): + """A couple of plain questions for the patient-subject fixture.""" + q = lambda *args, **kwargs: simple_question(slug, *args, **kwargs) # noqa: E731 + return [ + q("q-note", "string", "Note"), + q("q-detail", "text", "Additional detail"), + ] + + +def questionnaire_definition(slug, title, questions, subject_type="encounter"): + return { + "slug": slug, + "version": "1.0", + "title": title, + "description": f"E2E fixture questionnaire ({slug})", + "status": "active", + "subject_type": subject_type, + "styling_metadata": {}, + "questions": questions, + } + + +class QuestionnaireE2EFixtures(CareFixtureBase): + def list_all(self, url, params=None): + """Fetch every page of a paginated list endpoint.""" + params = {**(params or {}), "limit": 200, "offset": 0} + results = [] + while True: + page = self.get(url, params=params) + page_results = page.get("results", []) + results.extend(page_results) + params["offset"] += len(page_results) + if not page_results or params["offset"] >= page.get("count", 0): + return results + + def existing_questionnaires_by_slug(self): + """Head-revision questionnaires keyed by slug (archived rows excluded).""" + results = self.list_all(reverse("questionnaire-list")) + return {entry["slug"]: entry for entry in results} + + def find_facility(self, name): + for facility in self.list_all(reverse("facility-list"), {"name": name}): + if facility["name"] == name: + return facility + msg = f"Facility {name!r} not found — run the default fixtures first" + raise FixtureError(msg) + + def find_facility_organization(self, facility_id, name): + url = reverse( + "facility-organization-list", + kwargs={"facility_external_id": facility_id}, + ) + for org in self.list_all(url): + if org["name"] == name: + return org + msg = ( + f"Facility organization {name!r} not found — " + "run the default fixtures first" + ) + raise FixtureError(msg) + + def find_geo_organization(self, name): + params = {"name": name, "org_type": "govt"} + for org in self.list_all(reverse("organization-list"), params): + if org["name"] == name: + return org + msg = f"Organization {name!r} not found — run the default fixtures first" + raise FixtureError(msg) + + def find_patient_by_phone(self, phone_number): + results = self.list_all( + reverse("patient-list"), {"phone_number": phone_number} + ) + return results[0] if results else None + + def find_encounters(self, patient_id): + return self.list_all(reverse("encounter-list"), {"patient": patient_id}) + + def ensure_facility_org_membership(self, facility, organization, username, role): + """Add the user to the facility organization when not already a member.""" + url = reverse( + "facility-organization-users-list", + kwargs={ + "facility_external_id": facility["id"], + "facility_organizations_external_id": organization["id"], + }, + ) + user = self.get_user(username) + for membership in self.list_all(url): + if membership.get("user", {}).get("username") == username: + return + self.post(url, {"user": user["id"], "role": role["id"]}) + log(f" membership: added {username} to {organization['name']}") + + def set_facility_organizations(self, questionnaire_id, organization_ids): + url = reverse( + "questionnaire-set-facility-organizations", + kwargs={"external_id": questionnaire_id}, + ) + return self.post(url, {"facility_organizations": organization_ids}) + + def update_questionnaire(self, questionnaire_id, data): + url = reverse( + "questionnaire-detail", kwargs={"external_id": questionnaire_id} + ) + return self.put(url, data) + + +def seed_units_valueset(base): + """Instance valueset with three enumerated UCUM dose units (mg/g/kg). + + Referenced by slug from the ``e2e-units`` quantity question, so it must + exist before the questionnaires are created (the questionnaire spec + validates slug references against instance valuesets). The fixture + client is a superuser, which instance-valueset creation requires. + """ + if ValueSet.objects.filter( + slug=UNITS_VALUESET_SLUG, auth_context="instance", deleted=False + ).exists(): + log(f" valueset {UNITS_VALUESET_SLUG}: exists, skipping") + return + base.post( + reverse("value-set-list"), + { + "slug": UNITS_VALUESET_SLUG, + "name": "E2E Dose Units", + "description": "E2E fixture: bounded UCUM dose units (mg/g/kg)", + "status": "active", + "auth_context": "instance", + "inherited": False, + "compose": { + "include": [ + { + "system": "http://unitsofmeasure.org", + "concept": [ + {"code": code["code"], "display": code["display"]} + for code in (UCUM_MG, UCUM_G, UCUM_KG) + ], + } + ], + # Explicit empty exclude: ValueSet.create_composition iterates + # compose.exclude unguarded, so omitting it breaks $expand. + "exclude": [], + }, + }, + ) + log(f" valueset {UNITS_VALUESET_SLUG}: created") + + +def seed_questionnaires(base, existing, facility, general_medicine, admin_org, geo_org): + facility_org_ids = [admin_org["id"], general_medicine["id"]] + + def create(definition, *, auth_context, organizations=None, tag_orgs=False): + slug = definition["slug"] + if slug in existing: + log(f" {slug}: exists, skipping") + return existing[slug] + payload = {**definition, "auth_context": auth_context} + if auth_context == "facility": + payload["facility"] = facility["id"] + elif auth_context == "facility_organization": + payload["facility_organization"] = general_medicine["id"] + questionnaire = base.create_questionnaire(organizations or [], payload) + if tag_orgs and auth_context == "facility": + base.set_facility_organizations(questionnaire["id"], facility_org_ids) + log(f" {slug}: created") + return questionnaire + + create( + questionnaire_definition( + "e2e-kitchen-sink-instance", + "E2E Kitchen Sink (Instance)", + kitchen_sink_questions("e2e-kitchen-sink-instance"), + ), + auth_context="instance", + organizations=[geo_org["id"]], + ) + create( + questionnaire_definition( + "e2e-kitchen-sink-facility", + "E2E Kitchen Sink (Facility)", + kitchen_sink_questions("e2e-kitchen-sink-facility"), + ), + auth_context="facility", + tag_orgs=True, + ) + + create( + questionnaire_definition( + UNITS_QUESTIONNAIRE_SLUG, + "E2E Units Questionnaire", + units_questions(UNITS_QUESTIONNAIRE_SLUG), + ), + auth_context="facility", + tag_orgs=True, + ) + + for subject_type in ("location", "device", "facility"): + slug = f"e2e-subject-{subject_type}" + create( + questionnaire_definition( + slug, + f"E2E {subject_type.title()} Questionnaire", + [simple_question(slug, "q-note", "string", "Notes")], + subject_type=subject_type, + ), + auth_context="facility", + tag_orgs=True, + ) + + create( + questionnaire_definition( + "e2e-org-scope", + "E2E Facility Organization Scoped", + [ + simple_question( + "e2e-org-scope", "q-note", "string", "Department note" + ) + ], + ), + auth_context="facility_organization", + ) + + for index in range(1, PAGINATION_QUESTIONNAIRE_COUNT + 1): + slug = f"e2e-pagination-{index:03d}" + create( + questionnaire_definition( + slug, + f"E2E Pagination {index:03d}", + [simple_question(slug, "q-note", "string", "Note")], + ), + auth_context="facility", + ) + + +def seed_structured_questionnaires( + base, existing, facility, general_medicine, admin_org, geo_org +): + """Structured-rearchitecture fixtures (design doc §8): per-type + encounter-subject questionnaires for all 11 core structured types + (plus required variants for the five types getting full new test + coverage), an unknown-plugin-type fixture, a structured kitchen-sink, + and the patient-subject fixture.""" + facility_org_ids = [admin_org["id"], general_medicine["id"]] + + def create( + definition, *, auth_context="facility", organizations=None, tag_orgs=True + ): + slug = definition["slug"] + if slug in existing: + log(f" {slug}: exists, skipping") + return existing[slug] + payload = {**definition, "auth_context": auth_context} + if auth_context == "facility": + payload["facility"] = facility["id"] + questionnaire = base.create_questionnaire(organizations or [], payload) + if tag_orgs and auth_context == "facility": + base.set_facility_organizations(questionnaire["id"], facility_org_ids) + log(f" {slug}: created") + return questionnaire + + for structured_type, label in STRUCTURED_TYPES: + slug = f"e2e-structured-{structured_type}" + create( + questionnaire_definition( + slug, + f"E2E Structured {label}", + structured_type_questions(slug, structured_type, label, required=False), + ) + ) + if structured_type in STRUCTURED_REQUIRED_VARIANT_TYPES: + required_slug = f"{slug}-required" + create( + questionnaire_definition( + required_slug, + f"E2E Structured {label} (Required)", + structured_type_questions( + required_slug, structured_type, label, required=True + ), + ) + ) + + create( + questionnaire_definition( + "e2e-structured-unknown", + "E2E Structured Unknown Plugin", + unknown_structured_questions("e2e-structured-unknown", required=True), + ) + ) + create( + questionnaire_definition( + "e2e-structured-unknown-optional", + "E2E Structured Unknown Plugin (Optional)", + unknown_structured_questions( + "e2e-structured-unknown-optional", required=False + ), + ) + ) + + create( + questionnaire_definition( + "e2e-structured-kitchen-sink", + "E2E Structured Kitchen Sink", + structured_kitchen_sink_questions("e2e-structured-kitchen-sink"), + ) + ) + + create( + questionnaire_definition( + "e2e-subject-patient", + "E2E Patient Questionnaire", + patient_subject_questions("e2e-subject-patient"), + subject_type="patient", + ), + auth_context="instance", + organizations=[geo_org["id"]], + tag_orgs=False, + ) + + +def seed_user_scope_questionnaire(base, existing, facility, admin_org): + """Create the user-scoped questionnaire as care-fac-admin. + + care-doctor cannot be used here: the Doctor role does not have + can_write_questionnaire, so the API rejects the create. The Facility + Admin role does, and care-fac-admin is (ensured) a member of the + facility's Administration organization. + """ + slug = "e2e-user-scope" + if slug in existing: + log(f" {slug}: exists, skipping") + return + roles = base.get_roles() + base.ensure_facility_org_membership( + facility, admin_org, USER_SCOPE_USERNAME, roles["Facility Admin"] + ) + user = get_user_model().objects.get(username=USER_SCOPE_USERNAME) + client = APIClient() + client.force_authenticate(user=user) + user_base = CareFixtureBase(client) + payload = { + **questionnaire_definition( + slug, + "E2E User Scoped", + [simple_question(slug, "q-note", "string", "Personal note")], + ), + "auth_context": "user", + "facility": facility["id"], + } + user_base.post(reverse("questionnaire-list"), payload) + log(f" {slug}: created (as {USER_SCOPE_USERNAME})") + + +def seed_versioned_questionnaire(base, existing): + """Instance questionnaire with two archived revisions (internal_revision 3).""" + slug = "e2e-versioned" + revision_texts = { + 1: "Observation note (v1)", + 2: "Observation note (v2)", + 3: "Observation note (v3)", + } + + def definition(revision): + return questionnaire_definition( + slug, + "E2E Versioned Questionnaire", + [ + simple_question( + slug, "q-note", "string", revision_texts[revision] + ) + ], + ) + + questionnaire = existing.get(slug) + if questionnaire is None: + questionnaire = base.create_questionnaire([], { + **definition(1), + "auth_context": "instance", + }) + log(f" {slug}: created") + current_revision = questionnaire["internal_revision"] + if current_revision >= VERSIONED_TARGET_REVISION: + log(f" {slug}: already at revision {current_revision}, skipping") + return + for revision in range(current_revision + 1, VERSIONED_TARGET_REVISION + 1): + questionnaire = base.update_questionnaire( + questionnaire["id"], definition(revision) + ) + log(f" {slug}: bumped to revision {questionnaire['internal_revision']}") + + +def seed_patient_and_encounters(base, facility, general_medicine, geo_org): + patient = base.find_patient_by_phone(E2E_PATIENT_PHONE) + if patient is None: + patient = base.create_patient( + geo_org["id"], + name=E2E_PATIENT_NAME, + phone_number=E2E_PATIENT_PHONE, + ) + log(f" patient {E2E_PATIENT_NAME}: created") + else: + log(f" patient {E2E_PATIENT_NAME}: exists, skipping") + + freshness_cutoff = datetime.now(UTC) - timedelta(days=ENCOUNTER_FRESHNESS_DAYS) + + def has_fresh_encounter(encounters, status): + for encounter in encounters: + if encounter["status"] != status: + continue + created_date = encounter.get("created_date") + if not created_date: + return True + created = datetime.fromisoformat(created_date) + if created.tzinfo is None: + created = created.replace(tzinfo=UTC) + if created >= freshness_cutoff: + return True + return False + + encounters = base.find_encounters(patient["id"]) + for status in ( + EncounterStatusChoices.planned.value, + EncounterStatusChoices.in_progress.value, + ): + if has_fresh_encounter(encounters, status): + log(f" encounter ({status}): fresh one exists, skipping") + continue + base.create_encounter( + patient["id"], + facility["id"], + organizations=[general_medicine["id"]], + status=status, + encounter_class=ClassChoices.imp.value, + priority=EncounterPriorityChoices.routine.value, + ) + log(f" encounter ({status}): created") + + +def load_fixtures(base): + facility = base.find_facility(FACILITY_NAME) + general_medicine = base.find_facility_organization( + facility["id"], FACILITY_ORG_NAME + ) + admin_org = base.find_facility_organization( + facility["id"], ADMIN_FACILITY_ORG_NAME + ) + geo_org = base.find_geo_organization(GEO_ORG_NAME) + existing = base.existing_questionnaires_by_slug() + log("Resolved facility, organizations and existing questionnaires") + + seed_units_valueset(base) + log("Loading E2E units valueset completed") + + seed_questionnaires( + base, existing, facility, general_medicine, admin_org, geo_org + ) + log("Loading E2E questionnaires completed") + + seed_structured_questionnaires( + base, existing, facility, general_medicine, admin_org, geo_org + ) + log("Loading E2E structured questionnaires completed") + + seed_user_scope_questionnaire(base, existing, facility, admin_org) + log("Loading E2E user-scoped questionnaire completed") + + seed_versioned_questionnaire(base, existing) + log("Loading E2E versioned questionnaire completed") + + seed_patient_and_encounters(base, facility, general_medicine, geo_org) + log("Loading E2E patient and encounters completed") + + +if __name__ == "__main__": + with care_fixture_context(base_cls=QuestionnaireE2EFixtures) as base: + load_fixtures(base) diff --git a/data/questionnaire_fixtures.json b/data/questionnaire_fixtures.json index 1b0cc16f0f..6ad98c0827 100644 --- a/data/questionnaire_fixtures.json +++ b/data/questionnaire_fixtures.json @@ -7,6 +7,7 @@ "description": "", "status": "active", "subject_type": "encounter", + "auth_context": "instance", "styling_metadata": {}, "questions": [ { @@ -59,6 +60,7 @@ "description": "", "status": "active", "subject_type": "encounter", + "auth_context": "instance", "styling_metadata": {}, "questions": [ { @@ -83,6 +85,7 @@ "description": "", "status": "active", "subject_type": "encounter", + "auth_context": "instance", "styling_metadata": {}, "questions": [ { @@ -121,6 +124,7 @@ "description": "", "status": "active", "subject_type": "encounter", + "auth_context": "instance", "styling_metadata": {}, "questions": [ { @@ -152,6 +156,7 @@ "description": "", "status": "active", "subject_type": "encounter", + "auth_context": "instance", "styling_metadata": {}, "questions": [ { @@ -171,6 +176,7 @@ "description": "", "status": "active", "subject_type": "encounter", + "auth_context": "instance", "styling_metadata": {}, "questions": [ { @@ -217,6 +223,7 @@ "description": "A form to document respiratory status, including bilateral air entry and respiratory support details.", "status": "active", "subject_type": "encounter", + "auth_context": "instance", "styling_metadata": {}, "questions": [ { @@ -397,6 +404,7 @@ "description": "", "status": "active", "subject_type": "encounter", + "auth_context": "instance", "styling_metadata": {}, "questions": [ {