import json from unittest.mock import patch from django.core.cache import cache from django.contrib.auth import get_user_model from django.test import TestCase from django.urls import reverse from crm.models import Activity, Contact, Lead from schools.backups import build_backup, restore_backup from schools.cache import finder_rows from schools.forms import SchoolForm from schools.models import School class SchoolBackupTests(TestCase): def test_finder_rows_are_cached_per_user_and_invalidated_after_save(self): cache.clear() self.addCleanup(cache.clear) admin = get_user_model().objects.create_superuser(username="cache-admin", password="test") school = School.objects.create( organization_id=987649, name="Cached School", data={"organizationName": "Cached School"}, ) from schools.access import shaped_rows with patch("schools.cache.shaped_rows", wraps=shaped_rows) as build_rows: first = finder_rows(admin) second = finder_rows(admin) self.assertEqual(first, second) self.assertEqual(build_rows.call_count, 1) with self.captureOnCommitCallbacks(execute=True): school.data = {"organizationName": "Updated School"} school.save(update_fields=["data"]) updated = finder_rows(admin) self.assertEqual(build_rows.call_count, 2) self.assertEqual(updated[0]["organizationName"], "Updated School") other_user = get_user_model().objects.create_user(username="cache-user", password="test") self.assertEqual(finder_rows(other_user), []) def test_finder_metadata_is_cached_and_invalidated_after_school_change(self): cache.clear() self.addCleanup(cache.clear) admin = get_user_model().objects.create_superuser( username="metadata-cache-admin", email="metadata@example.com", password="metadata-cache-test-password", ) school = School.objects.create( organization_id=987648, name="Metadata Cache School", district_name="Before update", data={"organizationName": "Metadata Cache School"}, ) from schools.finder_data import _metadata, finder_result with patch("schools.finder_data._metadata", wraps=_metadata) as build_metadata: first = finder_result(admin, {"page": "1"}) second = finder_result(admin, {"page": "1"}) self.assertEqual(first["metadata"], second["metadata"]) self.assertEqual(build_metadata.call_count, 1) with self.captureOnCommitCallbacks(execute=True): school.district_name = "After update" school.save(update_fields=["district_name"]) updated = finder_result(admin, {"page": "1"}) self.assertEqual(build_metadata.call_count, 2) self.assertIn("After update", updated["metadata"]["districts"]) def test_admin_school_and_backup_pages_load_separately(self): admin = get_user_model().objects.create_superuser(username="site-admin", password="test") self.client.force_login(admin) school_response = self.client.get(reverse("school_admin")) new_school_response = self.client.get(reverse("school_admin"), {"new": "1"}) backup_response = self.client.get(reverse("data_backup")) self.assertEqual(school_response.status_code, 200) self.assertEqual(new_school_response.status_code, 200) self.assertEqual(backup_response.status_code, 200) self.assertContains(school_response, "Add school") self.assertNotContains(school_response, "Add new school") self.assertContains(school_response, "Recently deleted") self.assertContains(new_school_response, "Add new school") self.assertContains(new_school_response, "Cancel") self.assertContains(backup_response, "Download full backup") def test_school_and_crm_pages_load_compiled_tailwind_stylesheet(self): login_response = self.client.get(reverse("login")) self.assertContains(login_response, "schools/tailwind.css") admin = get_user_model().objects.create_superuser( username="tailwind-admin", email="tailwind@example.com", password="Test-password-123" ) self.client.force_login(admin) for url in (reverse("index"), reverse("crm_dashboard"), reverse("school_admin"), reverse("data_backup")): with self.subTest(url=url): response = self.client.get(url) self.assertEqual(response.status_code, 200) self.assertContains(response, "schools/tailwind.css") if url == reverse("index"): self.assertContains(response, "schools/finder-loading.css") self.assertContains(response, "initial-skeleton-row") self.assertContains(response, "results-skeleton-row") def test_school_admin_paginates_and_searches(self): admin = get_user_model().objects.create_superuser(username="list-admin", password="test") self.client.force_login(admin) School.objects.bulk_create([ School(organization_id=980000 + index, name=f"Pagination School {index}", data={}) for index in range(31) ]) response = self.client.get(reverse("school_admin")) self.assertContains(response, "Pagination School 24") self.assertNotContains(response, "Pagination School 25") self.assertContains(response, "Page 1 of 2") search_response = self.client.get(reverse("school_admin"), {"q": "Pagination School 30"}) self.assertContains(search_response, "Pagination School 30") self.assertNotContains(search_response, "Pagination School 29") def test_finder_api_returns_100_rows_and_filters_across_all_pages(self): admin = get_user_model().objects.create_superuser(username="finder-admin", password="test") self.client.force_login(admin) School.objects.bulk_create([ School( organization_id=970000 + index, name=f"Finder School {index}", district_name="Test District", total_students=index // 2, data={"organizationId": 970000 + index, "organizationName": f"Finder School {index}", "districtName": "Test District", "total_students": index // 2}, ) for index in range(205) ]) with patch("schools.cache.shaped_rows", side_effect=AssertionError("Finder should query the database")): first_page = self.client.get("/api/dump").json() second_page = self.client.get("/api/dump", {"page": "2"}).json() search = self.client.get("/api/dump", {"q": "Finder School 204"}).json() filtered = self.client.get( "/api/dump", {"dist": "Test District", "tmin": "10", "tmax": "11"} ).json() requested_full = self.client.get("/api/dump", {"all": "1"}).json() bulk_ids = self.client.get("/api/dump", {"ids_only": "1"}).json() export = self.client.get("/api/export", {"fmt": "csv", "q": "Finder School 204"}) self.assertEqual(len(first_page["results"]), 100) self.assertEqual(first_page["total"], 205) self.assertEqual(len(second_page["results"]), 100) self.assertEqual(second_page["page"], 2) expected_order = sorted(range(205), key=lambda index: index // 2, reverse=True) self.assertEqual([row["organizationId"] for row in first_page["results"]], [970000 + index for index in expected_order[:100]]) self.assertEqual([row["organizationId"] for row in second_page["results"]], [970000 + index for index in expected_order[100:200]]) self.assertEqual(search["total"], 1) self.assertEqual(search["results"][0]["organizationName"], "Finder School 204") self.assertEqual(filtered["total"], 4) self.assertEqual(len(requested_full["results"]), 100) self.assertEqual(len(bulk_ids["ids"]), 205) self.assertIn("Finder School 204", export.content.decode("utf-8-sig")) def test_finder_filters_schools_by_contact_number_type(self): admin = get_user_model().objects.create_superuser( username="phone-filter-admin", email="phone@example.com", password="phone-filter-test-password", ) self.client.force_login(admin) School.objects.bulk_create([ School(organization_id=975001, name="Landline School", data={"organizationName": "Landline School", "contactNumber": "123-4567"}), School(organization_id=975002, name="Mobile School", data={"organizationName": "Mobile School", "contactNumber": "+977 9812345678"}), School(organization_id=975003, name="Invalid Number School", data={"organizationName": "Invalid Number School", "contactNumber": "1234"}), ]) landlines = self.client.get("/api/dump", {"phone": "landline"}).json() mobiles = self.client.get("/api/dump", {"phone": "mobile"}).json() self.assertEqual(landlines["total"], 1) self.assertEqual(landlines["results"][0]["organizationName"], "Landline School") self.assertEqual(mobiles["total"], 1) self.assertEqual(mobiles["results"][0]["organizationName"], "Mobile School") def test_phone_filter_handles_more_rows_than_sqlite_parameter_limit(self): admin = get_user_model().objects.create_superuser( username="large-phone-filter-admin", email="phone@example.com", password="phone-filter-test-password", ) self.client.force_login(admin) School.objects.bulk_create([ School( organization_id=976000 + index, name=f"Landline School {index}", data={ "organizationName": f"Landline School {index}", "contactNumber": "123-4567", }, ) for index in range(1100) ]) response = self.client.get( "/api/dump", {"page": "1", "sort": "contactNumber", "dir": "1", "phone": "landline"}, ) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["total"], 1100) self.assertEqual(len(response.json()["results"]), 100) def test_school_form_validates_contact_number_length(self): School.objects.create( organization_id=987654, name="Phone Template School", data={"contactNumber": "12345"}, ) unbound_form = SchoolForm() phone_field = next( item["name"] for item in unbound_form.extra_fields if item["key"] == "contactNumber" ) form = SchoolForm(data={ "organization_id": "987655", "name": "Phone Test School", "district_id": "", "district_name": "", "municipality": "", "school_type": "", "total_students": "", "up_to_class": "", phone_field: "1234", }) self.assertFalse(form.is_valid()) self.assertIn("5-7 digits", str(form.errors)) def test_bulk_crm_update_changes_only_selected_schools(self): admin = get_user_model().objects.create_superuser( username="bulk-admin", email="bulk@example.com", password="test-password" ) self.client.force_login(admin) selected = [ School.objects.create(organization_id=960001 + index, name=f"Selected {index}", data={}) for index in range(2) ] unselected = School.objects.create(organization_id=960003, name="Not selected", data={}) response = self.client.post( reverse("crm_bulk"), data=json.dumps({ "ids": [school.organization_id for school in selected], "owner": admin.pk, }), content_type="application/json", ) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["updated"], 2) self.assertTrue(all(Lead.objects.get(school=school).owner == admin for school in selected)) self.assertFalse(Lead.objects.filter(school=unselected).exists()) def test_create_form_exposes_imported_fields_without_copying_template_values(self): School.objects.create( organization_id=987653, name="Template School", data={"electricity": True, "totalComputers": 3, "headTeacherName": "Template name"}, ) form = SchoolForm() names = {item["key"]: item["name"] for item in form.extra_fields} self.assertNotIn("data", form.fields) self.assertEqual(form.fields[names["electricity"]].initial, "") self.assertIsNone(form.fields[names["totalComputers"]].initial) submitted = { "organization_id": "987652", "name": "New School", "district_id": "", "district_name": "", "municipality": "", "school_type": "", "total_students": "", "up_to_class": "", names["electricity"]: "true", names["totalComputers"]: "4", names["headTeacherName"]: "New head", } form = SchoolForm(data=submitted) self.assertTrue(form.is_valid(), form.errors) school = form.save() self.assertTrue(school.data["electricity"]) self.assertEqual(school.data["totalComputers"], 4) self.assertEqual(school.data["headTeacherName"], "New head") def test_soft_deleted_school_keeps_crm_and_can_be_restored(self): school = School.objects.create(organization_id=987651, name="Recoverable School", data={}) lead = Lead.objects.create(school=school, stage="contacted") school.delete() self.assertFalse(School.objects.filter(pk=school.pk).exists()) self.assertIsNotNone(School.all_objects.get(pk=school.pk).deleted_at) self.assertTrue(Lead.objects.filter(pk=lead.pk).exists()) deleted_school = School.all_objects.get(pk=school.pk) self.assertTrue(deleted_school.restore()) self.assertTrue(School.objects.filter(pk=school.pk).exists()) def test_admin_can_delete_and_restore_school(self): admin = get_user_model().objects.create_superuser(username="recovery-admin", password="test") self.client.force_login(admin) school = School.objects.create(organization_id=987650, name="Admin Recovery School", data={}) lead = Lead.objects.create(school=school, stage="contacted") delete_response = self.client.post(reverse("school_admin"), { "action": "delete", "school_id": school.pk, }) self.assertEqual(delete_response.status_code, 302) self.assertFalse(School.objects.filter(pk=school.pk).exists()) self.assertTrue(Lead.objects.filter(pk=lead.pk).exists()) self.assertContains(self.client.get(reverse("school_admin")), "Admin Recovery School") restore_response = self.client.post(reverse("school_admin"), { "action": "restore", "school_id": school.pk, }) self.assertEqual(restore_response.status_code, 302) self.assertTrue(School.objects.filter(pk=school.pk).exists()) self.assertTrue(Lead.objects.filter(pk=lead.pk).exists()) def test_backup_round_trip_restores_school_and_crm(self): user = get_user_model().objects.create_user(username="backup-owner", password="test") school = School.objects.create( organization_id=987654, name="Backup School", data={"electricity": True, "unmapped_detail": "preserved"}, ) lead = Lead.objects.create( school=school, stage="contacted", priority="high", owner=user, attempts=2, next_action="Call again", ) Activity.objects.create( lead=lead, kind="call", outcome="connected", notes="Discussed enrollment", created_by=user, ) Contact.objects.create( school=school, name="A. Teacher", role="Head teacher", phone="12345", email="teacher@example.com", is_primary=True, ) backup = build_backup() School.objects.create(organization_id=987655, name="Keep This School", data={}) school.data = {"changed": True} school.save(update_fields=["data"]) Lead.objects.filter(school=school).delete() Contact.objects.filter(school=school).delete() counts = restore_backup(backup) school.refresh_from_db() restored_lead = Lead.objects.get(school=school) activity = Activity.objects.get(lead=restored_lead) contact = Contact.objects.get(school=school) self.assertEqual(school.data, {"electricity": True, "unmapped_detail": "preserved"}) self.assertEqual(restored_lead.stage, "contacted") self.assertEqual(restored_lead.owner, user) self.assertEqual(activity.notes, "Discussed enrollment") self.assertEqual(activity.created_by, user) self.assertEqual(contact.name, "A. Teacher") self.assertTrue(contact.is_primary) self.assertTrue(School.objects.filter(organization_id=987655).exists()) self.assertEqual(counts, {"schools": 1, "leads": 1, "activities": 1, "contacts": 1})