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, connection, transaction from django.test import TestCase from apps.access.models import Role, UserRoleAssignment from apps.organization.assignment_services import create_assignment from apps.organization.test_lifecycle import make_tree from apps.parts import services as parts from . import queries as q, services as s from .models import InventoryLocation, SerializedStockUnit, StockMovement, StockLedgerEntry, StockMovementUnit def grant(actor, company, *, center=None, permission_names=None): assignment = create_assignment(user=actor, company=company, region=center.region if center else None, service_center=center) role = Role.objects.create(code=f"ROLE-{uuid.uuid4().hex}", name="Inventory role") permissions = Permission.objects.filter(content_type__app_label="inventory") if permission_names is not None: permissions = permissions.filter(codename__in=permission_names) role.permissions.set(permissions) UserRoleAssignment.objects.create(user=actor, role=role, organization_assignment=assignment) return role def make_inventory_fixture(owner): owner.company, owner.region, owner.center, owner.department = make_tree("INV") owner.other_company, _, owner.other_center, _ = make_tree("OTHER-INV") owner.actor = get_user_model().objects.create_superuser(username="inventory-admin") owner.staff = get_user_model().objects.create_user(username="inventory-staff", is_staff=True) owner.category = parts.create_part_category(code="INV", name="Inventory parts") owner.part = parts.create_spare_part(part_code="INV-PART", name="Part", category=owner.category, serialization_policy="NOT_SERIALIZED") owner.serial_part = parts.create_spare_part(part_code="INV-SERIAL", name="Serialized part", category=owner.category, serialization_policy="REQUIRED_SERIAL") owner.location = s.create_location(actor=owner.actor, company=owner.company, code="MAIN", name="Main", location_type="WAREHOUSE") owner.destination = s.create_location(actor=owner.actor, company=owner.company, service_center=owner.center, code="CENTER-STORE", name="Center store", location_type="STORE") owner.foreign = s.create_location(actor=owner.actor, company=owner.other_company, code="MAIN", name="Other main", location_type="WAREHOUSE") class InventoryFixture(TestCase): @classmethod def setUpTestData(cls): make_inventory_fixture(cls) def receive(self, quantity=5, **kwargs): return s.receive_stock(**dict(actor=self.actor, destination=self.location, spare_part=self.part, quantity=quantity, reference="TEST-RECEIPT", idempotency_key=uuid.uuid4()) | kwargs) def move(self, quantity=1, **kwargs): return s.move_stock(**dict(actor=self.actor, source=self.location, destination=self.destination, spare_part=self.part, quantity=quantity, reference="TEST-MOVE", idempotency_key=uuid.uuid4()) | kwargs) def balance(self, location=None, part=None): return q.stock_on_hand(actor=self.actor, location=location or self.location, spare_part=part or self.part) class LocationTests(InventoryFixture): def test_canonical_company_scoped_codes(self): row = s.create_location(actor=self.actor, company=self.company, code=" secondary ", name=" Secondary ", location_type="WAREHOUSE") self.assertEqual((row.code, row.name), ("SECONDARY", "Secondary")) self.assertIsInstance(row.pk, uuid.UUID) with self.assertRaises(ValidationError): s.create_location(actor=self.actor, company=self.company, code=" main ", name="Duplicate", location_type="WAREHOUSE") def test_multiple_locations_per_center(self): row = s.create_location(actor=self.actor, company=self.company, service_center=self.center, code="SECOND-STORE", name="Second store", location_type="STORE") self.assertEqual(row.service_center, self.destination.service_center) def test_store_requires_center(self): with self.assertRaises(ValidationError): s.create_location(actor=self.actor, company=self.company, code="BAD", name="Bad", location_type="STORE") def test_cross_company_center_rejected(self): with self.assertRaises(ValidationError): s.create_location(actor=self.actor, company=self.company, service_center=self.other_center, code="BAD", name="Bad", location_type="STORE") def test_invalid_codes_names_and_type(self): for field, value in (("code", "bad code"), ("name", " "), ("location_type", "INVALID")): with self.subTest(field=field), self.assertRaises(ValidationError): s.create_location(**dict(actor=self.actor, company=self.company, code="NEW", name="New", location_type="WAREHOUSE") | {field: value}) def test_empty_location_lifecycle_and_stale_revision(self): revision = self.location.updated_at.isoformat() s.deactivate_location(actor=self.actor, location=self.location) with self.assertRaises(ValidationError): s.update_location(actor=self.actor, location=self.location, name="Stale", expected_revision=revision) self.assertTrue(s.reactivate_location(actor=self.actor, location=self.location).is_active) def test_nonempty_location_cannot_be_deactivated(self): self.receive() with self.assertRaises(ValidationError): s.deactivate_location(actor=self.actor, location=self.location) def test_database_identity_is_immutable(self): with self.assertRaises(IntegrityError), transaction.atomic(): InventoryLocation.objects.filter(pk=self.location.pk).update(company=self.other_company) class LedgerTests(InventoryFixture): def test_receipt_and_transfer_are_explainable(self): receipt = self.receive(7) move = self.move(3) self.assertEqual(self.balance(), 4) self.assertEqual(self.balance(self.destination), 3) self.assertEqual(receipt.entries.get().quantity_delta, 7) self.assertEqual(sorted(move.entries.values_list("quantity_delta", flat=True)), [-3, 3]) self.assertEqual(StockMovement.objects.count(), 2) def test_quantities_are_positive_integers(self): for quantity in (0, -1, True, 1.5, "2", 1000000001): with self.subTest(quantity=quantity), self.assertRaises(ValidationError): self.receive(quantity) self.assertEqual(self.balance(), 0) def test_insufficient_stock_rejected(self): self.receive(2) with self.assertRaises(ValidationError): self.move(3) self.assertEqual(self.balance(), 2) def test_same_location_rejected(self): self.receive() with self.assertRaises(ValidationError): self.move(destination=self.location) def test_cross_company_movement_rejected(self): self.receive() with self.assertRaises(ValidationError): self.move(destination=self.foreign) self.assertEqual(self.balance(self.foreign), 0) def test_inactive_location_rejected_freshly(self): s.deactivate_location(actor=self.actor, location=self.location) with self.assertRaises(ValidationError): self.receive() def test_inactive_part_and_category_rejected(self): parts.deactivate_spare_part(spare_part=self.part) with self.assertRaises(ValidationError): self.receive() parts.reactivate_spare_part(spare_part=self.part) parts.deactivate_part_category(part_category=self.category) with self.assertRaises(ValidationError): self.receive() def test_duplicate_command_rejected_without_duplicate_stock(self): key = uuid.uuid4() self.receive(idempotency_key=key) with self.assertRaises(ValidationError): self.receive(idempotency_key=key) self.assertEqual(self.balance(), 5) def test_reference_and_command_key_required(self): for kwargs in ({"reference": " "}, {"idempotency_key": None}, {"note": "x" * 2001}): with self.assertRaises(ValidationError): self.receive(**kwargs) def test_failed_ledger_write_rolls_back_header_and_entries(self): with patch.object(StockLedgerEntry, "_persist", side_effect=ValidationError("Injected write failure")), self.assertRaises(ValidationError): self.receive() self.assertEqual(StockMovement.objects.count(), 0) self.assertEqual(self.balance(), 0) def test_posted_history_rejects_supported_and_raw_updates(self): movement = self.receive() entry = movement.entries.get() for obj in (movement, entry): with self.assertRaises(ValidationError): obj.save() with self.assertRaises(ValidationError): obj.delete() with self.assertRaises(IntegrityError), transaction.atomic(): type(obj).objects.filter(pk=obj.pk).update(updated_at=obj.updated_at) def test_database_rejects_incomplete_posting(self): with self.assertRaises(IntegrityError), transaction.atomic(): movement = StockMovement(company=self.company, spare_part=self.part, kind="RECEIPT", quantity=1, destination=self.location, actor=self.actor, reference="RAW", idempotency_key=uuid.uuid4()) movement._persist() with connection.cursor() as cursor: cursor.execute("SET CONSTRAINTS ALL IMMEDIATE") def test_quarantine_not_usable_or_silently_promoted(self): quarantine = s.create_location(actor=self.actor, company=self.company, code="Q", name="Quarantine", location_type="QUARANTINE") self.receive(destination=quarantine) self.assertEqual(q.available_stock(actor=self.actor, location=quarantine, spare_part=self.part), 0) self.assertEqual(self.balance(quarantine), 5) with self.assertRaises(ValidationError): self.move(source=quarantine) class SerializedTests(InventoryFixture): def unit(self, identifier="SN-1", **kwargs): return s.register_serialized_unit(**dict(actor=self.actor, company=self.company, spare_part=self.serial_part, identifier=identifier) | kwargs) def test_identity_is_trimmed_case_preserving_and_permanent(self): unit = self.unit(" sn/001 ") self.assertEqual(unit.identifier, "sn/001") with self.assertRaises(ValidationError): self.unit("sn/001") with self.assertRaises(IntegrityError), transaction.atomic(): SerializedStockUnit.objects.filter(pk=unit.pk).update(identifier="REPLACED") def test_required_serial_rejects_anonymous_stock(self): with self.assertRaises(ValidationError): self.receive(spare_part=self.serial_part) def test_not_serialized_rejects_unit_registration(self): with self.assertRaises(ValidationError): self.unit(spare_part=self.part) def test_control_characters_and_empty_identity_rejected(self): for identifier in (" ", "SN\x00X", "SN\nX", "x" * 129): with self.assertRaises(ValidationError): self.unit(identifier) def test_serialized_receipt_and_move_preserve_identity_and_history(self): unit = self.unit() receipt = self.receive(1, spare_part=self.serial_part, units=[unit]) move = self.move(1, spare_part=self.serial_part, units=[unit]) unit.refresh_from_db() self.assertEqual(unit.current_location, self.destination) self.assertEqual(unit.current_movement, move) self.assertEqual(unit.movement_links.count(), 2) self.assertEqual(list(q.serialized_history(actor=self.actor, unit=unit).values_list("movement_id", flat=True)), [receipt.pk, move.pk]) def test_unit_cannot_be_received_twice(self): unit = self.unit() self.receive(1, spare_part=self.serial_part, units=[unit]) with self.assertRaises(ValidationError): self.receive(1, spare_part=self.serial_part, units=[unit]) def test_duplicate_unit_in_single_movement_rejected(self): unit = self.unit() with self.assertRaises(ValidationError): self.receive(2, spare_part=self.serial_part, units=[unit, unit]) def test_unit_wrong_company_or_part_rejected(self): unit = self.unit(company=self.other_company) with self.assertRaises(ValidationError): self.receive(1, spare_part=self.serial_part, units=[unit]) with self.assertRaises(ValidationError): self.receive(1, units=[unit]) def test_optional_serial_buckets_cannot_be_substituted(self): parts.update_spare_part(spare_part=self.part, serialization_policy="OPTIONAL_SERIAL") unit = self.unit(spare_part=self.part) self.receive(2, units=[unit]) with self.assertRaises(ValidationError): self.move(2) # One anonymous unit exists, not two. self.move(1) self.move(1, units=[unit]) self.assertEqual(self.balance(), 0) class PolicyIntegrationTests(InventoryFixture): def test_policy_changes_allowed_before_inventory(self): for policy in SparePartPolicies: self.assertEqual(parts.update_spare_part(spare_part=self.part, serialization_policy=policy).serialization_policy, policy) def test_registered_unit_alone_permanently_locks_policy(self): s.register_serialized_unit(actor=self.actor, company=self.company, spare_part=self.serial_part, identifier="REGISTERED") self.assertEqual(StockMovement.objects.count(), 0) with self.assertRaises(ValidationError): parts.update_spare_part(spare_part=self.serial_part, serialization_policy="OPTIONAL_SERIAL") def test_posted_anonymous_stock_locks_policy(self): self.receive() with self.assertRaises(ValidationError): parts.update_spare_part(spare_part=self.part, serialization_policy="REQUIRED_SERIAL") def test_same_policy_and_metadata_edits_remain_supported(self): self.receive() row = parts.update_spare_part(spare_part=self.part, name="Renamed", serialization_policy="NOT_SERIALIZED") self.assertEqual(row.name, "Renamed") def test_empty_original_location_does_not_unlock_policy(self): self.receive(1) self.move(1) self.assertEqual(self.balance(), 0) with self.assertRaises(ValidationError): parts.update_spare_part(spare_part=self.part, serialization_policy="OPTIONAL_SERIAL") def test_rolled_back_posting_does_not_lock_policy(self): with self.assertRaises(ValidationError): with transaction.atomic(): self.receive() raise ValidationError("Abort") self.assertEqual(parts.update_spare_part(spare_part=self.part, serialization_policy="REQUIRED_SERIAL").serialization_policy, "REQUIRED_SERIAL") SparePartPolicies = ("NOT_SERIALIZED", "OPTIONAL_SERIAL", "REQUIRED_SERIAL") class AuthorizationQueryTests(InventoryFixture): def test_staff_and_direct_permissions_do_not_create_scope(self): self.staff.user_permissions.set(Permission.objects.filter(content_type__app_label="inventory")) with self.assertRaises(PermissionDenied): self.receive(actor=self.staff) def test_company_scope_and_isolation(self): grant(self.staff, self.company) self.receive(actor=self.staff) with self.assertRaises(PermissionDenied): self.receive(actor=self.staff, destination=self.foreign) self.assertEqual(set(q.authorized_locations(actor=self.staff).values_list("company_id", flat=True)), {self.company.pk}) def test_center_scope_does_not_authorize_central_warehouse(self): grant(self.staff, self.company, center=self.center) self.receive(actor=self.staff, destination=self.destination) with self.assertRaises(PermissionDenied): self.receive(actor=self.staff) def test_inactive_actor_and_company_rejected(self): get_user_model().objects.filter(pk=self.actor.pk).update(is_active=False) with self.assertRaises(ValidationError): self.receive() def test_sql_aggregation_is_lazy_deterministic_and_bounded(self): self.receive() with self.assertNumQueries(0): positions = q.stock_positions_for_part(actor=self.actor, spare_part=self.part) with self.assertNumQueries(1): self.assertEqual(list(positions)[0]["on_hand"], 5) with self.assertNumQueries(3): self.assertEqual(self.balance(), 5) def test_available_excludes_inactive_parts_without_erasing_balance(self): self.receive() parts.deactivate_spare_part(spare_part=self.part) self.assertEqual(q.available_stock(actor=self.actor, location=self.location, spare_part=self.part), 0) self.assertEqual(self.balance(), 5) def test_cross_company_balance_query_denied(self): grant(self.staff, self.company) with self.assertRaises(PermissionDenied): q.stock_on_hand(actor=self.staff, location=self.foreign, spare_part=self.part)