"""Native Admin permission, CSRF, optimistic revision and history protection.""" from unittest.mock import patch from django.contrib.auth.models import Permission from django.test import Client, TestCase from django.urls import reverse from apps.service_catalog import services as taxonomy from . import repair_services as services from .models import ServiceRepairExecution, ServiceRepairAction from .repair_admin import RepairForm from .test_repair import setup_repair, begin, prepared, perform, complete, abandon, assert_repair_invariants class RepairAdminTests(TestCase): def setUp(self): setup_repair(self) self.engineer.is_staff = True self.engineer.save() self.engineer.user_permissions.add(Permission.objects.get(content_type__app_label="service", codename="change_servicecase")) self.client.force_login(self.engineer) self.url = reverse("admin:service_servicecase_repair", args=[self.case.pk]) def tearDown(self): assert_repair_invariants(self) def token(self): return self.client.get(self.url).context["form"].initial["revision"] def post(self, operation, **kwargs): return self.client.post(self.url, dict(operation=operation, revision=self.token(), **kwargs)) def test_full_workflow_delegates_to_services(self): with patch.object(services, "begin_service_case_repair", wraps=services.begin_service_case_repair) as begin_spy: self.assertEqual(self.post("begin").status_code, 302) self.assertEqual(begin_spy.call_args.kwargs["actor"], self.engineer) self.assertEqual(self.post("add", repair_action=self.action_type.pk).status_code, 302) action = ServiceRepairAction.objects.get() self.assertEqual(self.post("update", action=action.pk, repair_action=self.action_type2.pk, note="Synthetic plan").status_code, 302) self.assertEqual(self.post("note", execution_note="Synthetic repair").status_code, 302) self.assertEqual(self.post("perform", action=action.pk).status_code, 302) self.assertEqual(self.post("complete", outcome="REPAIRED", execution_note="Synthetic result").status_code, 302) self.assertEqual(ServiceRepairExecution.objects.get().completed_by, self.engineer) self.assertEqual(ServiceRepairAction.objects.get().performed_by, self.engineer) def test_remove_and_abandon(self): execution, action = prepared(self) self.assertEqual(self.post("remove", action=action.pk).status_code, 302) self.assertEqual(self.post("abandon", reason="Synthetic recovery").status_code, 302) execution.refresh_from_db() action.refresh_from_db() self.assertEqual(execution.status, "ABANDONED") self.assertFalse(action.is_active) self.assertIsNotNone(action.performed_at) def test_unsuccessful_outcome_then_begin(self): prepared(self) self.assertEqual(self.post("complete", outcome="NOT_REPAIRED").status_code, 302) self.assertEqual(self.post("begin").status_code, 302) self.assertEqual(ServiceRepairExecution.objects.count(), 2) def test_superuser_cannot_impersonate(self): self.user.is_staff = self.user.is_superuser = True self.user.save() self.client.force_login(self.user) self.assertContains(self.post("begin"), "currently eligible assigned engineer") self.assertFalse(ServiceRepairExecution.objects.exists()) def test_csrf_and_native_permission_gate(self): client = Client(enforce_csrf_checks=True) client.force_login(self.engineer) self.assertEqual(client.post(self.url, {}).status_code, 403) self.engineer2.is_staff = True self.engineer2.save() self.client.force_login(self.engineer2) self.assertEqual(self.client.get(self.url).status_code, 403) self.assertEqual(self.client.post(self.url, {}).status_code, 403) def test_tampered_signature_rejected(self): response = self.client.post(self.url, dict(operation="begin", revision=self.token() + "tampered")) self.assertContains(response, "Reload the repair form") self.assertFalse(ServiceRepairExecution.objects.exists()) def test_stale_action_and_completion_forms(self): execution, action = prepared(self, performed=False) token = self.token() perform(self, action) for operation in ("update", "remove", "perform", "complete"): response = self.client.post(self.url, dict(operation=operation, revision=token, action=action.pk, repair_action=self.action_type.pk, outcome="REPAIRED")) self.assertContains(response, "changed; reload") execution.refresh_from_db() self.assertEqual(execution.status, "OPEN") def test_fresh_choice_cannot_bypass_aggregate_revision(self): _, action = prepared(self, performed=False) token = self.token() original = RepairForm.clean def intervening_write(form): data = original(form) data["action"] = services.update_repair_action(action=action, actor=self.engineer, note="newer") return data with patch.object(RepairForm, "clean", intervening_write): response = self.client.post(self.url, dict(operation="update", revision=token, action=action.pk, repair_action=self.action_type.pk, note="stale overwrite")) self.assertContains(response, "changed; reload") action.refresh_from_db() self.assertEqual(action.note, "newer") def test_completed_case_blocks_stale_submission(self): execution, action = prepared(self) token = self.token() complete(self, execution) response = self.client.post(self.url, dict(operation="remove", revision=token, action=action.pk)) self.assertEqual(response.status_code, 403) action.refresh_from_db() self.assertTrue(action.is_active) def test_choices_and_submission_revalidate_taxonomy(self): begin(self) token = self.token() taxonomy.deactivate_repair_action(repair_action=self.action_type) taxonomy.set_repair_action_applicability(repair_action=self.action_type2, applies_to_all_product_categories=False, product_categories=[]) self.assertFalse(self.client.get(self.url).context["form"].fields["repair_action"].queryset.exists()) response = self.client.post(self.url, dict(operation="add", revision=token, repair_action=self.action_type.pk)) self.assertContains(response, "Select a valid choice") self.assertFalse(ServiceRepairAction.objects.exists()) def test_old_attempt_action_cannot_be_selected(self): execution, action = prepared(self) abandon(self, execution) begin(self) self.assertContains(self.post("remove", action=action.pk), "Select a valid choice") action.refresh_from_db() self.assertTrue(action.is_active) def test_history_readonly_add_and_delete_disabled(self): execution, action = prepared(self) complete(self, execution) self.user.is_staff = self.user.is_superuser = True self.user.save() self.client.force_login(self.user) for model, row in (("servicerepairexecution", execution), ("servicerepairaction", action)): url = reverse(f"admin:service_{model}_change", args=[row.pk]) self.assertEqual(self.client.get(url).status_code, 200) self.assertEqual(self.client.post(url, dict(note="overwrite")).status_code, 403) self.assertEqual(self.client.post(reverse(f"admin:service_{model}_delete", args=[row.pk])).status_code, 403) self.assertEqual(self.client.get(reverse(f"admin:service_{model}_add")).status_code, 403) response = self.client.get(reverse("admin:service_servicecase_change", args=[self.case.pk])) self.assertNotContains(response, 'name="status"') self.assertContains(response, "Repair and technical actions")