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

1from django.contrib.auth.models import User 

2from django.db import transaction 

3from django.utils import timezone 

4 

5from history.model_data import HistoryEventDict 

6from history.models import HistoryChildOperation, HistoryEvent 

7 

8 

9def get_last_event_query(event: HistoryEventDict): 

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

11 

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

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

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

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

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

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

18 

19 return query 

20 

21 

22@transaction.atomic 

23def insert_history_event(event: HistoryEventDict): 

24 last_event_query = get_last_event_query(event) 

25 last_event = last_event_query.first() 

26 # Legacy compat 

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

28 event["col"] = None 

29 

30 if last_event: 

31 if event.get("status", None) == "ERROR": 

32 last_event_query = last_event_query.filter(status="ERROR") 

33 last_event_query.delete() 

34 

35 user = None 

36 

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

38 userid = event["userid"] 

39 del event["userid"] 

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

41 

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

43 

44 children: list[HistoryChildOperation] = [] 

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

46 tasks = event["children"] 

47 for task in tasks: 

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

49 del event["children"] 

50 

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

52 

53 db_event.save() 

54 

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

56 op.event = db_event 

57 

58 HistoryChildOperation.objects.bulk_create(children) 

59 

60 

61def get_history_error_warning_counts(): 

62 error_count = HistoryEvent.objects.filter(status="ERROR").count() 

63 warning_count = HistoryEvent.objects.filter(status="WARNING").count() 

64 return error_count, warning_count 

65 

66 

67def delete_history_event(pk): 

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

69 

70 

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

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

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

74 last_events = last_events.filter(pid__startswith=pid) 

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

76 

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

78 return last_event 

79 

80 

81def get_gap(now, event): 

82 gap = "" 

83 if event: 

84 timelapse = now - event.created_on 

85 if timelapse.days > 0: 

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

87 elif timelapse.seconds > 3600: 

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

89 else: 

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

91 return gap 

92 

93 

94def get_last_unsolved_error(pid, strict): 

95 if strict: 

96 result = HistoryEvent.objects.filter(status="ERROR", pid=pid).latest("created_on") 

97 else: 

98 result = HistoryEvent.objects.filter(status="ERROR", pid__startswith=pid).latest( 

99 "created_on" 

100 ) 

101 return result