import uuid from unittest.mock import patch from django.core.exceptions import PermissionDenied, ValidationError from django.db import IntegrityError, transaction from apps.parts import services as parts from . import document_services as d, document_queries as dq, queries as q, services as s from .models import GoodsReceipt, GoodsReceiptLine, SerializedStockUnit, StockMovement, StockTransfer from .tests import InventoryFixture, grant def revision(record): record.refresh_from_db() return record.updated_at.isoformat() class DocumentFixture(InventoryFixture): def receipt(self, quantity=3, *, part=None, identifiers=(), destination=None): row = d.create_goods_receipt(actor=self.actor, destination=destination or self.location, external_reference="EXTERNAL-DOC") return d.set_goods_receipt_lines(actor=self.actor, receipt=row, expected_revision=revision(row), lines=[dict(spare_part=part or self.part, quantity=quantity, identifiers=list(identifiers))]) def transfer(self, quantity=1, *, part=None, units=(), source=None, destination=None): row = d.create_stock_transfer(actor=self.actor, source=source or self.location, destination=destination or self.destination) return d.set_stock_transfer_lines(actor=self.actor, transfer=row, expected_revision=revision(row), lines=[dict(spare_part=part or self.part, quantity=quantity, units=list(units))]) def post(self, receipt): return d.post_goods_receipt(actor=self.actor, receipt=receipt, expected_revision=revision(receipt)) def dispatch(self, transfer): return d.dispatch_stock_transfer(actor=self.actor, transfer=transfer, expected_revision=revision(transfer)) def receive_transfer(self, transfer): return d.receive_stock_transfer(actor=self.actor, transfer=transfer, expected_revision=revision(transfer)) class GoodsReceiptTests(DocumentFixture): def test_company_scoped_transactional_numbers(self): first, second = self.receipt(), self.receipt() foreign = self.receipt(destination=self.foreign) self.assertEqual((first.number, second.number, foreign.number), ("GRN-00000001", "GRN-00000002", "GRN-00000001")) def test_draft_has_no_stock_effect(self): row = self.receipt() self.assertEqual(row.status, "DRAFT") self.assertEqual(self.balance(), 0) self.assertEqual(StockMovement.objects.count(), 0) def test_draft_identifiers_do_not_create_units_or_lock_policy(self): row = self.receipt(1, part=self.serial_part, identifiers=["DRAFT-SERIAL"]) self.assertEqual(SerializedStockUnit.objects.count(), 0) parts.update_spare_part(spare_part=self.serial_part, serialization_policy="NOT_SERIALIZED") with self.assertRaises(ValidationError): self.post(row) self.assertEqual(SerializedStockUnit.objects.count(), 0) row.refresh_from_db() self.assertEqual(row.status, "DRAFT") def test_draft_and_cancelled_receipt_do_not_freeze_anonymous_policy(self): row = self.receipt() d.cancel_goods_receipt(actor=self.actor, receipt=row, expected_revision=revision(row), reason="Not received") self.assertEqual(parts.update_spare_part(spare_part=self.part, serialization_policy="REQUIRED_SERIAL").serialization_policy, "REQUIRED_SERIAL") def test_post_once_creates_immutable_receipt_evidence(self): row = self.post(self.receipt(4)) self.assertEqual(row.status, "POSTED") self.assertEqual(row.received_by, self.actor) self.assertIsNotNone(row.received_at) self.assertEqual(self.balance(), 4) self.assertEqual(row.lines.get().movement.reference, row.number) with self.assertRaises(ValidationError): self.post(row) def test_post_creates_real_serials_only_at_posting(self): row = self.post(self.receipt(2, part=self.serial_part, identifiers=["SER-A", "SER-B"])) self.assertEqual(SerializedStockUnit.objects.filter(state="IN_STOCK", current_location=self.location).count(), 2) self.assertEqual(row.lines.get().identifiers.filter(unit__isnull=False).count(), 2) with self.assertRaises(ValidationError): parts.update_spare_part(spare_part=self.serial_part, serialization_policy="OPTIONAL_SERIAL") def test_post_can_receive_pre_registered_unit(self): unit = s.register_serialized_unit(actor=self.actor, company=self.company, spare_part=self.serial_part, identifier="PRE-REGISTERED") row = self.post(self.receipt(1, part=self.serial_part, identifiers=[unit.identifier])) self.assertEqual(row.lines.get().identifiers.get().unit_id, unit.pk) self.assertEqual(SerializedStockUnit.objects.count(), 1) def test_duplicate_identifier_in_another_posted_receipt_rejected(self): self.post(self.receipt(1, part=self.serial_part, identifiers=["SAME"])) other = self.receipt(1, part=self.serial_part, identifiers=["SAME"]) with self.assertRaises(ValidationError): self.post(other) self.assertEqual(StockMovement.objects.count(), 1) def test_receipt_wrong_company_identifier_rejected(self): s.register_serialized_unit(actor=self.actor, company=self.other_company, spare_part=self.serial_part, identifier="FOREIGN") with self.assertRaises(ValidationError): self.post(self.receipt(1, part=self.serial_part, identifiers=["FOREIGN"])) def test_empty_receipt_cannot_post(self): row = d.create_goods_receipt(actor=self.actor, destination=self.location) with self.assertRaises(ValidationError): self.post(row) def test_serialized_receipt_requires_all_identifiers_at_posting(self): row = self.receipt(2, part=self.serial_part, identifiers=["ONLY-ONE"]) with self.assertRaises(ValidationError): self.post(row) self.assertEqual(SerializedStockUnit.objects.count(), 0) def test_posted_receipt_cannot_be_cancelled_or_edited(self): row = self.post(self.receipt()) for operation in ( lambda: d.cancel_goods_receipt(actor=self.actor, receipt=row, expected_revision=revision(row), reason="Erase"), lambda: d.update_goods_receipt(actor=self.actor, receipt=row, expected_revision=revision(row), note="Changed"), lambda: d.set_goods_receipt_lines(actor=self.actor, receipt=row, expected_revision=revision(row), lines=[])): with self.assertRaises(ValidationError): operation() def test_posted_header_and_line_updates_rejected_by_database(self): row = self.post(self.receipt()) for model, pk, changes in ((GoodsReceipt, row.pk, {"note": "Changed"}), (GoodsReceiptLine, row.lines.get().pk, {"quantity": 99})): with self.assertRaises(IntegrityError), transaction.atomic(): model.objects.filter(pk=pk).update(**changes) def test_stale_revision_rejects_post_and_edit(self): row = self.receipt() old = revision(row) d.update_goods_receipt(actor=self.actor, receipt=row, expected_revision=old, note="Changed") with self.assertRaises(ValidationError): d.post_goods_receipt(actor=self.actor, receipt=row, expected_revision=old) def test_failed_multi_line_post_rolls_back_all_stock_and_unit_creation(self): row = self.receipt() d.set_goods_receipt_lines(actor=self.actor, receipt=row, expected_revision=revision(row), lines=[ dict(spare_part=self.part, quantity=2), dict(spare_part=self.serial_part, quantity=1, identifiers=["ROLLBACK"]), ]) original = d._post calls = [] def fail_second(**kwargs): calls.append(1) if len(calls) == 2: raise ValidationError("Injected second-line failure") return original(**kwargs) with patch.object(d, "_post", side_effect=fail_second), self.assertRaises(ValidationError): self.post(row) self.assertEqual(StockMovement.objects.count(), 0) self.assertEqual(SerializedStockUnit.objects.count(), 0) row.refresh_from_db() self.assertEqual(row.status, "DRAFT") def test_line_replacement_retains_draft_history(self): row = self.receipt() old_id = row.lines.get().pk d.set_goods_receipt_lines(actor=self.actor, receipt=row, expected_revision=revision(row), lines=[]) self.assertFalse(row.lines.get().is_active) d.set_goods_receipt_lines(actor=self.actor, receipt=row, expected_revision=revision(row), lines=[dict(spare_part=self.part, quantity=1)]) self.assertEqual(row.lines.get().pk, old_id) self.assertTrue(row.lines.get().is_active) def test_duplicate_part_lines_and_duplicate_serials_rejected(self): row = self.receipt() for lines in ([dict(spare_part=self.part, quantity=1)] * 2, [dict(spare_part=self.serial_part, quantity=2, identifiers=["SAME", "SAME"])]): with self.assertRaises(ValidationError): d.set_goods_receipt_lines(actor=self.actor, receipt=row, expected_revision=revision(row), lines=lines) def test_center_scope_can_post_receipt_and_create_units_at_own_center(self): grant(self.staff, self.company, center=self.center) row = self.receipt(1, part=self.serial_part, identifiers=["SCOPED"], destination=self.destination) result = d.post_goods_receipt(actor=self.staff, receipt=row, expected_revision=revision(row)) self.assertEqual(result.received_by, self.staff) self.assertEqual(SerializedStockUnit.objects.get().registered_by, self.staff) class StockTransferTests(DocumentFixture): def test_draft_dispatch_receive_conserves_stock(self): self.post(self.receipt(5)) row = self.transfer(3) self.assertEqual(self.balance(), 5) row = self.dispatch(row) self.assertEqual((self.balance(), self.balance(self.destination), self.balance(row.transit_location)), (2, 0, 3)) self.assertEqual(q.available_stock(actor=self.actor, location=row.transit_location, spare_part=self.part), 0) row = self.receive_transfer(row) self.assertEqual((self.balance(), self.balance(self.destination), self.balance(row.transit_location)), (2, 3, 0)) self.assertEqual(row.status, "RECEIVED") def test_serialized_transfer_is_explicitly_in_transit(self): self.post(self.receipt(1, part=self.serial_part, identifiers=["SERIAL-TRANSFER"])) unit = SerializedStockUnit.objects.get() row = self.dispatch(self.transfer(1, part=self.serial_part, units=[unit])) unit.refresh_from_db() self.assertEqual((unit.state, unit.current_location_id), ("IN_TRANSIT", row.transit_location_id)) self.receive_transfer(row) unit.refresh_from_db() self.assertEqual((unit.state, unit.current_location_id), ("IN_STOCK", self.destination.pk)) self.assertEqual(unit.movement_links.count(), 3) def test_dispatched_transit_cannot_be_manually_received_or_moved(self): self.receive() row = self.dispatch(self.transfer()) with self.assertRaises(ValidationError): self.move(source=row.transit_location) with self.assertRaises(ValidationError): self.receive(destination=row.transit_location) with self.assertRaises(ValidationError): s.deactivate_location(actor=self.actor, location=row.transit_location) def test_pending_transfer_blocks_endpoint_deactivation(self): self.receive(1) row = self.dispatch(self.transfer()) for location in (self.location, self.destination): with self.assertRaises(ValidationError): s.deactivate_location(actor=self.actor, location=location) self.receive_transfer(row) self.assertFalse(s.deactivate_location(actor=self.actor, location=self.location).is_active) def test_cross_company_transfer_rejected(self): with self.assertRaises(ValidationError): self.transfer(destination=self.foreign) def test_non_serviceable_transfer_cannot_promote_stock(self): quarantine = s.create_location(actor=self.actor, company=self.company, code="Q", name="Q", location_type="QUARANTINE") with self.assertRaises(ValidationError): self.transfer(source=quarantine) def test_duplicate_dispatch_and_receipt_rejected(self): self.receive() row = self.dispatch(self.transfer()) with self.assertRaises(ValidationError): self.dispatch(row) self.receive_transfer(row) with self.assertRaises(ValidationError): self.receive_transfer(row) self.assertEqual(StockMovement.objects.count(), 3) def test_draft_cancel_has_no_stock_effect(self): row = self.transfer() result = d.cancel_stock_transfer(actor=self.actor, transfer=row, expected_revision=revision(row), reason="Not required") self.assertEqual(result.status, "CANCELLED") self.assertEqual(StockMovement.objects.count(), 0) def test_dispatched_transfer_cannot_cancel_or_edit(self): self.receive() row = self.dispatch(self.transfer()) with self.assertRaises(ValidationError): d.cancel_stock_transfer(actor=self.actor, transfer=row, expected_revision=revision(row), reason="Erase") with self.assertRaises(ValidationError): d.update_stock_transfer(actor=self.actor, transfer=row, expected_revision=revision(row), note="Changed") def test_draft_transfer_does_not_freeze_policy(self): self.transfer() parts.update_spare_part(spare_part=self.part, serialization_policy="REQUIRED_SERIAL") self.assertEqual(StockMovement.objects.count(), 0) def test_insufficient_stock_dispatch_rolls_back_transit_creation(self): row = self.transfer() with self.assertRaises(ValidationError): self.dispatch(row) row.refresh_from_db() self.assertIsNone(row.transit_location_id) self.assertEqual(row.status, "DRAFT") def test_transfer_requires_both_endpoints_scope(self): grant(self.staff, self.company, center=self.center) row = self.transfer() with self.assertRaises(PermissionDenied): d.dispatch_stock_transfer(actor=self.staff, transfer=row, expected_revision=revision(row)) self.assertEqual(list(dq.stock_transfers(actor=self.staff)), []) class DocumentQueryTests(DocumentFixture): def test_receipt_detail_query_budget(self): row = self.post(self.receipt(2, part=self.serial_part, identifiers=["Q1", "Q2"])) with self.assertNumQueries(4): detail = dq.goods_receipt_detail(actor=self.actor, receipt=row) self.assertEqual(len([(line.spare_part.name, item.unit.identifier) for line in detail.lines.all() for item in line.identifiers.all()]), 2) def test_transfer_detail_query_budget(self): self.post(self.receipt(1, part=self.serial_part, identifiers=["Q1"])) row = self.dispatch(self.transfer(1, part=self.serial_part, units=[SerializedStockUnit.objects.get()])) with self.assertNumQueries(4): detail = dq.stock_transfer_detail(actor=self.actor, transfer=row) self.assertEqual(len([(line.spare_part.name, entry.unit.identifier) for line in detail.lines.all() for entry in line.units.all()]), 1)