test: more scenarios

analog-nico · Jun 30, 2016 · c34b9bc · c34b9bc
1 parent 11df78d
commit c34b9bc
Show file tree

Hide file tree

Showing 4 changed files with 209 additions and 8 deletions.
diff --git a/lib/PencilPusher.js b/lib/PencilPusher.js
@@ -9,9 +9,9 @@ let TaskDispatcher = require('./task/TaskDispatcher.js')
 
 class PencilPusher {
 
-    constructor({ persistenceLayer } = {}) {
+    constructor({ persistenceLayer, maxConcurrentTasks } = {}) {
 
-        this._taskDispatcher = new TaskDispatcher({ persistenceLayer })
+        this._taskDispatcher = new TaskDispatcher({ persistenceLayer, maxConcurrentTasks })
 
         this.registerExecutionContext('main', new MainExecutionContext())
 

diff --git a/lib/task/TaskDispatcher.js b/lib/task/TaskDispatcher.js
@@ -203,7 +203,9 @@ class TaskDispatcher {
 
                 this.concurrentTasks -= 1
 
-                // FIXME: Next polling is not invoked
+                if (this.concurrentTasks === this.maxConcurrentTasks-1) {
+                    this._pollForTask()
+                }
 
             })
 

diff --git a/lib/task/TaskRunner.js b/lib/task/TaskRunner.js
@@ -64,7 +64,7 @@ class TaskRunner {
             })
             .then((output) => {
 
-                this._persistenceLayer.finishTaskProcessing({
+                return this._persistenceLayer.finishTaskProcessing({
                     task,
                     retain: taskDefinition.shallRetain(),
                     retainUntil: taskDefinition.shallRetainUntil(),
@@ -75,7 +75,7 @@ class TaskRunner {
             })
             .catch((err) => {
 
-                this._persistenceLayer.cancelTaskProcessing({
+                return this._persistenceLayer.cancelTaskProcessing({
                     task,
                     executionFailed: true,
                     err

diff --git a/test/spec/task.js b/test/spec/task.js
@@ -40,11 +40,11 @@ describe('PencilPusher\'s task management', () => {
 
         setTimeout(() => {
 
-            expect(spyGetNextPendingTask.callCount).to.eql(1)
+            expect(spyGetNextPendingTask.callCount).to.eql(2)
             expect(spySetTaskProcessingTime.callCount).to.eql(1)
             expect(spyCancelTaskProcessing.callCount).to.eql(0)
             expect(spyFinishTaskProcessing.callCount).to.eql(1)
-            expect(spyGetNextPollingTime.callCount).to.eql(0)
+            expect(spyGetNextPollingTime.callCount).to.eql(1)
             expect(spyStoreNewTask.callCount).to.eql(1)
 
             expect(taskWasExecuted).to.eql(true)
@@ -57,6 +57,205 @@ describe('PencilPusher\'s task management', () => {
 
     })
 
-    it('should cancel an undefined task')
+    it('should process two tasks in a sequence', (done) => {
+
+        let persistenceLayer = new PencilPusher.MemoryPersistenceLayer()
+        let spyGetNextPendingTask = sinon.spy(persistenceLayer, 'getNextPendingTask')
+        let spySetTaskProcessingTime = sinon.spy(persistenceLayer, 'setTaskProcessingTime')
+        let spyCancelTaskProcessing = sinon.spy(persistenceLayer, 'cancelTaskProcessing')
+        let spyFinishTaskProcessing = sinon.spy(persistenceLayer, 'finishTaskProcessing')
+        let spyGetNextPollingTime = sinon.spy(persistenceLayer, 'getNextPollingTime')
+        let spyStoreNewTask = sinon.spy(persistenceLayer, 'storeNewTask')
+
+        let pencilPusher = new PencilPusher({
+            persistenceLayer
+        })
+
+        let taskWasExecuted = [false, false]
+
+        pencilPusher.defineTask('simple', {
+            implementation: (id) => {
+                taskWasExecuted[id] = true
+            }
+        })
+
+        pencilPusher.scheduleTask({
+            name: 'simple',
+            input: 0,
+            due: moment().unix()-1
+        })
+
+        pencilPusher.scheduleTask({
+            name: 'simple',
+            input: 1,
+            due: moment().unix()
+        })
+
+        pencilPusher.start()
+
+        setTimeout(() => {
+
+            expect(spyGetNextPendingTask.callCount).to.eql(3)
+            expect(spySetTaskProcessingTime.callCount).to.eql(2)
+            expect(spyCancelTaskProcessing.callCount).to.eql(0)
+            expect(spyFinishTaskProcessing.callCount).to.eql(2)
+            expect(spyGetNextPollingTime.callCount).to.eql(1)
+            expect(spyStoreNewTask.callCount).to.eql(2)
+
+            expect(taskWasExecuted[0]).to.eql(true)
+            expect(taskWasExecuted[1]).to.eql(true)
+
+            expect(spyGetNextPendingTask.secondCall.calledAfter(spyFinishTaskProcessing.firstCall)).to.eql(true)
+
+            pencilPusher.stop()
+
+            done()
+
+        }, 10)
+
+    })
+
+    it('should process two tasks in parallel', (done) => {
+
+        let persistenceLayer = new PencilPusher.MemoryPersistenceLayer()
+        let spyGetNextPendingTask = sinon.spy(persistenceLayer, 'getNextPendingTask')
+        let spySetTaskProcessingTime = sinon.spy(persistenceLayer, 'setTaskProcessingTime')
+        let spyCancelTaskProcessing = sinon.spy(persistenceLayer, 'cancelTaskProcessing')
+        let spyFinishTaskProcessing = sinon.spy(persistenceLayer, 'finishTaskProcessing')
+        let spyGetNextPollingTime = sinon.spy(persistenceLayer, 'getNextPollingTime')
+        let spyStoreNewTask = sinon.spy(persistenceLayer, 'storeNewTask')
+
+        let pencilPusher = new PencilPusher({
+            persistenceLayer,
+            maxConcurrentTasks: 2
+        })
+
+        let taskWasExecuted = [false, false]
+
+        pencilPusher.defineTask('simple', {
+            implementation: (id) => {
+                taskWasExecuted[id] = true
+            }
+        })
+
+        pencilPusher.scheduleTask({
+            name: 'simple',
+            input: 0,
+            due: moment().unix()-1
+        })
+
+        pencilPusher.scheduleTask({
+            name: 'simple',
+            input: 1,
+            due: moment().unix()
+        })
+
+        pencilPusher.start()
+
+        setTimeout(() => {
+
+            expect(spyGetNextPendingTask.callCount).to.eql(3)
+            expect(spySetTaskProcessingTime.callCount).to.eql(2)
+            expect(spyCancelTaskProcessing.callCount).to.eql(0)
+            expect(spyFinishTaskProcessing.callCount).to.eql(2)
+            expect(spyGetNextPollingTime.callCount).to.eql(1)
+            expect(spyStoreNewTask.callCount).to.eql(2)
+
+            expect(taskWasExecuted[0]).to.eql(true)
+            expect(taskWasExecuted[1]).to.eql(true)
+
+            expect(spyGetNextPendingTask.secondCall.calledBefore(spyFinishTaskProcessing.firstCall)).to.eql(true)
+
+            pencilPusher.stop()
+
+            done()
+
+        }, 10)
+
+    })
+
+    it('should cancel an undefined task retrieved from the persistence layer', (done) => {
+
+        let persistenceLayer = new PencilPusher.MemoryPersistenceLayer()
+        let spyGetNextPendingTask = sinon.spy(persistenceLayer, 'getNextPendingTask')
+        let spySetTaskProcessingTime = sinon.spy(persistenceLayer, 'setTaskProcessingTime')
+        let spyCancelTaskProcessing = sinon.spy(persistenceLayer, 'cancelTaskProcessing')
+        let spyFinishTaskProcessing = sinon.spy(persistenceLayer, 'finishTaskProcessing')
+        let spyGetNextPollingTime = sinon.spy(persistenceLayer, 'getNextPollingTime')
+        let spyStoreNewTask = sinon.spy(persistenceLayer, 'storeNewTask')
+
+        let pencilPusher = new PencilPusher({
+            persistenceLayer
+        })
+
+        persistenceLayer.storeNewTask({ task: {
+            name: 'undefined',
+            input: null,
+            due: moment().unix()
+        }})
+
+        pencilPusher.start()
+
+        setTimeout(() => {
+
+            expect(spyGetNextPendingTask.callCount).to.eql(2)
+            expect(spySetTaskProcessingTime.callCount).to.eql(0)
+            expect(spyCancelTaskProcessing.callCount).to.eql(1)
+            expect(spyFinishTaskProcessing.callCount).to.eql(0)
+            expect(spyGetNextPollingTime.callCount).to.eql(1)
+            expect(spyStoreNewTask.callCount).to.eql(1)
+
+            pencilPusher.stop()
+
+            done()
+
+        }, 10)
+
+    })
+
+    it('should cancel a task that throws an error', (done) => {
+
+        let persistenceLayer = new PencilPusher.MemoryPersistenceLayer()
+        let spyGetNextPendingTask = sinon.spy(persistenceLayer, 'getNextPendingTask')
+        let spySetTaskProcessingTime = sinon.spy(persistenceLayer, 'setTaskProcessingTime')
+        let spyCancelTaskProcessing = sinon.spy(persistenceLayer, 'cancelTaskProcessing')
+        let spyFinishTaskProcessing = sinon.spy(persistenceLayer, 'finishTaskProcessing')
+        let spyGetNextPollingTime = sinon.spy(persistenceLayer, 'getNextPollingTime')
+        let spyStoreNewTask = sinon.spy(persistenceLayer, 'storeNewTask')
+
+        let pencilPusher = new PencilPusher({
+            persistenceLayer
+        })
+
+        pencilPusher.defineTask('simple', {
+            implementation: () => {
+                throw new Error('Thrown by task')
+            }
+        })
+
+        pencilPusher.scheduleTask({
+            name: 'simple',
+            input: null,
+            due: moment().unix()
+        })
+
+        pencilPusher.start()
+
+        setTimeout(() => {
+
+            expect(spyGetNextPendingTask.callCount).to.eql(2)
+            expect(spySetTaskProcessingTime.callCount).to.eql(1)
+            expect(spyCancelTaskProcessing.callCount).to.eql(1)
+            expect(spyFinishTaskProcessing.callCount).to.eql(0)
+            expect(spyGetNextPollingTime.callCount).to.eql(1)
+            expect(spyStoreNewTask.callCount).to.eql(1)
+
+            pencilPusher.stop()
+
+            done()
+
+        }, 10)
+
+    })
 
 })