mirror of
https://github.com/inventree/InvenTree
synced 2024-08-30 18:33:04 +00:00
754 lines
20 KiB
Python
754 lines
20 KiB
Python
"""
|
|
Unit testing for the Stock API
|
|
"""
|
|
|
|
# -*- coding: utf-8 -*-
|
|
from __future__ import unicode_literals
|
|
|
|
import os
|
|
|
|
from datetime import datetime, timedelta
|
|
|
|
from django.urls import reverse
|
|
from rest_framework import status
|
|
|
|
from InvenTree.status_codes import StockStatus
|
|
from InvenTree.api_tester import InvenTreeAPITestCase
|
|
|
|
from common.models import InvenTreeSetting
|
|
|
|
from .models import StockItem, StockLocation
|
|
from .tasks import delete_old_stock_items
|
|
|
|
|
|
class StockAPITestCase(InvenTreeAPITestCase):
|
|
|
|
fixtures = [
|
|
'category',
|
|
'part',
|
|
'company',
|
|
'location',
|
|
'supplier_part',
|
|
'stock',
|
|
'stock_tests',
|
|
]
|
|
|
|
roles = [
|
|
'stock.change',
|
|
'stock.add',
|
|
'stock_location.change',
|
|
'stock_location.add',
|
|
'stock.delete',
|
|
]
|
|
|
|
def setUp(self):
|
|
|
|
super().setUp()
|
|
|
|
|
|
class StockLocationTest(StockAPITestCase):
|
|
"""
|
|
Series of API tests for the StockLocation API
|
|
"""
|
|
list_url = reverse('api-location-list')
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
|
|
# Add some stock locations
|
|
StockLocation.objects.create(name='top', description='top category')
|
|
|
|
def test_list(self):
|
|
# Check that we can request the StockLocation list
|
|
response = self.client.get(self.list_url, format='json')
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertGreaterEqual(len(response.data), 1)
|
|
|
|
def test_add(self):
|
|
# Check that we can add a new StockLocation
|
|
data = {
|
|
'parent': 1,
|
|
'name': 'Location',
|
|
'description': 'Another location for stock'
|
|
}
|
|
response = self.client.post(self.list_url, data, format='json')
|
|
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
|
|
|
|
|
class StockItemListTest(StockAPITestCase):
|
|
"""
|
|
Tests for the StockItem API LIST endpoint
|
|
"""
|
|
|
|
list_url = reverse('api-stock-list')
|
|
|
|
def get_stock(self, **kwargs):
|
|
"""
|
|
Filter stock and return JSON object
|
|
"""
|
|
|
|
response = self.client.get(self.list_url, format='json', data=kwargs)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
|
|
# Return JSON-ified data
|
|
return response.data
|
|
|
|
def test_get_stock_list(self):
|
|
"""
|
|
List *all* StockItem objects.
|
|
"""
|
|
|
|
response = self.get_stock()
|
|
|
|
self.assertEqual(len(response), 20)
|
|
|
|
def test_filter_by_part(self):
|
|
"""
|
|
Filter StockItem by Part reference
|
|
"""
|
|
|
|
response = self.get_stock(part=25)
|
|
|
|
self.assertEqual(len(response), 8)
|
|
|
|
response = self.get_stock(part=10004)
|
|
|
|
self.assertEqual(len(response), 12)
|
|
|
|
def test_filter_by_IPN(self):
|
|
"""
|
|
Filter StockItem by IPN reference
|
|
"""
|
|
|
|
response = self.get_stock(IPN="R.CH")
|
|
self.assertEqual(len(response), 3)
|
|
|
|
def test_filter_by_location(self):
|
|
"""
|
|
Filter StockItem by StockLocation reference
|
|
"""
|
|
|
|
response = self.get_stock(location=5)
|
|
self.assertEqual(len(response), 1)
|
|
|
|
response = self.get_stock(location=1, cascade=0)
|
|
self.assertEqual(len(response), 0)
|
|
|
|
response = self.get_stock(location=1, cascade=1)
|
|
self.assertEqual(len(response), 2)
|
|
|
|
response = self.get_stock(location=7)
|
|
self.assertEqual(len(response), 16)
|
|
|
|
def test_filter_by_depleted(self):
|
|
"""
|
|
Filter StockItem by depleted status
|
|
"""
|
|
|
|
response = self.get_stock(depleted=1)
|
|
self.assertEqual(len(response), 1)
|
|
|
|
response = self.get_stock(depleted=0)
|
|
self.assertEqual(len(response), 19)
|
|
|
|
def test_filter_by_in_stock(self):
|
|
"""
|
|
Filter StockItem by 'in stock' status
|
|
"""
|
|
|
|
response = self.get_stock(in_stock=1)
|
|
self.assertEqual(len(response), 17)
|
|
|
|
response = self.get_stock(in_stock=0)
|
|
self.assertEqual(len(response), 3)
|
|
|
|
def test_filter_by_status(self):
|
|
"""
|
|
Filter StockItem by 'status' field
|
|
"""
|
|
|
|
codes = {
|
|
StockStatus.OK: 18,
|
|
StockStatus.DESTROYED: 1,
|
|
StockStatus.LOST: 1,
|
|
StockStatus.DAMAGED: 0,
|
|
StockStatus.REJECTED: 0,
|
|
}
|
|
|
|
for code in codes.keys():
|
|
num = codes[code]
|
|
|
|
response = self.get_stock(status=code)
|
|
self.assertEqual(len(response), num)
|
|
|
|
def test_filter_by_batch(self):
|
|
"""
|
|
Filter StockItem by batch code
|
|
"""
|
|
|
|
response = self.get_stock(batch='B123')
|
|
self.assertEqual(len(response), 1)
|
|
|
|
def test_filter_by_serialized(self):
|
|
"""
|
|
Filter StockItem by serialized status
|
|
"""
|
|
|
|
response = self.get_stock(serialized=1)
|
|
self.assertEqual(len(response), 12)
|
|
|
|
for item in response:
|
|
self.assertIsNotNone(item['serial'])
|
|
|
|
response = self.get_stock(serialized=0)
|
|
self.assertEqual(len(response), 8)
|
|
|
|
for item in response:
|
|
self.assertIsNone(item['serial'])
|
|
|
|
def test_filter_by_expired(self):
|
|
"""
|
|
Filter StockItem by expiry status
|
|
"""
|
|
|
|
# First, we can assume that the 'stock expiry' feature is disabled
|
|
response = self.get_stock(expired=1)
|
|
self.assertEqual(len(response), 20)
|
|
|
|
self.user.is_staff = True
|
|
self.user.save()
|
|
|
|
# Now, ensure that the expiry date feature is enabled!
|
|
InvenTreeSetting.set_setting('STOCK_ENABLE_EXPIRY', True, self.user)
|
|
|
|
response = self.get_stock(expired=1)
|
|
self.assertEqual(len(response), 1)
|
|
|
|
for item in response:
|
|
self.assertTrue(item['expired'])
|
|
|
|
response = self.get_stock(expired=0)
|
|
self.assertEqual(len(response), 19)
|
|
|
|
for item in response:
|
|
self.assertFalse(item['expired'])
|
|
|
|
# Mark some other stock items as expired
|
|
today = datetime.now().date()
|
|
|
|
for pk in [510, 511, 512]:
|
|
item = StockItem.objects.get(pk=pk)
|
|
item.expiry_date = today - timedelta(days=pk)
|
|
item.save()
|
|
|
|
response = self.get_stock(expired=1)
|
|
self.assertEqual(len(response), 4)
|
|
|
|
response = self.get_stock(expired=0)
|
|
self.assertEqual(len(response), 16)
|
|
|
|
def test_paginate(self):
|
|
"""
|
|
Test that we can paginate results correctly
|
|
"""
|
|
|
|
for n in [1, 5, 10]:
|
|
response = self.get_stock(limit=n)
|
|
|
|
self.assertIn('count', response)
|
|
self.assertIn('results', response)
|
|
|
|
self.assertEqual(len(response['results']), n)
|
|
|
|
|
|
class StockItemTest(StockAPITestCase):
|
|
"""
|
|
Series of API tests for the StockItem API
|
|
"""
|
|
|
|
list_url = reverse('api-stock-list')
|
|
|
|
def detail_url(self, pk):
|
|
return reverse('api-stock-detail', kwargs={'pk': pk})
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
# Create some stock locations
|
|
top = StockLocation.objects.create(name='A', description='top')
|
|
|
|
StockLocation.objects.create(name='B', description='location b', parent=top)
|
|
StockLocation.objects.create(name='C', description='location c', parent=top)
|
|
|
|
def test_create_default_location(self):
|
|
"""
|
|
Test the default location functionality,
|
|
if a 'location' is not specified in the creation request.
|
|
"""
|
|
|
|
# The part 'R_4K7_0603' (pk=4) has a default location specified
|
|
|
|
response = self.client.post(
|
|
self.list_url,
|
|
data={
|
|
'part': 4,
|
|
'quantity': 10
|
|
}
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
|
self.assertEqual(response.data['location'], 2)
|
|
|
|
# What if we explicitly set the location to a different value?
|
|
|
|
response = self.client.post(
|
|
self.list_url,
|
|
data={
|
|
'part': 4,
|
|
'quantity': 20,
|
|
'location': 1,
|
|
}
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
|
self.assertEqual(response.data['location'], 1)
|
|
|
|
# And finally, what if we set the location explicitly to None?
|
|
|
|
response = self.client.post(
|
|
self.list_url,
|
|
data={
|
|
'part': 4,
|
|
'quantity': 20,
|
|
'location': '',
|
|
}
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
|
self.assertEqual(response.data['location'], None)
|
|
|
|
def test_stock_item_create(self):
|
|
"""
|
|
Test creation of a StockItem via the API
|
|
"""
|
|
|
|
# POST with an empty part reference
|
|
|
|
response = self.client.post(
|
|
self.list_url,
|
|
data={
|
|
'quantity': 10,
|
|
'location': 1
|
|
}
|
|
)
|
|
|
|
self.assertContains(response, 'This field is required', status_code=status.HTTP_400_BAD_REQUEST)
|
|
|
|
# POST with an invalid part reference
|
|
|
|
response = self.client.post(
|
|
self.list_url,
|
|
data={
|
|
'quantity': 10,
|
|
'location': 1,
|
|
'part': 10000000,
|
|
}
|
|
)
|
|
|
|
self.assertContains(response, 'does not exist', status_code=status.HTTP_400_BAD_REQUEST)
|
|
|
|
# POST without quantity
|
|
response = self.post(
|
|
self.list_url,
|
|
{
|
|
'part': 1,
|
|
'location': 1,
|
|
},
|
|
expected_code=400
|
|
)
|
|
|
|
self.assertIn('Quantity is required', str(response.data))
|
|
|
|
# POST with quantity and part and location
|
|
response = self.post(
|
|
self.list_url,
|
|
data={
|
|
'part': 1,
|
|
'location': 1,
|
|
'quantity': 10,
|
|
},
|
|
expected_code=201
|
|
)
|
|
|
|
def test_default_expiry(self):
|
|
"""
|
|
Test that the "default_expiry" functionality works via the API.
|
|
|
|
- If an expiry_date is specified, use that
|
|
- Otherwise, check if the referenced part has a default_expiry defined
|
|
- If so, use that!
|
|
- Otherwise, no expiry
|
|
|
|
Notes:
|
|
- Part <25> has a default_expiry of 10 days
|
|
|
|
"""
|
|
|
|
# First test - create a new StockItem without an expiry date
|
|
data = {
|
|
'part': 4,
|
|
'quantity': 10,
|
|
}
|
|
|
|
response = self.client.post(self.list_url, data)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
|
|
|
self.assertIsNone(response.data['expiry_date'])
|
|
|
|
# Second test - create a new StockItem with an explicit expiry date
|
|
data['expiry_date'] = '2022-12-12'
|
|
|
|
response = self.client.post(self.list_url, data)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
|
|
|
self.assertIsNotNone(response.data['expiry_date'])
|
|
self.assertEqual(response.data['expiry_date'], '2022-12-12')
|
|
|
|
# Third test - create a new StockItem for a Part which has a default expiry time
|
|
data = {
|
|
'part': 25,
|
|
'quantity': 10
|
|
}
|
|
|
|
response = self.client.post(self.list_url, data)
|
|
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
|
|
|
# Expected expiry date is 10 days in the future
|
|
expiry = datetime.now().date() + timedelta(10)
|
|
|
|
self.assertEqual(response.data['expiry_date'], expiry.isoformat())
|
|
|
|
def test_purchase_price(self):
|
|
"""
|
|
Test that we can correctly read and adjust purchase price information via the API
|
|
"""
|
|
|
|
url = reverse('api-stock-detail', kwargs={'pk': 1})
|
|
|
|
data = self.get(url, expected_code=200).data
|
|
|
|
# Check fixture values
|
|
self.assertEqual(data['purchase_price'], '123.0000')
|
|
self.assertEqual(data['purchase_price_currency'], 'AUD')
|
|
self.assertEqual(data['purchase_price_string'], 'A$123.0000')
|
|
|
|
# Update just the amount
|
|
data = self.patch(
|
|
url,
|
|
{
|
|
'purchase_price': 456
|
|
},
|
|
expected_code=200
|
|
).data
|
|
|
|
self.assertEqual(data['purchase_price'], '456.0000')
|
|
self.assertEqual(data['purchase_price_currency'], 'AUD')
|
|
|
|
# Update the currency
|
|
data = self.patch(
|
|
url,
|
|
{
|
|
'purchase_price_currency': 'NZD',
|
|
},
|
|
expected_code=200
|
|
).data
|
|
|
|
self.assertEqual(data['purchase_price_currency'], 'NZD')
|
|
|
|
# Clear the price field
|
|
data = self.patch(
|
|
url,
|
|
{
|
|
'purchase_price': None,
|
|
},
|
|
expected_code=200
|
|
).data
|
|
|
|
self.assertEqual(data['purchase_price'], None)
|
|
self.assertEqual(data['purchase_price_string'], '-')
|
|
|
|
# Invalid currency code
|
|
data = self.patch(
|
|
url,
|
|
{
|
|
'purchase_price_currency': 'xyz',
|
|
},
|
|
expected_code=400
|
|
)
|
|
|
|
data = self.get(url).data
|
|
self.assertEqual(data['purchase_price_currency'], 'NZD')
|
|
|
|
|
|
class StocktakeTest(StockAPITestCase):
|
|
"""
|
|
Series of tests for the Stocktake API
|
|
"""
|
|
|
|
def test_action(self):
|
|
"""
|
|
Test each stocktake action endpoint,
|
|
for validation
|
|
"""
|
|
|
|
for endpoint in ['api-stock-count', 'api-stock-add', 'api-stock-remove']:
|
|
|
|
url = reverse(endpoint)
|
|
|
|
data = {}
|
|
|
|
# POST with a valid action
|
|
response = self.post(url, data)
|
|
|
|
self.assertIn("This field is required", str(response.data["items"]))
|
|
|
|
data['items'] = [{
|
|
'no': 'aa'
|
|
}]
|
|
|
|
# POST without a PK
|
|
response = self.post(url, data, expected_code=400)
|
|
|
|
self.assertIn('This field is required', str(response.data))
|
|
|
|
# POST with an invalid PK
|
|
data['items'] = [{
|
|
'pk': 10
|
|
}]
|
|
|
|
response = self.post(url, data, expected_code=400)
|
|
|
|
self.assertContains(response, 'object does not exist', status_code=status.HTTP_400_BAD_REQUEST)
|
|
|
|
# POST with missing quantity value
|
|
data['items'] = [{
|
|
'pk': 1234
|
|
}]
|
|
|
|
response = self.post(url, data, expected_code=400)
|
|
self.assertContains(response, 'This field is required', status_code=status.HTTP_400_BAD_REQUEST)
|
|
|
|
# POST with an invalid quantity value
|
|
data['items'] = [{
|
|
'pk': 1234,
|
|
'quantity': '10x0d'
|
|
}]
|
|
|
|
response = self.post(url, data)
|
|
self.assertContains(response, 'A valid number is required', status_code=status.HTTP_400_BAD_REQUEST)
|
|
|
|
data['items'] = [{
|
|
'pk': 1234,
|
|
'quantity': "-1.234"
|
|
}]
|
|
|
|
response = self.post(url, data)
|
|
self.assertContains(response, 'Ensure this value is greater than or equal to 0', status_code=status.HTTP_400_BAD_REQUEST)
|
|
|
|
def test_transfer(self):
|
|
"""
|
|
Test stock transfers
|
|
"""
|
|
|
|
data = {
|
|
'items': [
|
|
{
|
|
'pk': 1234,
|
|
'quantity': 10,
|
|
}
|
|
],
|
|
'location': 1,
|
|
'notes': "Moving to a new location"
|
|
}
|
|
|
|
url = reverse('api-stock-transfer')
|
|
|
|
# This should succeed
|
|
response = self.post(url, data, expected_code=201)
|
|
|
|
# Now try one which will fail due to a bad location
|
|
data['location'] = 'not a location'
|
|
|
|
response = self.post(url, data, expected_code=400)
|
|
|
|
self.assertContains(response, 'Incorrect type. Expected pk value', status_code=status.HTTP_400_BAD_REQUEST)
|
|
|
|
|
|
class StockItemDeletionTest(StockAPITestCase):
|
|
"""
|
|
Tests for stock item deletion via the API
|
|
"""
|
|
|
|
def test_delete(self):
|
|
|
|
# Check there are no stock items scheduled for deletion
|
|
self.assertEqual(
|
|
StockItem.objects.filter(scheduled_for_deletion=True).count(),
|
|
0
|
|
)
|
|
|
|
# Create and then delete a bunch of stock items
|
|
for idx in range(10):
|
|
|
|
# Create new StockItem via the API
|
|
response = self.post(
|
|
reverse('api-stock-list'),
|
|
{
|
|
'part': 1,
|
|
'location': 1,
|
|
'quantity': idx,
|
|
},
|
|
expected_code=201
|
|
)
|
|
|
|
pk = response.data['pk']
|
|
|
|
item = StockItem.objects.get(pk=pk)
|
|
|
|
self.assertFalse(item.scheduled_for_deletion)
|
|
|
|
# Request deletion via the API
|
|
self.delete(
|
|
reverse('api-stock-detail', kwargs={'pk': pk}),
|
|
expected_code=204
|
|
)
|
|
|
|
# There should be 100x StockItem objects marked for deletion
|
|
self.assertEqual(
|
|
StockItem.objects.filter(scheduled_for_deletion=True).count(),
|
|
10
|
|
)
|
|
|
|
# Perform the actual delete (will take some time)
|
|
delete_old_stock_items()
|
|
|
|
self.assertEqual(
|
|
StockItem.objects.filter(scheduled_for_deletion=True).count(),
|
|
0
|
|
)
|
|
|
|
|
|
class StockTestResultTest(StockAPITestCase):
|
|
|
|
def get_url(self):
|
|
return reverse('api-stock-test-result-list')
|
|
|
|
def test_list(self):
|
|
|
|
url = self.get_url()
|
|
response = self.client.get(url)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertGreaterEqual(len(response.data), 4)
|
|
|
|
response = self.client.get(url, data={'stock_item': 105})
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertGreaterEqual(len(response.data), 4)
|
|
|
|
def test_post_fail(self):
|
|
# Attempt to post a new test result without specifying required data
|
|
|
|
url = self.get_url()
|
|
|
|
response = self.client.post(
|
|
url,
|
|
data={
|
|
'test': 'A test',
|
|
'result': True,
|
|
},
|
|
format='json'
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
|
|
# This one should pass!
|
|
response = self.client.post(
|
|
url,
|
|
data={
|
|
'test': 'A test',
|
|
'stock_item': 105,
|
|
'result': True,
|
|
},
|
|
format='json'
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
|
|
|
def test_post(self):
|
|
# Test creation of a new test result
|
|
|
|
url = self.get_url()
|
|
|
|
response = self.client.get(url)
|
|
n = len(response.data)
|
|
|
|
data = {
|
|
'stock_item': 105,
|
|
'test': 'Checked Steam Valve',
|
|
'result': False,
|
|
'value': '150kPa',
|
|
'notes': 'I guess there was just too much pressure?',
|
|
}
|
|
|
|
response = self.client.post(url, data, format='json')
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
|
|
|
response = self.client.get(url)
|
|
self.assertEqual(len(response.data), n + 1)
|
|
|
|
# And read out again
|
|
response = self.client.get(url, data={'test': 'Checked Steam Valve'})
|
|
|
|
self.assertEqual(len(response.data), 1)
|
|
|
|
test = response.data[0]
|
|
self.assertEqual(test['value'], '150kPa')
|
|
self.assertEqual(test['user'], self.user.pk)
|
|
|
|
def test_post_bitmap(self):
|
|
"""
|
|
2021-08-25
|
|
|
|
For some (unknown) reason, prior to fix https://github.com/inventree/InvenTree/pull/2018
|
|
uploading a bitmap image would result in a failure.
|
|
|
|
This test has been added to ensure that there is no regression.
|
|
|
|
As a bonus this also tests the file-upload component
|
|
"""
|
|
|
|
here = os.path.dirname(__file__)
|
|
|
|
image_file = os.path.join(here, 'fixtures', 'test_image.bmp')
|
|
|
|
with open(image_file, 'rb') as bitmap:
|
|
|
|
data = {
|
|
'stock_item': 105,
|
|
'test': 'Checked Steam Valve',
|
|
'result': False,
|
|
'value': '150kPa',
|
|
'notes': 'I guess there was just too much pressure?',
|
|
"attachment": bitmap,
|
|
}
|
|
|
|
response = self.client.post(self.get_url(), data)
|
|
|
|
self.assertEqual(response.status_code, 201)
|
|
|
|
# Check that an attachment has been uploaded
|
|
self.assertIsNotNone(response.data['attachment'])
|