"""Native Admin gates, service delegation, confirmations and signed stale forms.""" from unittest.mock import patch from django.contrib.admin.models import LogEntry from django.contrib.auth.models import Permission from django.test import Client, TestCase from django.urls import reverse from apps.access import services as access from apps.customers import services as customers from apps.devices import services as devices from apps.organization import assignment_services as postings from . import handover_services as services from .handover_admin import HandoverForm from .models import ServiceCase, ServiceCaseDeliveryRelease, ServiceCaseHandover, ServiceCaseClosure from .test_handover import setup_handover, released, handover, delivered, close, invariants def staff(test): test.actor.is_staff = True test.actor.save() test.actor.user_permissions.add(Permission.objects.get(content_type__app_label="service", codename="change_servicecase")) test.url = reverse("admin:service_servicecase_handover", args=[test.case.pk]) def form_data(test, operation, token): data = dict(operation=operation, revision=token, device_confirmed="on", verification_confirmed="on", physical_handover_confirmed="on", recipient_type="CUSTOMER", verification_method="CUSTOMER_RECORD") data.update({f"returned_{a.pk.hex}": a.quantity for a in test.received}) return data class HandoverAdminTests(TestCase): @classmethod def setUpTestData(cls): setup_handover(cls) staff(cls) def setUp(self): self.client.force_login(self.actor) def tearDown(self): 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(form_data(self, operation, self.token()), **kwargs)) def test_three_explicit_steps_delegate_to_services(self): for operation, name in (("ready", "mark_service_case_ready_for_delivery"), ("handover", "handover_service_case"), ("close", "close_service_case")): with patch.object(services, name, wraps=getattr(services, name)) as spy: response = self.post(operation) self.assertEqual(response.status_code, 302, getattr(response, "context", None)) self.assertEqual(spy.call_args.kwargs["actor"], self.actor) self.assertEqual(ServiceCase.objects.get(pk=self.case.pk).status, "CLOSED") self.assertEqual(LogEntry.objects.count(), 3) self.assertFalse(LogEntry.objects.filter(change_message__contains=self.customer.display_name).exists()) def test_get_is_read_only(self): self.assertEqual(self.client.get(self.url).status_code, 200) self.assertFalse(ServiceCaseDeliveryRelease.objects.exists()) def test_csrf_required(self): client = Client(enforce_csrf_checks=True) client.force_login(self.actor) self.assertEqual(client.post(self.url, form_data(self, "ready", self.token())).status_code, 403) def test_native_permission_required_separately(self): self.actor2.is_staff = True self.actor2.save() self.client.force_login(self.actor2) self.assertEqual(self.client.get(self.url).status_code, 403) def test_view_only_cannot_mutate(self): self.actor.user_permissions.clear() self.actor.user_permissions.add(Permission.objects.get(content_type__app_label="service", codename="view_servicecase")) self.assertEqual(self.client.get(self.url).status_code, 200) self.assertEqual(self.client.post(self.url, {}).status_code, 403) def test_native_staff_permission_without_business_scope_denied(self): access.deactivate_role_assignment(assignment=self.delivery_role_assignment) self.assertEqual(self.client.get(self.url).status_code, 403) self.assertEqual(self.client.post(self.url, {}).status_code, 403) def test_tampered_revision_rejected(self): response = self.post("ready", revision=self.token() + "tampered") self.assertContains(response, "Reload the handover form") def test_explicit_physical_device_confirmation_required(self): response = self.post("ready", device_confirmed="") self.assertContains(response, "Explicit confirmation is required") self.assertFalse(ServiceCaseDeliveryRelease.objects.exists()) def test_recipient_and_physical_acceptance_confirmation_required(self): released(self) for field in ("verification_confirmed", "physical_handover_confirmed", "recipient_type", "verification_method", "device_confirmed"): with self.subTest(field=field): self.assertContains(self.post("handover", **{field: ""}), "Explicit confirmation is required") self.assertFalse(ServiceCaseHandover.objects.exists()) def test_accessories_not_silently_returned(self): released(self) form = self.client.get(self.url).context["form"] field = f"returned_{self.received[0].pk.hex}" self.assertIsNone(form.fields[field].initial) response = self.post("handover", **{field: ""}) self.assertContains(response, "This field is required") self.assertFalse(ServiceCaseHandover.objects.exists()) def test_stale_ready_form_after_release_rejected(self): token = self.token() released(self) self.assertContains(self.client.post(self.url, form_data(self, "ready", token)), "changed; reload") def test_stale_handover_after_other_collector_rejected(self): released(self) token = self.token() handover(self) self.assertContains(self.client.post(self.url, form_data(self, "handover", token)), "changed; reload") self.assertEqual(ServiceCaseHandover.objects.count(), 1) def test_stale_handover_after_customer_edit_rejected(self): released(self) token = self.token() customers.update_customer(customer=self.customer, full_name="Changed synthetic customer") self.assertContains(self.client.post(self.url, form_data(self, "handover", token)), "changed; reload") def test_stale_handover_after_identifier_change_rejected(self): released(self) token = self.token() devices.replace_device_identifier(device=self.device, identifier_type="SERIAL", new_value="SYNTHETIC-NEW-SERIAL") self.assertContains(self.client.post(self.url, form_data(self, "handover", token)), "changed; reload") def test_revoked_scope_rejects_stale_form(self): released(self) token = self.token() access.deactivate_role_assignment(assignment=self.delivery_role_assignment) self.assertEqual(self.client.post(self.url, form_data(self, "handover", token)).status_code, 403) def test_stale_closure_form_after_close_denied(self): delivered(self) token = self.token() close(self) self.assertEqual(self.client.post(self.url, form_data(self, "close", token)).status_code, 403) self.assertEqual(ServiceCaseClosure.objects.count(), 1) def test_customer_change_between_form_validation_and_service_rejected(self): released(self) original = HandoverForm.clean def intervening(form): data = original(form) customers.update_customer(customer=self.customer, full_name="Changed synthetic customer") return data with patch.object(HandoverForm, "clean", intervening): response = self.post("handover") self.assertContains(response, "changed; reload") self.assertFalse(ServiceCaseHandover.objects.exists()) def test_protected_actor_timestamp_and_case_cannot_be_posted(self): released(self) self.assertEqual(self.post("handover", handed_over_by=self.user.pk, delivered_at="2000-01-01", service_case="forged").status_code, 302) row = ServiceCaseHandover.objects.get() self.assertEqual(row.handed_over_by, self.actor) self.assertGreater(row.delivered_at.year, 2000) self.assertEqual(row.service_case_id, self.case.pk) def test_history_readonly_add_and_delete_disabled(self): row = delivered(self) closure = close(self) self.actor.is_superuser = True self.actor.save() for obj in (row.delivery_release, row, row.accessories.get(), closure): name = obj._meta.model_name url = reverse(f"admin:service_{name}_change", args=[obj.pk]) self.assertEqual(self.client.get(url).status_code, 200) self.assertEqual(self.client.post(url, {"note": "forged"}).status_code, 403) self.assertEqual(self.client.get(reverse(f"admin:service_{name}_add")).status_code, 403) self.assertEqual(self.client.post(reverse(f"admin:service_{name}_delete", args=[obj.pk])).status_code, 403) def test_cross_company_history_is_not_visible(self): delivered(self) access.deactivate_role_assignment(assignment=self.delivery_role_assignment) path = postings.create_assignment(user=self.actor, company=self.other_company) access.create_role_assignment(user=self.actor, role=self.delivery_role, organization_assignment=path) self.actor.user_permissions.add(Permission.objects.get(content_type__app_label="service", codename="view_servicecasehandover")) self.assertEqual(self.client.get(self.url).status_code, 403) response = self.client.get(reverse("admin:service_servicecasehandover_changelist")) self.assertEqual(response.status_code, 200) self.assertEqual(response.context["cl"].queryset.count(), 0) def test_superuser_uses_established_bypass(self): self.user.is_staff = self.user.is_superuser = True self.user.save() self.client.force_login(self.user) self.assertEqual(self.post("ready").status_code, 302)