mutapipe.spec.js

'use strict';
/* global it, describe */

var expect = require('expect.js'),
  es = require('event-stream'),
  fluent = require('fluent.js'),
  mp = require('./mutapipe'),
  R = require('ramda'),
  by = mp.by;

function append(str) {
  return es.map(function (data, cb) {
    cb(null, data + str);
  });
}

var write = fluent({
  initialData: '*',
  into: '*',
  andExpect: '*',
  then: '*'
}, function (initialData, stream, expected, done) {
  var result = '';
  stream.on('data', function (data) {
    result += data;
  });
  stream.on('end', function () {
    expect(result).to.eql(expected);
    done();
  });
  stream.write(initialData);
  stream.end();
});

describe('mp()', function () {
  describe('pipe()', function () {
    it('should handle simple streams', function (done) {
      write('abc').into(mp()()).andExpect('abc').then(done);
    });
    it('should let you pipe streams', function (done) {
      var pipeline = mp()
        .pipe(append, '1')
        .pipe(append, '2')
      ();

      write('abc').into(pipeline).andExpect('abc12').then(done);
    });
    it('should accept a function as a step', function (done) {
      var pipeline = mp()
        .pipe(append, '1')
      ();

      write('abc').into(pipeline).andExpect('abc1').then(done);
    });
    it('should accept a step object as a step', function (done) {
      var pipeline = mp()
        .pipe({task: append, args: ['1']})
      ();

      write('abc').into(pipeline).andExpect('abc1').then(done);
    });
    it('should allow you to supply your own step id', function () {
      var pipeline = mp()
        .pipe({id: 'otherId', task: append, args: ['1']})
        .pipe({id: 'someId', task: append, args: ['1']});

      var step = pipeline.first(mp.by.id('someId'));
      expect(step).to.eql({id: 'someId', task: append, args: ['1']});
    });
    it('should disallow duplicate ids', function () {
      var pipeline = mp()
        .pipe({id: 'someId', task: append, args: ['1']});

      expect(pipeline.pipe).withArgs({id: 'someId'}).to.throwException();
    });
    it('should support nesting', function (done) {
      var pipeline = mp()
        .pipe(mp()
          .pipe(append, 'Nested1')
          .pipe(mp()
            .pipe(append, 'Nested2')))
        .pipe(append, 'Regular');

      write('input').into(pipeline()).andExpect('inputNested1Nested2Regular').then(done);
    });
    it('should let you pipe other streams which may have duplicate ids', function (done) {
      var middlePipe = mp().pipe(append, 'Middle');
      var pipeline = mp()
        .pipe(append, 'Start')
        .pipe(middlePipe)
        .pipe(append, 'End');


      write('input').into(pipeline()).andExpect('inputStartMiddleEnd').then(done);
    });
    it('should make sure autogenerated ids are unique', function () {
      var middlePipe = mp().pipe(append, 'Middle');
      var pipeline = mp()
        .pipe(append, 'Start')
        .pipe(middlePipe)
        .pipe(append, 'End');

      var ids = pipeline._steps().map(R.prop('id'));
      var uniqueIds = R.uniqWith(R.equals, ids);
      expect(ids).to.eql(uniqueIds);
    });
    it('should support readable streams', function (done) {
      var received;
      mp().pipe(es.readArray, ['abc'])()
        .on('data', function (data) {
          received = data;
        })
        .on('end', function () {
          expect(received).to.eql('abc');
          done();
        });
    });
    it('should support built arguments', function (done) {
      var received = '';

      mp().pipe(es.merge,
        mp.build(mp().pipe(es.readArray, ['1'])),
        mp.build(mp().pipe(es.readArray, ['2']))
      )().on('data', function (data) {
        received += data;
      }).on('end', function () {
        expect(received).to.contain('1');
        expect(received).to.contain('2');
        done();
      });
    });
    it('should support built arguments using a custom build function', function (done) {
      var received = '';

      mp().pipe(es.merge,
        {build: mp().pipe(es.readArray, ['1'])}
      )().on('data', function (data) {
        received += data;
      }).on('end', function () {
        expect(received).to.contain('1');
        done();
      });
    });
  });

  describe('filter()', function () {
    function step1() {
    }

    function step2() {
    }

    it('should return all steps that match the predicate function', function () {
      var pipeline = mp()
        .pipe(step1)
        .pipe(step1)
        .pipe(step2);

      expect(pipeline.filter(function (step) {
        return step.task === step1;
      })).to.eql([
          {id: 0, task: step1, args: []},
          {id: 1, task: step1, args: []}
        ]);
    });
  });

  describe('first()', function () {
    function step1() {
    }

    function step2() {
    }

    it('should return the first step that matches the predicate function', function () {
      var pipeline = mp()
        .pipe(step1)
        .pipe(step1)
        .pipe(step2);

      expect(pipeline.first(function (step) {
        return step.task === step1;
      })).to.eql({id: 0, task: step1, args: []});
    });
    it('should return the first step if no predicate function is given', function () {
      var pipeline = mp()
        .pipe(step1)
        .pipe(step2);

      expect(pipeline.first()).to.eql({id: 0, task: step1, args: []});
    });
    it('should return undefined if no step matches the predicate', function () {
      var pipeline = mp()
        .pipe(step1)
        .pipe(step2);

      expect(pipeline.first(function (step) {
        return false;
      })).to.eql(undefined);
    });
  });

  describe('by', function () {
    describe('task()', function () {
      it('should return true when the task matches for a step', function () {
        expect(by.task(append)({task: append})).to.be(true);
      });
      it('should return false when the task does not match for a step', function () {
        expect(by.task(append)({
          task: function () {
          }
        })).to.be(false);
      });
    });
    describe('args()', function () {
      it('should return true when the args match exactly for a step', function () {
        expect(by.args('1')({args: ['1']})).to.be(true);
      });
      it('should return true when the args start with the supplied ones for a step', function () {
        expect(by.args('1')({args: ['1', '2']})).to.be(true);
      });
      it('should take multiple arguments', function () {
        expect(by.args('1', '2')({args: ['1', '3']})).to.be(false);
      });
      it('should return false when the args do not match for a step', function () {
        expect(by.args('1')({args: ['2']})).to.be(false);
      });
    });
    describe('id()', function () {
      it('should return true when the task matches for a step', function () {
        expect(by.id('some')({id: 'some'})).to.be(true);
      });
      it('should return false when the task does not match for a step', function () {
        expect(by.id('some')({id: 'other'})).to.be(false);
      });
    });
    describe('all()', function () {
      it('should return true when all the predicates match', function () {
        expect(by.all(
          by.task(append),
          by.args('2')
        )({task: append, args: ['2']})).to.be(true);
      });
      it('should return false when at least one of the predicates does not match', function () {
        expect(by.all(
          by.task(append),
          by.args('1')
        )({task: append, args: ['2']})).to.be(false);
      });
    });
  });

  describe('previous()', function () {
    function storeData() {
      var stored = '';
      var handler = es.map(function (data, cb) {
        stored = data;
        cb(null, data);
      });
      handler._stored = function (arg) {
        return es.map(function (data, cb) {
          cb(null, arg ? stored + arg : stored);
        });
      };
      return handler;
    }

    it('should let you pipe into previous step results', function (done) {
      var pipeline = mp()
        .pipe(append, '1')
        .pipe(storeData)
        .pipe(append, '2')
        .pipe(mp.previous(by.task(storeData), '_stored'));

      write('abc').into(pipeline()).andExpect('abc1').then(done);
    });

    it('should support arguments', function (done) {
      var pipeline = mp()
        .pipe(append, '1')
        .pipe(storeData)
        .pipe(append, '2')
        .pipe(mp.previous(by.task(storeData), '_stored'), 'Arg');

      write('abc').into(pipeline()).andExpect('abc1Arg').then(done);
    });

    it('should take the most recent step to restore', function (done) {
      var pipeline = mp()
        .pipe(append, '1')
        .pipe(storeData)
        .pipe(append, '2')
        .pipe(mp.previous(by.task(storeData), '_stored'), 'Earlier')
        .pipe(storeData)
        .pipe(append, '3')
        .pipe(mp.previous(by.task(storeData), '_stored'), 'Latest');

      write('abc').into(pipeline()).andExpect('abc1EarlierLatest').then(done);
    });
  });

  describe('insert()', function () {
    it('should be able to insert a task by constructor', function (done) {
      var pipeline = mp()
        .pipe(append, '1')
        .pipe(append, '3')
        .pipe(append, '4');

      pipeline = pipeline.insert(append, '2').after(pipeline.first(by.task(append)));

      write('abc').into(pipeline()).andExpect('abc1234').then(done);
    });
    it('should be able to replace with another mutapipe pipeline', function (done) {
      var pipeline = mp()
        .pipe(append, '1')
        .pipe(append, '4')
        .pipe(append, '5');

      pipeline = pipeline.insert(mp().pipe(append, '2').pipe(append, '3')).after(pipeline.first(by.task(append)));

      write('abc').into(pipeline()).andExpect('abc12345').then(done);
    });
    it('should be able to replace with a step object', function (done) {
      var pipeline = mp()
        .pipe(append, '1')
        .pipe(append, '3')
        .pipe(append, '4');

      pipeline = pipeline.insert({task: append, args: ['2']}).after(pipeline.first(by.task(append)));

      write('abc').into(pipeline()).andExpect('abc1234').then(done);
    });
    it('should leave the old pipeline intact', function (done) {
      var original = mp().pipe(append, '1').pipe(append, '3').pipe(append, '4');
      var replaced = original.insert(append, '2').after(original.first(by.task(append)));

      write('abc').into(replaced()).andExpect('abc1234').then(function () {
        write('abc').into(original()).andExpect('abc134').then(done);
      });
    });
    it('should support nesting', function (done) {
      var pipeline = mp()
        .pipe(append, 'After')
        .pipe(append, 'Regular')
        .pipe(append, 'End');

      pipeline = pipeline.insert(
        mp()
          .pipe(append, 'N1')
          .pipe(mp()
            .pipe(append, 'N2'))
      ).after(pipeline.first(by.args('After')));

      write('input').into(pipeline()).andExpect('inputAfterN1N2RegularEnd').then(done);
    });
    it('should work with selector function', function (done) {
      var pipeline = mp()
        .pipe(append, '1')
        .pipe(append, '3')
        .pipe(append, '4');

      pipeline = pipeline.insert(append, '2').after(by.task(append));

      write('abc').into(pipeline()).andExpect('abc1234').then(done);
    });
  });

  describe('without()', function () {
    it('should return the pipeline with the specified step excluded', function (done) {
      var pipeline = mp()
        .pipe(append, '1')
        .pipe(append, '2')
        .without(by.task(append));

      write('abc').into(pipeline()).andExpect('abc2').then(done);
    });
    it('should be immutable', function (done) {
      var original = mp().pipe(append, '1').pipe(append, '2');
      var replaced = original.without(original.first(by.task(append)));

      write('abc').into(replaced()).andExpect('abc2').then(function () {
        write('abc').into(original()).andExpect('abc12').then(done);
      });
    });
  });

  describe('replace()', function () {
    it('should be able to replace with constructor', function (done) {
      var pipeline = mp()
        .pipe(append, '1')
        .pipe(append, '2');

      pipeline = pipeline.replace(pipeline.first(by.task(append))).with(append, '3');

      write('abc').into(pipeline()).andExpect('abc32').then(done);
    });
    it('should be able to replace with a step object', function (done) {
      var pipeline = mp()
        .pipe(append, '1')
        .pipe(append, '2');

      pipeline = pipeline.replace(pipeline.first(by.task(append))).with({task: append, args: ['3']});

      write('abc').into(pipeline()).andExpect('abc32').then(done);
    });
    it('should be able to replace with another mutapipe pipeline', function (done) {
      var pipeline = mp()
        .pipe(append, '1')
        .pipe(append, '2');

      pipeline = pipeline.replace(pipeline.first(by.task(append)))
        .with(mp().pipe(append, '4').pipe(append, '3'));

      write('abc').into(pipeline()).andExpect('abc432').then(done);
    });
    it('should leave the old pipeline intact', function (done) {
      var original = mp().pipe(append, '1').pipe(append, '2');
      var replaced = original.replace(original.first(by.task(append))).with(append, '3');

      write('abc').into(replaced()).andExpect('abc32').then(function () {
        write('abc').into(original()).andExpect('abc12').then(done);
      });
    });
    it('should support nesting', function (done) {
      var pipeline = mp()
        .pipe(append, 'Replaced')
        .pipe(append, 'Regular')
        .pipe(append, 'End');

      pipeline = pipeline.replace(pipeline.first(by.args('Replaced'))).with(
        mp()
          .pipe(append, 'N1')
          .pipe(mp()
            .pipe(append, 'N2'))
      );

      write('input').into(pipeline()).andExpect('inputN1N2RegularEnd').then(done);
    });
    it('should work with selector function', function (done) {
      var pipeline = mp()
        .pipe(append, '1')
        .pipe(append, '2');

      pipeline = pipeline.replace(by.task(append)).with(append, '3');

      write('abc').into(pipeline()).andExpect('abc32').then(done);
    });
  });
});