import uuid from unittest.mock import patch from django.contrib.auth import get_user_model from django.contrib.auth.models import Permission from django.core.exceptions import PermissionDenied, ValidationError from django.db import IntegrityError, transaction from django.test import TestCase from apps.access import services as access from apps.parts import services as parts from apps.service import test_repair as repair_fixture from apps.service import services as case_services from . import request_services as s, services as stock, queries as q from .models import PartsRequest, PartsRequestLine, PartsRequestEvent, StockReservation, StockMovement def setup_requests(test): repair_fixture.setup_repair(test) test.actor = get_user_model().objects.create_superuser(username="parts-desk") access.set_role_permissions(role=test.role, permissions=[test.permission, Permission.objects.get(content_type__app_label="inventory", codename="request_parts")]) test.part_category = parts.create_part_category(code="JOB-PARTS", name="Job parts") test.part = parts.create_spare_part(category=test.part_category, part_code="JOB-PART", name="Replacement", serialization_policy="NOT_SERIALIZED") parts.set_spare_part_compatibility(spare_part=test.part, product_models=[test.model], product_variants=[]) test.location = stock.create_location(actor=test.actor, company=test.company, service_center=test.center, code="PARTS", name="Parts", location_type="STORE") test.destination = stock.create_location(actor=test.actor, company=test.company, code="DEST", name="Destination", location_type="WAREHOUSE") class RequestFixture: def request(self, **kwargs): return s.create_parts_request(**dict(actor=self.engineer, service_case=self.case, lines=[dict(spare_part=self.part, quantity=2)]) | kwargs) def approve(self, request): request.refresh_from_db() return s.approve_parts_request(actor=self.actor, request=request, expected_revision=request.updated_at.isoformat()) def reserve(self, request, **kwargs): request.refresh_from_db() return s.reserve_parts(**dict(actor=self.actor, line=request.lines.get(), location=self.location, quantity=1, expected_request_revision=request.updated_at.isoformat()) | kwargs) def receive(self, **kwargs): return stock.receive_stock(**dict(actor=self.actor, destination=self.location, spare_part=self.part, quantity=2, reference="REQUEST-TEST", idempotency_key=uuid.uuid4()) | kwargs) def move(self, **kwargs): return stock.move_stock(**dict(actor=self.actor, source=self.location, destination=self.destination, spare_part=self.part, quantity=1, reference="REQUEST-TEST", idempotency_key=uuid.uuid4()) | kwargs) def release(self, row): return s.release_reservation(actor=self.actor, reservation=row, reason="No longer required", expected_revision=row.updated_at.isoformat()) class RequestTests(RequestFixture, TestCase): @classmethod def setUpTestData(cls): setup_requests(cls) def test_request_preserves_demand_without_stock(self): row = self.request(note=" Synthetic job demand ") self.assertEqual(row.requester, self.engineer) self.assertEqual(row.note, "Synthetic job demand") self.assertEqual(row.lines.get().quantity, 2) self.assertEqual(row.events.get().status, "REQUESTED") self.assertFalse(StockMovement.objects.exists()) def test_current_engineer_and_business_permission_both_required(self): with self.assertRaises(ValidationError): self.request(actor=self.actor) with self.assertRaises(ValidationError): self.request(actor=self.engineer2) access.set_role_permissions(role=self.role, permissions=[self.permission]) with self.assertRaises(PermissionDenied): self.request() def test_incompatible_part_rejected(self): parts.set_spare_part_compatibility(spare_part=self.part, product_models=[], product_variants=[]) with self.assertRaises(ValidationError): self.request() def test_inactive_part_rejected(self): parts.deactivate_spare_part(spare_part=self.part) with self.assertRaises(ValidationError): self.request() def test_invalid_quantities_and_duplicate_lines_rejected(self): for quantity in (0, -1, True, 1.5, 1000000001): with self.subTest(quantity=quantity), self.assertRaises(ValidationError): self.request(lines=[dict(spare_part=self.part, quantity=quantity)]) with self.assertRaises(ValidationError): self.request(lines=[dict(spare_part=self.part, quantity=1)] * 2) def test_request_evidence_cannot_be_rewritten(self): row = self.request() for queryset, changes in [(PartsRequest.objects.filter(pk=row.pk), dict(note="Changed")), (PartsRequestLine.objects.filter(request=row), dict(quantity=3)), (PartsRequestEvent.objects.filter(request=row), dict(reason="Changed"))]: with self.assertRaises(IntegrityError), transaction.atomic(): queryset.update(**changes) def test_reserve_requires_approval(self): self.receive() with self.assertRaises(ValidationError): self.reserve(self.request()) def test_reservation_changes_availability_not_on_hand(self): self.receive() row = self.reserve(self.approve(self.request())) self.assertEqual(q.stock_on_hand(actor=self.actor, location=self.location, spare_part=self.part), 2) self.assertEqual(q.available_stock(actor=self.actor, location=self.location, spare_part=self.part), 1) self.assertEqual(StockMovement.objects.count(), 1) self.assertEqual(row.service_case, self.case) def test_release_retains_history_and_restores_available(self): self.receive() row = self.release(self.reserve(self.approve(self.request()))) self.assertEqual(row.status, "RELEASED") self.assertEqual(q.available_stock(actor=self.actor, location=self.location, spare_part=self.part), 2) with self.assertRaises(ValidationError): self.release(row) def test_no_over_reservation(self): self.receive(quantity=1) request = self.approve(self.request()) self.reserve(request) with self.assertRaises(ValidationError): self.reserve(request) def test_no_over_request_reservation(self): self.receive(quantity=5) request = self.approve(self.request()) self.reserve(request, quantity=2) with self.assertRaises(ValidationError): self.reserve(request) def test_transfers_cannot_spend_reserved_stock(self): self.receive() self.reserve(self.approve(self.request()), quantity=2) with self.assertRaises(ValidationError): self.move() def test_unreserved_remainder_can_move(self): self.receive() self.reserve(self.approve(self.request())) self.move() self.assertEqual(q.available_stock(actor=self.actor, location=self.location, spare_part=self.part), 0) def test_request_stale_revision_rejected(self): request = self.request() revision = request.updated_at.isoformat() self.approve(request) with self.assertRaises(ValidationError): s.approve_parts_request(actor=self.actor, request=request, expected_revision=revision) def test_reject_records_reason_and_terminal_state(self): row = self.request() row = s.reject_parts_request(actor=self.actor, request=row, reason="Not approved", expected_revision=row.updated_at.isoformat()) self.assertEqual(row.status, "REJECTED") self.assertEqual(row.events.count(), 2) with self.assertRaises(ValidationError): self.approve(row) def test_cancellation_requires_explicit_release(self): self.receive() request = self.approve(self.request()) row = self.reserve(request) request.refresh_from_db() with self.assertRaises(ValidationError): s.cancel_parts_request(actor=self.actor, request=request, reason="Cancel", expected_revision=request.updated_at.isoformat()) self.release(row) request.refresh_from_db() s.cancel_parts_request(actor=self.actor, request=request, reason="Cancel", expected_revision=request.updated_at.isoformat()) self.assertEqual(request.events.count(), 3) def test_case_cancellation_prevents_new_reservation_allows_release(self): self.receive() request = self.approve(self.request()) row = self.reserve(request) case_services.cancel_service_case(service_case=self.case, reason="Customer cancelled", cancelled_by=self.user) with self.assertRaises(ValidationError): self.reserve(request) self.release(row) def test_failed_reservation_rolls_back(self): self.receive() request = self.approve(self.request()) with patch.object(StockReservation, "_persist", side_effect=ValidationError("abort")), self.assertRaises(ValidationError): self.reserve(request) self.assertFalse(StockReservation.objects.exists()) self.assertEqual(q.available_stock(actor=self.actor, location=self.location, spare_part=self.part), 2) def test_variant_only_mapping_excluded_for_model_only_device(self): from apps.catalog.models import ProductVariant variant = ProductVariant.objects.create(product_model=self.model, code="ONLY", name="Only variant") parts.set_spare_part_compatibility(spare_part=self.part, product_models=[], product_variants=[variant]) with self.assertRaises(ValidationError): self.request() def test_compatibility_rechecked_before_approval_and_reservation(self): row = self.request() parts.set_spare_part_compatibility(spare_part=self.part, product_models=[], product_variants=[]) with self.assertRaises(ValidationError): self.approve(row) parts.set_spare_part_compatibility(spare_part=self.part, product_models=[self.model], product_variants=[]) row = self.approve(row) self.receive() parts.set_spare_part_compatibility(spare_part=self.part, product_models=[], product_variants=[]) with self.assertRaises(ValidationError): self.reserve(row) def test_cross_company_reservation_rejected(self): location = stock.create_location(actor=self.actor, company=self.other_company, code="OTHER", name="Other", location_type="WAREHOUSE") self.receive(destination=location) with self.assertRaises(ValidationError): self.reserve(self.approve(self.request()), location=location) def test_native_permissions_do_not_create_business_scope(self): self.user.is_staff = True self.user.save() self.user.user_permissions.set(Permission.objects.filter(content_type__app_label="inventory")) row = self.request() with self.assertRaises(PermissionDenied): s.approve_parts_request(actor=self.user, request=row, expected_revision=row.updated_at.isoformat()) def test_request_queries_are_scoped_and_prefetched(self): from . import request_queries as queries row = self.request() self.assertFalse(queries.parts_requests(actor=self.user).exists()) with self.assertNumQueries(6): detail = queries.parts_request_detail(actor=self.actor, request=row) for line in detail.lines.all(): str(line.spare_part) list(line.reservations.all()) for event in detail.events.all(): str(event.actor) self.assertEqual(list(queries.request_fulfillment(actor=self.actor, request=row))[0]["reserved_quantity"], 0) def test_serialized_reservation_requires_explicit_units(self): parts.update_spare_part(spare_part=self.part, serialization_policy="REQUIRED_SERIAL") unit = stock.register_serialized_unit(actor=self.actor, company=self.company, spare_part=self.part, identifier="JOB-SERIAL") self.receive(quantity=1, units=[unit]) row = self.approve(self.request()) with self.assertRaises(ValidationError): self.reserve(row) reservation = self.reserve(row, units=[unit]) self.assertEqual(reservation.units.get().unit, unit) self.release(reservation) self.move(units=[unit]) def test_optional_serial_buckets_cannot_be_oversubscribed(self): parts.update_spare_part(spare_part=self.part, serialization_policy="OPTIONAL_SERIAL") unit = stock.register_serialized_unit(actor=self.actor, company=self.company, spare_part=self.part, identifier="OPTIONAL-RESERVE") self.receive(quantity=2, units=[unit]) row = self.approve(self.request()) self.reserve(row) with self.assertRaises(ValidationError): self.reserve(row) self.reserve(row, units=[unit])