test.node.exec.js

/**
* @license Apache-2.0
*
* Copyright (c) 2018 The Stdlib Authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
*    http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

'use strict';

// MODULES //

var path = require( 'path' );
var EventEmitter = require( 'events' ).EventEmitter;
var tape = require( 'tape' );
var proxyquire = require( 'proxyquire' );
var objectKeys = require( '@stdlib/utils/keys' );
var noop = require( '@stdlib/utils/noop' );
var nextTick = require( '@stdlib/utils/next-tick' );
var rand = require( '@stdlib/random/base/minstd' );
var exec = require( './../lib/node/exec.js' );


// FIXTURES //

function DB() {
	EventEmitter.call( this );
	this._db = {};
	return this;
}
DB.prototype = Object.create( EventEmitter.prototype );
DB.prototype.constructor = DB;
DB.prototype.clean = function clean() {
	var keys;
	var i;
	keys = this.keys();
	for ( i = 0; i < keys.length; i++ ) {
		delete this._db[ keys[i] ];
	}
};
DB.prototype.keys = function keys() {
	return objectKeys( this._db );
};

// eslint-disable-next-line vars-on-top
var db = new DB();

function Fork() {
	EventEmitter.call( this );
	this.pid = rand();
	db[ this.pid ] = this;
	return this;
}
Fork.prototype = Object.create( EventEmitter.prototype );
Fork.prototype.constructor = Fork;
Fork.prototype.send = noop;

function files() {
	var out;
	var i;
	out = new Array( 4 );
	for ( i = 0; i < out.length; i++ ) {
		out[ i ] = path.resolve( __dirname, './fixtures/'+(i+1)+'.js' );
	}
	return out;
}


// TESTS //

tape( 'main export is a function', function test( t ) {
	t.ok( true, __filename );
	t.equal( typeof exec, 'function', 'main export is a function' );
	t.end();
});

tape( 'the function assigns scripts to workers', function test( t ) {
	var scripts;
	var count;
	var exec;
	var opts;
	var f;

	db.clean();

	exec = proxyquire( './../lib/node/exec.js', {
		'child_process': {
			'fork': fork
		}
	});

	f = files();
	opts = {
		'workers': 2,
		'concurrency': f.length
	};

	scripts = {};
	count = 0;
	exec( f, opts, done );

	function fork() {
		var f = new Fork();
		f.send = send;
		return f;
		function send( script ) {
			if ( script === 'close' ) {
				return f.emit( 'close', 0, null );
			}
			count += 1;
			if ( scripts[ script ] ) {
				t.ok( false, 'already allocated script: '+script );
			} else {
				scripts[ script ] = true;
			}
			f.emit( 'message', script );
		}
	}

	function done( error ) {
		if ( error ) {
			t.ok( false, error.message );
		} else {
			t.equal( count, f.length, 'runs all scripts' );
		}
		t.end();
	}
});

tape( 'the function returns the first encountered error to a provided callback', function test( t ) {
	var scripts;
	var count;
	var exec;
	var opts;
	var f;

	db.clean();

	exec = proxyquire( './../lib/node/exec.js', {
		'child_process': {
			'fork': fork
		}
	});

	f = files();
	opts = {
		'workers': 2,
		'concurrency': f.length
	};

	scripts = {};
	count = 0;
	exec( f, opts, done );

	function fork() {
		var f = new Fork();
		f.send = send;
		return f;
		function send( script ) {
			if ( script === 'close' ) {
				return f.emit( 'close', 0, null );
			}
			count += 1;
			if ( scripts[ script ] ) {
				t.ok( false, 'already allocated script: '+script );
			} else {
				scripts[ script ] = true;
			}
			if ( count > 2 ) {
				f.emit( 'error', new Error( 'error: '+count ) );
			} else {
				f.emit( 'message', script );
			}
		}
	}

	function done( error ) {
		t.ok( error, 'returns an error' );
		t.equal( error.message, 'error: 3', 'expected error message' );
		t.end();
	}
});

tape( 'the function returns an error to a provided callback if a worker exits with a non-zero exit code', function test( t ) {
	var exec;
	var opts;
	var f;

	db.clean();

	exec = proxyquire( './../lib/node/exec.js', {
		'child_process': {
			'fork': fork
		}
	});

	f = files();
	opts = {
		'workers': 2,
		'concurrency': f.length
	};

	exec( f, opts, done );

	function fork() {
		var f = new Fork();
		f.send = send;
		return f;
		function send( script ) {
			if ( script === 'close' ) {
				return f.emit( 'close', 0, null );
			}
			f.emit( 'exit', 1, null );
		}
	}

	function done( error ) {
		t.ok( error, 'returns an error' );
		t.equal( error.code, 1, 'sets exit code' );
		t.equal( error.signal, null, 'sets signal' );
		t.end();
	}
});

tape( 'the function returns an error to a provided callback if a worker exits due to a termination signal', function test( t ) {
	var exec;
	var opts;
	var f;

	db.clean();

	exec = proxyquire( './../lib/node/exec.js', {
		'child_process': {
			'fork': fork
		}
	});

	f = files();
	opts = {
		'workers': 2,
		'concurrency': f.length
	};

	exec( f, opts, done );

	function fork() {
		var f = new Fork();
		f.send = send;
		return f;
		function send( script ) {
			if ( script === 'close' ) {
				return f.emit( 'close', 0, null );
			}
			f.emit( 'exit', null, 'SIGINT' );
		}
	}

	function done( error ) {
		t.ok( error, 'returns an error' );
		t.equal( error.code, null, 'sets exit code' );
		t.equal( error.signal, 'SIGINT', 'sets signal' );
		t.end();
	}
});

tape( 'the function returns an error to a provided callback if a worker closes with a non-zero exit code', function test( t ) {
	var exec;
	var opts;
	var f;

	db.clean();

	exec = proxyquire( './../lib/node/exec.js', {
		'child_process': {
			'fork': fork
		}
	});

	f = files();
	opts = {
		'workers': 2,
		'concurrency': f.length
	};

	exec( f, opts, done );

	function fork() {
		var f = new Fork();
		f.send = send;
		return f;
		function send( script ) {
			if ( script === 'close' ) {
				return f.emit( 'close', 0, null );
			}
			f.emit( 'close', 1, null );
		}
	}

	function done( error ) {
		t.ok( error, 'returns an error' );
		t.equal( error.code, 1, 'sets exit code' );
		t.equal( error.signal, null, 'sets signal' );
		t.end();
	}
});

tape( 'the function returns an error to a provided callback if a worker closes due to a termination signal', function test( t ) {
	var exec;
	var opts;
	var f;

	db.clean();

	exec = proxyquire( './../lib/node/exec.js', {
		'child_process': {
			'fork': fork
		}
	});

	f = files();
	opts = {
		'workers': 2,
		'concurrency': f.length
	};

	exec( f, opts, done );

	function fork() {
		var f = new Fork();
		f.send = send;
		return f;
		function send( script ) {
			if ( script === 'close' ) {
				return f.emit( 'close', 0, null );
			}
			f.emit( 'close', null, 'SIGINT' );
		}
	}

	function done( error ) {
		t.ok( error, 'returns an error' );
		t.equal( error.code, null, 'sets exit code' );
		t.equal( error.signal, 'SIGINT', 'sets signal' );
		t.end();
	}
});

tape( 'the function does not error when a child process emits either a normal `disconnect` or `exit` event', function test( t ) {
	var count1;
	var count2;
	var exec;
	var opts;
	var f;

	db.clean();

	exec = proxyquire( './../lib/node/exec.js', {
		'child_process': {
			'fork': fork
		}
	});

	f = files();
	opts = {
		'workers': 2,
		'concurrency': f.length
	};

	count1 = 0;
	count2 = 0;
	exec( f, opts, done );

	function fork() {
		var f = new Fork();
		f.send = send;
		f.on( 'exit', onExit );
		f.on( 'disconnect', onDisconnect );
		return f;

		function send( script ) {
			if ( script === 'close' ) {
				f.emit( 'exit', 0, null );
				f.emit( 'disconnect' );
				return nextTick( onTick );
			}
			f.emit( 'message', script );
		}

		function onTick() {
			f.emit( 'close', 0, null );
		}
	}

	function onExit() {
		count1 += 1;
	}

	function onDisconnect() {
		count2 += 1;
	}

	function done( error ) {
		if ( error ) {
			t.ok( false, error.message );
		}
		t.equal( count1, opts.workers, 'does not error when `exit` emitted' );
		t.equal( count2, opts.workers, 'does not error when `disconnect` emitted' );
		t.end();
	}
});