435 lines
18 KiB
Python
435 lines
18 KiB
Python
"""
|
|
Tests for public complaint form and view.
|
|
"""
|
|
from datetime import date
|
|
|
|
from django.test import Client, TestCase
|
|
from django.urls import reverse
|
|
|
|
from apps.complaints.models import Complaint, ComplaintSourceType, Inquiry
|
|
from apps.organizations.models import Area, Department, Hospital, LocationType, Section
|
|
|
|
|
|
class PublicComplaintViewTests(TestCase):
|
|
def setUp(self):
|
|
self.client = Client()
|
|
self.hospital = Hospital.objects.create(
|
|
name="Test Hospital",
|
|
code="TEST",
|
|
status="active",
|
|
)
|
|
self.department = Department.objects.create(
|
|
hospital=self.hospital,
|
|
name="Emergency",
|
|
name_en="Emergency",
|
|
code="test_er",
|
|
status="active",
|
|
)
|
|
self.section = Section.objects.create(
|
|
department=self.department,
|
|
name_en="ER Section A",
|
|
code="test_er_a",
|
|
status="active",
|
|
)
|
|
self.area = Area.objects.create(
|
|
hospital=self.hospital,
|
|
name_en="Main Lobby",
|
|
code="lobby",
|
|
location_type=LocationType.OP,
|
|
status="active",
|
|
)
|
|
|
|
def test_public_complaint_form_get(self):
|
|
try:
|
|
response = self.client.get(reverse("complaints:public_complaint_submit"))
|
|
self.assertEqual(response.status_code, 200)
|
|
except ValueError:
|
|
pass
|
|
|
|
def test_public_complaint_post_saves_location_type(self):
|
|
data = {
|
|
"complainant_name": "John Doe",
|
|
"mobile_number": "0512345678",
|
|
"relation_to_patient": "patient",
|
|
"patient_name": "Jane Doe",
|
|
"national_id": "1234567890",
|
|
"incident_date": date.today().isoformat(),
|
|
"hospital": str(self.hospital.id),
|
|
"location_type": "OP",
|
|
"category": "medical",
|
|
"department": str(self.department.id),
|
|
"section": str(self.section.id),
|
|
"complaint_details": "Test complaint with enough detail.",
|
|
}
|
|
response = self.client.post(
|
|
reverse("complaints:public_complaint_submit"),
|
|
data,
|
|
HTTP_X_REQUESTED_WITH="XMLHttpRequest",
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
result = response.json()
|
|
self.assertTrue(result["success"])
|
|
|
|
complaint = Complaint.objects.first()
|
|
self.assertIsNotNone(complaint)
|
|
self.assertEqual(complaint.location_type, "OP")
|
|
self.assertEqual(complaint.department_id, self.department.id)
|
|
self.assertEqual(complaint.section_id, self.section.id)
|
|
self.assertEqual(complaint.hospital_id, self.hospital.id)
|
|
|
|
def test_public_complaint_post_missing_location_type_succeeds(self):
|
|
data = {
|
|
"complainant_name": "John Doe",
|
|
"mobile_number": "0512345678",
|
|
"relation_to_patient": "patient",
|
|
"patient_name": "Jane Doe",
|
|
"national_id": "1234567890",
|
|
"incident_date": date.today().isoformat(),
|
|
"hospital": str(self.hospital.id),
|
|
"location_type": "",
|
|
"category": "medical",
|
|
"department": str(self.department.id),
|
|
"complaint_details": "Test complaint.",
|
|
}
|
|
response = self.client.post(
|
|
reverse("complaints:public_complaint_submit"),
|
|
data,
|
|
HTTP_X_REQUESTED_WITH="XMLHttpRequest",
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
result = response.json()
|
|
self.assertFalse(result["success"])
|
|
|
|
def test_public_complaint_post_with_all_location_types(self):
|
|
for loc_type in ["OP", "IP", "ER", "GENERAL"]:
|
|
Complaint.objects.all().delete()
|
|
data = {
|
|
"complainant_name": "John Doe",
|
|
"mobile_number": "0512345678",
|
|
"relation_to_patient": "patient",
|
|
"patient_name": "Jane Doe",
|
|
"national_id": "1234567890",
|
|
"incident_date": date.today().isoformat(),
|
|
"hospital": str(self.hospital.id),
|
|
"location_type": loc_type,
|
|
"category": "medical",
|
|
"department": str(self.department.id),
|
|
"complaint_details": f"Test for {loc_type}",
|
|
}
|
|
response = self.client.post(
|
|
reverse("complaints:public_complaint_submit"),
|
|
data,
|
|
HTTP_X_REQUESTED_WITH="XMLHttpRequest",
|
|
)
|
|
self.assertEqual(response.status_code, 200, f"Failed for location_type={loc_type}")
|
|
complaint = Complaint.objects.first()
|
|
self.assertEqual(complaint.location_type, loc_type)
|
|
|
|
def test_public_complaint_post_invalid_hospital(self):
|
|
data = {
|
|
"complainant_name": "John Doe",
|
|
"mobile_number": "0512345678",
|
|
"hospital": "00000000-0000-0000-0000-000000000000",
|
|
"location_type": "OP",
|
|
"department": str(self.department.id),
|
|
"complaint_details": "Test.",
|
|
}
|
|
response = self.client.post(
|
|
reverse("complaints:public_complaint_submit"),
|
|
data,
|
|
HTTP_X_REQUESTED_WITH="XMLHttpRequest",
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_public_complaint_post_section_optional(self):
|
|
data = {
|
|
"complainant_name": "John Doe",
|
|
"mobile_number": "0512345678",
|
|
"relation_to_patient": "patient",
|
|
"patient_name": "Jane Doe",
|
|
"national_id": "1234567890",
|
|
"incident_date": date.today().isoformat(),
|
|
"hospital": str(self.hospital.id),
|
|
"location_type": "IP",
|
|
"category": "medical",
|
|
"department": str(self.department.id),
|
|
"complaint_details": "Test without section.",
|
|
}
|
|
response = self.client.post(
|
|
reverse("complaints:public_complaint_submit"),
|
|
data,
|
|
HTTP_X_REQUESTED_WITH="XMLHttpRequest",
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
complaint = Complaint.objects.first()
|
|
self.assertIsNone(complaint.section_id)
|
|
self.assertEqual(complaint.location_type, "IP")
|
|
|
|
def test_public_complaint_post_saves_contact_info(self):
|
|
data = {
|
|
"complainant_name": "John Doe",
|
|
"mobile_number": "0512345678",
|
|
"email": "john@example.com",
|
|
"relation_to_patient": "relative",
|
|
"patient_name": "Jane Doe",
|
|
"national_id": "1234567890",
|
|
"incident_date": date.today().isoformat(),
|
|
"hospital": str(self.hospital.id),
|
|
"location_type": "ER",
|
|
"category": "medical",
|
|
"department": str(self.department.id),
|
|
"complaint_details": "Contact info test.",
|
|
"staff_name": "Dr. Smith",
|
|
"expected_result": "Quick resolution",
|
|
}
|
|
response = self.client.post(
|
|
reverse("complaints:public_complaint_submit"),
|
|
data,
|
|
HTTP_X_REQUESTED_WITH="XMLHttpRequest",
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
complaint = Complaint.objects.first()
|
|
self.assertEqual(complaint.contact_name, "John Doe")
|
|
self.assertEqual(complaint.contact_phone, "0512345678")
|
|
self.assertEqual(complaint.contact_email, "john@example.com")
|
|
self.assertEqual(complaint.staff_name, "Dr. Smith")
|
|
self.assertEqual(complaint.expected_result, "Quick resolution")
|
|
|
|
|
|
class DepartmentNotificationTaskTests(TestCase):
|
|
"""The send-to-department notifications run as a Celery task (offloaded from the view)."""
|
|
|
|
def setUp(self):
|
|
from django.contrib.auth import get_user_model
|
|
from apps.organizations.models import Staff
|
|
|
|
User = get_user_model()
|
|
self.hospital = Hospital.objects.create(name="H1", code="H1", status="active")
|
|
self.complaint = Complaint.objects.create(
|
|
hospital=self.hospital,
|
|
title="T",
|
|
description="D",
|
|
status="in_progress",
|
|
reference_number="REF-TEST-1",
|
|
)
|
|
self.staff_user = User.objects.create_user(username="champ@x", email="champ@x", password="p")
|
|
self.staff = Staff.objects.create(
|
|
hospital=self.hospital,
|
|
first_name="Champ",
|
|
last_name="X",
|
|
status="active",
|
|
staff_type="other",
|
|
job_title="C",
|
|
employee_id="EMP-1",
|
|
user=self.staff_user,
|
|
email="champ@x",
|
|
)
|
|
|
|
def _payload(self, **overrides):
|
|
payload = {
|
|
"complaint_id": str(self.complaint.id),
|
|
"reference_number": self.complaint.reference_number,
|
|
"title": self.complaint.title,
|
|
"department_name": "Emergency",
|
|
"note": "",
|
|
"email_subject": "",
|
|
"email_body": "",
|
|
"targets": [
|
|
{
|
|
"email": "champ@x",
|
|
"phone": "+966500000000",
|
|
"link": "https://example.com/explain/token/",
|
|
"label": "Champion",
|
|
"display_name": "Champ X",
|
|
}
|
|
],
|
|
}
|
|
payload.update(overrides)
|
|
return payload
|
|
|
|
def test_task_sends_email_and_sms_to_targets(self):
|
|
from unittest.mock import patch
|
|
from apps.complaints.tasks import send_department_notification_task
|
|
|
|
with patch("apps.notifications.services.NotificationService.send_email") as mock_email, \
|
|
patch("apps.notifications.services.NotificationService.send_sms") as mock_sms:
|
|
result = send_department_notification_task.apply(args=[self._payload()]).get()
|
|
|
|
self.assertEqual(result, {"emailed": 1, "smsed": 1})
|
|
mock_email.assert_called_once()
|
|
self.assertEqual(mock_email.call_args.kwargs["email"], "champ@x")
|
|
mock_sms.assert_called_once()
|
|
|
|
def test_task_skips_missing_complaint(self):
|
|
import uuid
|
|
from apps.complaints.tasks import send_department_notification_task
|
|
|
|
payload = self._payload(complaint_id=str(uuid.uuid4()))
|
|
result = send_department_notification_task.apply(args=[payload]).get()
|
|
self.assertEqual(result, {"emailed": 0, "smsed": 0})
|
|
|
|
def test_task_continues_if_one_send_fails(self):
|
|
from unittest.mock import patch
|
|
from apps.complaints.tasks import send_department_notification_task
|
|
|
|
payload = self._payload(targets=[
|
|
{"email": "a@x", "phone": "", "link": "l", "label": "Champion", "display_name": "A"},
|
|
{"email": "", "phone": "+9665", "link": "l", "label": "Manager", "display_name": "B"},
|
|
])
|
|
with patch("apps.notifications.services.NotificationService.send_email", side_effect=Exception("smtp down")), \
|
|
patch("apps.notifications.services.NotificationService.send_sms") as mock_sms:
|
|
result = send_department_notification_task.apply(args=[payload]).get()
|
|
|
|
# Email failed (best-effort), SMS still sent; task did not raise.
|
|
self.assertEqual(result, {"emailed": 0, "smsed": 1})
|
|
mock_sms.assert_called_once()
|
|
|
|
|
|
class InvestigationReviewOtpTests(TestCase):
|
|
"""The investigation-review verification code goes to the logged-in reviewer."""
|
|
|
|
def setUp(self):
|
|
from django.contrib.auth import get_user_model
|
|
from apps.organizations.models import Staff
|
|
from apps.complaints.models import ComplaintExplanation, ChampionInvestigation
|
|
|
|
User = get_user_model()
|
|
self.hospital = Hospital.objects.create(name="H1", code="H1", status="active")
|
|
self.complaint = Complaint.objects.create(
|
|
hospital=self.hospital, title="T", description="D",
|
|
status="in_progress", reference_number="REF-INV-1",
|
|
)
|
|
# Manager = investigator of record (explanation.staff / investigation.champion)
|
|
self.manager_user = User.objects.create_user(username="mgr@x", email="mgr@x", password="p")
|
|
self.manager = Staff.objects.create(
|
|
hospital=self.hospital, first_name="Mgr", last_name="X",
|
|
status="active", staff_type="other", job_title="M",
|
|
employee_id="EMP-M", user=self.manager_user, email="mgr@x",
|
|
)
|
|
# Champion = the actual reviewer (logged in)
|
|
self.champ_user = User.objects.create_user(username="champ@x", email="champ@x", password="p")
|
|
self.champion = Staff.objects.create(
|
|
hospital=self.hospital, first_name="Champ", last_name="X",
|
|
status="active", staff_type="other", job_title="C",
|
|
employee_id="EMP-C", user=self.champ_user, email="champ@x",
|
|
)
|
|
self.token = "manager-token-abc"
|
|
self.explanation = ComplaintExplanation.objects.create(
|
|
complaint=self.complaint, staff=self.manager, token=self.token,
|
|
)
|
|
self.investigation = ChampionInvestigation.objects.create(
|
|
complaint=self.complaint, explanation=self.explanation,
|
|
champion=self.manager, status="answers_received",
|
|
)
|
|
|
|
def test_otp_goes_to_logged_in_reviewer_not_investigator_of_record(self):
|
|
from unittest.mock import patch, Mock
|
|
|
|
self.client.force_login(self.champ_user)
|
|
url = reverse("complaints:champion_review_answers", args=[self.complaint.id, self.token])
|
|
|
|
with patch("apps.notifications.services.NotificationService.send_email",
|
|
return_value=Mock(status="sent")) as mock_email, \
|
|
patch("apps.notifications.services.NotificationService.send_sms",
|
|
return_value=Mock(status="sent")) as mock_sms:
|
|
self.client.post(url, {
|
|
"action": "send_code",
|
|
"final_reply": "My final reply.",
|
|
"consent": "on",
|
|
})
|
|
|
|
# The code MUST go to the logged-in champion, not the manager of record.
|
|
mock_email.assert_called_once()
|
|
self.assertEqual(mock_email.call_args.kwargs["email"], "champ@x")
|
|
# Champion has no phone on file -> SMS not used (and manager's phone not pulled in).
|
|
mock_sms.assert_not_called()
|
|
|
|
def test_send_code_persists_assessment_draft(self):
|
|
"""The assessment must be persisted at send_code so a return visit keeps it.
|
|
|
|
Regression: previously only otp_code/otp_sent_at were saved, so leaving the
|
|
page and coming back lost the final reply / findings, and verify_submit then
|
|
rejected the submission as an empty assessment.
|
|
"""
|
|
from unittest.mock import patch, Mock
|
|
|
|
self.client.force_login(self.champ_user)
|
|
url = reverse("complaints:champion_review_answers", args=[self.complaint.id, self.token])
|
|
with patch("apps.notifications.services.NotificationService.send_email",
|
|
return_value=Mock(status="sent")), \
|
|
patch("apps.notifications.services.NotificationService.send_sms",
|
|
return_value=Mock(status="sent")):
|
|
self.client.post(url, {
|
|
"action": "send_code",
|
|
"final_reply": "Persisted final reply.",
|
|
"negligence_finding": "no",
|
|
"policy_issue_finding": "yes",
|
|
"requires_improvement_project": "yes",
|
|
"improvement_project_note": "Need a project.",
|
|
"consent": "on",
|
|
})
|
|
|
|
self.investigation.refresh_from_db()
|
|
self.assertEqual(self.investigation.final_reply, "Persisted final reply.")
|
|
self.assertEqual(self.investigation.negligence_finding, "no")
|
|
self.assertEqual(self.investigation.policy_issue_finding, "yes")
|
|
self.assertEqual(self.investigation.requires_improvement_project, "yes")
|
|
self.assertEqual(self.investigation.improvement_project_note, "Need a project.")
|
|
self.assertTrue(self.investigation.otp_code)
|
|
|
|
|
|
class InquiryTimestampTests(TestCase):
|
|
"""resolved_at / closed_at are auto-stamped on status transition."""
|
|
|
|
def setUp(self):
|
|
self.hospital = Hospital.objects.create(name="H1", code="H1", status="active")
|
|
self.inquiry = Inquiry.objects.create(
|
|
hospital=self.hospital,
|
|
subject="Test",
|
|
message="M",
|
|
status="in_progress",
|
|
)
|
|
|
|
def test_resolved_at_stamped_on_resolve(self):
|
|
self.inquiry.status = "resolved"
|
|
self.inquiry.save()
|
|
self.inquiry.refresh_from_db()
|
|
self.assertIsNotNone(self.inquiry.resolved_at)
|
|
self.assertIsNone(self.inquiry.closed_at)
|
|
|
|
def test_closed_at_stamped_on_close(self):
|
|
self.inquiry.status = "closed"
|
|
self.inquiry.save()
|
|
self.inquiry.refresh_from_db()
|
|
self.assertIsNotNone(self.inquiry.closed_at)
|
|
|
|
def test_reopen_preserves_timestamps(self):
|
|
self.inquiry.status = "resolved"
|
|
self.inquiry.save()
|
|
stamped = self.inquiry.resolved_at
|
|
self.inquiry.status = "in_progress"
|
|
self.inquiry.save()
|
|
self.inquiry.refresh_from_db()
|
|
self.assertEqual(self.inquiry.resolved_at, stamped)
|
|
|
|
def test_reresolve_updates_timestamp(self):
|
|
self.inquiry.status = "resolved"
|
|
self.inquiry.save()
|
|
first = self.inquiry.resolved_at
|
|
self.inquiry.status = "in_progress"
|
|
self.inquiry.save()
|
|
self.inquiry.status = "resolved"
|
|
self.inquiry.save()
|
|
self.inquiry.refresh_from_db()
|
|
self.assertGreater(self.inquiry.resolved_at, first)
|
|
|
|
def test_resolved_by_stamped_from_acting_user(self):
|
|
from django.contrib.auth import get_user_model
|
|
User = get_user_model()
|
|
resolver = User.objects.create_user(username="resolver@x", email="resolver@x", password="p")
|
|
self.inquiry._acting_user = resolver
|
|
self.inquiry.status = "resolved"
|
|
self.inquiry.save()
|
|
self.inquiry.refresh_from_db()
|
|
self.assertEqual(self.inquiry.resolved_by, resolver)
|