<commit_before>'use strict';

const _ = require('lodash')
const Helpers = require('../helpers.js')
const chai = require('chai')
const sinon = require('sinon')

chai
  .use(require('sinon-chai'))
  .use(require('chai-as-promised'))
  .should()

const { expect } = chai

const th = new Helpers()
const { allways, sideEffect, withTasksRef, withQueueWorkerFor, validBasicTaskSpec, validTaskSpecWithFinishedState, chain, pushTasks, waitForState, waitForStates, echo } = th
const tasksRef = th.tasksRef
const _tasksRef = tasksRef

const nonBooleans             = ['', 'foo', NaN, Infinity,              0, 1, ['foo', 'bar'], { foo: 'bar' }, null,            { foo: { bar: { baz: true } } }, _.noop                ]
const nonStrings              = [           NaN, Infinity, true, false, 0, 1, ['foo', 'bar'], { foo: 'bar' }, null,            { foo: { bar: { baz: true } } }, _.noop                ]
const nonFunctions            = ['', 'foo', NaN, Infinity, true, false, 0, 1, ['foo', 'bar'], { foo: 'bar' }, null,            { foo: { bar: { baz: true } } }                        ]
const nonPlainObjects         = ['', 'foo', NaN, Infinity, true, false, 0, 1, ['foo', 'bar'],                 null,                                             _.noop                ]
const nonPositiveIntegers     = ['', 'foo', NaN, Infinity, true, false, 0,    ['foo', 'bar'], { foo: 'bar' },                  { foo: { bar: { baz: true } } }, _.noop, -1, 1.1       ]
const invalidPercentageValues = ['', 'foo', NaN, Infinity, true, false,       ['foo', 'bar'], { foo: 'bar' },                  { foo: { bar: { baz: true } } }, _.noop, -1,      100.1]
const invalidTaskSpecs        = ['', 'foo', NaN, Infinity, true, false, 0, 1, ['foo', 'bar'], { foo: 'bar' }, null, undefined, { foo: { bar: { baz: true } } }, _.noop, -1            ]

const nonStringsWithoutNull = nonStrings.filter(x => x !== null)
const nonPositiveIntegersWithout0 = nonPositiveIntegers.filter(x => x !== 0)

function now() { return new Date().getTime() }
function serverNow() { return now() + th.offset }

