2 import { expect } from 'chai'
3 import { wait } from '@shared/core-utils'
4 import { JobState, JobType } from '../../models'
5 import { PeerTubeServer } from './server'
7 async function waitJobs (
8 serversArg: PeerTubeServer[] | PeerTubeServer,
10 skipDelayed?: boolean // default false
13 const { skipDelayed = false } = options
15 const pendingJobWait = process.env.NODE_PENDING_JOB_WAIT
16 ? parseInt(process.env.NODE_PENDING_JOB_WAIT, 10)
19 let servers: PeerTubeServer[]
21 if (Array.isArray(serversArg) === false) servers = [ serversArg as PeerTubeServer ]
22 else servers = serversArg as PeerTubeServer[]
24 const states: JobState[] = [ 'waiting', 'active' ]
25 if (!skipDelayed) states.push('delayed')
27 const repeatableJobs: JobType[] = [ 'videos-views-stats', 'activitypub-cleaner' ]
28 let pendingRequests: boolean
30 function tasksBuilder () {
31 const tasks: Promise<any>[] = []
33 // Check if each server has pending request
34 for (const server of servers) {
35 for (const state of states) {
36 const p = server.jobs.list({
41 }).then(body => body.data)
42 .then(jobs => jobs.filter(j => !repeatableJobs.includes(j.type)))
44 if (jobs.length !== 0) {
45 pendingRequests = true
52 const p = server.debug.getDebug()
54 if (obj.activityPubMessagesWaiting !== 0) {
55 pendingRequests = true
66 pendingRequests = false
67 await Promise.all(tasksBuilder())
69 // Retry, in case of new jobs were created
70 if (pendingRequests === false) {
71 await wait(pendingJobWait)
72 await Promise.all(tasksBuilder())
75 if (pendingRequests) {
76 await wait(pendingJobWait)
78 } while (pendingRequests)
81 async function expectNoFailedTranscodingJob (server: PeerTubeServer) {
82 const { data } = await server.jobs.listFailed({ jobType: 'video-transcoding' })
83 expect(data).to.have.lengthOf(0)
86 // ---------------------------------------------------------------------------
90 expectNoFailedTranscodingJob