Merge pull request #1643 from lebauce/workflow-tests

tests: add tests for workflows

Author: safchain

cmd/client/workflow.go

@@ -18,14 +18,11 @@
 package client
 
 import (
-	"fmt"
 	"io/ioutil"
 	"os"
 
 	"gopkg.in/yaml.v2"
 
-	"github.com/robertkrimen/otto"
-
 	"github.com/skydive-project/skydive/api/client"
 	"github.com/skydive-project/skydive/api/types"
 	"github.com/skydive-project/skydive/js"
@@ -173,40 +170,16 @@ var WorkflowCall = &cobra.Command{
 		runtime.Start()
 		runtime.RegisterAPIClient(client)
 
-		result, err := runtime.Exec("(" + workflow.Source + ")")
-		if err != nil {
-			exitOnError(fmt.Errorf("Error while compile workflow %s: %s", workflow.Source, result.String()))
-		}
-
 		params := make([]interface{}, len(args)-1)
 		for i, arg := range args[1:] {
 			params[i] = arg
 		}
 
-		result, err = result.Call(result, params...)
+		result, err := runtime.ExecPromise(workflow.Source, params...)
 		if err != nil {
-			exitOnError(fmt.Errorf("Error while executing workflow: %s", result.String()))
-		}
-
-		if !result.IsObject() {
-			exitOnError(fmt.Errorf("Workflow is expected to return a promise, returned %s", result.Class()))
+			exitOnError(err)
 		}
 
-		done := make(chan otto.Value)
-		promise := result.Object()
-
-		finally, err := runtime.ToValue(func(call otto.FunctionCall) otto.Value {
-			result := call.Argument(0)
-			done <- result
-			return result
-		})
-
-		result, _ = promise.Call("then", finally)
-		promise = result.Object()
-		promise.Call("catch", finally)
-
-		result = <-done
-
 		runtime.Set("result", result)
 		runtime.Exec("console.log(JSON.stringify(result))")
 	},

graffiti/graph/graph.go

@@ -103,14 +103,14 @@ type graphElement struct {
 
 // Node of the graph
 type Node struct {
-	graphElement
+	graphElement `mapstructure:",squash"`
 }
 
 // Edge of the graph linked by a parent and a child
 type Edge struct {
-	graphElement
-	Parent Identifier
-	Child  Identifier
+	graphElement `mapstructure:",squash"`
+	Parent       Identifier
+	Child        Identifier
 }
 
 // Graph errors

js/otto.js

@@ -31,4 +31,6 @@ function require(m) {
     return {}
 }
 
+var api = require("./api");
+
 Promise.prototype.finally = Promise.prototype.done

js/runtime.go

@@ -27,6 +27,7 @@ import (
 	"math/rand"
 	"os"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/robertkrimen/otto"
@@ -55,6 +56,7 @@ type jsTimer struct {
 // Runtime is a Skydive JavaScript runtime environment
 type Runtime struct {
 	*otto.Otto
+	sync.Mutex
 	evalQueue     chan *evalReq
 	stopEventLoop chan bool
 	closed        chan struct{}
@@ -386,19 +388,70 @@ func (r *Runtime) Do(fn func(*otto.Otto)) {
 	<-done
 }
 
-// Exec executes some JavaScript code
+// Exec queues the execution of some JavaScript code
 func (r *Runtime) Exec(code string) (v otto.Value, err error) {
 	r.Do(func(vm *otto.Otto) { v, err = vm.Run(code) })
 	return v, err
 }
 
+// ExecFunction queues a CallFunction method
+func (r *Runtime) ExecFunction(source string, params ...interface{}) (v otto.Value, err error) {
+	r.Do(func(vm *otto.Otto) { v, err = r.CallFunction(source, params...) })
+	return v, err
+}
+
+// ExecPromise executes a promise and return its result
+func (r *Runtime) ExecPromise(source string, params ...interface{}) (v otto.Value, err error) {
+	var done chan otto.Value
+	r.Do(func(vm *otto.Otto) { done, err = r.CallPromise(source, params...) })
+	v = <-done
+	return v, err
+}
+
+// CallFunction takes the source of a function and evaluate it with the specifed parameters
+func (r *Runtime) CallFunction(source string, params ...interface{}) (otto.Value, error) {
+	result, err := r.Run("(" + source + ")")
+	if err != nil {
+		return otto.UndefinedValue(), fmt.Errorf("Error while compile source %s: %s", source, result.String())
+	}
+
+	return result.Call(result, params...)
+}
+
+// CallPromise takes the source of a promise and evaluate it with the specifed parameters
+func (r *Runtime) CallPromise(source string, params ...interface{}) (chan otto.Value, error) {
+	result, err := r.CallFunction(source, params...)
+	if err != nil {
+		return nil, fmt.Errorf("Error while executing function: %s", err)
+	}
+
+	if !result.IsObject() {
+		return nil, fmt.Errorf("Workflow is expected to return a promise, returned %s", result.Class())
+	}
+
+	done := make(chan otto.Value)
+	promise := result.Object()
+	finally, err := r.ToValue(func(call otto.FunctionCall) otto.Value {
+		result = call.Argument(0)
+		done <- result
+		return result
+	})
+
+	result, _ = promise.Call("then", finally)
+	promise = result.Object()
+	promise.Call("catch", finally)
+
+	return done, nil
+}
+
 // Start the runtime evaluation loop
 func (r *Runtime) Start() {
 	go r.runEventLoop()
 }
 
 // Stop the runtime evaluation loop
 func (r *Runtime) Stop() {
+	r.stopEventLoop <- true
 }
 
 // NewRuntime returns a new JavaScript runtime environment

statics/workflows/flow-matrix.yaml

@@ -20,6 +20,8 @@ parameters:
     type: boolean
 source: |
     function FlowMatrix(useFlows, format, debug) {
+        format = format || "graph"
+
         var G = client.gremlin.G()
         var at;
         var listenPorts = []
@@ -30,8 +32,8 @@ source: |
         var hostAddresses = {}
 
         var isLoopback = function(addr) {
-            return addr.startsWith("127.") ||
-                   addr.startsWith("::1/")
+            return addr.indexOf("127.") == 0 ||
+                   addr.indexOf("::1/") == 0
         }
 
         var isListenSocket = function(socket) {
@@ -251,7 +253,7 @@ source: |
             }
         }
 
-        var buildGraph = function() {
+        var outputGraph = function() {
             log("Matrix", matrix)
 
             var endpoints = buildEndpoints()
@@ -311,7 +313,7 @@ source: |
                 case "dot":
                     return outputDot()
                 case "graph":
-                    return buildGraph()
+                    return outputGraph()
             }
         }
 

tests/tests.go

@@ -84,6 +84,7 @@ agent:
       - lxd
       - lldp
       - runc
+      - socketinfo
       {{.OpencontrailProbe}}
     netlink:
       metrics_update: 5