Add fission integration tests

pkg/fnenv/fission/runtime.go

@@ -40,8 +40,9 @@ const (
 
 func NewFunctionEnv(executor *executor.Client, routerURL string) *FunctionEnv {
 	return &FunctionEnv{
-		executor:  executor,
-		routerURL: routerURL,
+		executor:         executor,
+		routerURL:        routerURL,
+		timedExecService: newTimedExecPool(),
 	}
 }
 
@@ -110,7 +111,7 @@ func (fe *FunctionEnv) Invoke(spec *types.TaskInvocationSpec) (*types.TaskInvoca
 }
 
 // Notify signals the Fission runtime that a function request is expected at a specific time.
-func (fe *FunctionEnv) Notify(taskID string, fn types.FnRef, expectedAt time.Time) error {
+func (fe *FunctionEnv) Notify(fn types.FnRef, expectedAt time.Time) error {
 	reqURL, err := fe.getFnURL(fn)
 	if err != nil {
 		return err

pkg/fnenv/fission/timed.go

@@ -24,7 +24,7 @@ func newTimedExecPool() *timedExecPool {
 func (ds *timedExecPool) Submit(fn func(), execAt time.Time) {
 	ds.fnsLock.Lock()
 	defer ds.fnsLock.Unlock()
-	ds.fnQueue.Push(timedFn{
+	ds.fnQueue.Push(&timedFn{
 		execAt: execAt,
 		fn:     fn,
 	})

pkg/fnenv/fnenv.go

@@ -88,7 +88,7 @@ type Notifier interface {
 	// a signal that the function invocation will come (almost) immediately. fnId is an optional
 	// identifier for the signal, which the implementation can use this to identify signals.
 	// By default, if fnId is empty, it is not possible to later update the notification.
-	Notify(taskID string, fn types.FnRef, expectedAt time.Time) error
+	Notify(fn types.FnRef, expectedAt time.Time) error
 }
 
 // Resolver resolves a reference to a function to a deterministic, unique function id.

test/integration/bundle/bundle_test.go

@@ -3,7 +3,6 @@ package bundle
 
 import (
 	"context"
-	"fmt"
 	"os"
 	"strings"
 	"testing"
@@ -15,6 +14,7 @@ import (
 	"github.com/fission/fission-workflows/pkg/types/typedvalues"
 	"github.com/fission/fission-workflows/test/integration"
 	"github.com/golang/protobuf/ptypes/empty"
+	log "github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
 	"google.golang.org/grpc"
 )
@@ -25,7 +25,7 @@ const (
 
 func TestMain(m *testing.M) {
 	if testing.Short() {
-		fmt.Println("Skipping bundle integration tests...")
+		log.Info("Short test; skipping bundle integration tests.")
 		return
 	}
 

test/integration/fission/fn.js

@@ -0,0 +1,8 @@
+module.exports = async (context) => {
+    const body = JSON.stringify(context.request.body);
+    const headers = JSON.stringify(context.request.headers);
+    return {
+        status: 200,
+        body: `body:${body}\nheaders:${headers}`
+    };
+}

test/integration/fission/runtime_test.go

@@ -0,0 +1,105 @@
+package fission
+
+import (
+	"context"
+	"os"
+	"os/exec"
+	"testing"
+	"time"
+
+	"github.com/fission/fission-workflows/pkg/fnenv/fission"
+	"github.com/fission/fission-workflows/pkg/types"
+	"github.com/fission/fission-workflows/pkg/types/typedvalues"
+	controllerclient "github.com/fission/fission/controller/client"
+	executorclient "github.com/fission/fission/executor/client"
+	log "github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+)
+
+// Future: discover executor of fission deployment
+const (
+	executorURL   = "http://executor.fission"
+	routerURL     = "http://router.fission"
+	controllerURL = "http://controller.fission"
+)
+
+var executor = executorclient.MakeClient(executorURL)
+var controller = controllerclient.MakeClient(controllerURL)
+var testFnName = "fission-runtime-test"
+
+// Currently we assume that fission is present (along with the CLI) and kubectl.
+func TestMain(m *testing.M) {
+	if testing.Short() {
+		log.Info("Short test; skipping Fission integration tests")
+		return
+	}
+
+	ctx, cancelFn := context.WithTimeout(context.Background(), 10*time.Second)
+
+	// Test if Fission is present
+	if err := exec.CommandContext(ctx, "fission", "fn", "list").Run(); err != nil {
+		log.Fatalf("Fission is not present: %v", err)
+	}
+
+	// Test if cluster is accessible
+	if err := exec.CommandContext(ctx, "curl", controllerURL).Run(); err != nil {
+		log.Fatalf("Fission cluster (%v) is not accessible: %v", controllerURL, err)
+	}
+
+	// Setup test environment and function (use fission CLI because client is not exposed correctly)
+	if err := exec.Command("fission", "spec", "apply").Run(); err != nil {
+		log.Fatalf("Failed to create test resources in fission: %v", err)
+	}
+	log.Info("Fission integration test resources setup.")
+
+	status := m.Run()
+
+	// Clean up test function and env
+	cancelFn()
+	if err := exec.Command("fission", "spec", "destroy").Run(); err != nil {
+		log.Fatalf("Failed to destroy test resources in fission: %v", err)
+	}
+	log.Info("Cleaned up fission test resources.")
+	os.Exit(status)
+}
+
+func TestFnenvResolve(t *testing.T) {
+	resolver := fission.NewResolver(controller)
+	resolved, err := resolver.Resolve(testFnName)
+	assert.NoError(t, err)
+	assert.Equal(t, testFnName, resolved)
+}
+
+func TestFnenvNotify(t *testing.T) {
+	fnref := types.NewFnRef(fission.Name, testFnName)
+	fnenv := fission.NewFunctionEnv(executor, routerURL)
+	err := fnenv.Notify(fnref, time.Now().Add(100*time.Millisecond))
+	assert.NoError(t, err)
+}
+
+func TestFnenvInvoke(t *testing.T) {
+	fnref := types.NewFnRef(fission.Name, testFnName)
+	fnenv := fission.NewFunctionEnv(executor, routerURL)
+	body := "stubBodyVal"
+	headerVal := "stub-header-val"
+	headerKey := "stub-header-key"
+
+	result, err := fnenv.Invoke(&types.TaskInvocationSpec{
+		TaskId:       "fooTask",
+		InvocationId: "fooInvocation",
+		Inputs: types.Inputs{
+			"default": typedvalues.MustParse(body),
+			"headers": typedvalues.MustParse(map[string]interface{}{
+				headerKey: headerVal,
+			}),
+		},
+		FnRef: &fnref,
+	})
+	output := typedvalues.MustFormat(result.Output)
+	assert.NoError(t, err)
+	assert.True(t, result.Finished())
+	assert.NotEmpty(t, output)
+	assert.Contains(t, output, body)
+	assert.Contains(t, output, headerVal)
+	assert.Contains(t, output, headerKey)
+}

test/integration/fission/specs/env-fission-runtime-test.yaml

@@ -0,0 +1,17 @@
+apiVersion: fission.io/v1
+kind: Environment
+metadata:
+  creationTimestamp: null
+  name: fission-runtime-test
+  namespace: default
+spec:
+  TerminationGracePeriod: 360
+  builder: {}
+  poolsize: 3
+  resources: {}
+  runtime:
+    functionendpointport: 0
+    image: fission/node-env
+    loadendpointpath: ""
+    loadendpointport: 0
+  version: 1

test/integration/fission/specs/fission-deployment-config.yaml

@@ -0,0 +1,7 @@
+# This file is generated by the 'fission spec init' command.
+# See the README in this directory for background and usage information.
+# Do not edit the UID below: that will break 'fission spec apply'
+apiVersion: fission.io/v1
+kind: DeploymentConfig
+name: fission
+uid: 3aacd049-9a2c-45d1-b9db-16ad4d6635aa

test/integration/fission/specs/function-fission-runtime-test.yaml

@@ -0,0 +1,50 @@
+include:
+- fn.js
+kind: ArchiveUploadSpec
+name: fn-js
+
+---
+apiVersion: fission.io/v1
+kind: Package
+metadata:
+  creationTimestamp: null
+  name: fn-js-faqo
+  namespace: default
+spec:
+  deployment:
+    checksum: {}
+    type: url
+    url: archive://fn-js
+  environment:
+    name: fission-runtime-test
+    namespace: default
+  source:
+    checksum: {}
+status:
+  buildstatus: none
+
+---
+apiVersion: fission.io/v1
+kind: Function
+metadata:
+  creationTimestamp: null
+  name: fission-runtime-test
+  namespace: default
+spec:
+  InvokeStrategy:
+    ExecutionStrategy:
+      ExecutorType: poolmgr
+      MaxScale: 1
+      MinScale: 0
+      TargetCPUPercent: 80
+    StrategyType: execution
+  configmaps: null
+  environment:
+    name: fission-runtime-test
+    namespace: default
+  package:
+    packageref:
+      name: fn-js-faqo
+      namespace: default
+  resources: {}
+  secrets: null