""" Tests for Healthie webhook processing pipeline. """ import hashlib import hmac import json from datetime import date from unittest.mock import MagicMock, patch from django.test import RequestFactory, TestCase, override_settings from access_core.intake.models import IntakeIdentifier, IntakeRecord, RawIngestionRecord from access_core.services.healthie_webhook import ( WebhookPayload, WebhookSignatureError, verify_webhook_signature, ) # ------------------------------------------------------------------ # Signature verification tests # ------------------------------------------------------------------ class TestWebhookSignatureVerification(TestCase): """Test HMAC-SHA256 signature verification.""" def setUp(self): self.factory = RequestFactory() self.secret = "whsec_test-secret-key" def _make_signed_request( self, body: dict, path: str = "/api/intake/webhooks/healthie/" ) -> MagicMock: """Create a mock request with valid Healthie webhook signature.""" # Match production code exactly: compact JSON for content-length computation body_obj = body # already a dict compact_json = json.dumps(body_obj, separators=(",", ":")) body_bytes = compact_json.encode("utf-8") content_digest_hash = hashlib.sha256(body_bytes).hexdigest() # Production code extracts hash-only from Content-Digest (split on "=") content_digest_full = f"SHA-256={content_digest_hash}" # Data string matches Healthie's JS reference: uses hash portion only, compact JSON length data_to_sign = ( f"post {path} {content_digest_hash} application/json {len(body_bytes)}" ) signature = hmac.new( self.secret.encode("utf-8"), data_to_sign.encode("utf-8"), hashlib.sha256, ).hexdigest() return self.factory.post( path, data=body_bytes, content_type="application/json", HTTP_CONTENT_DIGEST=content_digest_full, HTTP_SIGNATURE=f"sig1={signature}", ) @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_valid_signature(self): body = { "resource_id": "12345", "resource_id_type": "FormAnswerGroup", "event_type": "form_answer_group.created", "changed_fields": [], } request = self._make_signed_request(body) self.assertTrue(verify_webhook_signature(request)) @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_wrong-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_invalid_signature(self): body = { "resource_id": "12345", "resource_id_type": "FormAnswerGroup", "event_type": "form_answer_group.created", } request = self._make_signed_request(body) with self.assertRaises(WebhookSignatureError): verify_webhook_signature(request) @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_valid_signature_non_ascii_payload(self): """Non-ASCII names must verify: JS JSON.stringify keeps unicode chars.""" body = { "resource_id": "12345", "resource_id_type": "FormAnswerGroup", "event_type": "form_answer_group.created", "changed_fields": ["patient_name"], "data": {"patient_first_name": "José", "patient_last_name": "Müller"}, } # Sign exactly like Healthie's JS reference: contentLength is the string # length of JSON.stringify(body) (unicode preserved), digest over UTF-8 bytes compact_json = json.dumps(body, separators=(",", ":"), ensure_ascii=False) body_bytes = compact_json.encode("utf-8") content_digest_hash = hashlib.sha256(body_bytes).hexdigest() # note double space: empty query string in the f-string data_to_sign = ( f"post /api/intake/webhooks/healthie/ {content_digest_hash} " f"application/json {len(compact_json)}" ) signature = hmac.new( self.secret.encode("utf-8"), data_to_sign.encode("utf-8"), hashlib.sha256, ).hexdigest() request = self.factory.post( "/api/intake/webhooks/healthie/", data=body_bytes, content_type="application/json", HTTP_CONTENT_DIGEST=f"SHA-256={content_digest_hash}", HTTP_SIGNATURE=f"sig1={signature}", ) self.assertTrue(verify_webhook_signature(request)) def test_verification_disabled(self): """When verification is disabled, any request passes.""" body = {"resource_id": "1", "event_type": "test"} request = self.factory.post( "/test", data=json.dumps(body), content_type="application/json" ) with override_settings(HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=False): self.assertTrue(verify_webhook_signature(request)) @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_missing_headers(self): body = {"resource_id": "1", "event_type": "test"} request = self.factory.post( "/test", data=json.dumps(body), content_type="application/json", ) with self.assertRaises(WebhookSignatureError): verify_webhook_signature(request) @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_malformed_content_digest_raises_signature_error(self): """A Content-Digest without '=' must be a clean signature failure, not an IndexError escaping to the view's 500 handler.""" body = {"resource_id": "1", "event_type": "test"} request = self.factory.post( "/api/intake/webhooks/healthie/", data=json.dumps(body), content_type="application/json", HTTP_CONTENT_DIGEST="SHA-256", # no '=' at all HTTP_SIGNATURE="sig1=abc", ) with self.assertRaises(WebhookSignatureError): verify_webhook_signature(request) @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_malformed_content_digest_view_returns_401(self): """The endpoint returns 401 (not 500) for an unparseable digest.""" body = {"resource_id": "1", "event_type": "test"} request = self.factory.post( "/api/intake/webhooks/healthie/", data=json.dumps(body), content_type="application/json", HTTP_CONTENT_DIGEST="SHA-256=", # empty hash value HTTP_SIGNATURE="sig1=abc", ) from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 401) # ------------------------------------------------------------------ # WebhookPayload tests # ------------------------------------------------------------------ class TestWebhookPayload(TestCase): """Test WebhookPayload dataclass.""" def test_is_create_event(self): p = WebhookPayload("123", "FormAnswerGroup", "form_answer_group.created") self.assertTrue(p.is_create_event) self.assertFalse(p.is_update_event) self.assertFalse(p.is_delete_event) def test_is_update_event(self): p = WebhookPayload("123", "User", "patient.updated", changed_fields=["email"]) self.assertTrue(p.is_update_event) self.assertFalse(p.is_create_event) def test_is_delete_event(self): p = WebhookPayload("123", "Document", "document.deleted") self.assertTrue(p.is_delete_event) def test_changed_fields_default(self): p = WebhookPayload("123", "FormAnswerGroup", "form_answer_group.created") self.assertEqual(p.changed_fields, []) # ------------------------------------------------------------------ # Webhook endpoint view tests # ------------------------------------------------------------------ class TestWebhookView(TestCase): """Test the Django webhook endpoint view.""" def setUp(self): self.factory = RequestFactory() self.secret = "whsec_test-secret-key" def _make_signed_request(self, body: dict) -> MagicMock: # Match production code: compact JSON for content-length computation compact_json = json.dumps(body, separators=(",", ":")) body_bytes = compact_json.encode("utf-8") content_digest_hash = hashlib.sha256(body_bytes).hexdigest() # Production extracts hash-only from Content-Digest header content_digest_full = f"SHA-256={content_digest_hash}" # Data string: uses hash portion only, compact JSON length data_to_sign = f"post /api/intake/webhooks/healthie/ {content_digest_hash} application/json {len(body_bytes)}" signature = hmac.new( self.secret.encode("utf-8"), data_to_sign.encode("utf-8"), hashlib.sha256, ).hexdigest() return self.factory.post( "/api/intake/webhooks/healthie/", data=body_bytes, content_type="application/json", HTTP_CONTENT_DIGEST=content_digest_full, HTTP_SIGNATURE=f"sig1={signature}", ) @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_valid_webhook_stores_raw_record(self): """A valid webhook creates a RawIngestionRecord.""" body = { "resource_id": "999", "resource_id_type": "FormAnswerGroup", "event_type": "form_answer_group.created", "changed_fields": [], } request = self._make_signed_request(body) # Mock the HealthieClient factory mock_client = MagicMock() mock_client.query.return_value = { "data": { "formAnswerGroup": { "id": "999", "finished": True, "created_at": "2026-01-15T10:00:00Z", "user": { "id": "6263587", "first_name": "Jane", "last_name": "Doe", "email": "jane@example.com", "dob": "1990-05-15", "gender": "Female", }, "custom_module_form": {"id": "2360739", "name": "Intake"}, "form_answers": [], } } } with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) # Verify raw record was created raw = RawIngestionRecord.objects.get(source_record_id="999") self.assertEqual(raw.source_system, "healthie") self.assertEqual(raw.processing_status, "normalized") self.assertTrue(raw.webhook_signature_valid) # Verify intake record was created intake = IntakeRecord.objects.get(external_form_id="999") self.assertEqual(intake.first_name, "Jane") self.assertEqual(intake.last_name, "Doe") self.assertEqual(intake.sex, "female") self.assertEqual(intake.status, "intake_in_progress") # Verify identifiers email_id = IntakeIdentifier.objects.get( intake_record=intake, identifier_type="email" ) self.assertEqual(email_id.identifier_value, "jane@example.com") # Verify sub-models were created from access_core.intake.models import ( IntakeAssessment, IntakeFemaleHealth, IntakeLifestyle, IntakeMSK, ParticipationReadiness, SafetyScreening, ) self.assertEqual( IntakeAssessment.objects.filter(intake_record=intake).count(), 1 ) self.assertEqual( IntakeLifestyle.objects.filter(intake_record=intake).count(), 1 ) # Gender is Female, so female health should be created self.assertEqual( IntakeFemaleHealth.objects.filter(intake_record=intake).count(), 1 ) # MSK and safety screening always created self.assertEqual(IntakeMSK.objects.filter(intake_record=intake).count(), 1) self.assertEqual( SafetyScreening.objects.filter(intake_record=intake).count(), 1 ) # Participation readiness created with default pending status self.assertEqual( ParticipationReadiness.objects.filter(intake_record=intake).count(), 1 ) # Verify SafetyScreening field values (no safety stop answers in mock data) safety = SafetyScreening.objects.get(intake_record=intake) self.assertFalse(safety.safety_stop_applies) self.assertEqual(safety.screening_status, "needs_review") # Verify ParticipationReadiness field values (no acknowledgment in mock data) readiness = ParticipationReadiness.objects.get(intake_record=intake) self.assertFalse(readiness.acknowledged) self.assertEqual(readiness.status, "pending") @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_unhandled_event_stores_raw_only(self): """An unhandled event type stores raw record but doesn't crash.""" body = { "resource_id": "555", "resource_id_type": "Appointment", "event_type": "appointment.updated", "changed_fields": ["pm_status"], } request = self._make_signed_request(body) from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) raw = RawIngestionRecord.objects.get(source_record_id="555") self.assertEqual(raw.processing_status, "received") @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_invalid_signature_returns_401(self): """An invalid signature returns 401.""" body = {"resource_id": "1", "event_type": "test"} request = self.factory.post( "/api/intake/webhooks/healthie/", data=json.dumps(body), content_type="application/json", ) from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 401) def test_non_post_returns_405(self): """Only POST method is allowed.""" request = self.factory.get("/api/intake/webhooks/healthie/") from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 405) @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_handler_failure_returns_500(self): """If the handler raises an exception, return 500 for retry.""" body = { "resource_id": "777", "resource_id_type": "FormAnswerGroup", "event_type": "form_answer_group.created", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() mock_client.query.side_effect = Exception("API timeout") with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 500) # Raw ingestion record should exist (created before GraphQL fetch) raw = RawIngestionRecord.objects.get(source_record_id="777") # Status remains "processing" since the exception occurred after raw record creation self.assertEqual(raw.processing_status, "processing") self.assertTrue(raw.webhook_signature_valid) @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_transient_transport_error_returns_500_for_retry(self): """Connection errors should return 500 so Healthie retries the event.""" import requests body = { "resource_id": "781", "resource_id_type": "FormAnswerGroup", "event_type": "form_answer_group.created", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() mock_client.query.side_effect = requests.exceptions.ConnectionError( "network down" ) with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 500) # Record stays "processing" — Healthie will retry and the store resets it raw = RawIngestionRecord.objects.get(source_record_id="781") self.assertEqual(raw.processing_status, "processing") @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_permanent_http_error_marks_failed(self): """A 4xx from Healthie is permanent — mark the record failed and ack with 200.""" import requests body = { "resource_id": "782", "resource_id_type": "FormAnswerGroup", "event_type": "form_answer_group.created", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() http_error = requests.exceptions.HTTPError("401 Client Error") http_error.response = MagicMock(status_code=401) mock_client.query.side_effect = http_error with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) raw = RawIngestionRecord.objects.get(source_record_id="782") self.assertEqual(raw.processing_status, "failed") @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_patient_created_creates_intake_record(self): """patient.created should create an IntakeRecord.""" body = { "resource_id": "6000999", "resource_id_type": "User", "event_type": "patient.created", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() mock_client.query.return_value = { "data": { "user": { "id": "6000999", "first_name": "Test", "last_name": "Patient", "email": "test.patient@example.com", "phone_number": "+1-555-0999", "dob": "1980-06-15", "gender": "Male", } } } with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) # Verify intake record was created intake = IntakeRecord.objects.get(external_form_id="healthie-user-6000999") self.assertEqual(intake.first_name, "Test") self.assertEqual(intake.last_name, "Patient") self.assertEqual(intake.sex, "male") self.assertEqual(intake.status, "prospective") # Verify identifiers email_id = IntakeIdentifier.objects.get( intake_record=intake, identifier_type="email" ) self.assertEqual(email_id.identifier_value, "test.patient@example.com") phone_id = IntakeIdentifier.objects.get( intake_record=intake, identifier_type="phone" ) self.assertEqual(phone_id.identifier_value, "+1-555-0999") # Verify raw ingestion raw = RawIngestionRecord.objects.get(source_record_id="6000999") self.assertEqual(raw.processing_status, "normalized") @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_safety_flagged_intake(self): """Intake with safety stop = 'Yes' should flag screening.""" body = { "resource_id": "1000999", "resource_id_type": "FormAnswerGroup", "event_type": "form_answer_group.created", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() mock_client.query.return_value = { "data": { "formAnswerGroup": { "id": "1000999", "finished": True, "created_at": "2026-08-01T12:00:00Z", "user": { "id": "6000002", "first_name": "Safety", "last_name": "Flagged", "email": "safety@example.com", "dob": "1965-01-01", "gender": "Female", }, "custom_module_form": {"id": "2360739", "name": "Intake"}, "form_answers": [ { "id": "ans-safety", "label": "Do any of the above current conditions apply to you?", "answer": "Yes", "displayed_answer": "Yes", "custom_module": { "id": "mod_safety", "label": "Do any of the above current conditions apply to you?", "mod_type": "select", "required": True, }, }, { "id": "ans-ack", "label": "I have read and understand the Eligibility Assessment", "answer": "Yes", "displayed_answer": "Yes", "custom_module": { "id": "mod_ack", "label": "I have read and understand the Eligibility Assessment", "mod_type": "checkbox", "required": True, }, }, ], } } } with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) intake = IntakeRecord.objects.get(external_form_id="1000999") from access_core.intake.models import ParticipationReadiness, SafetyScreening safety = SafetyScreening.objects.get(intake_record=intake) self.assertTrue(safety.safety_stop_applies) self.assertEqual(safety.screening_status, "flagged") readiness = ParticipationReadiness.objects.get(intake_record=intake) self.assertTrue(readiness.acknowledged) self.assertEqual(readiness.status, "acknowledged") # ------------------------------------------------------------------ # Care plan lifecycle handler tests # ------------------------------------------------------------------ class TestCarePlanHandlers(TestCase): """Test care_plan.* webhook handlers.""" def setUp(self): self.factory = RequestFactory() self.secret = "whsec_test-secret-key" def _make_signed_request(self, body: dict) -> MagicMock: # Match production code: compact JSON for content-length computation compact_json = json.dumps(body, separators=(",", ":")) body_bytes = compact_json.encode("utf-8") content_digest_hash = hashlib.sha256(body_bytes).hexdigest() # Production extracts hash-only from Content-Digest header content_digest_full = f"SHA-256={content_digest_hash}" # Data string: uses hash portion only, compact JSON length data_to_sign = f"post /api/intake/webhooks/healthie/ {content_digest_hash} application/json {len(body_bytes)}" signature = hmac.new( self.secret.encode("utf-8"), data_to_sign.encode("utf-8"), hashlib.sha256, ).hexdigest() return self.factory.post( "/api/intake/webhooks/healthie/", data=body_bytes, content_type="application/json", HTTP_CONTENT_DIGEST=content_digest_full, HTTP_SIGNATURE=f"sig1={signature}", ) @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_care_plan_created_stores_reference(self): """care_plan.created stores care plan data in raw_payload.""" body = { "resource_id": "4001", "resource_id_type": "CarePlan", "event_type": "care_plan.created", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() mock_client.query.return_value = { "data": { "carePlan": { "id": "4001", "name": "CKM Standard Protocol v2.1", "is_active": True, "is_template": False, "patient_id": "6000001", "recommendations": [ { "id": "r1", "recommendation_body": "Walk 30 min/day", "recommendation_type": "lifestyle", }, { "id": "r2", "recommendation_body": "Monitor BP twice daily", "recommendation_type": "monitoring", }, { "id": "r3", "recommendation_body": "Lab work in 3 months", "recommendation_type": "lab", }, ], } } } with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) raw = RawIngestionRecord.objects.get(source_record_id="4001") self.assertEqual(raw.processing_status, "normalized") ref = raw.raw_payload.get("care_plan_reference") self.assertEqual(ref["id"], "4001") self.assertEqual(ref["name"], "CKM Standard Protocol v2.1") self.assertTrue(ref["is_active"]) self.assertFalse(ref["is_template"]) self.assertEqual(ref["patient_id"], "6000001") self.assertEqual(ref["event_type"], "care_plan.created") self.assertEqual(len(ref["recommendations"]), 3) @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_care_plan_activated_marks_active(self): """care_plan.activated stores reference with active=True.""" body = { "resource_id": "4002", "resource_id_type": "CarePlan", "event_type": "care_plan.activated", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() mock_client.query.return_value = { "data": { "carePlan": { "id": "4002", "name": "eCKM Protocol", "is_active": True, "is_template": False, "patient_id": "6000002", "recommendations": [ { "id": "r1", "recommendation_body": "Diet change", "recommendation_type": "lifestyle", } ], } } } with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) raw = RawIngestionRecord.objects.get(source_record_id="4002") self.assertEqual(raw.processing_status, "normalized") ref = raw.raw_payload.get("care_plan_reference") self.assertTrue(ref["is_active"]) self.assertEqual(ref["event_type"], "care_plan.activated") @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_care_plan_deactivated_marks_inactive(self): """care_plan.deactivated stores reference with active=False.""" body = { "resource_id": "4003", "resource_id_type": "CarePlan", "event_type": "care_plan.deactivated", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() mock_client.query.return_value = { "data": { "carePlan": { "id": "4003", "name": "Old Plan", "is_active": False, "is_template": False, "patient_id": "6000003", "recommendations": [], } } } with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) raw = RawIngestionRecord.objects.get(source_record_id="4003") ref = raw.raw_payload.get("care_plan_reference") self.assertFalse(ref["is_active"]) self.assertEqual(ref["event_type"], "care_plan.deactivated") @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_care_plan_deleted_stores_event(self): """care_plan.deleted stores event reference even with no GraphQL data.""" body = { "resource_id": "4004", "resource_id_type": "CarePlan", "event_type": "care_plan.deleted", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() # deleted care plans return no data from GraphQL mock_client.query.return_value = {"data": {"carePlan": None}} with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) raw = RawIngestionRecord.objects.get(source_record_id="4004") self.assertEqual(raw.processing_status, "failed") # ------------------------------------------------------------------ # Patient merge handler test # ------------------------------------------------------------------ class TestPatientMergeHandler(TestCase): """Test patient.merged webhook handler.""" def setUp(self): self.factory = RequestFactory() self.secret = "whsec_test-secret-key" def _make_signed_request(self, body: dict) -> MagicMock: # Match production code: compact JSON for content-length computation compact_json = json.dumps(body, separators=(",", ":")) body_bytes = compact_json.encode("utf-8") content_digest_hash = hashlib.sha256(body_bytes).hexdigest() # Production extracts hash-only from Content-Digest header content_digest_full = f"SHA-256={content_digest_hash}" # Data string: uses hash portion only, compact JSON length data_to_sign = f"post /api/intake/webhooks/healthie/ {content_digest_hash} application/json {len(body_bytes)}" signature = hmac.new( self.secret.encode("utf-8"), data_to_sign.encode("utf-8"), hashlib.sha256, ).hexdigest() return self.factory.post( "/api/intake/webhooks/healthie/", data=body_bytes, content_type="application/json", HTTP_CONTENT_DIGEST=content_digest_full, HTTP_SIGNATURE=f"sig1={signature}", ) @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_patient_merged_stores_raw(self): """patient.merged stores raw record with merge info.""" body = { "resource_id": "6000001", "resource_id_type": "User", "event_type": "patient.merged", "changed_fields": [], "source_user_id": "6000099", "destination_user_id": "6000001", } request = self._make_signed_request(body) from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) raw = RawIngestionRecord.objects.get(source_record_id="6000001") self.assertEqual(raw.processing_status, "normalized") self.assertEqual(raw.processing_status, "normalized") self.assertEqual(raw.source_system, "healthie") # ------------------------------------------------------------------ # Daily check-in handler tests # ------------------------------------------------------------------ class TestDailyCheckinHandler(TestCase): """Test daily check-in form normalization and safety signal detection.""" def setUp(self): self.factory = RequestFactory() self.secret = "whsec_test-secret-key" def _make_signed_request(self, body: dict) -> MagicMock: # Match production code: compact JSON for content-length computation compact_json = json.dumps(body, separators=(",", ":")) body_bytes = compact_json.encode("utf-8") content_digest_hash = hashlib.sha256(body_bytes).hexdigest() # Production extracts hash-only from Content-Digest header content_digest_full = f"SHA-256={content_digest_hash}" # Data string: uses hash portion only, compact JSON length data_to_sign = f"post /api/intake/webhooks/healthie/ {content_digest_hash} application/json {len(body_bytes)}" signature = hmac.new( self.secret.encode("utf-8"), data_to_sign.encode("utf-8"), hashlib.sha256, ).hexdigest() return self.factory.post( "/api/intake/webhooks/healthie/", data=body_bytes, content_type="application/json", HTTP_CONTENT_DIGEST=content_digest_full, HTTP_SIGNATURE=f"sig1={signature}", ) @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) @override_settings(HEALTHIE_INTAKE_FORM_ID="2360739") def test_daily_checkin_non_beneficiary_stores_raw(self): """Daily check-in from non-beneficiary user stores RawDailyCheckinSubmission.""" body = { "resource_id": "100010", "resource_id_type": "FormAnswerGroup", "event_type": "form_answer_group.created", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() mock_client.query.return_value = { "data": { "formAnswerGroup": { "id": "100010", "finished": True, "created_at": "2026-08-01T08:00:00Z", "user": { "id": "6000999", "first_name": "New", "last_name": "Patient", "email": "new@example.com", "dob": "1990-01-01", "gender": "Female", }, "custom_module_form": {"id": "999888", "name": "Daily Check-in"}, "form_answers": [ { "id": "ans1", "label": "How are you feeling today?", "answer": "Good", "displayed_answer": "Good", "custom_module": { "id": "m1", "label": "How are you feeling today?", "mod_type": "text", "required": True, }, }, ], } } } with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) raw = RawIngestionRecord.objects.get(source_record_id="100010") self.assertEqual(raw.processing_status, "normalized") # Non-beneficiary user gets raw submission stored from access_core.monitoring.models import RawDailyCheckinSubmission raw_sub = RawDailyCheckinSubmission.objects.get(source_record_id="100010") self.assertEqual(raw_sub.source_system, "healthie") self.assertEqual(raw_sub.raw_payload["user_id"], "6000999") @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) @override_settings(HEALTHIE_INTAKE_FORM_ID="2360739") def test_daily_checkin_safety_signal_red(self): """Red safety keyword triggers SafetySignalEvent + CareTeamReviewTask.""" from access_core.identity.models import Beneficiary, ExternalSystemIdentifier from access_core.monitoring.models import CarePeriod, ClinicalTrack beneficiary = Beneficiary.objects.create( first_name="Test", last_name="Patient", dob="1990-01-01", sex="female" ) ExternalSystemIdentifier.objects.create( beneficiary=beneficiary, system_name="healthie", external_id="6000999" ) track = ClinicalTrack.objects.create( track_name="CKM", description="Chronic Kidney Disease" ) CarePeriod.objects.create( beneficiary=beneficiary, clinical_track=track, start_date="2026-01-01" ) body = { "resource_id": "100020", "resource_id_type": "FormAnswerGroup", "event_type": "form_answer_group.created", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() mock_client.query.return_value = { "data": { "formAnswerGroup": { "id": "100020", "finished": True, "created_at": "2026-08-01T08:00:00Z", "user": { "id": "6000999", "first_name": "Test", "last_name": "Patient", "email": "test@example.com", "dob": "1990-01-01", "gender": "Female", }, "custom_module_form": {"id": "999888", "name": "Daily Check-in"}, "form_answers": [ { "id": "ans1", "label": "Any symptoms today?", "answer": "I have chest pain and shortness of breath", "displayed_answer": "I have chest pain and shortness of breath", "custom_module": { "id": "m1", "label": "Any symptoms today?", "mod_type": "text", "required": True, }, }, ], } } } with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) from access_core.governance.models import CareTeamReviewTask, SafetySignalEvent from access_core.monitoring.models import ( DailyCheckinSubmission, ) # Daily check-in submission created submission = DailyCheckinSubmission.objects.get(beneficiary=beneficiary) self.assertEqual(submission.source_system, "healthie") self.assertEqual(submission.responses.count(), 1) # Red safety signal created signal = SafetySignalEvent.objects.get( beneficiary=beneficiary, signal_severity="red" ) self.assertEqual(signal.signal_type, "adverse_event") self.assertTrue(signal.auto_generated) self.assertEqual(signal.review_status, "pending") # Care team review task created task = CareTeamReviewTask.objects.get(beneficiary=beneficiary) self.assertEqual(task.trigger_type, "safety_signal") self.assertEqual(task.priority, "urgent") self.assertEqual(task.status, "pending") @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) @override_settings(HEALTHIE_INTAKE_FORM_ID="2360739") def test_daily_checkin_safety_signal_yellow(self): """Yellow safety keyword creates safety signal with high priority.""" from access_core.identity.models import Beneficiary, ExternalSystemIdentifier from access_core.monitoring.models import CarePeriod, ClinicalTrack beneficiary = Beneficiary.objects.create( first_name="Test", last_name="Patient", dob="1990-01-01", sex="female" ) ExternalSystemIdentifier.objects.create( beneficiary=beneficiary, system_name="healthie", external_id="6000999" ) track = ClinicalTrack.objects.create( track_name="CKM", description="Chronic Kidney Disease" ) CarePeriod.objects.create( beneficiary=beneficiary, clinical_track=track, start_date="2026-01-01" ) body = { "resource_id": "100030", "resource_id_type": "FormAnswerGroup", "event_type": "form_answer_group.created", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() mock_client.query.return_value = { "data": { "formAnswerGroup": { "id": "100030", "finished": True, "created_at": "2026-08-01T08:00:00Z", "user": { "id": "6000999", "first_name": "Test", "last_name": "Patient", "email": "test@example.com", "dob": "1990-01-01", "gender": "Female", }, "custom_module_form": {"id": "999888", "name": "Daily Check-in"}, "form_answers": [ { "id": "ans1", "label": "Any symptoms today?", "answer": "I have some dizziness and swelling in feet", "displayed_answer": "I have some dizziness and swelling in feet", "custom_module": { "id": "m1", "label": "Any symptoms today?", "mod_type": "text", "required": True, }, }, ], } } } with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) from access_core.governance.models import CareTeamReviewTask, SafetySignalEvent signal = SafetySignalEvent.objects.get( beneficiary=beneficiary, signal_severity="yellow" ) self.assertEqual(signal.signal_type, "symptom_escalation") task = CareTeamReviewTask.objects.get(beneficiary=beneficiary) self.assertEqual(task.priority, "high") # ------------------------------------------------------------------ # Document handler test # ------------------------------------------------------------------ class TestDocumentHandler(TestCase): """Test document.created webhook handler.""" def setUp(self): self.factory = RequestFactory() self.secret = "whsec_test-secret-key" def _make_signed_request(self, body: dict) -> MagicMock: # Match production code: compact JSON for content-length computation compact_json = json.dumps(body, separators=(",", ":")) body_bytes = compact_json.encode("utf-8") content_digest_hash = hashlib.sha256(body_bytes).hexdigest() # Production extracts hash-only from Content-Digest header content_digest_full = f"SHA-256={content_digest_hash}" # Data string: uses hash portion only, compact JSON length data_to_sign = f"post /api/intake/webhooks/healthie/ {content_digest_hash} application/json {len(body_bytes)}" signature = hmac.new( self.secret.encode("utf-8"), data_to_sign.encode("utf-8"), hashlib.sha256, ).hexdigest() return self.factory.post( "/api/intake/webhooks/healthie/", data=body_bytes, content_type="application/json", HTTP_CONTENT_DIGEST=content_digest_full, HTTP_SIGNATURE=f"sig1={signature}", ) @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_document_created_stores_raw(self): """document.created fetches doc data and stores normalized record.""" body = { "resource_id": "3001", "resource_id_type": "Document", "event_type": "document.created", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() mock_client.query.return_value = { "data": { "document": { "id": "3001", "display_name": "Lab Report.pdf", "file_content_type": "application/pdf", "expiring_url": "https://example.com/url", "owner": {"id": "6000001", "email": "patient@example.com"}, } } } with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) raw = RawIngestionRecord.objects.get(source_record_id="3001") self.assertEqual(raw.processing_status, "normalized") # ------------------------------------------------------------------ # Metric entry handler test # ------------------------------------------------------------------ class TestMetricEntryHandler(TestCase): """Test metric_entry.created webhook handler.""" def setUp(self): self.factory = RequestFactory() self.secret = "whsec_test-secret-key" def _make_signed_request(self, body: dict) -> MagicMock: # Match production code: compact JSON for content-length computation compact_json = json.dumps(body, separators=(",", ":")) body_bytes = compact_json.encode("utf-8") content_digest_hash = hashlib.sha256(body_bytes).hexdigest() # Production extracts hash-only from Content-Digest header content_digest_full = f"SHA-256={content_digest_hash}" # Data string: uses hash portion only, compact JSON length data_to_sign = f"post /api/intake/webhooks/healthie/ {content_digest_hash} application/json {len(body_bytes)}" signature = hmac.new( self.secret.encode("utf-8"), data_to_sign.encode("utf-8"), hashlib.sha256, ).hexdigest() return self.factory.post( "/api/intake/webhooks/healthie/", data=body_bytes, content_type="application/json", HTTP_CONTENT_DIGEST=content_digest_full, HTTP_SIGNATURE=f"sig1={signature}", ) @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_metric_entry_created_stores_raw(self): """metric_entry.created fetches entry and stores normalized record.""" body = { "resource_id": "2001", "resource_id_type": "Entry", "event_type": "metric_entry.created", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() mock_client.query.return_value = { "data": { "entry": { "id": "2001", "type": "MetricEntry", "category": "Blood Pressure", "metric_stat": {"systolic": 120, "diastolic": 80}, "created_at": "2026-08-01T07:30:00Z", "client_id": "6000001", "added_by_user": {"id": "6000001", "is_patient": True}, } } } with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) raw = RawIngestionRecord.objects.get(source_record_id="2001") self.assertEqual(raw.processing_status, "normalized") # ------------------------------------------------------------------ # Entry handler tests (symptom, generic) # ------------------------------------------------------------------ class TestEntryHandlers(TestCase): """Test entry.created for symptom and generic entries.""" def setUp(self): self.factory = RequestFactory() self.secret = "whsec_test-secret-key" def _make_signed_request(self, body: dict) -> MagicMock: # Match production code: compact JSON for content-length computation compact_json = json.dumps(body, separators=(",", ":")) body_bytes = compact_json.encode("utf-8") content_digest_hash = hashlib.sha256(body_bytes).hexdigest() # Production extracts hash-only from Content-Digest header content_digest_full = f"SHA-256={content_digest_hash}" # Data string: uses hash portion only, compact JSON length data_to_sign = f"post /api/intake/webhooks/healthie/ {content_digest_hash} application/json {len(body_bytes)}" signature = hmac.new( self.secret.encode("utf-8"), data_to_sign.encode("utf-8"), hashlib.sha256, ).hexdigest() return self.factory.post( "/api/intake/webhooks/healthie/", data=body_bytes, content_type="application/json", HTTP_CONTENT_DIGEST=content_digest_full, HTTP_SIGNATURE=f"sig1={signature}", ) @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_entry_symptom_stores_raw(self): """SymptomEntry is routed to _store_symptom_entry.""" body = { "resource_id": "2002", "resource_id_type": "Entry", "event_type": "entry.created", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() mock_client.query.return_value = { "data": { "entry": { "id": "2002", "type": "SymptomEntry", "description": "Lower back pain, level 6/10", "created_at": "2026-08-01T09:15:00Z", "client_id": "6000001", "added_by_user": {"id": "6000001", "is_patient": True}, } } } with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) raw = RawIngestionRecord.objects.get(source_record_id="2002") self.assertEqual(raw.processing_status, "normalized") @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_entry_metric_skips_duplicate(self): """MetricEntry via entry.created is skipped to avoid duplication.""" body = { "resource_id": "2003", "resource_id_type": "Entry", "event_type": "entry.created", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() mock_client.query.return_value = { "data": { "entry": { "id": "2003", "type": "MetricEntry", "category": "Weight", "metric_stat": {"value": 160, "unit": "lbs"}, "created_at": "2026-08-01T10:00:00Z", "client_id": "6000001", "added_by_user": {"id": "6000001", "is_patient": True}, } } } with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) raw = RawIngestionRecord.objects.get(source_record_id="2003") self.assertEqual(raw.processing_status, "normalized") # ------------------------------------------------------------------ # Tag event handler tests # ------------------------------------------------------------------ class TestTagEventHandlers(TestCase): """Test applied_tag.created and applied_tag.deleted handlers.""" def setUp(self): self.factory = RequestFactory() self.secret = "whsec_test-secret-key" def _make_signed_request(self, body: dict) -> MagicMock: # Match production code: compact JSON for content-length computation compact_json = json.dumps(body, separators=(",", ":")) body_bytes = compact_json.encode("utf-8") content_digest_hash = hashlib.sha256(body_bytes).hexdigest() # Production extracts hash-only from Content-Digest header content_digest_full = f"SHA-256={content_digest_hash}" # Data string: uses hash portion only, compact JSON length data_to_sign = f"post /api/intake/webhooks/healthie/ {content_digest_hash} application/json {len(body_bytes)}" signature = hmac.new( self.secret.encode("utf-8"), data_to_sign.encode("utf-8"), hashlib.sha256, ).hexdigest() return self.factory.post( "/api/intake/webhooks/healthie/", data=body_bytes, content_type="application/json", HTTP_CONTENT_DIGEST=content_digest_full, HTTP_SIGNATURE=f"sig1={signature}", ) @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_tag_created_stores_raw(self): """applied_tag.created stores normalized raw record.""" body = { "resource_id": "5001", "resource_id_type": "AppliedTag", "event_type": "applied_tag.created", "changed_fields": [], "tag_id": "tag-ckm-enrolled", "user_id": "6000001", } request = self._make_signed_request(body) from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) raw = RawIngestionRecord.objects.get(source_record_id="5001") self.assertEqual(raw.processing_status, "normalized") @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_tag_deleted_stores_raw(self): """applied_tag.deleted stores normalized raw record.""" body = { "resource_id": "5002", "resource_id_type": "AppliedTag", "event_type": "applied_tag.deleted", "changed_fields": [], "tag_id": "tag-ckm-enrolled", "user_id": "6000001", } request = self._make_signed_request(body) from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) raw = RawIngestionRecord.objects.get(source_record_id="5002") self.assertEqual(raw.processing_status, "normalized") # ------------------------------------------------------------------ # Form signed handler test # ------------------------------------------------------------------ class TestFormSignedHandler(TestCase): """Test form_answer_group.signed handler.""" def setUp(self): self.factory = RequestFactory() self.secret = "whsec_test-secret-key" def _make_signed_request(self, body: dict) -> MagicMock: # Match production code: compact JSON for content-length computation compact_json = json.dumps(body, separators=(",", ":")) body_bytes = compact_json.encode("utf-8") content_digest_hash = hashlib.sha256(body_bytes).hexdigest() # Production extracts hash-only from Content-Digest header content_digest_full = f"SHA-256={content_digest_hash}" # Data string: uses hash portion only, compact JSON length data_to_sign = f"post /api/intake/webhooks/healthie/ {content_digest_hash} application/json {len(body_bytes)}" signature = hmac.new( self.secret.encode("utf-8"), data_to_sign.encode("utf-8"), hashlib.sha256, ).hexdigest() return self.factory.post( "/api/intake/webhooks/healthie/", data=body_bytes, content_type="application/json", HTTP_CONTENT_DIGEST=content_digest_full, HTTP_SIGNATURE=f"sig1={signature}", ) @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_form_signed_stores_normalized(self): """form_answer_group.signed normalizes even when GraphQL has no data.""" body = { "resource_id": "1001", "resource_id_type": "FormAnswerGroup", "event_type": "form_answer_group.signed", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() mock_client.query.return_value = {"data": {}} with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) raw = RawIngestionRecord.objects.get(source_record_id="1001") self.assertEqual(raw.processing_status, "normalized") # ------------------------------------------------------------------ # IntakeRecord model tests # ------------------------------------------------------------------ class TestIntakeRecord(TestCase): """Test IntakeRecord model behavior.""" def test_default_status(self): record = IntakeRecord.objects.create( external_form_id="test-1", first_name="Test", last_name="User", dob="1990-01-01", sex="male", ) self.assertEqual(record.status, "prospective") def test_all_statuses_available(self): """Verify all 10 statuses from client spec exist.""" expected = { "prospective", "intake_in_progress", "pending_clinical_review", "pending_cms_eligibility", "pending_consent", "pending_alignment", "active_access_beneficiary", "rejected", "withdrawn", "inactive", } actual = {c[0] for c in IntakeRecord.STATUS_CHOICES} self.assertEqual(expected, actual) # ------------------------------------------------------------------ # Handler registration tests # ------------------------------------------------------------------ class TestHandlerRegistration(TestCase): """Verify all expected event handlers are registered.""" def test_intake_handlers_registered(self): """Core intake-related handlers should be registered.""" from access_core.intake.handlers import _HANDLERS expected_events = { "form_answer_group.created", "patient.created", "patient.updated", "patient.merged", "document.created", "document.updated", "metric_entry.created", "metric_entry.updated", "entry.created", "entry.updated", "care_plan.created", "care_plan.updated", "care_plan.activated", "care_plan.deactivated", "care_plan.deleted", "applied_tag.created", "applied_tag.deleted", "form_answer_group.signed", } registered = set(_HANDLERS.keys()) self.assertTrue( expected_events.issubset(registered), f"Missing handlers: {expected_events - registered}", ) # ------------------------------------------------------------------ # Metric normalization tests # ------------------------------------------------------------------ class TestMetricNormalization(TestCase): """Test metric_entry normalization creates ClinicalObservation + OAPMeasureObservation.""" def setUp(self): self.factory = RequestFactory() self.secret = "whsec_test-secret-key" from access_core.identity.models import Beneficiary, ExternalSystemIdentifier from access_core.monitoring.models import CarePeriod, ClinicalTrack self.track = ClinicalTrack.objects.create(track_name="all", is_active=True) self.beneficiary = Beneficiary.objects.create( first_name="Test", last_name="User", dob="1990-01-01", sex="male", is_synthetic=True, ) self.care_period = CarePeriod.objects.create( beneficiary=self.beneficiary, clinical_track=self.track, start_date="2026-01-01", operational_status="active", ) self.ext_id = ExternalSystemIdentifier.objects.create( beneficiary=self.beneficiary, system_name="healthie", external_id="6000001", entity_type="patient", ) def _make_signed_request(self, body: dict): # Match production code: compact JSON for content-length computation compact_json = json.dumps(body, separators=(",", ":")) body_bytes = compact_json.encode("utf-8") content_digest_hash = hashlib.sha256(body_bytes).hexdigest() # Production extracts hash-only from Content-Digest header content_digest_full = f"SHA-256={content_digest_hash}" # Data string: uses hash portion only, compact JSON length data_to_sign = f"post /api/intake/webhooks/healthie/ {content_digest_hash} application/json {len(body_bytes)}" signature = hmac.new( self.secret.encode("utf-8"), data_to_sign.encode("utf-8"), hashlib.sha256, ).hexdigest() return self.factory.post( "/api/intake/webhooks/healthie/", data=body_bytes, content_type="application/json", HTTP_CONTENT_DIGEST=content_digest_full, HTTP_SIGNATURE=f"sig1={signature}", ) @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_metric_entry_creates_clinical_observation(self): """metric_entry.created creates ClinicalObservation and OAPMeasureObservation.""" from access_core.clinical.models import ( ClinicalObservation, OAPMeasureObservation, ) body = { "resource_id": "3001", "resource_id_type": "Entry", "event_type": "metric_entry.created", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() mock_client.query.return_value = { "data": { "entry": { "id": "3001", "type": "MetricEntry", "category": "Blood Pressure", "metric_stat": {"systolic": 120, "diastolic": 80}, "created_at": "2026-08-01T07:30:00Z", "client_id": "6000001", "added_by_user": {"id": "6000001", "is_patient": True}, } } } with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) obs = ClinicalObservation.objects.get(beneficiary=self.beneficiary) self.assertEqual(obs.measure_name, "Blood Pressure") self.assertEqual(obs.observation_type, "vital_sign") self.assertEqual(obs.value_text, "120/80 mmHg") self.assertEqual(obs.value_numeric, None) self.assertEqual(obs.unit, "mmHg") self.assertEqual(obs.collection_date, date(2026, 8, 1)) self.assertEqual(obs.data_source, "patient_upload") self.assertEqual(obs.status, "imported") oap = OAPMeasureObservation.objects.get(canonical_observation=obs) self.assertEqual(oap.beneficiary, self.beneficiary) self.assertEqual(oap.care_period, self.care_period) self.assertEqual(oap.reporting_type, "baseline") self.assertEqual(oap.status, "draft") @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_metric_entry_simple_value(self): """Weight metric with simple value creates observation with numeric value.""" from access_core.clinical.models import ClinicalObservation body = { "resource_id": "3002", "resource_id_type": "Entry", "event_type": "metric_entry.created", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() mock_client.query.return_value = { "data": { "entry": { "id": "3002", "type": "MetricEntry", "category": "Weight", "metric_stat": {"value": 160, "unit": "lbs"}, "created_at": "2026-08-01T08:00:00Z", "client_id": "6000001", "added_by_user": {"id": "6000001", "is_patient": True}, } } } with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) obs = ClinicalObservation.objects.get(beneficiary=self.beneficiary) self.assertEqual(obs.value_numeric, 160.0) self.assertEqual(obs.value_text, "160 lbs") self.assertEqual(obs.unit, "lbs") @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_metric_no_beneficiary_stores_pending(self): """When beneficiary cannot be resolved, data stored in raw_payload.""" body = { "resource_id": "3003", "resource_id_type": "Entry", "event_type": "metric_entry.created", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() mock_client.query.return_value = { "data": { "entry": { "id": "3003", "type": "MetricEntry", "category": "Blood Pressure", "metric_stat": {"systolic": 130, "diastolic": 85}, "created_at": "2026-08-01T09:00:00Z", "client_id": "9999999", "added_by_user": {"id": "9999999", "is_patient": True}, } } } from access_core.clinical.models import ClinicalObservation with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) self.assertEqual(ClinicalObservation.objects.count(), 0) raw = RawIngestionRecord.objects.get(source_record_id="3003") self.assertIn("metric_pending_normalization", raw.raw_payload) # ------------------------------------------------------------------ # Symptom normalization tests # ------------------------------------------------------------------ class TestSymptomNormalization(TestCase): """Test symptom_entry normalization creates SafetySignalEvent.""" def setUp(self): self.factory = RequestFactory() self.secret = "whsec_test-secret-key" from access_core.identity.models import Beneficiary, ExternalSystemIdentifier from access_core.monitoring.models import CarePeriod, ClinicalTrack self.track = ClinicalTrack.objects.create(track_name="all", is_active=True) self.beneficiary = Beneficiary.objects.create( first_name="Test", last_name="User", dob="1990-01-01", sex="male", is_synthetic=True, ) self.care_period = CarePeriod.objects.create( beneficiary=self.beneficiary, clinical_track=self.track, start_date="2026-01-01", operational_status="active", ) self.ext_id = ExternalSystemIdentifier.objects.create( beneficiary=self.beneficiary, system_name="healthie", external_id="6000001", entity_type="patient", ) def _make_signed_request(self, body: dict): # Match production code: compact JSON for content-length computation compact_json = json.dumps(body, separators=(",", ":")) body_bytes = compact_json.encode("utf-8") content_digest_hash = hashlib.sha256(body_bytes).hexdigest() # Production extracts hash-only from Content-Digest header content_digest_full = f"SHA-256={content_digest_hash}" # Data string: uses hash portion only, compact JSON length data_to_sign = f"post /api/intake/webhooks/healthie/ {content_digest_hash} application/json {len(body_bytes)}" signature = hmac.new( self.secret.encode("utf-8"), data_to_sign.encode("utf-8"), hashlib.sha256, ).hexdigest() return self.factory.post( "/api/intake/webhooks/healthie/", data=body_bytes, content_type="application/json", HTTP_CONTENT_DIGEST=content_digest_full, HTTP_SIGNATURE=f"sig1={signature}", ) @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_symptom_red_severity(self): """Red keyword in symptom description creates red SafetySignalEvent.""" from access_core.governance.models import SafetySignalEvent body = { "resource_id": "4001", "resource_id_type": "Entry", "event_type": "entry.created", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() mock_client.query.return_value = { "data": { "entry": { "id": "4001", "type": "SymptomEntry", "description": "Severe chest pain and shortness of breath", "created_at": "2026-08-01T09:15:00Z", "client_id": "6000001", "added_by_user": {"id": "6000001", "is_patient": True}, } } } with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) signal = SafetySignalEvent.objects.get(beneficiary=self.beneficiary) self.assertEqual(signal.signal_severity, "red") self.assertEqual(signal.signal_type, "adverse_event") self.assertEqual( signal.source_record_id, RawIngestionRecord.objects.get(source_record_id="4001").id, ) self.assertTrue(signal.auto_generated) @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_symptom_yellow_severity(self): """Yellow keyword creates yellow SafetySignalEvent.""" from access_core.governance.models import SafetySignalEvent body = { "resource_id": "4002", "resource_id_type": "Entry", "event_type": "entry.created", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() mock_client.query.return_value = { "data": { "entry": { "id": "4002", "type": "SymptomEntry", "description": "Persistent headache and dizziness", "created_at": "2026-08-01T10:00:00Z", "client_id": "6000001", "added_by_user": {"id": "6000001", "is_patient": True}, } } } with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) signal = SafetySignalEvent.objects.get(beneficiary=self.beneficiary) self.assertEqual(signal.signal_severity, "yellow") self.assertEqual(signal.signal_type, "symptom_escalation") @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_symptom_green_severity(self): """No keywords produces green SafetySignalEvent.""" from access_core.governance.models import SafetySignalEvent body = { "resource_id": "4003", "resource_id_type": "Entry", "event_type": "entry.created", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() mock_client.query.return_value = { "data": { "entry": { "id": "4003", "type": "SymptomEntry", "description": "Feeling tired today, slept poorly", "created_at": "2026-08-01T11:00:00Z", "client_id": "6000001", "added_by_user": {"id": "6000001", "is_patient": True}, } } } with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) signal = SafetySignalEvent.objects.get(beneficiary=self.beneficiary) self.assertEqual(signal.signal_severity, "green") @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_symptom_no_beneficiary_stores_pending(self): """When beneficiary cannot be resolved, data stored in raw_payload.""" body = { "resource_id": "4004", "resource_id_type": "Entry", "event_type": "entry.created", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() mock_client.query.return_value = { "data": { "entry": { "id": "4004", "type": "SymptomEntry", "description": "Mild headache", "created_at": "2026-08-01T12:00:00Z", "client_id": "9999999", "added_by_user": {"id": "9999999", "is_patient": True}, } } } from access_core.governance.models import SafetySignalEvent with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) self.assertEqual(SafetySignalEvent.objects.count(), 0) raw = RawIngestionRecord.objects.get(source_record_id="4004") self.assertIn("symptom_pending_normalization", raw.raw_payload) # ------------------------------------------------------------------ # Care plan normalization tests # ------------------------------------------------------------------ class TestCarePlanNormalization(TestCase): """Test care_plan normalization creates MonitoringPlan + items.""" def setUp(self): self.factory = RequestFactory() self.secret = "whsec_test-secret-key" from access_core.identity.models import Beneficiary, ExternalSystemIdentifier from access_core.monitoring.models import CarePeriod, ClinicalTrack self.track = ClinicalTrack.objects.create(track_name="all", is_active=True) self.beneficiary = Beneficiary.objects.create( first_name="Test", last_name="User", dob="1990-01-01", sex="male", is_synthetic=True, ) self.care_period = CarePeriod.objects.create( beneficiary=self.beneficiary, clinical_track=self.track, start_date="2026-01-01", operational_status="active", ) self.ext_id = ExternalSystemIdentifier.objects.create( beneficiary=self.beneficiary, system_name="healthie", external_id="6000001", entity_type="patient", ) def _make_signed_request(self, body: dict): # Match production code: compact JSON for content-length computation compact_json = json.dumps(body, separators=(",", ":")) body_bytes = compact_json.encode("utf-8") content_digest_hash = hashlib.sha256(body_bytes).hexdigest() # Production extracts hash-only from Content-Digest header content_digest_full = f"SHA-256={content_digest_hash}" # Data string: uses hash portion only, compact JSON length data_to_sign = f"post /api/intake/webhooks/healthie/ {content_digest_hash} application/json {len(body_bytes)}" signature = hmac.new( self.secret.encode("utf-8"), data_to_sign.encode("utf-8"), hashlib.sha256, ).hexdigest() return self.factory.post( "/api/intake/webhooks/healthie/", data=body_bytes, content_type="application/json", HTTP_CONTENT_DIGEST=content_digest_full, HTTP_SIGNATURE=f"sig1={signature}", ) @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_care_plan_creates_monitoring_plan(self): """care_plan.created creates MonitoringPlan + MonitoringPlanItems.""" from access_core.monitoring.models import MonitoringPlan, MonitoringPlanItem body = { "resource_id": "5001", "resource_id_type": "CarePlan", "event_type": "care_plan.created", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() mock_client.query.return_value = { "data": { "carePlan": { "id": "5001", "name": "CKD Monitoring Plan", "description": "Chronic kidney disease monitoring protocol", "is_active": True, "is_template": False, "patient_id": "6000001", "recommendations": [ { "id": "r1", "recommendation_body": "Monthly kidney function panel", "recommendation_type": "lab", }, { "id": "r2", "recommendation_body": "Weekly BP monitoring", "recommendation_type": "vital_sign", }, ], } } } with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) plan = MonitoringPlan.objects.get(beneficiary=self.beneficiary) self.assertEqual(plan.clinical_trigger, "CKD Monitoring Plan") self.assertEqual(plan.status, "active") self.assertEqual( plan.baseline_abnormality, "Chronic kidney disease monitoring protocol" ) items = MonitoringPlanItem.objects.filter(monitoring_plan=plan) self.assertEqual(items.count(), 2) item_names = {i.item_description for i in items} self.assertIn("Monthly kidney function panel", item_names) self.assertIn("Weekly BP monitoring", item_names) @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_care_plan_template_skipped(self): """Templates (is_template=True) do not create MonitoringPlan.""" from access_core.monitoring.models import MonitoringPlan body = { "resource_id": "5002", "resource_id_type": "CarePlan", "event_type": "care_plan.created", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() mock_client.query.return_value = { "data": { "carePlan": { "id": "5002", "name": "Template Plan", "description": "A template", "is_active": True, "is_template": True, "patient_id": "6000001", "recommendations": [], } } } with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook response = healthie_webhook(request) self.assertEqual(response.status_code, 200) self.assertEqual(MonitoringPlan.objects.count(), 0) @override_settings( HEALTHIE_WEBHOOK_SECRET="whsec_test-secret-key", HEALTHIE_WEBHOOK_VERIFY_SIGNATURE=True, ) def test_care_plan_deleted_archives_plan(self): """care_plan.deleted archives existing MonitoringPlan.""" from access_core.monitoring.models import MonitoringPlan body = { "resource_id": "5001", "resource_id_type": "CarePlan", "event_type": "care_plan.created", "changed_fields": [], } request = self._make_signed_request(body) mock_client = MagicMock() mock_client.query.return_value = { "data": { "carePlan": { "id": "5001", "name": "CKD Monitoring Plan", "description": "Test plan", "is_active": True, "is_template": False, "patient_id": "6000001", "recommendations": [], } } } with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): from access_core.intake.views import healthie_webhook healthie_webhook(request) self.assertEqual(MonitoringPlan.objects.count(), 1) body2 = { "resource_id": "5001-del", "resource_id_type": "CarePlan", "event_type": "care_plan.deleted", "changed_fields": [], } request2 = self._make_signed_request(body2) mock_client.query.return_value = { "data": { "carePlan": { "id": "5001", "name": "CKD Monitoring Plan", "description": "Test plan", "is_active": False, "is_template": False, "patient_id": "6000001", "recommendations": [], } } } with patch( "access_core.intake.consumers._get_healthie_client", return_value=mock_client, ): healthie_webhook(request2) plan = MonitoringPlan.objects.get(beneficiary=self.beneficiary) self.assertEqual(plan.status, "archived")