1 files changed, 541 insertions, 0 deletions

diff --git a/server/tests/api/runners/runner-vod-transcoding.ts b/server/tests/api/runners/runner-vod-transcoding.ts
new file mode 100644
index 000000000..92a47ac3b
--- /dev/null
+++ b/server/tests/api/runners/runner-vod-transcoding.ts
@@ -0,0 +1,541 @@
+/* eslint-disable @typescript-eslint/no-unused-expressions,@typescript-eslint/require-await */
+
+import { expect } from 'chai'
+import { readFile } from 'fs-extra'
+import { completeCheckHlsPlaylist } from '@server/tests/shared'
+import { buildAbsoluteFixturePath } from '@shared/core-utils'
+import {
+  HttpStatusCode,
+  RunnerJobSuccessPayload,
+  RunnerJobVODAudioMergeTranscodingPayload,
+  RunnerJobVODHLSTranscodingPayload,
+  RunnerJobVODPayload,
+  RunnerJobVODWebVideoTranscodingPayload,
+  VideoState,
+  VODAudioMergeTranscodingSuccess,
+  VODHLSTranscodingSuccess,
+  VODWebVideoTranscodingSuccess
+} from '@shared/models'
+import {
+  cleanupTests,
+  createMultipleServers,
+  doubleFollow,
+  makeGetRequest,
+  makeRawRequest,
+  PeerTubeServer,
+  setAccessTokensToServers,
+  setDefaultVideoChannel,
+  waitJobs
+} from '@shared/server-commands'
+
+async function processAllJobs (server: PeerTubeServer, runnerToken: string) {
+  do {
+    const { availableJobs } = await server.runnerJobs.requestVOD({ runnerToken })
+    if (availableJobs.length === 0) break
+
+    const { job } = await server.runnerJobs.accept<RunnerJobVODPayload>({ runnerToken, jobUUID: availableJobs[0].uuid })
+
+    const payload: RunnerJobSuccessPayload = {
+      videoFile: `video_short_${job.payload.output.resolution}p.mp4`,
+      resolutionPlaylistFile: `video_short_${job.payload.output.resolution}p.m3u8`
+    }
+    await server.runnerJobs.success({ runnerToken, jobUUID: job.uuid, jobToken: job.jobToken, payload })
+  } while (true)
+
+  await waitJobs([ server ])
+}
+
+describe('Test runner VOD transcoding', function () {
+  let servers: PeerTubeServer[] = []
+  let runnerToken: string
+
+  before(async function () {
+    this.timeout(120_000)
+
+    servers = await createMultipleServers(2)
+
+    await setAccessTokensToServers(servers)
+    await setDefaultVideoChannel(servers)
+
+    await doubleFollow(servers[0], servers[1])
+
+    await servers[0].config.enableRemoteTranscoding()
+    runnerToken = await servers[0].runners.autoRegisterRunner()
+  })
+
+  describe('Without transcoding', function () {
+
+    before(async function () {
+      this.timeout(60000)
+
+      await servers[0].config.disableTranscoding()
+      await servers[0].videos.quickUpload({ name: 'video' })
+
+      await waitJobs(servers)
+    })
+
+    it('Should not have available jobs', async function () {
+      const { availableJobs } = await servers[0].runnerJobs.requestVOD({ runnerToken })
+      expect(availableJobs).to.have.lengthOf(0)
+    })
+  })
+
+  describe('With classic transcoding enabled', function () {
+
+    before(async function () {
+      this.timeout(60000)
+
+      await servers[0].config.enableTranscoding(true, true)
+    })
+
+    it('Should error a transcoding job', async function () {
+      this.timeout(60000)
+
+      await servers[0].runnerJobs.cancelAllJobs()
+      const { uuid } = await servers[0].videos.quickUpload({ name: 'video' })
+      await waitJobs(servers)
+
+      const { availableJobs } = await servers[0].runnerJobs.request({ runnerToken })
+      const jobUUID = availableJobs[0].uuid
+
+      const { job } = await servers[0].runnerJobs.accept({ runnerToken, jobUUID })
+      const jobToken = job.jobToken
+
+      await servers[0].runnerJobs.error({ runnerToken, jobUUID, jobToken, message: 'Error' })
+
+      const video = await servers[0].videos.get({ id: uuid })
+      expect(video.state.id).to.equal(VideoState.TRANSCODING_FAILED)
+    })
+
+    it('Should cancel a transcoding job', async function () {
+      await servers[0].runnerJobs.cancelAllJobs()
+      const { uuid } = await servers[0].videos.quickUpload({ name: 'video' })
+      await waitJobs(servers)
+
+      const { availableJobs } = await servers[0].runnerJobs.request({ runnerToken })
+      const jobUUID = availableJobs[0].uuid
+
+      await servers[0].runnerJobs.cancelByAdmin({ jobUUID })
+
+      const video = await servers[0].videos.get({ id: uuid })
+      expect(video.state.id).to.equal(VideoState.PUBLISHED)
+    })
+  })
+
+  describe('Web video transcoding only', function () {
+    let videoUUID: string
+    let jobToken: string
+    let jobUUID: string
+
+    before(async function () {
+      this.timeout(60000)
+
+      await servers[0].runnerJobs.cancelAllJobs()
+      await servers[0].config.enableTranscoding(true, false)
+
+      const { uuid } = await servers[0].videos.quickUpload({ name: 'web video', fixture: 'video_short.webm' })
+      videoUUID = uuid
+
+      await waitJobs(servers)
+    })
+
+    it('Should have jobs available for remote runners', async function () {
+      const { availableJobs } = await servers[0].runnerJobs.requestVOD({ runnerToken })
+      expect(availableJobs).to.have.lengthOf(1)
+
+      jobUUID = availableJobs[0].uuid
+    })
+
+    it('Should have a valid first transcoding job', async function () {
+      const { job } = await servers[0].runnerJobs.accept<RunnerJobVODWebVideoTranscodingPayload>({ runnerToken, jobUUID })
+      jobToken = job.jobToken
+
+      expect(job.type === 'vod-web-video-transcoding')
+      expect(job.payload.input.videoFileUrl).to.exist
+      expect(job.payload.output.resolution).to.equal(720)
+      expect(job.payload.output.fps).to.equal(25)
+
+      const { body } = await servers[0].runnerJobs.getInputFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
+      const inputFile = await readFile(buildAbsoluteFixturePath('video_short.webm'))
+
+      expect(body).to.deep.equal(inputFile)
+    })
+
+    it('Should transcode the max video resolution and send it back to the server', async function () {
+      this.timeout(60000)
+
+      const payload: VODWebVideoTranscodingSuccess = {
+        videoFile: 'video_short.mp4'
+      }
+      await servers[0].runnerJobs.success({ runnerToken, jobUUID, jobToken, payload })
+
+      await waitJobs(servers)
+    })
+
+    it('Should have the video updated', async function () {
+      for (const server of servers) {
+        const video = await server.videos.get({ id: videoUUID })
+        expect(video.files).to.have.lengthOf(1)
+        expect(video.streamingPlaylists).to.have.lengthOf(0)
+
+        const { body } = await makeRawRequest({ url: video.files[0].fileUrl, expectedStatus: HttpStatusCode.OK_200 })
+        expect(body).to.deep.equal(await readFile(buildAbsoluteFixturePath('video_short.mp4')))
+      }
+    })
+
+    it('Should have 4 lower resolution to transcode', async function () {
+      const { availableJobs } = await servers[0].runnerJobs.requestVOD({ runnerToken })
+      expect(availableJobs).to.have.lengthOf(4)
+
+      for (const resolution of [ 480, 360, 240, 144 ]) {
+        const job = availableJobs.find(j => j.payload.output.resolution === resolution)
+        expect(job).to.exist
+        expect(job.type).to.equal('vod-web-video-transcoding')
+
+        if (resolution === 240) jobUUID = job.uuid
+      }
+    })
+
+    it('Should process one of these transcoding jobs', async function () {
+      const { job } = await servers[0].runnerJobs.accept<RunnerJobVODWebVideoTranscodingPayload>({ runnerToken, jobUUID })
+      jobToken = job.jobToken
+
+      const { body } = await servers[0].runnerJobs.getInputFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
+      const inputFile = await readFile(buildAbsoluteFixturePath('video_short.mp4'))
+
+      expect(body).to.deep.equal(inputFile)
+
+      const payload: VODWebVideoTranscodingSuccess = { videoFile: 'video_short_240p.mp4' }
+      await servers[0].runnerJobs.success({ runnerToken, jobUUID, jobToken, payload })
+    })
+
+    it('Should process all other jobs', async function () {
+      const { availableJobs } = await servers[0].runnerJobs.requestVOD({ runnerToken })
+      expect(availableJobs).to.have.lengthOf(3)
+
+      for (const resolution of [ 480, 360, 144 ]) {
+        const availableJob = availableJobs.find(j => j.payload.output.resolution === resolution)
+        expect(availableJob).to.exist
+        jobUUID = availableJob.uuid
+
+        const { job } = await servers[0].runnerJobs.accept<RunnerJobVODWebVideoTranscodingPayload>({ runnerToken, jobUUID })
+        jobToken = job.jobToken
+
+        const { body } = await servers[0].runnerJobs.getInputFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
+        const inputFile = await readFile(buildAbsoluteFixturePath('video_short.mp4'))
+        expect(body).to.deep.equal(inputFile)
+
+        const payload: VODWebVideoTranscodingSuccess = { videoFile: `video_short_${resolution}p.mp4` }
+        await servers[0].runnerJobs.success({ runnerToken, jobUUID, jobToken, payload })
+      }
+    })
+
+    it('Should have the video updated', async function () {
+      for (const server of servers) {
+        const video = await server.videos.get({ id: videoUUID })
+        expect(video.files).to.have.lengthOf(5)
+        expect(video.streamingPlaylists).to.have.lengthOf(0)
+
+        const { body } = await makeRawRequest({ url: video.files[0].fileUrl, expectedStatus: HttpStatusCode.OK_200 })
+        expect(body).to.deep.equal(await readFile(buildAbsoluteFixturePath('video_short.mp4')))
+
+        for (const file of video.files) {
+          await makeRawRequest({ url: file.fileUrl, expectedStatus: HttpStatusCode.OK_200 })
+          await makeRawRequest({ url: file.torrentUrl, expectedStatus: HttpStatusCode.OK_200 })
+        }
+      }
+    })
+
+    it('Should not have available jobs anymore', async function () {
+      const { availableJobs } = await servers[0].runnerJobs.requestVOD({ runnerToken })
+      expect(availableJobs).to.have.lengthOf(0)
+    })
+  })
+
+  describe('HLS transcoding only', function () {
+    let videoUUID: string
+    let jobToken: string
+    let jobUUID: string
+
+    before(async function () {
+      this.timeout(60000)
+
+      await servers[0].config.enableTranscoding(false, true)
+
+      const { uuid } = await servers[0].videos.quickUpload({ name: 'hls video', fixture: 'video_short.webm' })
+      videoUUID = uuid
+
+      await waitJobs(servers)
+    })
+
+    it('Should run the optimize job', async function () {
+      this.timeout(60000)
+
+      await servers[0].runnerJobs.autoProcessWebVideoJob(runnerToken)
+    })
+
+    it('Should have 5 HLS resolution to transcode', async function () {
+      const { availableJobs } = await servers[0].runnerJobs.requestVOD({ runnerToken })
+      expect(availableJobs).to.have.lengthOf(5)
+
+      for (const resolution of [ 720, 480, 360, 240, 144 ]) {
+        const job = availableJobs.find(j => j.payload.output.resolution === resolution)
+        expect(job).to.exist
+        expect(job.type).to.equal('vod-hls-transcoding')
+
+        if (resolution === 480) jobUUID = job.uuid
+      }
+    })
+
+    it('Should process one of these transcoding jobs', async function () {
+      this.timeout(60000)
+
+      const { job } = await servers[0].runnerJobs.accept<RunnerJobVODHLSTranscodingPayload>({ runnerToken, jobUUID })
+      jobToken = job.jobToken
+
+      const { body } = await servers[0].runnerJobs.getInputFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
+      const inputFile = await readFile(buildAbsoluteFixturePath('video_short.mp4'))
+
+      expect(body).to.deep.equal(inputFile)
+
+      const payload: VODHLSTranscodingSuccess = {
+        videoFile: 'video_short_480p.mp4',
+        resolutionPlaylistFile: 'video_short_480p.m3u8'
+      }
+      await servers[0].runnerJobs.success({ runnerToken, jobUUID, jobToken, payload })
+
+      await waitJobs(servers)
+    })
+
+    it('Should have the video updated', async function () {
+      for (const server of servers) {
+        const video = await server.videos.get({ id: videoUUID })
+
+        expect(video.files).to.have.lengthOf(1)
+        expect(video.streamingPlaylists).to.have.lengthOf(1)
+
+        const hls = video.streamingPlaylists[0]
+        expect(hls.files).to.have.lengthOf(1)
+
+        await completeCheckHlsPlaylist({ videoUUID, hlsOnly: false, servers, resolutions: [ 480 ] })
+      }
+    })
+
+    it('Should process all other jobs', async function () {
+      this.timeout(60000)
+
+      const { availableJobs } = await servers[0].runnerJobs.requestVOD({ runnerToken })
+      expect(availableJobs).to.have.lengthOf(4)
+
+      let maxQualityFile = 'video_short.mp4'
+
+      for (const resolution of [ 720, 360, 240, 144 ]) {
+        const availableJob = availableJobs.find(j => j.payload.output.resolution === resolution)
+        expect(availableJob).to.exist
+        jobUUID = availableJob.uuid
+
+        const { job } = await servers[0].runnerJobs.accept<RunnerJobVODHLSTranscodingPayload>({ runnerToken, jobUUID })
+        jobToken = job.jobToken
+
+        const { body } = await servers[0].runnerJobs.getInputFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
+        const inputFile = await readFile(buildAbsoluteFixturePath(maxQualityFile))
+        expect(body).to.deep.equal(inputFile)
+
+        const payload: VODHLSTranscodingSuccess = {
+          videoFile: `video_short_${resolution}p.mp4`,
+          resolutionPlaylistFile: `video_short_${resolution}p.m3u8`
+        }
+        await servers[0].runnerJobs.success({ runnerToken, jobUUID, jobToken, payload })
+
+        if (resolution === 720) {
+          maxQualityFile = 'video_short_720p.mp4'
+        }
+      }
+
+      await waitJobs(servers)
+    })
+
+    it('Should have the video updated', async function () {
+      for (const server of servers) {
+        const video = await server.videos.get({ id: videoUUID })
+
+        expect(video.files).to.have.lengthOf(0)
+        expect(video.streamingPlaylists).to.have.lengthOf(1)
+
+        const hls = video.streamingPlaylists[0]
+        expect(hls.files).to.have.lengthOf(5)
+
+        await completeCheckHlsPlaylist({ videoUUID, hlsOnly: true, servers, resolutions: [ 720, 480, 360, 240, 144 ] })
+      }
+    })
+
+    it('Should not have available jobs anymore', async function () {
+      const { availableJobs } = await servers[0].runnerJobs.requestVOD({ runnerToken })
+      expect(availableJobs).to.have.lengthOf(0)
+    })
+  })
+
+  describe('Web video and HLS transcoding', function () {
+
+    before(async function () {
+      this.timeout(60000)
+
+      await servers[0].config.enableTranscoding(true, true)
+
+      await servers[0].videos.quickUpload({ name: 'web video and hls video', fixture: 'video_short.webm' })
+
+      await waitJobs(servers)
+    })
+
+    it('Should process the first optimize job', async function () {
+      this.timeout(60000)
+
+      await servers[0].runnerJobs.autoProcessWebVideoJob(runnerToken)
+    })
+
+    it('Should have 9 jobs to process', async function () {
+      const { availableJobs } = await servers[0].runnerJobs.requestVOD({ runnerToken })
+
+      expect(availableJobs).to.have.lengthOf(9)
+
+      const webVideoJobs = availableJobs.filter(j => j.type === 'vod-web-video-transcoding')
+      const hlsJobs = availableJobs.filter(j => j.type === 'vod-hls-transcoding')
+
+      expect(webVideoJobs).to.have.lengthOf(4)
+      expect(hlsJobs).to.have.lengthOf(5)
+    })
+
+    it('Should process all available jobs', async function () {
+      await processAllJobs(servers[0], runnerToken)
+    })
+  })
+
+  describe('Audio merge transcoding', function () {
+    let videoUUID: string
+    let jobToken: string
+    let jobUUID: string
+
+    before(async function () {
+      this.timeout(60000)
+
+      await servers[0].config.enableTranscoding(true, true)
+
+      const attributes = { name: 'audio_with_preview', previewfile: 'preview.jpg', fixture: 'sample.ogg' }
+      const { uuid } = await servers[0].videos.upload({ attributes, mode: 'legacy' })
+      videoUUID = uuid
+
+      await waitJobs(servers)
+    })
+
+    it('Should have an audio merge transcoding job', async function () {
+      const { availableJobs } = await servers[0].runnerJobs.requestVOD({ runnerToken })
+      expect(availableJobs).to.have.lengthOf(1)
+
+      expect(availableJobs[0].type).to.equal('vod-audio-merge-transcoding')
+
+      jobUUID = availableJobs[0].uuid
+    })
+
+    it('Should have a valid remote audio merge transcoding job', async function () {
+      const { job } = await servers[0].runnerJobs.accept<RunnerJobVODAudioMergeTranscodingPayload>({ runnerToken, jobUUID })
+      jobToken = job.jobToken
+
+      expect(job.type === 'vod-audio-merge-transcoding')
+      expect(job.payload.input.audioFileUrl).to.exist
+      expect(job.payload.input.previewFileUrl).to.exist
+      expect(job.payload.output.resolution).to.equal(480)
+
+      {
+        const { body } = await servers[0].runnerJobs.getInputFile({ url: job.payload.input.audioFileUrl, jobToken, runnerToken })
+        const inputFile = await readFile(buildAbsoluteFixturePath('sample.ogg'))
+        expect(body).to.deep.equal(inputFile)
+      }
+
+      {
+        const { body } = await servers[0].runnerJobs.getInputFile({ url: job.payload.input.previewFileUrl, jobToken, runnerToken })
+
+        const video = await servers[0].videos.get({ id: videoUUID })
+        const { body: inputFile } = await makeGetRequest({
+          url: servers[0].url,
+          path: video.previewPath,
+          expectedStatus: HttpStatusCode.OK_200
+        })
+
+        expect(body).to.deep.equal(inputFile)
+      }
+    })
+
+    it('Should merge the audio', async function () {
+      this.timeout(60000)
+
+      const payload: VODAudioMergeTranscodingSuccess = { videoFile: 'video_short_480p.mp4' }
+      await servers[0].runnerJobs.success({ runnerToken, jobUUID, jobToken, payload })
+
+      await waitJobs(servers)
+    })
+
+    it('Should have the video updated', async function () {
+      for (const server of servers) {
+        const video = await server.videos.get({ id: videoUUID })
+        expect(video.files).to.have.lengthOf(1)
+        expect(video.streamingPlaylists).to.have.lengthOf(0)
+
+        const { body } = await makeRawRequest({ url: video.files[0].fileUrl, expectedStatus: HttpStatusCode.OK_200 })
+        expect(body).to.deep.equal(await readFile(buildAbsoluteFixturePath('video_short_480p.mp4')))
+      }
+    })
+
+    it('Should have 7 lower resolutions to transcode', async function () {
+      const { availableJobs } = await servers[0].runnerJobs.requestVOD({ runnerToken })
+      expect(availableJobs).to.have.lengthOf(7)
+
+      for (const resolution of [ 360, 240, 144 ]) {
+        const jobs = availableJobs.filter(j => j.payload.output.resolution === resolution)
+        expect(jobs).to.have.lengthOf(2)
+      }
+
+      jobUUID = availableJobs.find(j => j.payload.output.resolution === 480).uuid
+    })
+
+    it('Should process one other job', async function () {
+      this.timeout(60000)
+
+      const { job } = await servers[0].runnerJobs.accept<RunnerJobVODHLSTranscodingPayload>({ runnerToken, jobUUID })
+      jobToken = job.jobToken
+
+      const { body } = await servers[0].runnerJobs.getInputFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
+      const inputFile = await readFile(buildAbsoluteFixturePath('video_short_480p.mp4'))
+      expect(body).to.deep.equal(inputFile)
+
+      const payload: VODHLSTranscodingSuccess = {
+        videoFile: `video_short_480p.mp4`,
+        resolutionPlaylistFile: `video_short_480p.m3u8`
+      }
+      await servers[0].runnerJobs.success({ runnerToken, jobUUID, jobToken, payload })
+
+      await waitJobs(servers)
+    })
+
+    it('Should have the video updated', async function () {
+      for (const server of servers) {
+        const video = await server.videos.get({ id: videoUUID })
+
+        expect(video.files).to.have.lengthOf(1)
+        expect(video.streamingPlaylists).to.have.lengthOf(1)
+
+        const hls = video.streamingPlaylists[0]
+        expect(hls.files).to.have.lengthOf(1)
+
+        await completeCheckHlsPlaylist({ videoUUID, hlsOnly: false, servers, resolutions: [ 480 ] })
+      }
+    })
+
+    it('Should process all available jobs', async function () {
+      await processAllJobs(servers[0], runnerToken)
+    })
+  })
+
+  after(async function () {
+    await cleanupTests(servers)
+  })
+})