service-core/test/core.test.integration.mjs

import { Eltro as t, assert} from 'eltro'
import fs from 'fs/promises'
import http from 'http'
import Util from '../core/util.mjs'
import { request } from '../core/client.mjs'
import { setTimeout } from 'timers/promises'
import { prettyPrintMessage } from './helpers.mjs'
import { pipeline } from 'stream'
import getLog from '../core/log.mjs'

const util = new Util(import.meta.url)
const port = 61412

t.timeout(10000).describe('', function() {
  let server = null
  let prefix = `http://localhost:${port}/`
  let files = []
  let logs = []
  let versions = []
  let processor
  let integrationLog = getLog('test.integration', [])
  let compressorPath = util.getPathFromRoot('./7za.exe')
  if (process.platform !== 'win32') {
    compressorPath = util.getPathFromRoot('./7zas')
  }

  t.before(function(cb) {
    server = http.createServer(function(req, res) {
      req.on('error', function(err) {
        integrationLog.error(err, 'error')
      })
      res.on('error', function(err) {
        integrationLog.error(err, 'error')
      })

      integrationLog.info('[SERVER] got request ' + req.url)

      if (req.url === '/releases') {
        res.statusCode = 200
        let output = versions.map(x => {
          return {
            name: x[0],
            body: x[1],
            assets: [{
              name: x[2],
              browser_download_url: prefix + 'files/' + x[2]
            }]
          }
        })
        res.end(JSON.stringify(output));
        return
      } else if (req.url.startsWith('/files')) {
        let filename =  req.url.substring(req.url.lastIndexOf('/') + 1)
        return fs.open(util.getPathFromRoot('./' + filename))
        .then(function(file) {
          pipeline(file.createReadStream(), res, function(err) {
            if (err) {
              console.log(err)        
              res.statusCode = 404
              res.end(JSON.stringify({ error: 'unknown url' }))
            }
          })
        }).catch(function(err) {
          console.log(err)        
          res.statusCode = 404
          res.end(JSON.stringify({ error: 'unknown url' }))
        })
      }
      
      res.statusCode = 404
      res.end(JSON.stringify({ error: 'unknown url' }))
    })

    fs.rm(util.getPathFromRoot('./db.json'), { force: true }).then(function() {
      server.listen(port, cb)
    }, cb)
  })

  t.after(function() {
    return Promise.all(files.map(function(file) {
      return fs.rm(file, { force: true, recursive: true })
    }))
    .then(function() {
      if (processor && !processor.exitCode) {
        processor.kill()
      }
    })
  })

  const version_1_stable = `
  export function start(http, port, ctx) {
    const server = http.createServer(function (req, res) {
      res.writeHead(200);
      res.end(JSON.stringify({ version: 'v1' }))
    })

    return server.listenAsync(port, '0.0.0.0')
      .then(() => {
        ctx.log.info({ port: port, listening: true }, \`Server is listening on \${port} serving v1\`)
      })
  }
  `

  const version_2_nolisten = `
  export function start(http, port, ctx) {
  }
  `

  const version_3_crashing = `
  export function start(http, port, ctx) {
    process.exit(1)
  }
  `

  const version_4_stable = `
  export function start(http, port, ctx) {
    const server = http.createServer(function (req, res) {
      res.writeHead(200);
      res.end(JSON.stringify({ version: 'v4' }))
    })

    return server.listenAsync(port, '0.0.0.0')
      .then(() => {
        ctx.log.info({ port: port, listening: true }, \`Server is listening on \${port} serving v4\`)
      })
  }
  `

  function file(relative) {
    let file = util.getPathFromRoot(relative)
    files.push(file)
    return file
  }

  function log(message) {
    let lines = message.split('\n')
    for (let line of lines) {
      if (!line.trim()) continue
      logs.push(line)
    }
  }

  function parseLine(line) {
    if (line[0] === '{') {
      return JSON.parse(line)
    }
    return {
      msg: line
    }
  }


  let logIndex = 0
  function catchupLog() {
    if (logs.length > logIndex) {
      for (; logIndex < logs.length; logIndex++) {
        prettyPrintMessage(logs[logIndex])
      }
    }
  }

  integrationLog.on('newlog', function(record) {
    prettyPrintMessage(JSON.stringify(record))
  })

  let logWaitIndex = 0
  function hasLogLine(regMatch) {
    if (logs.length > logWaitIndex) {
      for (; logWaitIndex < logs.length; logWaitIndex++) {
        if (typeof(regMatch) === 'function') {
          let res = regMatch(parseLine(logs[logWaitIndex]))
          if (res) return true
        }
        else if (logs[logWaitIndex].match(regMatch)) {
          return true
        }
      }
    }
    return false
  }
  
  function findInLogs(regMatch) {
    for (let i = 0; i < logs.length; i++) {
      if (typeof(regMatch) === 'function') {
        let res = regMatch(parseLine(logs[i]))
        if (res) return true
      }
      else if (logs[i].match(regMatch)) {
        return true
      }
    }
  }

  async function waitUntilListening() {
    let listeningLine = null
    while (processor.exitCode == null
        && !hasLogLine((rec) => { listeningLine = rec; return rec.listening && rec.port })) {
      catchupLog()
      await setTimeout(10)
    }
    catchupLog()
    return listeningLine 
  }

  async function waitUntilClosed(listening) {
    while (true) {
      catchupLog()
      try {
        await request({}, `http://localhost:${listening.port}/`)
      } catch (err) {
        break
      }
      await setTimeout(25)
    }
    catchupLog()

    logs.splice(0, logs.length); logIndex = 0; logWaitIndex = 0; console.log('\n-------\n')
  }

  function startRunner() {
    return util.runCommandBackground('node', ['runner.mjs'], util.getPathFromRoot('./'), log)
  }

  t.test('should be fully operational', async function() {
    console.log()

    let index = file('./index.mjs')
    await fs.writeFile(index, version_1_stable)
    await util.runCommand(compressorPath, ['a', file('./v1-sc.7z'), index], util.getPathFromRoot('./testapp'))

    processor = startRunner()


    while (processor.exitCode == null) {
      catchupLog()
      await setTimeout(10)
    }

    catchupLog()

    let secondLast = parseLine(logs[logs.length - 2])
    let last = parseLine(logs[logs.length - 1])
    assert.match(secondLast.msg, /creating/i)
    assert.match(secondLast.msg, /application/i)
    assert.match(secondLast.msg, /testapp/i)
    assert.match(secondLast.msg, /0 releases/i)
    assert.match(last.err.message, /none/i)
    assert.match(last.err.message, /successful/i)

    // Reset our log
    logs.splice(0, logs.length); logIndex = 0; logWaitIndex = 0; console.log('\n-------\n')

    const assertNameVersion1 = 'v1_ok'

    file(`./testapp/${assertNameVersion1}`)

    versions.splice(0, 0, [assertNameVersion1, 'ok version', 'v1-sc.7z'])
    processor = startRunner()

    let listening = await waitUntilListening()

    let checkListening = await request({}, `http://localhost:${listening.port}/`)
    assert.strictEqual(checkListening.body.version, 'v1')

    while (!hasLogLine(/core is running/)) {
      catchupLog()
      await setTimeout(10)
    }

    catchupLog()

    let db = JSON.parse(await fs.readFile(util.getPathFromRoot('./db.json')))
    assert.strictEqual(db.core.testapp.active, assertNameVersion1)
    assert.strictEqual(db.core.testapp.versions.length, 1)
    assert.strictEqual(db.core.testapp.versions[0].stable, 1)
    assert.strictEqual(db.core.testapp.versions[0].installed, true)

    // Create our second version
    await fs.writeFile(index, version_2_nolisten)
    await util.runCommand(compressorPath, ['a', file('./v2-sc.7z'), index], util.getPathFromRoot('./testapp'))
    
    const assertNameVersion2 = 'v2_nolisten'
    file(`./testapp/${assertNameVersion2}`)
    versions.splice(0, 0, [assertNameVersion2, 'no listen version', 'v2-sc.7z'])

    // wait a second for it to trigger an update

    await setTimeout(500)

    while (!hasLogLine(/Error starting v2_nolisten/)) {
      catchupLog()
      await setTimeout(10)
    }

    while (!hasLogLine(/Attempting to run version v1_ok/)) {
      catchupLog()
      await setTimeout(10)
    }

    while (!hasLogLine(/Server is listening on 31313 serving v1/)) {
      catchupLog()
      await setTimeout(10)
    }

    catchupLog()

    checkListening = await request({}, `http://localhost:${listening.port}/`)
    assert.strictEqual(checkListening.body.version, 'v1')

    await setTimeout(10)
    
    db = JSON.parse(await fs.readFile(util.getPathFromRoot('./db.json')))
    assert.strictEqual(db.core.testapp.active, assertNameVersion1)
    assert.strictEqual(db.core.testapp.versions.length, 2)
    assert.strictEqual(db.core.testapp.versions[0].stable, -1)
    assert.strictEqual(db.core.testapp.versions[0].installed, true)
    assert.strictEqual(db.core.testapp.versions[1].stable, 1)
    assert.strictEqual(db.core.testapp.versions[1].installed, true)

    processor.kill()
    // Wait for ports to be marked as closed
    await waitUntilClosed()

    processor = startRunner()

    listening = await waitUntilListening()

    assert.ok(listening)

    checkListening = await request({}, `http://localhost:${listening.port}/`)
    assert.strictEqual(checkListening.body.version, 'v1')

    while (!hasLogLine(/core is running/)) {
      await setTimeout(10)
    }

    db = JSON.parse(await fs.readFile(util.getPathFromRoot('./db.json')))
    assert.strictEqual(db.core.testapp.active, assertNameVersion1)
    assert.strictEqual(db.core.testapp.versions.length, 2)
    assert.strictEqual(db.core.testapp.versions[0].stable, -2)
    assert.strictEqual(db.core.testapp.versions[1].stable, 1)

    assert.ok(findInLogs(/Attempting to run version v2_nolisten/))
    assert.ok(findInLogs(/Error starting v2_nolisten/))

    processor.kill()

    await waitUntilClosed()
    processor = startRunner()

    listening = await waitUntilListening()

    assert.ok(listening)

    checkListening = await request({}, `http://localhost:${listening.port}/`)
    assert.strictEqual(checkListening.body.version, 'v1')
    
    while (!hasLogLine(/core is running/)) {
      await setTimeout(10)
    }

    db = JSON.parse(await fs.readFile(util.getPathFromRoot('./db.json')))
    assert.strictEqual(db.core.testapp.active, assertNameVersion1)
    assert.strictEqual(db.core.testapp.versions.length, 2)
    assert.strictEqual(db.core.testapp.versions[0].stable, -2)
    assert.strictEqual(db.core.testapp.versions[1].stable, 1)

    assert.notOk(findInLogs(/Attempting to run version v2_nolisten/))
    assert.notOk(findInLogs(/Error starting v2_nolisten/))
    
    // Create our third version
    await fs.writeFile(index, version_3_crashing)
    await util.runCommand(compressorPath, ['a', file('./v3-sc.7z'), index], util.getPathFromRoot('./testapp'))
    
    const assertNameVersion3 = 'v3_crash'
    file(`./testapp/${assertNameVersion3}`)
    versions.splice(0, 0, [assertNameVersion3, 'crash version', 'v3-sc.7z'])

    // wait a second for it to trigger an update

    await setTimeout(500)

    while (!hasLogLine(/Attempting to run version v3_crash/)) {
      catchupLog()
      await setTimeout(10)
    }
    
    while (processor.exitCode == null) {
      catchupLog()
      await setTimeout(10)
    }

    db = JSON.parse(await fs.readFile(util.getPathFromRoot('./db.json')))
    assert.strictEqual(db.core.testapp.active, assertNameVersion3)
    assert.strictEqual(db.core.testapp.versions.length, 3)
    assert.strictEqual(db.core.testapp.versions[0].stable, -2)
    assert.strictEqual(db.core.testapp.versions[1].stable, -2)
    assert.strictEqual(db.core.testapp.versions[2].stable, 1)

    catchupLog()

    // Should recover afterwards
    await waitUntilClosed()
    processor = startRunner()

    listening = await waitUntilListening()

    assert.ok(listening)

    checkListening = await request({}, `http://localhost:${listening.port}/`)
    assert.strictEqual(checkListening.body.version, 'v1')
    
    while (!hasLogLine(/core is running/)) {
      await setTimeout(10)
    }

    // Create our fourth version
    await fs.writeFile(index, version_4_stable)
    await util.runCommand(compressorPath, ['a', file('./v4-sc.7z'), index], util.getPathFromRoot('./testapp'))
    
    const assertNameVersion4 = 'v4_stable'
    file(`./testapp/${assertNameVersion4}`)
    versions.splice(0, 0, [assertNameVersion4, 'no listen version', 'v4-sc.7z'])

    // wait a second for it to trigger an update

    await setTimeout(500)

    while (!hasLogLine(/Attempting to run version v4_stable/)) {
      catchupLog()
      await setTimeout(10)
    }

    while (!hasLogLine(/Server is listening on 31313 serving v4/)) {
      catchupLog()
      await setTimeout(10)
    }

    catchupLog()

    checkListening = await request({}, `http://localhost:${listening.port}/`)
    assert.strictEqual(checkListening.body.version, 'v4')

    await setTimeout(10)
    
    db = JSON.parse(await fs.readFile(util.getPathFromRoot('./db.json')))
    assert.strictEqual(db.core.testapp.active, assertNameVersion4)
    assert.strictEqual(db.core.testapp.versions.length, 4)
    assert.strictEqual(db.core.testapp.versions[0].stable, 1)
    assert.strictEqual(db.core.testapp.versions[1].stable, -2)
  })
})