import uuid from unittest.mock import patch from django.contrib.auth.models import Permission from django.test import Client from django.urls import reverse from apps.parts import services as parts from . import services as s from .models import InventoryLocation, SerializedStockUnit, StockMovement from .tests import InventoryFixture, grant class InventoryAdminTests(InventoryFixture): def setUp(self): self.client.force_login(self.actor) self.location_url = reverse("admin:inventory_inventorylocation_change", args=[self.location.pk]) self.movement_url = reverse("admin:inventory_stockmovement_add") def movement_data(self): page = self.client.get(self.movement_url) self.assertEqual(page.status_code, 200) return dict(source="", destination=str(self.location.pk), spare_part=str(self.part.pk), quantity=1, reference="ADMIN-RECEIPT", note="", idempotency_key=str(uuid.uuid4()), location_revisions=page.context["adminform"].form.initial["location_revisions"], units=[], _save="Save") def location_data(self): page = self.client.get(self.location_url) return dict(name="New name", revision=page.context["adminform"].form.initial["revision"], _save="Save") def test_add_location_service_delegation(self): with patch.object(s, "create_location", wraps=s.create_location) as called: response = self.client.post(reverse("admin:inventory_inventorylocation_add"), dict( company=self.company.pk, service_center="", code=" new ", name="New", location_type="WAREHOUSE", _save="Save")) self.assertEqual(response.status_code, 302) called.assert_called_once() self.assertTrue(InventoryLocation.objects.filter(code="NEW").exists()) def test_edit_location_uses_service_and_signed_revision(self): with patch.object(s, "update_location", wraps=s.update_location) as called: response = self.client.post(self.location_url, self.location_data()) self.assertEqual(response.status_code, 302) self.assertEqual(called.call_args.kwargs["expected_revision"], self.location.updated_at.isoformat()) def test_location_identity_and_lifecycle_fields_protected(self): data = self.location_data() | dict(company=str(self.other_company.pk), code="REPLACED", location_type="DEFECTIVE", is_active="") self.assertEqual(self.client.post(self.location_url, data).status_code, 302) self.location.refresh_from_db() self.assertEqual((self.location.company_id, self.location.code, self.location.location_type, self.location.is_active), (self.company.pk, "MAIN", "WAREHOUSE", True)) def test_stale_location_form_rejected(self): data = self.location_data() s.deactivate_location(actor=self.actor, location=self.location) self.assertEqual(self.client.post(self.location_url, data).status_code, 200) self.location.refresh_from_db() self.assertEqual(self.location.name, "Main") def test_tampered_location_form_rejected(self): self.assertEqual(self.client.post(self.location_url, self.location_data() | {"revision": "forged"}).status_code, 200) def test_stock_posting_service_delegation(self): data = self.movement_data() with patch.object(s, "receive_stock", wraps=s.receive_stock) as called: response = self.client.post(self.movement_url, data) self.assertEqual(response.status_code, 302) called.assert_called_once() self.assertEqual(StockMovement.objects.count(), 1) def test_stale_movement_location_revision_rejected(self): data = self.movement_data() s.deactivate_location(actor=self.actor, location=self.location) self.assertEqual(self.client.post(self.movement_url, data).status_code, 200) self.assertEqual(StockMovement.objects.count(), 0) def test_unit_registration_uses_service(self): with patch.object(s, "register_serialized_unit", wraps=s.register_serialized_unit) as called: response = self.client.post(reverse("admin:inventory_serializedstockunit_add"), dict(company=str(self.company.pk), spare_part=str(self.serial_part.pk), identifier=" ADMIN-SERIAL ", _save="Save")) self.assertEqual(response.status_code, 302) called.assert_called_once() self.assertEqual(SerializedStockUnit.objects.get().identifier, "ADMIN-SERIAL") def test_ledger_readonly_and_delete_forbidden(self): movement = self.receive() change = reverse("admin:inventory_stockmovement_change", args=[movement.pk]) self.assertEqual(self.client.get(change).status_code, 200) self.assertEqual(self.client.post(change, {"quantity": 999}).status_code, 403) self.assertEqual(self.client.post(reverse("admin:inventory_stockmovement_delete", args=[movement.pk]), {"post": "yes"}).status_code, 403) def test_csrf_enforced(self): client = Client(enforce_csrf_checks=True) client.force_login(self.actor) self.assertEqual(client.post(self.movement_url, self.movement_data()).status_code, 403) self.assertEqual(client.post(self.location_url, self.location_data()).status_code, 403) def test_native_permissions_do_not_grant_business_scope(self): self.staff.user_permissions.set(Permission.objects.filter(content_type__app_label="inventory")) self.client.force_login(self.staff) response = self.client.get(reverse("admin:inventory_inventorylocation_changelist")) self.assertEqual(response.context["cl"].result_count, 0) def test_company_scope_blocks_other_company_idor(self): self.staff.user_permissions.set(Permission.objects.filter(content_type__app_label="inventory")) grant(self.staff, self.company) self.client.force_login(self.staff) response = self.client.get(reverse("admin:inventory_inventorylocation_changelist")) self.assertEqual(response.context["cl"].result_count, 2) self.assertEqual(self.client.get(reverse("admin:inventory_inventorylocation_change", args=[self.foreign.pk])).status_code, 302) def test_lifecycle_actions_delegate(self): url = reverse("admin:inventory_inventorylocation_changelist") data = {"action": "deactivate_selected", "_selected_action": [str(self.location.pk)]} preview = self.client.post(url, data) self.assertEqual(preview.status_code, 200) with patch.object(s, "deactivate_location", wraps=s.deactivate_location) as called: response = self.client.post(url, data | {"confirm_transition": "1", "transition_revision": preview.context_data["transition_revision"]}) self.assertEqual(response.status_code, 302) called.assert_called_once() def test_existing_part_admin_cannot_change_locked_policy(self): self.receive() url = reverse("admin:parts_sparepart_change", args=[self.part.pk]) token = self.client.get(url).context["adminform"].form.initial["revision"] response = self.client.post(url, dict(name="Part", description="", category=str(self.category.pk), manufacturer_part_number="", serialization_policy="REQUIRED_SERIAL", product_models=[], product_variants=[], revision=token, _save="Save"), follow=True) self.assertContains(response, "Save rejected") self.part.refresh_from_db() self.assertEqual(self.part.serialization_policy, "NOT_SERIALIZED")