Coverage for core/tables/tasks.py: 92%

1import django_tables2 as tables 

2from django.utils.translation import gettext_lazy as _ 

3from django_tables2.utils import A 

4 

5from core.constants import RQ_TASK_STATUSES 

6from core.tables.columns import BadgeColumn 

7from netbox.tables import BaseTable, columns 

8 

9 

10class BackgroundQueueTable(BaseTable): 

11 name = tables.Column( 

12 verbose_name=_("Name") 

13 ) 

14 jobs = tables.Column( 

15 linkify=("core:background_task_list", [A("index"), "queued"]), 

16 verbose_name=_("Queued") 

17 ) 

18 oldest_job_timestamp = tables.Column( 

19 verbose_name=_("Oldest Task") 

20 ) 

21 started_jobs = tables.Column( 

22 linkify=("core:background_task_list", [A("index"), "started"]), 

23 verbose_name=_("Active") 

24 ) 

25 deferred_jobs = tables.Column( 

26 linkify=("core:background_task_list", [A("index"), "deferred"]), 

27 verbose_name=_("Deferred") 

28 ) 

29 finished_jobs = tables.Column( 

30 linkify=("core:background_task_list", [A("index"), "finished"]), 

31 verbose_name=_("Finished") 

32 ) 

33 failed_jobs = tables.Column( 

34 linkify=("core:background_task_list", [A("index"), "failed"]), 

35 verbose_name=_("Failed") 

36 ) 

37 scheduled_jobs = tables.Column( 

38 linkify=("core:background_task_list", [A("index"), "scheduled"]), 

39 verbose_name=_("Scheduled") 

40 ) 

41 workers = tables.Column( 

42 linkify=("core:worker_list", [A("index")]), 

43 verbose_name=_("Workers") 

44 ) 

45 host = tables.Column( 

46 accessor="connection_kwargs__host", 

47 verbose_name=_("Host") 

48 ) 

49 port = tables.Column( 

50 accessor="connection_kwargs__port", 

51 verbose_name=_("Port") 

52 ) 

53 db = tables.Column( 

54 accessor="connection_kwargs__db", 

55 verbose_name=_("DB") 

56 ) 

57 pid = tables.Column( 

58 accessor="scheduler__pid", 

59 verbose_name=_("Scheduler PID") 

60 ) 

61 

62 class Meta(BaseTable.Meta): 

63 empty_text = _('No queues found') 

64 fields = ( 

65 'name', 'jobs', 'oldest_job_timestamp', 'started_jobs', 'deferred_jobs', 'finished_jobs', 'failed_jobs', 

66 'scheduled_jobs', 'workers', 'host', 'port', 'db', 'pid', 

67 ) 

68 default_columns = ( 

69 'name', 'jobs', 'started_jobs', 'deferred_jobs', 'finished_jobs', 'failed_jobs', 'scheduled_jobs', 

70 'workers', 

71 ) 

72 

73 

74class BackgroundTaskTable(BaseTable): 

75 id = tables.Column( 

76 linkify=("core:background_task", [A("id")]), 

77 verbose_name=_("ID") 

78 ) 

79 created_at = columns.DateTimeColumn( 

80 verbose_name=_("Created") 

81 ) 

82 enqueued_at = columns.DateTimeColumn( 

83 verbose_name=_("Enqueued") 

84 ) 

85 ended_at = columns.DateTimeColumn( 

86 verbose_name=_("Ended") 

87 ) 

88 status = BadgeColumn( 

89 badges=RQ_TASK_STATUSES, 

90 verbose_name=_("Status"), 

91 accessor='get_status' 

92 ) 

93 callable = tables.Column( 

94 empty_values=(), 

95 verbose_name=_("Callable") 

96 ) 

97 

98 class Meta(BaseTable.Meta): 

99 empty_text = _('No tasks found') 

100 fields = ( 

101 'id', 'created_at', 'enqueued_at', 'ended_at', 'status', 'callable', 

102 ) 

103 default_columns = ( 

104 'id', 'created_at', 'enqueued_at', 'ended_at', 'status', 'callable', 

105 ) 

106 

107 def render_callable(self, value, record): 

108 try: 

109 return record.func_name 

110 except Exception as e: 

111 return repr(e) 

112 

113 

114class WorkerTable(BaseTable): 

115 name = tables.Column( 

116 linkify=("core:worker", [A("name")]), 

117 verbose_name=_("Name") 

118 ) 

119 state = tables.Column( 

120 verbose_name=_("State") 

121 ) 

122 birth_date = columns.DateTimeColumn( 

123 verbose_name=_("Birth") 

124 ) 

125 pid = tables.Column( 

126 verbose_name=_("PID") 

127 ) 

128 

129 class Meta(BaseTable.Meta): 

130 empty_text = _('No workers found') 

131 fields = ( 

132 'name', 'state', 'birth_date', 'pid', 

133 ) 

134 default_columns = ( 

135 'name', 'state', 'birth_date', 'pid', 

136 )