from unittest.mock import patch from django.contrib.auth import get_user_model from django.contrib.auth.models import Permission from django.core import signing from django.core.exceptions import ValidationError from django.test import Client from django.urls import reverse from . import services as s from .admin import REVISION_SALT from .models import PartCategory, SparePart, SparePartCompatibility as Mapping from .tests import PartsFixture class PartsAdminTests(PartsFixture): @classmethod def setUpTestData(cls): super().setUpTestData() cls.admin_user = get_user_model().objects.create_superuser(username="parts-admin", password="test-only") cls.staff = get_user_model().objects.create_user(username="parts-staff", is_staff=True) def setUp(self): self.client.force_login(self.admin_user) self.part_url = reverse("admin:parts_sparepart_change", args=[self.part.pk]) self.category_url = reverse("admin:parts_partcategory_change", args=[self.category.pk]) def part_data(self): response = self.client.get(self.part_url) self.assertEqual(response.status_code, 200) return dict(name="Edited", description="Notes", category=str(self.category.pk), manufacturer_part_number="MPN", serialization_policy="REQUIRED_SERIAL", product_models=[str(self.model.pk)], product_variants=[], revision=response.context["adminform"].form.initial["revision"], _save="Save") def category_data(self): response = self.client.get(self.category_url) return dict(name="Edited category", description="Notes", revision=response.context["adminform"].form.initial["revision"], _save="Save") def test_add_category_uses_service_and_normalizes(self): with patch.object(s, "create_part_category", wraps=s.create_part_category) as called: response = self.client.post(reverse("admin:parts_partcategory_add"), {"code": " battery ", "name": " Battery ", "description": "", "_save": "Save"}) self.assertEqual(response.status_code, 302) called.assert_called_once() self.assertEqual(PartCategory.objects.get(code="BATTERY").name, "Battery") def test_add_part_uses_atomic_service(self): data = self.part_data() data.update(part_code=" new-01 ", revision="") with patch.object(s, "create_spare_part", wraps=s.create_spare_part) as called: response = self.client.post(reverse("admin:parts_sparepart_add"), data) self.assertEqual(response.status_code, 302) called.assert_called_once() self.assertTrue(SparePart.objects.get(part_code="NEW-01").compatibilities.get().is_active) def test_edit_part_uses_service_with_signed_precondition(self): data = self.part_data() old_revision = s.revision(self.part) with patch.object(s, "update_spare_part", wraps=s.update_spare_part) as called: response = self.client.post(self.part_url, data) self.assertEqual(response.status_code, 302) self.assertEqual(called.call_args.kwargs["expected_revision"], old_revision) self.part.refresh_from_db() self.assertEqual(self.part.name, "Edited") self.assertEqual(Mapping.objects.get().product_model, self.model) def test_category_edit_uses_service(self): with patch.object(s, "update_part_category", wraps=s.update_part_category) as called: response = self.client.post(self.category_url, self.category_data()) self.assertEqual(response.status_code, 302) called.assert_called_once() def test_protected_fields_ignore_forged_post(self): data = self.part_data() data.update(part_code="REWRITTEN", is_active="", id="00000000-0000-0000-0000-000000000000", created_at="2000-01-01", updated_at="2000-01-01") original_created = self.part.created_at self.assertEqual(self.client.post(self.part_url, data).status_code, 302) self.part.refresh_from_db() self.assertEqual(self.part.part_code, "DISPLAY-01") self.assertTrue(self.part.is_active) self.assertEqual(self.part.created_at, original_created) def test_invalid_or_missing_revision_rejected(self): data = self.part_data() for token in ("", "forged"): data["revision"] = token response = self.client.post(self.part_url, data) self.assertEqual(response.status_code, 200) self.assertTrue(response.context["adminform"].form.errors) self.part.refresh_from_db() self.assertEqual(self.part.name, "Display assembly") def test_revision_from_different_record_rejected(self): data = self.part_data() data["revision"] = signing.dumps(["parts.sparepart", str(self.category.pk), s.revision(self.part)], salt=REVISION_SALT) self.assertEqual(self.client.post(self.part_url, data).status_code, 200) self.assertEqual(Mapping.objects.count(), 0) def test_stale_form_after_lifecycle_rejected(self): data = self.part_data() s.deactivate_spare_part(spare_part=self.part) self.assertEqual(self.client.post(self.part_url, data).status_code, 200) self.part.refresh_from_db() self.assertFalse(self.part.is_active) self.assertEqual(self.part.name, "Display assembly") def test_stale_category_form_rejected(self): data = self.category_data() s.deactivate_part_category(part_category=self.category) self.assertEqual(self.client.post(self.category_url, data).status_code, 200) self.category.refresh_from_db() self.assertEqual(self.category.name, "Display") self.assertFalse(self.category.is_active) def test_stale_form_after_compatibility_edit_rejected(self): data = self.part_data() self.configure(variants=[self.variant]) self.assertEqual(self.client.post(self.part_url, data).status_code, 200) self.assertIsNone(Mapping.objects.get().product_model) def test_race_after_form_validation_is_rejected_by_locked_service(self): data = self.part_data() original = s.update_spare_part def competing_edit(**kwargs): s.deactivate_spare_part(spare_part=self.part) return original(**kwargs) with patch.object(s, "update_spare_part", side_effect=competing_edit): response = self.client.post(self.part_url, data, follow=True) self.assertContains(response, "Save rejected") self.part.refresh_from_db() self.assertEqual(self.part.name, "Display assembly") self.assertEqual(Mapping.objects.count(), 0) def test_contradictory_admin_configuration_rejected(self): data = self.part_data() data["product_variants"] = [str(self.variant.pk)] response = self.client.post(self.part_url, data) self.assertEqual(response.status_code, 200) self.assertTrue(response.context["adminform"].form.errors) self.assertEqual(Mapping.objects.count(), 0) def test_atomic_admin_rollback_on_compatibility_failure(self): data = self.part_data() with patch.object(Mapping, "_persist", side_effect=ValidationError("Failure")): response = self.client.post(self.part_url, data, follow=True) self.assertContains(response, "Save rejected") self.part.refresh_from_db() self.assertEqual(self.part.name, "Display assembly") def test_lifecycle_actions_use_services(self): self.configure(models=[self.model]) for model, obj, argument, deactivation, reactivation in ( ("sparepart", self.part, "spare_part", "deactivate_spare_part", "reactivate_spare_part"), ("partcategory", self.category, "part_category", "deactivate_part_category", "reactivate_part_category")): url = reverse(f"admin:parts_{model}_changelist") for action, operation, active in (("deactivate_selected", deactivation, False), ("reactivate_selected", reactivation, True)): with patch.object(s, operation, wraps=getattr(s, operation)) as called: response = self.client.post(url, {"action": action, "_selected_action": [str(obj.pk)]}) self.assertEqual(response.status_code, 302) self.assertIn(argument, called.call_args.kwargs) obj.refresh_from_db() self.assertEqual(obj.is_active, active) self.assertEqual(Mapping.objects.count(), 1) def test_deletion_disabled_and_mapping_readonly(self): self.configure(models=[self.model]) for model, obj in (("sparepart", self.part), ("partcategory", self.category), ("sparepartcompatibility", Mapping.objects.get())): self.assertEqual(self.client.post(reverse(f"admin:parts_{model}_delete", args=[obj.pk]), {"post": "yes"}).status_code, 403) self.assertEqual(self.client.post(reverse("admin:parts_sparepartcompatibility_add"), {}).status_code, 403) self.assertEqual(self.client.post(reverse("admin:parts_sparepartcompatibility_change", args=[Mapping.objects.get().pk]), {}).status_code, 403) def test_csrf_rejects_mutations_and_actions(self): client = Client(enforce_csrf_checks=True) client.force_login(self.admin_user) for url, data in ((self.part_url, self.part_data()), (reverse("admin:parts_sparepart_changelist"), {"action": "deactivate_selected", "_selected_action": [str(self.part.pk)]})): self.assertEqual(client.post(url, data).status_code, 403) def test_staff_requires_native_permissions(self): self.client.force_login(self.staff) self.assertEqual(self.client.get(self.part_url).status_code, 403) self.staff.user_permissions.add(Permission.objects.get(content_type__app_label="parts", codename="view_sparepart")) self.assertEqual(self.client.get(self.part_url).status_code, 200) self.assertEqual(self.client.post(self.part_url, {"name": "Denied"}).status_code, 403) def test_native_change_permission_allows_without_business_scope(self): self.staff.user_permissions.add(Permission.objects.get(content_type__app_label="parts", codename="change_sparepart")) self.client.force_login(self.staff) data = self.part_data() self.assertEqual(self.client.post(self.part_url, data).status_code, 302) self.part.refresh_from_db() self.assertEqual(self.part.name, "Edited") def test_xss_output_escaped(self): s.update_spare_part(spare_part=self.part, name="") response = self.client.get(reverse("admin:parts_sparepart_changelist")) self.assertContains(response, "<script>alert(1)</script>") self.assertNotContains(response, "")