server/tests/api/runners/runner-live-transcoding.ts

   1 /* eslint-disable @typescript-eslint/no-unused-expressions,@typescript-eslint/require-await */
   2
   3 import { expect } from 'chai'
   4 import { FfmpegCommand } from 'fluent-ffmpeg'
   5 import { readFile } from 'fs-extra'
   6 import { buildAbsoluteFixturePath, wait } from '@shared/core-utils'
   7 import {
   8   HttpStatusCode,
   9   LiveRTMPHLSTranscodingUpdatePayload,
  10   LiveVideo,
  11   LiveVideoError,
  12   RunnerJob,
  13   RunnerJobLiveRTMPHLSTranscodingPayload,
  14   Video,
  15   VideoPrivacy,
  16   VideoState
  17 } from '@shared/models'
  18 import {
  19   cleanupTests,
  20   createSingleServer,
  21   makeRawRequest,
  22   PeerTubeServer,
  23   sendRTMPStream,
  24   setAccessTokensToServers,
  25   setDefaultVideoChannel,
  26   stopFfmpeg,
  27   testFfmpegStreamError,
  28   waitJobs
  29 } from '@shared/server-commands'
  30
  31 describe('Test runner live transcoding', function () {
  32   let server: PeerTubeServer
  33   let runnerToken: string
  34   let baseUrl: string
  35
  36   before(async function () {
  37     this.timeout(120_000)
  38
  39     server = await createSingleServer(1)
  40
  41     await setAccessTokensToServers([ server ])
  42     await setDefaultVideoChannel([ server ])
  43
  44     await server.config.enableRemoteTranscoding()
  45     await server.config.enableTranscoding()
  46     runnerToken = await server.runners.autoRegisterRunner()
  47
  48     baseUrl = server.url + '/static/streaming-playlists/hls'
  49   })
  50
  51   describe('Without transcoding enabled', function () {
  52
  53     before(async function () {
  54       await server.config.enableLive({
  55         allowReplay: false,
  56         resolutions: 'min',
  57         transcoding: false
  58       })
  59     })
  60
  61     it('Should not have available jobs', async function () {
  62       this.timeout(120000)
  63
  64       const { live, video } = await server.live.quickCreate({ permanentLive: true, saveReplay: false, privacy: VideoPrivacy.PUBLIC })
  65
  66       const ffmpegCommand = sendRTMPStream({ rtmpBaseUrl: live.rtmpUrl, streamKey: live.streamKey })
  67       await server.live.waitUntilPublished({ videoId: video.id })
  68
  69       await waitJobs([ server ])
  70
  71       const { availableJobs } = await server.runnerJobs.requestLive({ runnerToken })
  72       expect(availableJobs).to.have.lengthOf(0)
  73
  74       await stopFfmpeg(ffmpegCommand)
  75     })
  76   })
  77
  78   describe('With transcoding enabled on classic live', function () {
  79     let live: LiveVideo
  80     let video: Video
  81     let ffmpegCommand: FfmpegCommand
  82     let jobUUID: string
  83     let acceptedJob: RunnerJob & { jobToken: string }
  84
  85     async function testPlaylistFile (fixture: string, expected: string) {
  86       const text = await server.streamingPlaylists.get({ url: `${baseUrl}/${video.uuid}/${fixture}` })
  87       expect(await readFile(buildAbsoluteFixturePath(expected), 'utf-8')).to.equal(text)
  88
  89     }
  90
  91     async function testTSFile (fixture: string, expected: string) {
  92       const { body } = await makeRawRequest({ url: `${baseUrl}/${video.uuid}/${fixture}`, expectedStatus: HttpStatusCode.OK_200 })
  93       expect(await readFile(buildAbsoluteFixturePath(expected))).to.deep.equal(body)
  94     }
  95
  96     before(async function () {
  97       await server.config.enableLive({
  98         allowReplay: true,
  99         resolutions: 'max',
 100         transcoding: true
 101       })
 102     })
 103
 104     it('Should publish a a live and have available jobs', async function () {
 105       this.timeout(120000)
 106
 107       const data = await server.live.quickCreate({ permanentLive: false, saveReplay: false, privacy: VideoPrivacy.PUBLIC })
 108       live = data.live
 109       video = data.video
 110
 111       ffmpegCommand = sendRTMPStream({ rtmpBaseUrl: live.rtmpUrl, streamKey: live.streamKey })
 112       await waitJobs([ server ])
 113
 114       const job = await server.runnerJobs.requestLiveJob(runnerToken)
 115       jobUUID = job.uuid
 116
 117       expect(job.type).to.equal('live-rtmp-hls-transcoding')
 118       expect(job.payload.input.rtmpUrl).to.exist
 119
 120       expect(job.payload.output.toTranscode).to.have.lengthOf(5)
 121
 122       for (const { resolution, fps } of job.payload.output.toTranscode) {
 123         expect([ 720, 480, 360, 240, 144 ]).to.contain(resolution)
 124
 125         expect(fps).to.be.above(25)
 126         expect(fps).to.be.below(70)
 127       }
 128     })
 129
 130     it('Should update the live with a new chunk', async function () {
 131       this.timeout(120000)
 132
 133       const { job } = await server.runnerJobs.accept<RunnerJobLiveRTMPHLSTranscodingPayload>({ jobUUID, runnerToken })
 134       acceptedJob = job
 135
 136       {
 137         const payload: LiveRTMPHLSTranscodingUpdatePayload = {
 138           masterPlaylistFile: 'live/master.m3u8',
 139           resolutionPlaylistFile: 'live/0.m3u8',
 140           resolutionPlaylistFilename: '0.m3u8',
 141           type: 'add-chunk',
 142           videoChunkFile: 'live/0-000067.ts',
 143           videoChunkFilename: '0-000067.ts'
 144         }
 145         await server.runnerJobs.update({ jobUUID, runnerToken, jobToken: job.jobToken, payload, progress: 50 })
 146
 147         const updatedJob = await server.runnerJobs.getJob({ uuid: job.uuid })
 148         expect(updatedJob.progress).to.equal(50)
 149       }
 150
 151       {
 152         const payload: LiveRTMPHLSTranscodingUpdatePayload = {
 153           resolutionPlaylistFile: 'live/1.m3u8',
 154           resolutionPlaylistFilename: '1.m3u8',
 155           type: 'add-chunk',
 156           videoChunkFile: 'live/1-000068.ts',
 157           videoChunkFilename: '1-000068.ts'
 158         }
 159         await server.runnerJobs.update({ jobUUID, runnerToken, jobToken: job.jobToken, payload })
 160       }
 161
 162       await wait(1000)
 163
 164       await testPlaylistFile('master.m3u8', 'live/master.m3u8')
 165       await testPlaylistFile('0.m3u8', 'live/0.m3u8')
 166       await testPlaylistFile('1.m3u8', 'live/1.m3u8')
 167
 168       await testTSFile('0-000067.ts', 'live/0-000067.ts')
 169       await testTSFile('1-000068.ts', 'live/1-000068.ts')
 170     })
 171
 172     it('Should replace existing m3u8 on update', async function () {
 173       this.timeout(120000)
 174
 175       const payload: LiveRTMPHLSTranscodingUpdatePayload = {
 176         masterPlaylistFile: 'live/1.m3u8',
 177         resolutionPlaylistFilename: '0.m3u8',
 178         resolutionPlaylistFile: 'live/1.m3u8',
 179         type: 'add-chunk',
 180         videoChunkFile: 'live/1-000069.ts',
 181         videoChunkFilename: '1-000068.ts'
 182       }
 183       await server.runnerJobs.update({ jobUUID, runnerToken, jobToken: acceptedJob.jobToken, payload })
 184       await wait(1000)
 185
 186       await testPlaylistFile('master.m3u8', 'live/1.m3u8')
 187       await testPlaylistFile('0.m3u8', 'live/1.m3u8')
 188       await testTSFile('1-000068.ts', 'live/1-000069.ts')
 189     })
 190
 191     it('Should update the live with removed chunks', async function () {
 192       this.timeout(120000)
 193
 194       const payload: LiveRTMPHLSTranscodingUpdatePayload = {
 195         resolutionPlaylistFile: 'live/0.m3u8',
 196         resolutionPlaylistFilename: '0.m3u8',
 197         type: 'remove-chunk',
 198         videoChunkFilename: '1-000068.ts'
 199       }
 200       await server.runnerJobs.update({ jobUUID, runnerToken, jobToken: acceptedJob.jobToken, payload })
 201
 202       await wait(1000)
 203
 204       await server.streamingPlaylists.get({ url: `${baseUrl}/${video.uuid}/master.m3u8` })
 205       await server.streamingPlaylists.get({ url: `${baseUrl}/${video.uuid}/0.m3u8` })
 206       await server.streamingPlaylists.get({ url: `${baseUrl}/${video.uuid}/1.m3u8` })
 207       await makeRawRequest({ url: `${baseUrl}/${video.uuid}/0-000067.ts`, expectedStatus: HttpStatusCode.OK_200 })
 208       await makeRawRequest({ url: `${baseUrl}/${video.uuid}/1-000068.ts`, expectedStatus: HttpStatusCode.NOT_FOUND_404 })
 209     })
 210
 211     it('Should complete the live and save the replay', async function () {
 212       this.timeout(120000)
 213
 214       for (const segment of [ '0-000069.ts', '0-000070.ts' ]) {
 215         const payload: LiveRTMPHLSTranscodingUpdatePayload = {
 216           masterPlaylistFile: 'live/master.m3u8',
 217           resolutionPlaylistFilename: '0.m3u8',
 218           resolutionPlaylistFile: 'live/0.m3u8',
 219           type: 'add-chunk',
 220           videoChunkFile: 'live/' + segment,
 221           videoChunkFilename: segment
 222         }
 223         await server.runnerJobs.update({ jobUUID, runnerToken, jobToken: acceptedJob.jobToken, payload })
 224
 225         await wait(1000)
 226       }
 227
 228       await waitJobs([ server ])
 229
 230       {
 231         const { state } = await server.videos.get({ id: video.uuid })
 232         expect(state.id).to.equal(VideoState.PUBLISHED)
 233       }
 234
 235       await stopFfmpeg(ffmpegCommand)
 236
 237       await server.runnerJobs.success({ jobUUID, runnerToken, jobToken: acceptedJob.jobToken, payload: {} })
 238
 239       await wait(1500)
 240       await waitJobs([ server ])
 241
 242       {
 243         const { state } = await server.videos.get({ id: video.uuid })
 244         expect(state.id).to.equal(VideoState.LIVE_ENDED)
 245
 246         const session = await server.live.findLatestSession({ videoId: video.uuid })
 247         expect(session.error).to.be.null
 248       }
 249     })
 250   })
 251
 252   describe('With transcoding enabled on cancelled/aborted/errored live', function () {
 253     let live: LiveVideo
 254     let video: Video
 255     let ffmpegCommand: FfmpegCommand
 256
 257     async function prepare () {
 258       ffmpegCommand = sendRTMPStream({ rtmpBaseUrl: live.rtmpUrl, streamKey: live.streamKey })
 259       await server.runnerJobs.requestLiveJob(runnerToken)
 260
 261       const { job } = await server.runnerJobs.autoAccept({ runnerToken, type: 'live-rtmp-hls-transcoding' })
 262
 263       return job
 264     }
 265
 266     async function checkSessionError (error: LiveVideoError) {
 267       await wait(1500)
 268       await waitJobs([ server ])
 269
 270       const session = await server.live.findLatestSession({ videoId: video.uuid })
 271       expect(session.error).to.equal(error)
 272     }
 273
 274     before(async function () {
 275       await server.config.enableLive({
 276         allowReplay: true,
 277         resolutions: 'max',
 278         transcoding: true
 279       })
 280
 281       const data = await server.live.quickCreate({ permanentLive: true, saveReplay: false, privacy: VideoPrivacy.PUBLIC })
 282       live = data.live
 283       video = data.video
 284     })
 285
 286     it('Should abort a running live', async function () {
 287       this.timeout(120000)
 288
 289       const job = await prepare()
 290
 291       await Promise.all([
 292         server.runnerJobs.abort({ jobUUID: job.uuid, runnerToken, jobToken: job.jobToken, reason: 'abort' }),
 293         testFfmpegStreamError(ffmpegCommand, true)
 294       ])
 295
 296       // Abort is not supported
 297       await checkSessionError(LiveVideoError.RUNNER_JOB_ERROR)
 298     })
 299
 300     it('Should cancel a running live', async function () {
 301       this.timeout(120000)
 302
 303       const job = await prepare()
 304
 305       await Promise.all([
 306         server.runnerJobs.cancelByAdmin({ jobUUID: job.uuid }),
 307         testFfmpegStreamError(ffmpegCommand, true)
 308       ])
 309
 310       await checkSessionError(LiveVideoError.RUNNER_JOB_CANCEL)
 311     })
 312
 313     it('Should error a running live', async function () {
 314       this.timeout(120000)
 315
 316       const job = await prepare()
 317
 318       await Promise.all([
 319         server.runnerJobs.error({ jobUUID: job.uuid, runnerToken, jobToken: job.jobToken, message: 'error' }),
 320         testFfmpegStreamError(ffmpegCommand, true)
 321       ])
 322
 323       await checkSessionError(LiveVideoError.RUNNER_JOB_ERROR)
 324     })
 325   })
 326
 327   after(async function () {
 328     await cleanupTests([ server ])
 329   })
 330 })