Complete test coverage on flow engine refactor

red/nodes/Node.js

@@ -39,6 +39,7 @@ function Node(n) {
 util.inherits(Node, EventEmitter);
 
 Node.prototype.updateWires = function(wires) {
+    //console.log("UPDATE",this.id);
     this.wires = wires || [];
     delete this._wire;
 

red/nodes/credentials.js

@@ -68,7 +68,7 @@ module.exports = {
         storage = _storage;
         redApp = _app;
     },
-    
+
     /**
      * Loads the credentials from storage.
      */
@@ -79,7 +79,7 @@ module.exports = {
             log.warn(log._("nodes.credentials.error",{message: err}));
         });
     },
-    
+
     /**
      * Adds a set of credentials for the given node id.
      * @param id the node id for the credentials
@@ -118,7 +118,7 @@ module.exports = {
     clean: function (config) {
         var existingIds = {};
         config.forEach(function(n) {
-            existingIds[n.id] = true;     
+            existingIds[n.id] = true;
         });
         var deletedCredentials = false;
         for (var c in credentialCache) {
@@ -135,7 +135,7 @@ module.exports = {
             return when.resolve();
         }
     },
-    
+
     /**
      * Registers a node credential definition.
      * @param type the node type
@@ -146,14 +146,14 @@ module.exports = {
         credentialsDef[dashedType] = definition;
         registerEndpoint(dashedType);
     },
-    
+
     /**
      * Extracts and stores any credential updates in the provided node.
      * The provided node may have a .credentials property that contains
      * new credentials for the node.
      * This function loops through the credentials in the definition for
      * the node-type and applies any of the updates provided in the node.
-     * 
+     *
      * This function does not save the credentials to disk as it is expected
      * to be called multiple times when a new flow is deployed.
      *
@@ -171,7 +171,7 @@ module.exports = {
                 log.warn(log._("nodes.credentials.not-registered",{type:nodeType}));
                 return;
             }
-            
+
             for (var cred in definition) {
                 if (definition.hasOwnProperty(cred)) {
                     if (newCreds[cred] === undefined) {
@@ -191,7 +191,7 @@ module.exports = {
             delete node.credentials;
         }
     },
-    
+
     /**
      * Saves the credentials to storage
      * @return a promise for the saving of credentials to storage
@@ -199,7 +199,7 @@ module.exports = {
     save: function () {
         return storage.saveCredentials(credentialCache);
     },
-    
+
     /**
      * Gets the credential definition for the given node type
      * @param type the node type

red/nodes/flows/Flow.js

@@ -212,6 +212,7 @@ function Flow(global,flow) {
                         source: {
                             id: node.id,
                             type: node.type,
+                            name: node.name,
                             count: count
                         }
                     };
@@ -258,11 +259,12 @@ function mapEnvVarProperties(obj,prop) {
 }
 
 function createNode(type,config) {
-    //console.log("CREATE",type,config.id);
+    // console.log("CREATE",type,config.id);
     var nn = null;
     var nt = typeRegistry.get(type);
     if (nt) {
         var conf = clone(config);
+        delete conf.credentials;
         for (var p in conf) {
             if (conf.hasOwnProperty(p)) {
                 mapEnvVarProperties(conf,p);
@@ -371,9 +373,7 @@ function createSubflow(sf,sfn,subflows,globalSubflows,activeNodes) {
         if (sf.out) {
             var node,wires,i,j;
             // Restore the original wiring to the internal nodes
-
             subflowInstance.wires = clone(subflowInstance._originalWires);
-
             for (i=0;i<sf.out.length;i++) {
                 wires = sf.out[i].wires;
                 for (j=0;j<wires.length;j++) {

red/nodes/flows/index.js

@@ -39,22 +39,30 @@ var started = false;
 
 var activeNodesToFlow = {};
 
+var typeEventRegistered = false;
+
 function init(_settings, _storage) {
+    if (started) {
+        throw new Error("Cannot init without a stop");
+    }
     settings = _settings;
     storage = _storage;
     started = false;
-    events.on('type-registered',function(type) {
+    if (!typeEventRegistered) {
-        if (activeFlowConfig && activeFlowConfig.missingTypes.length > 0) {
+        events.on('type-registered',function(type) {
-            var i = activeFlowConfig.missingTypes.indexOf(type);
+            if (activeFlowConfig && activeFlowConfig.missingTypes.length > 0) {
-            if (i != -1) {
+                var i = activeFlowConfig.missingTypes.indexOf(type);
-                log.info(log._("nodes.flows.registered-missing", {type:type}));
+                if (i != -1) {
-                activeFlowConfig.missingTypes.splice(i,1);
+                    log.info(log._("nodes.flows.registered-missing", {type:type}));
-                if (activeFlowConfig.missingTypes.length === 0 && started) {
+                    activeFlowConfig.missingTypes.splice(i,1);
-                    start();
+                    if (activeFlowConfig.missingTypes.length === 0 && started) {
+                        start();
+                    }
                 }
             }
-        }
+        });
-    });
+        typeEventRegistered = true;
+    }
 }
 function load() {
     return storage.getFlows().then(function(flows) {
@@ -67,21 +75,25 @@ function load() {
     });
 }
 
-function setConfig(config,type) {
+function setConfig(_config,type) {
+    var config = clone(_config);
     type = type||"full";
 
     var credentialsChanged = false;
     var credentialSavePromise = null;
     var configSavePromise = null;
 
-    var cleanConfig = clone(config);
+    var diff;
-    cleanConfig.forEach(function(node) {
+    var newFlowConfig = flowUtil.parseConfig(clone(config));
+    if (type !== 'full' && type !== 'load') {
+        diff = flowUtil.diffConfigs(activeFlowConfig,newFlowConfig);
+    }
+    config.forEach(function(node) {
         if (node.credentials) {
             credentials.extract(node);
             credentialsChanged = true;
         }
     });
-
     if (credentialsChanged) {
         credentialSavePromise = credentials.save();
     } else {
@@ -92,27 +104,19 @@ function setConfig(config,type) {
         type = 'full';
     } else {
         configSavePromise = credentialSavePromise.then(function() {
-            return storage.saveFlows(cleanConfig);
+            return storage.saveFlows(config);
         });
     }
 
     return configSavePromise
         .then(function() {
-            var diff;
+            activeConfig = config;
-            activeConfig = cleanConfig;
+            activeFlowConfig = newFlowConfig;
-            if (type === 'full') {
+            return credentials.clean(activeConfig).then(function() {
-                activeFlowConfig = flowUtil.parseConfig(clone(config));
-            } else {
-                var newConfig = flowUtil.parseConfig(clone(config));
-                diff = flowUtil.diffConfigs(activeFlowConfig,newConfig);
-                activeFlowConfig = newConfig;
-            }
-            credentials.clean(activeConfig).then(function() {
                 if (started) {
                     return stop(type,diff).then(function() {
                         start(type,diff);
                     }).otherwise(function(err) {
-                        console.log(err);
                     })
                 }
             });
@@ -136,9 +140,9 @@ function getNode(id) {
 }
 
 function eachNode(cb) {
-    for (var id in activeFlowConfig.nodes) {
+    for (var id in activeFlowConfig.allNodes) {
-        if (activeFlowConfig.nodes.hasOwnProperty(id)) {
+        if (activeFlowConfig.allNodes.hasOwnProperty(id)) {
-            cb(activeFlowConfig.nodes[id]);
+            cb(activeFlowConfig.allNodes[id]);
         }
     }
 }

test/red/nodes/flows/index_spec.js

@@ -26,30 +26,10 @@ var credentials = require("../../../../red/nodes/credentials");
 var typeRegistry = require("../../../../red/nodes/registry");
 var Flow = require("../../../../red/nodes/flows/Flow");
 
-var settings = {
-    available: function() { return false; }
-}
-
-function loadFlows(testFlows, cb) {
-    var storage = {
-        getFlows: function() {
-            return when.resolve(testFlows);
-        },
-        getCredentials: function() {
-            return when.resolve({});
-        }
-    };
-    RED.init(settings, storage);
-    flows.load().then(function() {
-        should.deepEqual(testFlows, flows.getFlows());
-        cb();
-    });
-}
-
 describe('flows/index', function() {
 
-    var eventsOn;
     var storage;
+    var eventsOn;
     var credentialsExtract;
     var credentialsSave;
     var credentialsClean;
@@ -60,7 +40,7 @@ describe('flows/index', function() {
 
     before(function() {
         getType = sinon.stub(typeRegistry,"get",function(type) {
-            return type!=='missing';
+            return type.indexOf('missing') === -1;
         });
     });
     after(function() {
@@ -69,7 +49,7 @@ describe('flows/index', function() {
 
 
     beforeEach(function() {
-        eventsOn = sinon.stub(events,"on",function(evt,handler) {});
+        eventsOn = sinon.spy(events,"on");
         credentialsExtract = sinon.stub(credentials,"extract",function(conf) {
             delete conf.credentials;
         });
@@ -91,9 +71,17 @@ describe('flows/index', function() {
                 id = flow.id;
             }
             flowCreate.flows[id] = {
+                flow: flow,
+                global: global,
                 start: sinon.spy(),
                 update: sinon.spy(),
-                stop: sinon.spy()
+                stop: sinon.spy(),
+                getActiveNodes: function() {
+                    return flow.nodes||{};
+                },
+                handleError: sinon.spy(),
+                handleStatus: sinon.spy()
+
             }
             return flowCreate.flows[id];
         });
@@ -107,20 +95,23 @@ describe('flows/index', function() {
         }
     });
 
-    afterEach(function() {
+    afterEach(function(done) {
         eventsOn.restore();
         credentialsExtract.restore();
         credentialsSave.restore();
         credentialsClean.restore();
         credentialsLoad.restore();
         flowCreate.restore();
+
+        flows.stopFlows().then(done);
+
     });
-    describe('#init',function() {
+    // describe('#init',function() {
-        it('registers the type-registered handler', function() {
+    //     it('registers the type-registered handler', function() {
-            flows.init({},{});
+    //         flows.init({},{});
-            eventsOn.calledOnce.should.be.true;
+    //         eventsOn.calledOnce.should.be.true;
-        });
+    //     });
-    });
+    // });
 
     describe('#setFlows', function() {
         it('sets the full flow', function(done) {
@@ -166,12 +157,70 @@ describe('flows/index', function() {
                 credentialsClean.called.should.be.true;
                 storage.hasOwnProperty('conf').should.be.true;
                 var cleanedFlows = flows.getFlows();
+                storage.conf.should.eql(cleanedFlows);
                 cleanedFlows.should.not.eql(originalConfig);
                 cleanedFlows[0].credentials = {};
                 cleanedFlows.should.eql(originalConfig);
                 done();
             });
+        });
 
+        it('updates existing flows with partial deployment - nodes', function(done) {
+            var originalConfig = [
+                {id:"t1-1",x:10,y:10,z:"t1",type:"test",wires:[]},
+                {id:"t1",type:"tab"}
+            ];
+            var newConfig = clone(originalConfig);
+            newConfig.push({id:"t1-2",x:10,y:10,z:"t1",type:"test",wires:[]});
+            newConfig.push({id:"t2",type:"tab"});
+            newConfig.push({id:"t2-1",x:10,y:10,z:"t2",type:"test",wires:[]});
+            storage.getFlows = function() {
+                return when.resolve(originalConfig);
+            }
+
+            events.once('nodes-started',function() {
+                flows.setFlows(newConfig,"nodes").then(function() {
+                    flows.getFlows().should.eql(newConfig);
+                    flowCreate.flows['t1'].update.called.should.be.true;
+                    flowCreate.flows['t2'].start.called.should.be.true;
+                    flowCreate.flows['_GLOBAL_'].update.called.should.be.true;
+                    done();
+                })
+            });
+
+            flows.init({},storage);
+            flows.load().then(function() {
+                flows.startFlows();
+            });
+        });
+
+        it('updates existing flows with partial deployment - flows', function(done) {
+            var originalConfig = [
+                {id:"t1-1",x:10,y:10,z:"t1",type:"test",wires:[]},
+                {id:"t1",type:"tab"}
+            ];
+            var newConfig = clone(originalConfig);
+            newConfig.push({id:"t1-2",x:10,y:10,z:"t1",type:"test",wires:[]});
+            newConfig.push({id:"t2",type:"tab"});
+            newConfig.push({id:"t2-1",x:10,y:10,z:"t2",type:"test",wires:[]});
+            storage.getFlows = function() {
+                return when.resolve(originalConfig);
+            }
+
+            events.once('nodes-started',function() {
+                flows.setFlows(newConfig,"nodes").then(function() {
+                    flows.getFlows().should.eql(newConfig);
+                    flowCreate.flows['t1'].update.called.should.be.true;
+                    flowCreate.flows['t2'].start.called.should.be.true;
+                    flowCreate.flows['_GLOBAL_'].update.called.should.be.true;
+                    flows.stopFlows().then(done);
+                })
+            });
+
+            flows.init({},storage);
+            flows.load().then(function() {
+                flows.startFlows();
+            });
         });
 
     });
@@ -195,12 +244,84 @@ describe('flows/index', function() {
                 flows.getFlows().should.eql(originalConfig);
                 done();
             });
-
         });
     });
 
     describe('#startFlows', function() {
-        it.skip('starts the loaded config', function(done) {
+        it('starts the loaded config', function(done) {
+            var originalConfig = [
+                {id:"t1-1",x:10,y:10,z:"t1",type:"test",wires:[]},
+                {id:"t1",type:"tab"}
+            ];
+            storage.getFlows = function() {
+                return when.resolve(originalConfig);
+            }
+
+            events.once('nodes-started',function() {
+                Object.keys(flowCreate.flows).should.eql(['_GLOBAL_','t1']);
+                done();
+            });
+
+            flows.init({},storage);
+            flows.load().then(function() {
+                flows.startFlows();
+            });
+        });
+        it('does not start if nodes missing', function(done) {
+            var originalConfig = [
+                {id:"t1-1",x:10,y:10,z:"t1",type:"test",wires:[]},
+                {id:"t1-2",x:10,y:10,z:"t1",type:"missing",wires:[]},
+                {id:"t1",type:"tab"}
+            ];
+            storage.getFlows = function() {
+                return when.resolve(originalConfig);
+            }
+
+            flows.init({},storage);
+            flows.load().then(function() {
+                flows.startFlows();
+                flowCreate.called.should.be.false;
+                done();
+            });
+        });
+
+        it('starts when missing nodes registered', function(done) {
+            var originalConfig = [
+                {id:"t1-1",x:10,y:10,z:"t1",type:"test",wires:[]},
+                {id:"t1-2",x:10,y:10,z:"t1",type:"missing",wires:[]},
+                {id:"t1-3",x:10,y:10,z:"t1",type:"missing2",wires:[]},
+                {id:"t1",type:"tab"}
+            ];
+            storage.getFlows = function() {
+                return when.resolve(originalConfig);
+            }
+            flows.init({},storage);
+            flows.load().then(function() {
+                flows.startFlows();
+                flowCreate.called.should.be.false;
+
+                events.emit("type-registered","missing");
+                setTimeout(function() {
+                    flowCreate.called.should.be.false;
+                    events.emit("type-registered","missing2");
+                    setTimeout(function() {
+                        flowCreate.called.should.be.true;
+                        done();
+                    },10);
+                },10);
+            });
+        });
+
+
+
+    });
+
+    describe('#get',function() {
+
+    });
+
+    describe('#eachNode', function() {
+        it('iterates the flow nodes', function(done) {
             var originalConfig = [
                 {id:"t1-1",x:10,y:10,z:"t1",type:"test",wires:[]},
                 {id:"t1",type:"tab"}
@@ -210,29 +331,62 @@ describe('flows/index', function() {
             }
             flows.init({},storage);
             flows.load().then(function() {
-                flows.startFlows();
+                var c = 0;
-                // TODO: PICK IT UP FROM HERE
+                flows.eachNode(function(node) {
+                    c++
+                })
+                c.should.equal(2);
                 done();
             });
         });
     });
 
-    describe('#get',function() {
-
-    });
-
-    describe('#eachNode', function() {
-
-    });
-
     describe('#stopFlows', function() {
 
     });
     describe('#handleError', function() {
+        it('passes error to correct flow', function(done) {
+            var originalConfig = [
+                {id:"t1-1",x:10,y:10,z:"t1",type:"test",wires:[]},
+                {id:"t1",type:"tab"}
+            ];
+            storage.getFlows = function() {
+                return when.resolve(originalConfig);
+            }
 
+            events.once('nodes-started',function() {
+                flows.handleError(originalConfig[0],"message",{});
+                flowCreate.flows['t1'].handleError.called.should.be.true;
+                done();
+            });
+
+            flows.init({},storage);
+            flows.load().then(function() {
+                flows.startFlows();
+            });
+        });
     });
     describe('#handleStatus', function() {
+        it('passes status to correct flow', function(done) {
+            var originalConfig = [
+                {id:"t1-1",x:10,y:10,z:"t1",type:"test",wires:[]},
+                {id:"t1",type:"tab"}
+            ];
+            storage.getFlows = function() {
+                return when.resolve(originalConfig);
+            }
 
+            events.once('nodes-started',function() {
+                flows.handleStatus(originalConfig[0],"message");
+                flowCreate.flows['t1'].handleStatus.called.should.be.true;
+                done();
+            });
+
+            flows.init({},storage);
+            flows.load().then(function() {
+                flows.startFlows();
+            });
+        });
     });
 });
 

test/red/nodes/index_spec.js

@@ -21,8 +21,15 @@ var when = require("when");
 var sinon = require('sinon');
 
 var index = require("../../../red/nodes/index");
+var flows = require("../../../red/nodes/flows");
 
 describe("red/nodes/index", function() {
+    before(function() {
+        sinon.stub(flows,"startFlows");
+    });
+    after(function() {
+        flows.startFlows.restore();
+    });
 
     afterEach(function() {
         index.clearRegistry();