Coverage for src/backend/InvenTree/plugin/base/locate/api.py: 47%

1"""API for location plugins.""" 

2 

3from rest_framework import serializers 

4from rest_framework.exceptions import NotFound, ParseError, ValidationError 

5from rest_framework.generics import GenericAPIView 

6from rest_framework.response import Response 

7 

8import InvenTree.permissions 

9from InvenTree.exceptions import log_error 

10from InvenTree.tasks import offload_task 

11from plugin import PluginMixinEnum, registry 

12from plugin.registry import call_plugin_function 

13from stock.models import StockItem, StockLocation 

14 

15 

16class LocatePluginSerializer(serializers.Serializer): 

17 """Serializer for the LocatePluginView API endpoint.""" 

18 

19 plugin = serializers.CharField( 

20 help_text='Plugin to use for location identification' 

21 ) 

22 item = serializers.IntegerField(required=False, help_text='StockItem to identify') 

23 location = serializers.IntegerField( 

24 required=False, help_text='StockLocation to identify' 

25 ) 

26 

27 

28class LocatePluginView(GenericAPIView): 

29 """Endpoint for using a custom plugin to identify or 'locate' a stock item or location.""" 

30 

31 permission_classes = [InvenTree.permissions.IsAuthenticatedOrReadScope] 

32 serializer_class = LocatePluginSerializer 

33 

34 def post(self, request, *args, **kwargs): 

35 """Identify or 'locate' a stock item or location with a plugin.""" 

36 # Check inputs and offload the task to the plugin 

37 # Which plugin to we wish to use? 

38 plugin = request.data.get('plugin', None) 

39 

40 if not plugin: 

41 raise ParseError("'plugin' field must be supplied") 

42 

43 # Check that the plugin exists, and supports the 'locate' mixin 

44 plugins = registry.with_mixin(PluginMixinEnum.LOCATE) 

45 

46 if plugin not in [p.slug for p in plugins]: 46 ↛ 52line 46 didn't jump to line 52 because the condition on line 46 was always true

47 raise ParseError( 

48 f"Plugin '{plugin}' is not installed, or does not support the location mixin" 

49 ) 

50 

51 # StockItem to identify 

52 item_pk = request.data.get('item', None) 

53 

54 # StockLocation to identify 

55 location_pk = request.data.get('location', None) 

56 

57 data = {'success': 'Identification plugin activated', 'plugin': plugin} 

58 

59 # StockItem takes priority 

60 if item_pk: 

61 try: 

62 StockItem.objects.get(pk=item_pk) 

63 

64 offload_task( 

65 call_plugin_function, 

66 plugin, 

67 'locate_stock_item', 

68 item_pk, 

69 group='plugin', 

70 ) 

71 

72 data['item'] = item_pk 

73 

74 return Response(data) 

75 

76 except (ValueError, StockItem.DoesNotExist): 

77 raise NotFound(f"StockItem matching PK '{item_pk}' not found") 

78 except Exception: 

79 log_error('locate_stock_item', plugin=plugin.slug) 

80 return ValidationError('Error locating stock item') 

81 

82 elif location_pk: 

83 try: 

84 StockLocation.objects.get(pk=location_pk) 

85 

86 offload_task( 

87 call_plugin_function, 

88 plugin, 

89 'locate_stock_location', 

90 location_pk, 

91 group='plugin', 

92 ) 

93 

94 data['location'] = location_pk 

95 

96 return Response(data) 

97 

98 except (ValueError, StockLocation.DoesNotExist): 

99 raise NotFound(f"StockLocation matching PK '{location_pk}' not found") 

100 except Exception: 

101 log_error('locate_stock_location', plugin=plugin.slug) 

102 return ValidationError('Error locating stock location') 

103 else: 

104 raise ParseError("Must supply either 'item' or 'location' parameter")