Coverage for core/tables/jobs.py: 81%

1import django_tables2 as tables 

2from django.utils.html import conditional_escape 

3from django.utils.safestring import mark_safe 

4from django.utils.translation import gettext_lazy as _ 

5 

6from core.constants import JOB_LOG_ENTRY_LEVELS 

7from core.models import Job 

8from core.tables.columns import BadgeColumn 

9from netbox.tables import BaseTable, NetBoxTable, columns 

10from utilities.string import humanize_duration 

11 

12 

13class JobTable(NetBoxTable): 

14 id = tables.Column( 

15 verbose_name=_('ID'), 

16 linkify=True 

17 ) 

18 name = tables.Column( 

19 verbose_name=_('Name'), 

20 linkify=True 

21 ) 

22 object_type = columns.ContentTypeColumn( 

23 verbose_name=_('Type') 

24 ) 

25 object = tables.Column( 

26 verbose_name=_('Object'), 

27 linkify=True, 

28 orderable=False 

29 ) 

30 status = columns.ChoiceFieldColumn( 

31 verbose_name=_('Status'), 

32 ) 

33 created = columns.DateTimeColumn( 

34 verbose_name=_('Created'), 

35 ) 

36 scheduled = columns.DateTimeColumn( 

37 verbose_name=_('Scheduled'), 

38 ) 

39 interval = columns.DurationColumn( 

40 verbose_name=_('Interval'), 

41 ) 

42 started = columns.DateTimeColumn( 

43 verbose_name=_('Started'), 

44 ) 

45 completed = columns.DateTimeColumn( 

46 verbose_name=_('Completed'), 

47 ) 

48 execution_time = tables.Column( 

49 verbose_name=_('Execution Time'), 

50 ) 

51 queue_name = tables.Column( 

52 verbose_name=_('Queue'), 

53 ) 

54 log_entries = tables.Column( 

55 verbose_name=_('Log Entries'), 

56 ) 

57 actions = columns.ActionsColumn( 

58 actions=('delete',) 

59 ) 

60 

61 class Meta(NetBoxTable.Meta): 

62 model = Job 

63 fields = ( 

64 'pk', 'id', 'object_type', 'object', 'name', 'status', 'created', 'scheduled', 'interval', 'started', 

65 'completed', 'execution_time', 'user', 'queue_name', 'log_entries', 'error', 'job_id', 

66 ) 

67 default_columns = ( 

68 'pk', 'id', 'object_type', 'object', 'name', 'status', 'created', 'started', 'completed', 'execution_time', 

69 'user', 

70 ) 

71 

72 def render_log_entries(self, value): 

73 return len(value) 

74 

75 def render_execution_time(self, value): 

76 return humanize_duration(value) 

77 

78 def value_execution_time(self, value): 

79 # Export the recorded execution time as a raw number of seconds, rather than the humanized 

80 # string, as an export is intended for analysis 

81 return round(value.total_seconds(), 3) 

82 

83 

84class JobLogEntryTable(BaseTable): 

85 timestamp = columns.DateTimeColumn( 

86 timespec='milliseconds', 

87 verbose_name=_('Time'), 

88 ) 

89 level = BadgeColumn( 

90 badges=JOB_LOG_ENTRY_LEVELS, 

91 verbose_name=_('Level'), 

92 ) 

93 message = tables.Column( 

94 verbose_name=_('Message'), 

95 ) 

96 

97 class Meta(BaseTable.Meta): 

98 empty_text = _('No log entries') 

99 fields = ('timestamp', 'level', 'message') 

100 

101 def render_message(self, record, value): 

102 if record.get('level') == 'error' and '\n' in value: 

103 value = conditional_escape(value) 

104 return mark_safe(f'<pre class="p-0">{value}</pre>') 

105 return value