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

import { Eltro as t, assert} from 'eltro'
import fs from 'fs/promises'
import HttpServer from '../core/http.mjs'
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
const turnDebuggingOn = false

const runners = [
  ['runner.mjs', 'testapp'],
  ['runner_cluster.mjs', 'testappcluster'],
]

runners.forEach(function([runnerName, appname]) {
  t.timeout(10000).describe(runnerName, function() {
    let wasSuccessful = false
    let http = null
    let server = null
    let prefix = `http://localhost:${port}/`
    let files = [util.getPathFromRoot('./testappcluster')]
    let logs = []
    let allLogs = []
    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() {
      http = new HttpServer()
  
      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' }))
      })
  
      return fs.rm(util.getPathFromRoot('./db.json'), { force: true })
      .then(function() {
        return server.listenAsync(port)
      })
    })
  
    t.after(function() {
      if (!turnDebuggingOn && !wasSuccessful) {
        for (let i = 0; i < allLogs.length; i++) {
          prettyPrintMessage(allLogs[i])
        }
      }

      return Promise.all(files.map(function(file) {
        return fs.rm(file, { force: true, recursive: true })
      }))
      .then(function() {
        if (processor && !processor.exitCode) {
          processor.kill()
      
          return waitUntilClosed()
        }
      }).then(function() {
        return http.closeServer()
      })
    })
  
    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)
        allLogs.push(line)
      }
    }
  
    function parseLine(line) {
      if (line[0] === '{') {
        return JSON.parse(line)
      }
      return {
        msg: line
      }
    }
  
  
    let logIndex = 0
    function catchupLog(ms = 0) {
      if (logs.length > logIndex) {
        for (; logIndex < logs.length; logIndex++) {
          if (turnDebuggingOn) {
            prettyPrintMessage(logs[logIndex])
          }
        }
      }
      if (ms > 0) {
        return setTimeout(ms)
      }
    }
  
    integrationLog.on('newlog', function(record) {
      allLogs.push(JSON.stringify(record))
      if (turnDebuggingOn) {
        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 })) {
        await catchupLog(10)
      }
      catchupLog()
      if (listeningLine.listening && listeningLine.port) {
        return listeningLine 
      } else {
        return null
      }
    }

    async function sendRequestToApplication(listening) {
      let lastErr = null
      for (let i = 0; i < 4; i++) {
        try {
          let checkListening = await request({}, `http://localhost:${listening.port}/`)
          return checkListening
        } catch (err) {
          lastErr = err
          integrationLog.info(`Request http://localhost:${listening.port}/ failed with ${err.message} trying again in 250ms`)
          await setTimeout(250)
        }
      }
      log('-- core.test.integration.mjs crash here --')
      log(lastErr.toString())
      throw lastErr
    }
  
    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;
      if (turnDebuggingOn) { console.log('\n-------\n') }
    }
  
    function startRunner() {
      return util.runCommandBackground('node', [runnerName], util.getPathFromRoot('./'), log)
    }
  
    t.test('should be fully operational', async function() {
      console.log()
      if (!turnDebuggingOn) { console.log('Running empty test') }

      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) {
        await catchupLog(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;
      if (turnDebuggingOn) { console.log('\n-------\n') }

      const assertNameVersion1 = 'v1_ok'

      if (!turnDebuggingOn) { console.log(`Running update ${assertNameVersion1} test`) }

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

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

      let listening = await waitUntilListening()

      let checkListening = await sendRequestToApplication(listening)
      assert.strictEqual(checkListening.body.version, 'v1')

      while (!hasLogLine(/core is running/)) {
        await catchupLog(10)
        if (processor.exitCode !== null) {
          throw new Error('Process exited with ' + processor.exitCode)
        }
      }

      catchupLog()

      let db = JSON.parse(await fs.readFile(util.getPathFromRoot('./db.json')))
      assert.strictEqual(db.core[appname].active, assertNameVersion1)
      assert.strictEqual(db.core[appname].versions.length, 1)
      assert.strictEqual(db.core[appname].versions[0].stable, 1)
      assert.strictEqual(db.core[appname].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'
      if (!turnDebuggingOn) { console.log(`Running update ${assertNameVersion2} test`) }
      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/)) {
        await catchupLog(10)
      }

      if (appname !== 'testappcluster') {
        while (!hasLogLine(/restart request.*v2_nolisten.*dirty/)) {
          await catchupLog(10)
        }

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

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

        // Since application was in dirty state, on next attempt should attempt to
        // run v2 again and then falling back to v1

        await waitUntilClosed()
        
        if (!turnDebuggingOn) { console.log(`Running fresh ${assertNameVersion2} test`) }
        processor = startRunner()
        await catchupLog(10)

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

      while (!hasLogLine(/Attempting to run version v1_ok/)) {
        await catchupLog(10)
      }
      
      listening = await waitUntilListening()
      assert.ok(listening)
      checkListening = await sendRequestToApplication(listening)
      assert.strictEqual(checkListening.body.version, 'v1')
      
      while (!hasLogLine(/is up and running/)) {
        await setTimeout(10)
      }

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

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

      processor.kill()

      if (!turnDebuggingOn) { console.log(`Running version stability check test`) }

      await waitUntilClosed()
      processor = startRunner()

      listening = await waitUntilListening()

      assert.ok(listening)

      checkListening = await sendRequestToApplication(listening)
      assert.strictEqual(checkListening.body.version, 'v1')
      
      while (!hasLogLine(/is up and running/)) {
        await setTimeout(10)
      }

      db = JSON.parse(await fs.readFile(util.getPathFromRoot('./db.json')))
      assert.strictEqual(db.core[appname].active, assertNameVersion1)
      assert.strictEqual(db.core[appname].versions.length, 2)
      assert.strictEqual(db.core[appname].versions[0].stable, -2)
      assert.strictEqual(db.core[appname].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'
      if (!turnDebuggingOn) { console.log(`Running update ${assertNameVersion3} test`) }
      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/)) {
        await catchupLog(10)
      }
      
      if (appname !== 'testappcluster') {
        while (processor.exitCode == null) {
          await catchupLog(10)
        }

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

        catchupLog()

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

        listening = await waitUntilListening()

        assert.ok(listening)

        checkListening = await sendRequestToApplication(listening)
        assert.strictEqual(checkListening.body.version, 'v1')
        
        while (!hasLogLine(/core is running/)) {
          await setTimeout(10)
        }
      } else {
        while (!hasLogLine(/Attempting to run version v1_ok/)) {
          await catchupLog(10)
        }
        while (!hasLogLine(/is up and running/)) {
          await setTimeout(10)
        }
        
        db = JSON.parse(await fs.readFile(util.getPathFromRoot('./db.json')))
        assert.strictEqual(db.core[appname].active, assertNameVersion1)
        assert.strictEqual(db.core[appname].versions.length, 3)
        assert.strictEqual(db.core[appname].versions[0].stable, -2)
        assert.strictEqual(db.core[appname].versions[1].stable, -2)
        assert.strictEqual(db.core[appname].versions[2].stable, 1)
      }

      // 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'
      if (!turnDebuggingOn) { console.log(`Running update ${assertNameVersion4} test`) }
      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/)) {
        await catchupLog(10)
      }

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

      catchupLog()

      checkListening = await sendRequestToApplication(listening)
      assert.strictEqual(checkListening.body.version, 'v4')

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