Merge pull request #124 from inventree/stock-adjustment-actions

Stock adjustment actions

inventree/stock.py

@@ -3,6 +3,7 @@
 import os
 import logging
 
+import inventree.api
 import inventree.base
 import inventree.part
 
@@ -47,6 +48,148 @@ class StockItem(inventree.base.MetadataMixin, inventree.base.InventreeObject):
 
     URL = 'stock'
 
+    @classmethod
+    def adjustStockItems(cls, api: inventree.api.InvenTreeAPI, method: str, items: list, **kwargs):
+        """Perform a generic stock 'adjustment' action.
+        
+        Arguments:
+            api: InvenTreeAPI instance
+            method: Adjument method, e.g. 'count' / 'add'
+            items: List of items to include in the adjustment (see below)
+            kwargs: Additional kwargs to send with the adjustment
+        
+        Items:
+            Each 'item' in the 'items' list must be a dict object, containing the following fields:
+
+            pk: The 'pk' (primary key) identifier for a StockItem instance
+            quantity: The quantity of each stock item for the particular action
+        """
+
+        if method not in ['count', 'add', 'remove', 'transfer']:
+            raise ValueError(f"Stock adjustment method '{method}' not supported")
+
+        url = f"stock/{method}/"
+
+        data = kwargs
+        data['items'] = items
+
+        return api.post(url, data=data)
+
+    @classmethod
+    def countStockItems(cls, api: inventree.api.InvenTreeAPI, items: list, **kwargs):
+        """Perform 'count' adjustment for multiple stock items"""
+
+        return cls.adjustStockItems(
+            api,
+            'count',
+            items,
+            **kwargs
+        )
+
+    @classmethod
+    def addStockItems(cls, api: inventree.api.InvenTreeAPI, items: list, **kwargs):
+        """Perform 'add' adjustment for multiple stock items"""
+
+        return cls.adjustStockItems(
+            api,
+            'add',
+            items,
+            **kwargs
+        )
+
+    @classmethod
+    def removeStockItems(cls, api: inventree.api.InvenTreeAPI, items: list, **kwargs):
+        """Perform 'remove' adjustment for multiple stock items"""
+
+        return cls.adjustStockItems(
+            api,
+            'remove',
+            items,
+            **kwargs
+        )
+
+    @classmethod
+    def transferStockItems(cls, api: inventree.api.InvenTreeAPI, items: list, location: int, **kwargs):
+        """Perform 'transfer' adjustment for multiple stock items"""
+
+        kwargs['location'] = location
+
+        return cls.adjustStockItems(
+            api,
+            'transfer',
+            items,
+            **kwargs
+        )
+
+    def countStock(self, quantity, **kwargs):
+        """Perform a count (stocktake) action for this StockItem"""
+
+        self.countStockItems(
+            self._api,
+            [
+                {
+                    'pk': self.pk,
+                    'quantity': quantity,
+                }
+            ],
+            **kwargs
+        )
+
+    def addStock(self, quantity, **kwargs):
+        """Manually add the specified quantity to this StockItem"""
+
+        self.addStockItems(
+            self._api,
+            [
+                {
+                    'pk': self.pk,
+                    'quantity': quantity,
+                }
+            ],
+            **kwargs
+        )
+
+    def removeStock(self, quantity, **kwargs):
+        """Manually remove the specified quantity to this StockItem"""
+
+        self.removeStockItems(
+            self._api,
+            [
+                {
+                    'pk': self.pk,
+                    'quantity': quantity,
+                }
+            ],
+            **kwargs
+        )
+
+    def transferStock(self, location, quantity=None, **kwargs):
+        """Transfer this StockItem into the specified location.
+        
+        Arguments:
+            location: A StockLocation instance or integer ID value
+            quantity: Optionally specify quantity to transfer. If None, entire quantity is transferred
+            notes: Optional transaction notes
+        """
+
+        if isinstance(location, StockLocation):
+            location = location.pk
+
+        if quantity is None:
+            quantity = self.quantity
+
+        self.transferStockItems(
+            self._api,
+            [
+                {
+                    'pk': self.pk,
+                    'quantity': quantity,
+                }
+            ],
+            location=location,
+            **kwargs
+        )
+
     def getPart(self):
         """ Return the base Part object associated with this StockItem """
         return inventree.part.Part(self._api, self.part)
@@ -63,6 +206,11 @@ def getLocation(self):
 
         return StockLocation(self._api, self.location)
 
+    def getTrackingEntries(self, **kwargs):
+        """Return list of StockItemTracking instances associated with this StockItem"""
+
+        return StockItemTracking.list(self._api, item=self.pk, **kwargs)
+
     def getTestResults(self, **kwargs):
         """ Return all the test results associated with this StockItem """
 

test/test_stock.py

@@ -1,8 +1,10 @@
 # -*- coding: utf-8 -*-
 
 import os
+import requests
 import sys
 
