Coverage for src / history / utils.py: 56%

1from django.contrib.auth.models import User 

2from django.db import transaction 

3from django.db.models import Q 

4from django.utils import timezone 

5 

6from history.model_data import HistoryEventDict, HistoryEventStatus 

7from history.models import HistoryChildOperation, HistoryEvent 

8 

9 

10def get_last_event_query(event: HistoryEventDict): 

11 query = HistoryEvent.objects.filter(type=event["type"], pid=event["pid"]) 

12 

13 if "source" in event and event["source"] != "": 13 ↛ 14line 13 didn't jump to line 14 because the condition on line 13 was never true

14 query = query.filter(source=event["source"]) 

15 if "unique_id" in event and event["unique_id"]: 15 ↛ 16line 15 didn't jump to line 16 because the condition on line 15 was never true

16 query = query.filter(unique_id=event["unique_id"]) 

17 if "parent_id" in event: 17 ↛ 18line 17 didn't jump to line 18 because the condition on line 17 was never true

18 query = query.filter(pk=event["parent_id"]) 

19 

20 return query 

21 

22 

23@transaction.atomic 

24def insert_history_event(event: HistoryEventDict): 

25 last_event_query = get_last_event_query(event) 

26 last_event = last_event_query.first() 

27 # Legacy compat 

28 if event["col"] == "ALL": 28 ↛ 29line 28 didn't jump to line 29 because the condition on line 28 was never true

29 event["col"] = None 

30 

31 if last_event: 

32 # Always keep last successfull event 

33 # If current event is an error, only delete previous errors. 

34 if event.get("status", None) in ( 

35 HistoryEventStatus.ERROR, 

36 HistoryEventStatus.PENDING, 

37 ): 

38 last_event_query = last_event_query.filter( 

39 Q(status=HistoryEventStatus.ERROR) | Q(status=HistoryEventStatus.PENDING) 

40 ) 

41 last_event_query.delete() 

42 

43 user = None 

44 

45 if "userid" in event: 45 ↛ 46line 45 didn't jump to line 46 because the condition on line 45 was never true

46 userid = event["userid"] 

47 del event["userid"] 

48 user = User.objects.get(pk=userid) 

49 

50 event.setdefault("created_on", timezone.now()) 

51 

52 children: list[HistoryChildOperation] = [] 

53 if "children" in event: 53 ↛ 54line 53 didn't jump to line 54 because the condition on line 53 was never true

54 tasks = event["children"] 

55 for task in tasks: 

56 children.append(HistoryChildOperation(**task)) 

57 del event["children"] 

58 

59 db_event = HistoryEvent(**event, user=user) 

60 

61 db_event.save() 

62 

63 for op in children: 63 ↛ 64line 63 didn't jump to line 64 because the loop on line 63 never started

64 op.event = db_event 

65 

66 HistoryChildOperation.objects.bulk_create(children) 

67 

68 

69def get_history_error_warning_counts(): 

70 error_count = HistoryEvent.objects.filter(status=HistoryEventStatus.ERROR).count() 

71 warning_count = HistoryEvent.objects.filter(status=HistoryEventStatus.WARNING).count() 

72 return error_count, warning_count 

73 

74 

75def delete_history_event(pk): 

76 HistoryEvent.objects.get(pk=pk).delete() 

77 

78 

79def get_history_last_event_by(type, pid=""): 

80 last_events = HistoryEvent.objects.filter(type=type) 

81 if len(pid) > 0: 81 ↛ 83line 81 didn't jump to line 83 because the condition on line 81 was always true

82 last_events = last_events.filter(pid__startswith=pid) 

83 last_events = last_events.order_by("-pk")[:1] 

84 

85 last_event = last_events[0] if len(last_events) > 0 else None 

86 return last_event 

87 

88 

89def get_gap(now, event): 

90 gap = "" 

91 if event: 

92 timelapse = now - event.created_on 

93 if timelapse.days > 0: 

94 gap = str(timelapse.days) + " days ago" 

95 elif timelapse.seconds > 3600: 

96 gap = str(int(timelapse.seconds // 3600)) + " hours ago" 

97 else: 

98 gap = str(int(timelapse.seconds // 60)) + " minutes ago" 

99 return gap 

100 

101 

102def get_last_unsolved_error(pid, strict): 

103 result = HistoryEvent.objects.filter(status=HistoryEventStatus.ERROR) 

104 if strict: 

105 return result.filter(pid=pid).latest("created_on") 

106 return result.filter(pid__startswith=pid).latest("created_on")