Coverage for src/backend/InvenTree/stock/migrations/0108_auto_20240219_0252.py: 24%

1# Generated by Django 4.2.10 on 2024-02-19 02:52 

2 

3from django.db import migrations 

4from django.db.models import F, OuterRef, Subquery, IntegerField 

5 

6 

7def update_templates(apps, schema_editor): 

8 """Run data migration to fix potentially mis-applied data migration. 

9  

10 Ref: https://github.com/inventree/InvenTree/pull/6514 

11 

12 The previous data migration (stock.0106_auto_20240207_0353) had a bug, 

13 where it would look for any matching PartTestTemplate objects for a given StockItemTestResult, 

14 as long as the "part tree ID" was the same. 

15 

16 However, if the template was defined for a part on a different *branch* of the tree, 

17 the wrong template could be applied. 

18 

19 This is really only the case where the user has a very complex set of nested part variants, 

20 but still there is a potential for a mis-match. 

21 

22 This data migration will attempt to fix any mis-applied templates. 

23 """ 

24 

25 PartTestTemplate = apps.get_model('part', 'PartTestTemplate') 

26 StockItemTestResult = apps.get_model('stock', 'StockItemTestResult') 

27 

28 # Find any StockItemTestResult objects which match a "bad" template 

29 # Here a "bad" template points to a Part which is not *above* the part in the tree 

30 bad_results = StockItemTestResult.objects.exclude( 

31 stock_item__part__tree_id=F('template__part__tree_id'), 

32 stock_item__part__lft__gte=F('template__part__lft'), 

33 stock_item__part__rght__lte=F('template__part__rght'), 

34 ) 

35 

36 n = bad_results.count() 

37 

38 if n == 0: 38 ↛ 42line 38 didn't jump to line 42 because the condition on line 38 was always true

39 # Escape early - no bad results! 

40 return 

41 

42 print(f"Found {n} StockItemTestResult objects with bad templates...") 

43 

44 # For each bad result, attempt to find a matching template 

45 # Here, a matching template must point to a part *above* the part in the tree 

46 # Annotate the queryset with a "matching template" 

47 

48 template_query = PartTestTemplate.objects.filter( 

49 part__tree_id=OuterRef('stock_item__part__tree_id'), 

50 part__lft__lte=OuterRef('stock_item__part__lft'), 

51 part__rght__gte=OuterRef('stock_item__part__rght'), 

52 key=OuterRef('template__key') 

53 ).order_by('part__level').values('pk') 

54 

55 bad_results = bad_results.annotate( 

56 matching_template=Subquery(template_query[:1], output_field=IntegerField()) 

57 ) 

58 

59 # Update the results for which we have a "good" matching template 

60 matching_results = bad_results.filter(matching_template__isnull=False) 

61 missing_results = bad_results.filter(matching_template__isnull=True) 

62 

63 results_to_update = [] 

64 

65 for result in matching_results: 

66 if result.template.pk != result.matching_template: 

67 result.template = PartTestTemplate.objects.get(pk=result.matching_template) 

68 results_to_update.append(result) 

69 

70 if len(results_to_update) > 0: 

71 # Update any results which point to the wrong template, but have a matching template 

72 print("Updating", len(results_to_update), "matching templates...") 

73 StockItemTestResult.objects.bulk_update(results_to_update, ['template']) 

74 

75 results_to_update = [] 

76 

77 # For the remaining results, we need to create a new template 

78 for result in missing_results: 

79 # Check that a template does *not* exist already 

80 if template := PartTestTemplate.objects.filter( 

81 part__tree_id=result.stock_item.part.tree_id, 

82 part__lft__lte=result.stock_item.part.lft, 

83 part__rght__gte=result.stock_item.part.rght, 

84 key=result.template.key 

85 ).first(): 

86 pass 

87 else: 

88 # Create a new template (by copying the old one) 

89 template = result.template 

90 template.part = result.stock_item.part 

91 template.pk = None 

92 template.save() 

93 template.refresh_from_db() 

94 

95 result.template = template 

96 results_to_update.append(result) 

97 

98 if len(results_to_update) > 0: 

99 print("Updating", len(results_to_update), "missing templates...") 

100 StockItemTestResult.objects.bulk_update(results_to_update, ['template']) 

101 

102 # Finall, check that there are no longer any "bad" results 

103 assert(bad_results.order_by('pk').count() == 0) 

104 

105 

106class Migration(migrations.Migration): 

107 

108 dependencies = [ 

109 ('stock', '0107_remove_stockitemtestresult_test_and_more'), 

110 ] 

111 

112 operations = [ 

113 migrations.RunPython(update_templates, reverse_code=migrations.RunPython.noop) 

114 ]