+
 sys.path.append(os.path.abspath(os.path.dirname(__file__)))
 
 from inventree.stock import StockItem, StockLocation  # noqa: E402
@@ -177,17 +179,155 @@ def test_get_stock_item(self):
         item = StockItem(self.api, pk=1)
 
         self.assertEqual(item.pk, 1)
-        self.assertEqual(item.location, 3)
+        self.assertIsNotNone(item.location)
 
         # Get the Part reference
         prt = item.getPart()
 
         self.assertEqual(type(prt), part.Part)
         self.assertEqual(prt.pk, 1)
 
-        # Get the Location reference
+        # Move the item to a known location
+        item.transferStock(3)
+        item.reload()
+
         location = item.getLocation()
 
         self.assertEqual(type(location), StockLocation)
         self.assertEqual(location.pk, 3)
         self.assertEqual(location.name, "Dining Room")
+
+
+class StockAdjustTest(InvenTreeTestCase):
+    """Unit tests for stock 'adjustment' actions"""
+
+    def test_count(self):
+        """Test the 'count' action"""
+
+        item = StockItem(self.api, pk=1)
+
+        # Count number of tracking entries
+        n_tracking = len(item.getTrackingEntries())
+
+        q = item.quantity
+
+        item.countStock(q + 100)
+        item.reload()
+
+        self.assertEqual(item.quantity, q + 100)
+
+        item.countStock(q, notes='Why hello there')
+        item.reload()
+        self.assertEqual(item.quantity, q)
+
+        # 2 tracking entries should have been added
+        self.assertEqual(
+            len(item.getTrackingEntries()),
+            n_tracking + 2
+        )
+
+        # The most recent tracking entry should have a note
+        t = item.getTrackingEntries()[0]
+        self.assertEqual(t.label, 'Stock counted')
+
+        # Check error conditions
+        with self.assertRaises(requests.exceptions.HTTPError):
+            item.countStock('not a number')
+
+        with self.assertRaises(requests.exceptions.HTTPError):
+            item.countStock(-1)
+
+    def test_add_remove(self):
+        """Test the 'add' and 'remove' actions"""
+
+        item = StockItem(self.api, pk=1)
+
+        n_tracking = len(item.getTrackingEntries())
+
+        q = item.quantity
+
+        # Add some items
+        item.addStock(10)
+        item.reload()
+        self.assertEqual(item.quantity, q + 10)
+
+        # Remove the items again
+        item.removeStock(10)
+        item.reload()
+        self.assertEqual(item.quantity, q)
+
+        # 2 additional tracking entries should have been added
+        self.assertTrue(len(item.getTrackingEntries()) > n_tracking)
+
+        # Test error conditions
+        for v in [-1, 'gg', None]:
+            with self.assertRaises(requests.exceptions.HTTPError):
+                item.addStock(v)
+            with self.assertRaises(requests.exceptions.HTTPError):
+                item.removeStock(v)
+
+    def test_transfer(self):
+        """Unit test for 'transfer' action"""
+
+        item = StockItem(self.api, pk=2)
+
+        n_tracking = len(item.getTrackingEntries())
+
+        # Transfer to a StockLocation instance
+        location = StockLocation(self.api, pk=1)
+
+        item.transferStock(location)
+        item.reload()
+        self.assertEqual(item.location, 1)
+
+        # Transfer with a location ID
+        item.transferStock(2)
+        item.reload()
+        self.assertEqual(item.location, 2)
+
+        # 2 additional tracking entries should have been added
+        self.assertTrue(len(item.getTrackingEntries()) > n_tracking)
+
+        # Attempt to transfer to an invalid location
+        for loc in [-1, 'qqq', 99999, None]:
+            with self.assertRaises(requests.exceptions.HTTPError):
+                item.transferStock(loc)
+
+        # Attempt to transfer with an invalid quantity
+        for q in [-1, None, 'hhhh']:
+            with self.assertRaises(requests.exceptions.HTTPError):
+                item.transferStock(loc, quantity=q)
+
+    def test_transfer_multiple(self):
+        """Test transfer of *multiple* items"""
+
+        items = StockItem.list(self.api, location=1)
+        self.assertTrue(len(items) > 1)
+
+        # Construct data to send
+        data = []
+
+        for item in items:
+            data.append({
+                'pk': item.pk,
+                'quantity': item.quantity,
+            })
+
+        # Transfer all items into a new location
+        StockItem.transferStockItems(self.api, data, 2)
+
+        for item in items:
+            item.reload()
+            self.assertEqual(item.location, 2)
+
+        # Transfer back to the original location
+        StockItem.transferStockItems(self.api, data, 1)
+
+        for item in items:
+            item.reload()
+            self.assertEqual(item.location, 1)
+
+            history = item.getTrackingEntries()
+
+            self.assertTrue(len(history) >= 2)
+            self.assertEqual(history[0].label, 'Location changed')