describe('QueueWorker', () => {
  
  describe('initialize', () => {

    function initialize({ tasksRef = _tasksRef, processId = '0', sanitize = true, suppressStack = false, processingFunction = _.noop } = {}) {
      return new th.QueueWorker(tasksRef, processId, sanitize, suppressStack, processingFunction)
    }

    // we should probably replace the `throws(strings)` with `throws(CONSTANT)`

    it('should not create a QueueWorker with no parameters', () => 
      expect(() => new th.QueueWorker()).to.throw('No tasks reference provided.')
    )

    it('should not create a QueueWorker with only a tasksRef', () =>
      expect(() => new th.QueueWorker(tasksRef)).to.throw('Invalid process ID provided.')
    )

    it('should not create a QueueWorker with only a tasksRef, process ID, sanitize and suppressStack option', () =>
      expect(() => new th.QueueWorker(tasksRef, '0', true, false)).to.throw('No processing function provided.')
    )

    it('should not create a QueueWorker with a tasksRef, processId, sanitize option and an invalid processing function', () => 
      nonFunctions.forEach(nonFunctionObject =>
        expect(() => initialize({ processingFunction: nonFunctionObject })).to.throw('No processing function provided.')
      )
    )

    it('should create a QueueWorker with a tasksRef, processId, sanitize option and a processing function', () =>
      expect(() => initialize()).to.not.throw(Error)
    )

    it('should not create a QueueWorker with a non-string processId specified', () => 
      nonStrings.forEach(nonStringObject =>
        expect(() => initialize({ processId: nonStringObject })).to.throw('Invalid process ID provided.')
      )
    )

    it('should not create a QueueWorker with a non-boolean sanitize option specified', () =>
      nonBooleans.forEach(nonBooleanObject =>
        expect(() => initialize({ sanitize: nonBooleanObject })).to.throw('Invalid sanitize option.')
      )
    )

    it('should not create a QueueWorker with a non-boolean suppressStack option specified', () =>
      nonBooleans.forEach(nonBooleanObject =>
        expect(() => initialize({ suppressStack: nonBooleanObject })).to.throw('Invalid suppressStack option.')
      )
    )
  })

  describe('# Resetting tasks', () => {

    it('should reset a task when another task is currently being processed', () =>
      withTasksRef(tasksRef => 
        withQueueWorkerFor(tasksRef, echo, qw => {
          qw.setTaskSpec(validTaskSpecWithFinishedState)
          const { startState, inProgressState, finishedState } = validTaskSpecWithFinishedState
          return chain(
            pushTasks(tasksRef, { task: 1 }, { task: 2 }),
            ([task1, task2]) => waitForState([task1, task2], inProgressState),
            ([_, task2]) => waitForState(task2, startState),
            task2 => {
              const task = task2.val()
              expect(task).to.not.have.any.keys('_owner', '_progress', '_state')
              expect(task).to.have.property('task').that.equals(2)
              expect(task).to.have.property('_state_changed').that.is.closeTo(serverNow(), 250)
            }
          )
        })
      )
    )
  })

  describe('#_resetTask', () => {
    let qw
    let testRef

    beforeEach(() => {
      qw = new th.QueueWorker(tasksRef, '0', true, false, _.noop)
    })

    afterEach(done => {
      testRef.off()
      qw.shutdown()
        .then(_ => tasksRef.set(null))
        .then(done)
    })

    function resetTask({ task, forceReset, spec = th.validBasicTaskSpec }) {
      qw._setTaskSpec(spec)
      testRef = tasksRef.push()
      return testRef.set(task)
        .then(_ => qw._resetTask(testRef, forceReset))
        .then(_ => testRef.once('value'))
    }

    it('should not reset a task if `forceReset` set but no longer owned by current worker', () => {
      const task = {
        '_state': th.validBasicTaskSpec.inProgressState,
        '_state_changed': now(),
        '_owner': 'someone-else'
      }
      return resetTask({ task, forceReset: true })
        .then(snapshot => {
          expect(snapshot.val()).to.deep.equal(task)
        })
    })

    it('should not reset a task if `forceReset` not set and it is has changed state recently', () => {
      const task = {
        '_state': th.validBasicTaskSpec.inProgressState,
        '_state_changed': now(),
        '_owner': 'someone'
      }
      return resetTask({ task, forceReset: false })
        .then(snapshot => {
          expect(snapshot.val()).to.deep.equal(task)<commit_msg>reintroduced a previously removed unit test<commit_after>'use strict';

const _ = require('lodash')
const Helpers = require('../helpers.js')
const chai = require('chai')
const sinon = require('sinon')

chai
  .use(require('sinon-chai'))
  .use(require('chai-as-promised'))
  .should()

const { expect } = chai

const th = new Helpers()
const { allways, sideEffect, withTasksRef, withQueueWorkerFor, validBasicTaskSpec, validTaskSpecWithFinishedState, chain, pushTasks, waitForState, waitForStates, echo } = th
const tasksRef = th.tasksRef
const _tasksRef = tasksRef

const nonBooleans             = ['', 'foo', NaN, Infinity,              0, 1, ['foo', 'bar'], { foo: 'bar' }, null,            { foo: { bar: { baz: true } } }, _.noop                ]
const nonStrings              = [           NaN, Infinity, true, false, 0, 1, ['foo', 'bar'], { foo: 'bar' }, null,            { foo: { bar: { baz: true } } }, _.noop                ]
const nonFunctions            = ['', 'foo', NaN, Infinity, true, false, 0, 1, ['foo', 'bar'], { foo: 'bar' }, null,            { foo: { bar: { baz: true } } }                        ]
const nonPlainObjects         = ['', 'foo', NaN, Infinity, true, false, 0, 1, ['foo', 'bar'],                 null,                                             _.noop                ]
const nonPositiveIntegers     = ['', 'foo', NaN, Infinity, true, false, 0,    ['foo', 'bar'], { foo: 'bar' },                  { foo: { bar: { baz: true } } }, _.noop, -1, 1.1       ]
const invalidPercentageValues = ['', 'foo', NaN, Infinity, true, false,       ['foo', 'bar'], { foo: 'bar' },                  { foo: { bar: { baz: true } } }, _.noop, -1,      100.1]
const invalidTaskSpecs        = ['', 'foo', NaN, Infinity, true, false, 0, 1, ['foo', 'bar'], { foo: 'bar' }, null, undefined, { foo: { bar: { baz: true } } }, _.noop, -1            ]

const nonStringsWithoutNull = nonStrings.filter(x => x !== null)
const nonPositiveIntegersWithout0 = nonPositiveIntegers.filter(x => x !== 0)

function now() { return new Date().getTime() }
function serverNow() { return now() + th.offset }

describe('QueueWorker', () => {
  
  describe('initialize', () => {

    function initialize({ tasksRef = _tasksRef, processId = '0', sanitize = true, suppressStack = false, processingFunction = _.noop } = {}) {
      return new th.QueueWorker(tasksRef, processId, sanitize, suppressStack, processingFunction)
    }

    // we should probably replace the `throws(strings)` with `throws(CONSTANT)`

    it('should not create a QueueWorker with no parameters', () => 
      expect(() => new th.QueueWorker()).to.throw('No tasks reference provided.')
    )

    it('should not create a QueueWorker with only a tasksRef', () =>
      expect(() => new th.QueueWorker(tasksRef)).to.throw('Invalid process ID provided.')
    )

    it('should not create a QueueWorker with only a tasksRef, process ID, sanitize and suppressStack option', () =>
      expect(() => new th.QueueWorker(tasksRef, '0', true, false)).to.throw('No processing function provided.')
    )

    it('should not create a QueueWorker with a tasksRef, processId, sanitize option and an invalid processing function', () => 
      nonFunctions.forEach(nonFunctionObject =>
        expect(() => initialize({ processingFunction: nonFunctionObject })).to.throw('No processing function provided.')
      )
    )

    it('should create a QueueWorker with a tasksRef, processId, sanitize option and a processing function', () =>
      expect(() => initialize()).to.not.throw(Error)
    )

    it('should not create a QueueWorker with a non-string processId specified', () => 
      nonStrings.forEach(nonStringObject =>
        expect(() => initialize({ processId: nonStringObject })).to.throw('Invalid process ID provided.')
      )
    )

    it('should not create a QueueWorker with a non-boolean sanitize option specified', () =>
      nonBooleans.forEach(nonBooleanObject =>
        expect(() => initialize({ sanitize: nonBooleanObject })).to.throw('Invalid sanitize option.')
      )
    )

    it('should not create a QueueWorker with a non-boolean suppressStack option specified', () =>
      nonBooleans.forEach(nonBooleanObject =>
        expect(() => initialize({ suppressStack: nonBooleanObject })).to.throw('Invalid suppressStack option.')
      )
    )
  })

  describe('# Resetting tasks', () => {

    it('should reset a task when another task is currently being processed', () =>
      withTasksRef(tasksRef => 
        withQueueWorkerFor(tasksRef, echo, qw => {
          qw.setTaskSpec(validTaskSpecWithFinishedState)
          const { startState, inProgressState, finishedState } = validTaskSpecWithFinishedState
          return chain(
            pushTasks(tasksRef, { task: 1 }, { task: 2 }),
            ([task1, task2]) => waitForState([task1, task2], inProgressState),
            ([_, task2]) => waitForState(task2, startState),
            task2 => {
              const task = task2.val()
              expect(task).to.not.have.any.keys('_owner', '_progress', '_state')
              expect(task).to.have.property('task').that.equals(2)
              expect(task).to.have.property('_state_changed').that.is.closeTo(serverNow(), 250)
            }
          )
        })
      )
    )
  })

  describe('#_resetTask', () => {
    let qw
    let testRef

    beforeEach(() => {
      qw = new th.QueueWorker(tasksRef, '0', true, false, _.noop)
    })

    afterEach(done => {
      testRef.off()
      qw.shutdown()
        .then(_ => tasksRef.set(null))
        .then(done)
    })

    function resetTask({ task, forceReset, spec = th.validBasicTaskSpec }) {
      qw._setTaskSpec(spec)
      testRef = tasksRef.push()
      return testRef.set(task)
        .then(_ => qw._resetTask(testRef, forceReset))
        .then(_ => testRef.once('value'))
    }

    it('should reset a task that is currently in progress', () => {
      return resetTask({
        forceReset: true,
        task: {
          '_state': th.validBasicTaskSpec.inProgressState,
          '_state_changed': now(),
          '_owner': qw._currentId()
        }
      }).then(snapshot => {
          const task = snapshot.val()
          expect(task).to.not.have.any.keys('_owner', '_progress', '_state')
          expect(task).to.have.property('_state_changed').that.is.closeTo(serverNow(), 250)
        })
    })

    it('should not reset a task if `forceReset` set but no longer owned by current worker', () => {
      const task = {
        '_state': th.validBasicTaskSpec.inProgressState,
        '_state_changed': now(),
        '_owner': 'someone-else'
      }
      return resetTask({ task, forceReset: true })
        .then(snapshot => {
          expect(snapshot.val()).to.deep.equal(task)
        })
    })

    it('should not reset a task if `forceReset` not set and it is has changed state recently', () => {
      const task = {
        '_state': th.validBasicTaskSpec.inProgressState,
        '_state_changed': now(),
        '_owner': 'someone'
      }
      return resetTask({ task, forceReset: false })
        .then(snapshot => {
          expect(snapshot.val()).to.deep.equal(task)
{'fraction_non_alphanumeric': 0.152428810720268, 'fraction_numerical': 0.00589906051999126, 'mean_word_length': 2.6193990511333687, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 87, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2949704', 'n_tokens_mistral': 4086, 'n_tokens_neox': 3946, 'n_words': 1057}