diff --git a/backend/inventory/constants.py b/backend/inventory/constants.py new file mode 100644 index 0000000..4505ad8 --- /dev/null +++ b/backend/inventory/constants.py @@ -0,0 +1,6 @@ +""" +Constants for the inventory app. +""" + +# Events that actually change the physical location of an item +LOCATION_CHANGING_EVENTS = ["INITIAL", "ARRIVED", "VERIFIED", "LOCATION_CORRECTION"] diff --git a/backend/inventory/management/__init__.py b/backend/inventory/management/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/inventory/management/commands/__init__.py b/backend/inventory/management/commands/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/inventory/management/commands/rebuild_item_locations.py b/backend/inventory/management/commands/rebuild_item_locations.py new file mode 100644 index 0000000..148b076 --- /dev/null +++ b/backend/inventory/management/commands/rebuild_item_locations.py @@ -0,0 +1,42 @@ +""" +Simple management command to rebuild item locations from history. +""" + +from django.core.management.base import BaseCommand +from inventory.models import CollectionItem + + +class Command(BaseCommand): + help = "Rebuild current_location and is_on_floor for all CollectionItems based on their history" + + def add_arguments(self, parser): + parser.add_argument( + "--item-id", + type=int, + help="Update only the specified item ID", + ) + + def handle(self, *args, **options): + item_id = options.get("item_id") + + if item_id: + # Single item + try: + item = CollectionItem.objects.get(id=item_id) + item.update_location_from_history() + self.stdout.write(self.style.SUCCESS(f"Updated item {item_id}")) + except CollectionItem.DoesNotExist: + self.stdout.write(self.style.ERROR(f"Item {item_id} not found")) + else: + # All items + items = CollectionItem.objects.all() + updated_count = 0 + + for item in items: + try: + item.update_location_from_history() + updated_count += 1 + except Exception as e: + self.stdout.write(self.style.ERROR(f"Error updating item {item.id}: {e}")) + + self.stdout.write(self.style.SUCCESS(f"Updated {updated_count} items")) diff --git a/backend/inventory/models.py b/backend/inventory/models.py index 2736293..4af4c0c 100644 --- a/backend/inventory/models.py +++ b/backend/inventory/models.py @@ -1,5 +1,12 @@ from django.db import models from django.conf import settings +from django.db.models.signals import post_save +from django.dispatch import receiver +import logging + +from .constants import LOCATION_CHANGING_EVENTS + +logger = logging.getLogger(__name__) class Location(models.Model): @@ -143,3 +150,19 @@ class Meta: def __str__(self): return f"{self.item.item_code} - {self.get_event_type_display()} at {self.created_at}" + + +@receiver(post_save, sender=ItemHistory) +def update_item_location_on_history_change(sender, instance, created, **kwargs): + """ + Update item location when a new history event is created. + Only triggers for location-changing events (INITIAL, ARRIVED, VERIFIED, LOCATION_CORRECTION). + """ + if created and instance.event_type in LOCATION_CHANGING_EVENTS: + # Use try/except to prevent cascading failures + try: + item: CollectionItem = instance.item + item.update_location_from_history() + except Exception as e: + # Log error but don't raise to prevent disrupting the original save + logger.error(f"Failed to update item location for item {instance.item_id}: {e}") diff --git a/backend/inventory/tests.py b/backend/inventory/tests.py index 7ce503c..89a61f7 100644 --- a/backend/inventory/tests.py +++ b/backend/inventory/tests.py @@ -1,3 +1,97 @@ from django.test import TestCase +from django.core.management import call_command +from io import StringIO -# Create your tests here. +from .models import CollectionItem, Location, ItemHistory +from users.models import User + + +class ItemLocationTest(TestCase): + """Test the item location functionality.""" + + def setUp(self): + self.user = User.objects.create_user(email="test@example.com", name="Test User", password="testpass", role="VOLUNTEER") + self.location_storage = Location.objects.create(name="Storage A", location_type="STORAGE") + self.location_floor = Location.objects.create(name="Main Floor", location_type="FLOOR") + self.item = CollectionItem.objects.create( + item_code="TEST001", title="Test Item", current_location=self.location_storage + ) + + def test_update_location_from_history_initial_event(self): + """Test location update with INITIAL event.""" + # Create INITIAL history event + ItemHistory.objects.create(item=self.item, event_type="INITIAL", to_location=self.location_storage, acted_by=self.user) + + # Update location using model method + self.item.update_location_from_history() + + self.assertEqual(self.item.current_location, self.location_storage) + self.assertFalse(self.item.is_on_floor) + + def test_update_location_from_history_floor_move(self): + """Test location update when item moves to floor.""" + # Create events + ItemHistory.objects.create(item=self.item, event_type="INITIAL", to_location=self.location_storage) + ItemHistory.objects.create( + item=self.item, event_type="ARRIVED", from_location=self.location_storage, to_location=self.location_floor + ) + + self.item.update_location_from_history() + + self.assertEqual(self.item.current_location, self.location_floor) + self.assertTrue(self.item.is_on_floor) + + def test_signal_triggers_on_location_changing_event(self): + """Test that signal updates item location for location-changing events.""" + # Create ARRIVED event - should trigger signal + ItemHistory.objects.create( + item=self.item, event_type="ARRIVED", to_location=self.location_floor, from_location=self.location_storage + ) + + # Refresh item from database + self.item.refresh_from_db() + + # Verify location was updated by signal + self.assertEqual(self.item.current_location, self.location_floor) + self.assertTrue(self.item.is_on_floor) + + def test_signal_does_not_update_location_for_workflow_events(self): + """Test that signal triggers but doesn't update location for workflow-only events.""" + original_location = self.item.current_location + original_is_on_floor = self.item.is_on_floor + + # Create MOVE_REQUESTED event - should NOT trigger signal + ItemHistory.objects.create( + item=self.item, event_type="MOVE_REQUESTED", to_location=self.location_floor, from_location=self.location_storage + ) + + # Refresh item from database + self.item.refresh_from_db() + + # Verify location was NOT updated + self.assertEqual(self.item.current_location, original_location) + self.assertEqual(self.item.is_on_floor, original_is_on_floor) + + +class RebuildItemLocationsCommandTest(TestCase): + """Test the management command.""" + + def setUp(self): + self.location = Location.objects.create(name="Test Location", location_type="STORAGE") + self.item = CollectionItem.objects.create(item_code="TEST001", title="Test Item", current_location=self.location) + + def test_command_all_items(self): + """Test command rebuilds all items.""" + out = StringIO() + call_command("rebuild_item_locations", stdout=out) + + output = out.getvalue() + self.assertIn("Updated", output) + + def test_command_single_item(self): + """Test command with specific item ID.""" + out = StringIO() + call_command("rebuild_item_locations", "--item-id", self.item.id, stdout=out) + + output = out.getvalue() + self.assertIn(f"Updated item {self.item.id}", output) diff --git a/backend/inventory/utils.py b/backend/inventory/utils.py index df94e8e..03a8088 100644 --- a/backend/inventory/utils.py +++ b/backend/inventory/utils.py @@ -3,6 +3,7 @@ """ from .models import ItemHistory, Location +from .constants import LOCATION_CHANGING_EVENTS def get_current_location(item_id): @@ -21,9 +22,6 @@ def get_current_location(item_id): Returns: Location object or None """ - # Events that actually change the physical location - LOCATION_CHANGING_EVENTS = ["INITIAL", "ARRIVED", "VERIFIED", "LOCATION_CORRECTION"] - # Get the most recent location-changing event last_event = ( ItemHistory.objects.filter(item_id=item_id, event_type__in=LOCATION_CHANGING_EVENTS).order_by("-created_at").first()