func TestPassUndeclaredFlowParameter(t *testing.T) {
	flow := mocks.MakeFlow("flow")
	flow.Flow.Parameters = map[string]client.FlowParameter{
		"arg1": mocks.MakeFlowParameter("p"),
	}
	c := mocks.NewClient(
		mocks.MakeFlow("flow"),
		mocks.MakeResourceDefinition("job/ready-1"),
		mocks.MakeDependency("flow/flow", "job/ready-1", "flow=flow"),
	)
	_, err := New(c, nil, 0).BuildDependencyGraph(
		interfaces.DependencyGraphOptions{
			ReplicaCount: 1,
			FlowName:     "flow",
			Args: map[string]string{
				"arg2": "x",
			}})
	if err == nil {
		t.Error("pass of undeclared argument should not succeed")
	} else if err.Error() != "unexpected argument arg2" {
		t.Errorf("unexpected error: %v", err)
	}
	_, err = New(c, nil, 0).BuildDependencyGraph(
		interfaces.DependencyGraphOptions{
			ReplicaCount:        1,
			FlowName:            "flow",
			AllowUndeclaredArgs: true,
			Args: map[string]string{
				"arg2": "x",
			}})
	if err != nil {
		t.Error("pass of undeclared argument should succeed if it was explicitly permited")
	}
}
TestPassUndeclaredFlowParameter tests that by-default only declared flow arguments can be passed
package scheduler
import (
	"errors"
	"strings"
	"testing"
	"github.com/Mirantis/k8s-AppController/pkg/client"
	"github.com/Mirantis/k8s-AppController/pkg/interfaces"
	"github.com/Mirantis/k8s-AppController/pkg/mocks"
	"k8s.io/client-go/pkg/api"
	api_v1 "k8s.io/client-go/pkg/api/v1"
	batchapiv1 "k8s.io/client-go/pkg/apis/batch/v1"
	"k8s.io/client-go/pkg/runtime"
	k8stesting "k8s.io/client-go/testing"
)
func TestFlowResourcesIdentified(t *testing.T) {
	c := mocks.NewClient(
		mocks.MakeFlow("test"),
		mocks.MakeResourceDefinition("job/1"),
		mocks.MakeResourceDefinition("job/2"),
		mocks.MakeResourceDefinition("job/3"),
		mocks.MakeResourceDefinition("job/4"),
		mocks.MakeDependency("flow/test", "job/1", "flow=test"),
		mocks.MakeDependency("job/1", "job/2", "flow=test"),
		mocks.MakeDependency("job/2", "job/3"),  
	)
	depGraph, err := New(c, nil, 0).BuildDependencyGraph(
		interfaces.DependencyGraphOptions{
			FlowName:     "test",
			ReplicaCount: 1,
		})
	if err != nil {
		t.Fatal(err)
	}
	graph := depGraph.(*dependencyGraph).graph
	expectedLength := 2
	if len(graph) != expectedLength {
		t.Errorf("wrong length of a graph %d != %d", expectedLength, len(graph))
	}
	if _, found := graph["job/1"]; !found {
		t.Error("job/1 is not in the graph")
	}
	if _, found := graph["job/2"]; !found {
		t.Error("job/2 is not in the graph")
	}
}
func TestTopLevelResourcesAssignedToDefaultFlow(t *testing.T) {
	c := mocks.NewClient(
		mocks.MakeFlow("test"),
		mocks.MakeResourceDefinition("job/1"),
		mocks.MakeResourceDefinition("job/2"),
		mocks.MakeResourceDefinition("job/3"),
		mocks.MakeResourceDefinition("job/4"),
		mocks.MakeResourceDefinition("job/5"),
		mocks.MakeDependency("flow/test", "job/1", "flow=test"),
		mocks.MakeDependency("job/1", "job/2", "flow=test"),
		mocks.MakeDependency("job/2", "job/3"),
		mocks.MakeDependency("job/4", "job/5"),
	)
	depGraph, err := New(c, nil, 0).BuildDependencyGraph(
		interfaces.DependencyGraphOptions{ReplicaCount: 1})
	if err != nil {
		t.Fatal(err)
	}
	graph := depGraph.(*dependencyGraph).graph
	expectedLength := 2
	if len(graph) != expectedLength {
		t.Errorf("wrong length of a graph %d != %d", expectedLength, len(graph))
	}
	if _, found := graph["job/4"]; !found {
		t.Error("job/4 is not in the graph")
	}
	if _, found := graph["job/5"]; !found {
		t.Error("job/5 is not in the graph")
	}
}
func TestTriggerFlowIndependently(t *testing.T) {
	c := mocks.NewClient(
		mocks.MakeFlow("flow1"),
		mocks.MakeFlow("flow2"),
		mocks.MakeResourceDefinition("job/ready-flow1-1"),
		mocks.MakeResourceDefinition("job/ready-flow1-2"),
		mocks.MakeResourceDefinition("job/ready-flow2-1"),
		mocks.MakeResourceDefinition("job/ready-flow2-2"),
		mocks.MakeDependency("flow/flow1", "job/ready-flow1-1", "flow=flow1"),
		mocks.MakeDependency("job/ready-flow1-1", "flow/flow2", "flow=flow1"),
		mocks.MakeDependency("flow/flow2", "job/ready-flow1-2", "flow=flow1"),
		mocks.MakeDependency("flow/flow2", "job/ready-flow2-1", "flow=flow2"),
		mocks.MakeDependency("job/ready-flow2-1", "job/ready-flow2-2", "flow=flow2"),
	)
	depGraph, err := New(c, nil, 0).BuildDependencyGraph(
		interfaces.DependencyGraphOptions{ReplicaCount: 1, FlowName: "flow2"})
	if err != nil {
		t.Fatal(err)
	}
	jobs, err := c.Jobs().List(api_v1.ListOptions{})
	if err != nil {
		t.Fatal(err)
	}
	if len(jobs.Items) != 0 {
		t.Fatal("Job list is not empty")
	}
	stopChan := make(chan struct{})
	depGraph.Deploy(stopChan)
	jobs, err = c.Jobs().List(api_v1.ListOptions{})
	if err != nil {
		t.Fatal(err)
	}
	jobNames := map[string]bool{
		"ready-flow2-1": true,
		"ready-flow2-2": true,
	}
	for _, job := range jobs.Items {
		if !jobNames[job.Name] {
			t.Error("found unexpected job", job.Name)
		}
		delete(jobNames, job.Name)
	}
	if len(jobNames) != 0 {
		t.Errorf("not all jobs were created: %d jobs were not found", len(jobNames))
	}
}
func TestUseNotExportedFlow(t *testing.T) {
	flow := mocks.MakeFlow("test")
	flow.Flow.Exported = false
	c := mocks.NewClient(flow)
	sched := New(c, nil, 0)
	_, err := sched.BuildDependencyGraph(
		interfaces.DependencyGraphOptions{ReplicaCount: 1, FlowName: "test", ExportedOnly: true})
	if err == nil {
		t.Error("not exported flows cannot be run if ExportedOnly is true")
	} else if err.Error() != "flow test is not exported" {
		t.Errorf("unexpected error: %v", err)
	}
	_, err = sched.BuildDependencyGraph(
		interfaces.DependencyGraphOptions{ReplicaCount: 1, FlowName: "test"})
	if err != nil {
		t.Fatal(err)
	}
}
func TestTriggerOneFlowFromAnother(t *testing.T) {
	flow2 := mocks.MakeFlow("flow2")
	flow2.Flow.Exported = false
	c := mocks.NewClient(
		mocks.MakeFlow("flow1"),
		flow2,
		mocks.MakeResourceDefinition("job/ready-flow1-1"),
		mocks.MakeResourceDefinition("job/ready-flow1-2"),
		mocks.MakeResourceDefinition("job/ready-flow2-1"),
		mocks.MakeResourceDefinition("job/ready-flow2-2"),
		mocks.MakeDependency("flow/flow1", "job/ready-flow1-1", "flow=flow1"),
		mocks.MakeDependency("job/ready-flow1-1", "flow/flow2", "flow=flow1"),
		mocks.MakeDependency("flow/flow2", "job/ready-flow1-2", "flow=flow1"),
		mocks.MakeDependency("flow/flow2", "job/ready-flow2-1", "flow=flow2"),
		mocks.MakeDependency("job/ready-flow2-1", "job/ready-flow2-2", "flow=flow2"),
	)
	depGraph, err := New(c, nil, 0).BuildDependencyGraph(
		interfaces.DependencyGraphOptions{ReplicaCount: 1, FlowName: "flow1"})
	if err != nil {
		t.Fatal(err)
	}
	stopChan := make(chan struct{})
	depGraph.Deploy(stopChan)
	jobs, err := c.Jobs().List(api_v1.ListOptions{})
	if err != nil {
		t.Fatal(err)
	}
	jobNames := map[string]bool{
		"ready-flow1-1": true,
		"ready-flow1-2": true,
		"ready-flow2-1": true,
		"ready-flow2-2": true,
	}
	for _, job := range jobs.Items {
		if !jobNames[job.Name] {
			t.Error("found unexpected job", job.Name)
		}
		delete(jobNames, job.Name)
	}
	if len(jobNames) != 0 {
		t.Errorf("not all jobs were created: %d jobs were not found", len(jobNames))
	}
}
func TestParameterPassing(t *testing.T) {
	dep := mocks.MakeDependency("job/ready-1", "job/ready-$arg1-$arg2")
	dep.Args = map[string]string{
		"arg1": "x",
		"arg2": "y",
	}
	c := mocks.NewClient(
		mocks.MakeResourceDefinition("job/ready-1"),
		mocks.MakeResourceDefinition("job/ready-$arg1-$arg2"),
		dep,
	)
	depGraph, err := New(c, nil, 0).BuildDependencyGraph(
		interfaces.DependencyGraphOptions{ReplicaCount: 1})
	if err != nil {
		t.Fatal(err)
	}
	stopChan := make(chan struct{})
	depGraph.Deploy(stopChan)
	jobs, err := c.Jobs().List(api_v1.ListOptions{})
	if err != nil {
		t.Fatal(err)
	}
	jobNames := map[string]bool{
		"ready-1":   true,
		"ready-x-y": true,
	}
	for _, job := range jobs.Items {
		if !jobNames[job.Name] {
			t.Error("found unexpected job", job.Name)
		}
		delete(jobNames, job.Name)
	}
	if len(jobNames) != 0 {
		t.Errorf("not all jobs were created: %d jobs were not found", len(jobNames))
	}
}
func TestMultipathParameterPassing(t *testing.T) {
	dep1 := mocks.MakeDependency("job/ready-1", "job/ready-$arg")
	dep1.Args = map[string]string{"arg": "x"}
	dep2 := mocks.MakeDependency("job/ready-1", "job/ready-$arg")
	dep2.Args = map[string]string{"arg": "y"}
	c := mocks.NewClient(
		mocks.MakeResourceDefinition("job/ready-1"),
		mocks.MakeResourceDefinition("job/ready-$arg"),
		dep1,
		dep2,
	)
	depGraph, err := New(c, nil, 0).BuildDependencyGraph(
		interfaces.DependencyGraphOptions{ReplicaCount: 1})
	if err != nil {
		t.Fatal(err)
	}
	stopChan := make(chan struct{})
	depGraph.Deploy(stopChan)
	jobs, err := c.Jobs().List(api_v1.ListOptions{})
	if err != nil {
		t.Fatal(err)
	}
	jobNames := map[string]bool{
		"ready-1": true,
		"ready-x": true,
		"ready-y": true,
	}
	for _, job := range jobs.Items {
		if !jobNames[job.Name] {
			t.Error("found unexpected job", job.Name)
		}
		delete(jobNames, job.Name)
	}
	if len(jobNames) != 0 {
		t.Errorf("not all jobs were created: %d jobs were not found", len(jobNames))
	}
}
func TestParametrizedFlow(t *testing.T) {
	flow := mocks.MakeFlow("flow")
	flow.Flow.Parameters = map[string]client.FlowParameter{
		"arg1": mocks.MakeFlowParameter("p"),
		"arg2": mocks.MakeFlowParameter("q"),
	}
	c := mocks.NewClient(
		flow,
		mocks.MakeResourceDefinition("job/ready-$arg1-$arg2"),
		mocks.MakeDependency("flow/flow", "job/ready-$arg1-$arg2", "flow=flow"),
	)
	depGraph, err := New(c, nil, 0).BuildDependencyGraph(
		interfaces.DependencyGraphOptions{
			ReplicaCount: 1,
			FlowName:     "flow",
			Args: map[string]string{
				"arg1": "x",
			}})
	if err != nil {
		t.Fatal(err)
	}
	stopChan := make(chan struct{})
	depGraph.Deploy(stopChan)
	jobs, err := c.Jobs().List(api_v1.ListOptions{})
	if err != nil {
		t.Fatal(err)
	}
	expectedJobName := "ready-x-q"
	if len(jobs.Items) != 1 {
		t.Errorf("wrong number %d of jobs were created", len(jobs.Items))
	} else if jobs.Items[0].Name != expectedJobName {
		t.Errorf("expected job name: %s, found: %s", expectedJobName, jobs.Items[0].Name)
	}
}
func TestAcNameParameter(t *testing.T) {
	flow := mocks.MakeFlow("flow2")
	flow.Flow.Parameters = map[string]client.FlowParameter{
		"name": mocks.MakeFlowParameter(""),
	}
	dep := mocks.MakeDependency("flow/flow1", "flow/flow2", "flow=flow1")
	dep.Args = map[string]string{"name": "$AC_NAME"}
	c := mocks.NewClient(
		mocks.MakeFlow("flow1"),
		flow,
		dep,
		mocks.MakeResourceDefinition("job/ready-$name"),
		mocks.MakeDependency("flow/flow2", "job/ready-$name", "flow=flow2"),
	)
	depGraph, err := New(c, nil, 0).BuildDependencyGraph(
		interfaces.DependencyGraphOptions{
			ReplicaCount: 1,
			FlowName:     "flow1",
		})
	if err != nil {
		t.Fatal(err)
	}
	stopChan := make(chan struct{})
	depGraph.Deploy(stopChan)
	replicas, err := c.Replicas().List(api.ListOptions{})
	if err != nil {
		t.Fatal(err)
	}
	if len(replicas.Items) != 2 {
		t.Fatalf("there should be 2 replicas - 1 for each flow. Found %d replicas", len(replicas.Items))
	}
	for _, replica := range replicas.Items {
		if replica.FlowName == "flow1" {
			acName := replica.ReplicaName()
			jobName := "ready-" + acName
			_, err := c.Jobs().Get(jobName)
			if err != nil {
				t.Error(err)
			}
			return
		}
	}
	t.Error("no flow1 replica found")
}
func TestUseUndeclaredFlowParameter(t *testing.T) {
	flow := mocks.MakeFlow("flow")
	flow.Flow.Parameters = map[string]client.FlowParameter{
		"arg1": mocks.MakeFlowParameter("p"),
	}
	c := mocks.NewClient(
		flow,
		mocks.MakeResourceDefinition("job/ready-$arg1-$arg2"),
		mocks.MakeDependency("flow/flow", "job/ready-$arg1-$arg2", "flow=flow"),
	)
	depGraph, err := New(c, nil, 0).BuildDependencyGraph(
		interfaces.DependencyGraphOptions{
			ReplicaCount: 1,
			FlowName:     "flow",
		})
	if err != nil {
		t.Fatal(err)
	}
	stopChan := make(chan struct{})
	depGraph.Deploy(stopChan)
	jobs, err := c.Jobs().List(api_v1.ListOptions{})
	if err != nil {
		t.Fatal(err)
	}
	expectedJobName := "ready-p-$arg2"
	if len(jobs.Items) != 1 {
		t.Errorf("wrong number %d of jobs were created", len(jobs.Items))
	} else if jobs.Items[0].Name != expectedJobName {
		t.Errorf("expected job name: %s, found: %s", expectedJobName, jobs.Items[0].Name)
	}
}