// ----------------------------- Medium Tests ----------------------------
func TestStopTask(t *testing.T) {
	logrus.SetLevel(logrus.FatalLevel)
	s := newScheduler()
	s.Start()
	w := newMockWorkflowMap()
	tsk, _ := s.CreateTask(schedule.NewSimpleSchedule(time.Millisecond*100), w, false)
	task := s.tasks.Get(tsk.ID())
	task.Spin()
	err := s.StopTask(tsk.ID())

	Convey("Calling StopTask a running task", t, func() {
		Convey("Should not return an error", func() {
			So(err, ShouldBeNil)
		})
		time.Sleep(100 * time.Millisecond)
		Convey("State of the task should be TaskStopped", func() {
			So(task.state, ShouldEqual, core.TaskStopped)
		})
	})

	tskStopped, _ := s.CreateTask(schedule.NewSimpleSchedule(time.Millisecond*100), w, false)
	err = s.StopTask(tskStopped.ID())
	Convey("Calling StopTask on a stopped task", t, func() {
		Convey("Should return an error", func() {
			So(err, ShouldNotBeNil)
		})
		Convey("Error should read: Task is already stopped.", func() {
			So(err[0].Error(), ShouldResemble, "Task is already stopped.")
		})
	})

	tskDisabled, _ := s.CreateTask(schedule.NewSimpleSchedule(time.Millisecond*100), w, false)
	taskDisabled := s.tasks.Get(tskDisabled.ID())
	taskDisabled.state = core.TaskDisabled
	err = s.StopTask(tskDisabled.ID())
	Convey("Calling StopTask on a disabled task", t, func() {
		Convey("Should return an error", func() {
			So(err, ShouldNotBeNil)
		})
		Convey("Error should read: Task is disabled. Only running tasks can be stopped.", func() {
			So(err[0].Error(), ShouldResemble, "Task is disabled. Only running tasks can be stopped.")
		})
	})

	s.Stop()
}

func makeSchedule(s Schedule) (schedule.Schedule, error) {
	switch s.Type {
	case "simple":
		if s.Interval == "" {
			return nil, errors.New("missing `interval` in configuration of simple schedule")
		}

		d, err := time.ParseDuration(s.Interval)
		if err != nil {
			return nil, err
		}
		sch := schedule.NewSimpleSchedule(d)

		err = sch.Validate()
		if err != nil {
			return nil, err
		}
		return sch, nil
	case "windowed":
		if s.StartTimestamp == nil || s.StopTimestamp == nil || s.Interval == "" {
			errmsg := fmt.Sprintf("missing parameter/parameters in configuration of windowed schedule,"+
				"start_timestamp: %s, stop_timestamp: %s, interval: %s",
				s.StartTimestamp, s.StopTimestamp, s.Interval)
			return nil, errors.New(errmsg)
		}

		d, err := time.ParseDuration(s.Interval)
		if err != nil {
			return nil, err
		}

		sch := schedule.NewWindowedSchedule(
			d,
			s.StartTimestamp,
			s.StopTimestamp,
		)

		err = sch.Validate()
		if err != nil {
			return nil, err
		}
		return sch, nil
	case "cron":
		if s.Interval == "" {
			return nil, errors.New("missing `interval` in configuration of cron schedule")
		}
		sch := schedule.NewCronSchedule(s.Interval)

		err := sch.Validate()
		if err != nil {
			return nil, err
		}
		return sch, nil
	default:
		return nil, errors.New("unknown schedule type " + s.Type)
	}
}

func makeSchedule(s request.Schedule) (cschedule.Schedule, error) {
	switch s.Type {
	case "simple":
		d, err := time.ParseDuration(s.Interval)
		if err != nil {
			return nil, err
		}
		sch := cschedule.NewSimpleSchedule(d)

		err = sch.Validate()
		if err != nil {
			return nil, err
		}
		return sch, nil
	case "windowed":
		d, err := time.ParseDuration(s.Interval)
		if err != nil {
			return nil, err
		}

		var start, stop *time.Time
		if s.StartTimestamp != nil {
			t := time.Unix(*s.StartTimestamp, 0)
			start = &t
		}
		if s.StopTimestamp != nil {
			t := time.Unix(*s.StopTimestamp, 0)
			stop = &t
		}
		sch := cschedule.NewWindowedSchedule(
			d,
			start,
			stop,
		)

		err = sch.Validate()
		if err != nil {
			return nil, err
		}
		return sch, nil
	case "cron":
		if s.Interval == "" {
			return nil, errors.New("missing cron entry ")
		}
		sch := cschedule.NewCronSchedule(s.Interval)

		err := sch.Validate()
		if err != nil {
			return nil, err
		}
		return sch, nil
	default:
		return nil, errors.New("unknown schedule type " + s.Type)
	}
}

func TestCollectPublishWorkflow(t *testing.T) {
	log.SetLevel(log.FatalLevel)
	Convey("Given a started plugin control", t, func() {

		c := control.New()
		c.Start()
		s := New()
		s.SetMetricManager(c)
		Convey("create a workflow", func() {
			rp, err := core.NewRequestedPlugin(snap_collector_mock2_path)
			So(err, ShouldBeNil)
			_, err = c.Load(rp)
			So(err, ShouldBeNil)
			rp2, err := core.NewRequestedPlugin(snap_publisher_file_path)
			So(err, ShouldBeNil)
			_, err = c.Load(rp2)
			So(err, ShouldBeNil)
			rp3, err := core.NewRequestedPlugin(snap_processor_passthru_path)
			So(err, ShouldBeNil)
			_, err = c.Load(rp3)
			So(err, ShouldBeNil)
			time.Sleep(100 * time.Millisecond)

			metrics, err2 := c.MetricCatalog()
			So(err2, ShouldBeNil)
			So(metrics, ShouldNotBeEmpty)

			w := wmap.NewWorkflowMap()
			w.CollectNode.AddMetric("/intel/mock/foo", 2)
			w.CollectNode.AddConfigItem("/intel/mock/foo", "password", "secret")

			pu := wmap.NewPublishNode("file", 3)
			pu.AddConfigItem("file", "/tmp/snap-TestCollectPublishWorkflow.out")

			pr := wmap.NewProcessNode("passthru", 1)
			time.Sleep(100 * time.Millisecond)

			pr.Add(pu)
			w.CollectNode.Add(pr)

			Convey("Start scheduler", func() {
				err := s.Start()
				So(err, ShouldBeNil)
				Convey("Create task", func() {
					t, err := s.CreateTask(schedule.NewSimpleSchedule(time.Millisecond*500), w, false)
					So(err.Errors(), ShouldBeEmpty)
					So(t, ShouldNotBeNil)
					t.(*task).Spin()
					time.Sleep(3 * time.Second)

				})
			})
		})
	})
}

func getSchedule(s *core.Schedule) schedule.Schedule {
	switch s.Type {
	case "simple":
		d, e := time.ParseDuration(s.Interval)
		if e != nil {
			log.WithField("_block", "get-schedule").Error(e)
			return nil
		}
		return schedule.NewSimpleSchedule(d)
	}
	return nil
}

func TestTask(t *testing.T) {
	log.SetLevel(log.FatalLevel)
	Convey("Task", t, func() {
		sampleWFMap := wmap.Sample()
		wf, errs := wmapToWorkflow(sampleWFMap)
		So(errs, ShouldBeEmpty)
		c := &mockMetricManager{}
		c.setAcceptedContentType("rabbitmq", core.PublisherPluginType, 5, []string{plugin.SnapGOBContentType})
		mgrs := newManagers(c)
		err := wf.BindPluginContentTypes(&mgrs)
		So(err, ShouldBeNil)
		Convey("task + simple schedule", func() {
			sch := schedule.NewSimpleSchedule(time.Millisecond * 100)
			task, err := newTask(sch, wf, newWorkManager(), c, emitter)
			So(err, ShouldBeNil)
			task.Spin()
			time.Sleep(time.Millisecond * 10) // it is a race so we slow down the test
			So(task.state, ShouldEqual, core.TaskSpinning)

			task.Stop()
		})

		Convey("Task specified-name test", func() {
			sch := schedule.NewSimpleSchedule(time.Millisecond * 100)
			task, err := newTask(sch, wf, newWorkManager(), c, emitter, core.SetTaskName("My name is unique"))
			So(err, ShouldBeNil)
			task.Spin()
			So(task.GetName(), ShouldResemble, "My name is unique")

		})
		Convey("Task default-name test", func() {
			sch := schedule.NewSimpleSchedule(time.Millisecond * 100)
			task, err := newTask(sch, wf, newWorkManager(), c, emitter)
			So(err, ShouldBeNil)
			task.Spin()
			So(task.GetName(), ShouldResemble, "Task-"+task.ID())

		})

		Convey("Task deadline duration test", func() {
			sch := schedule.NewSimpleSchedule(time.Millisecond * 100)
			task, err := newTask(sch, wf, newWorkManager(), c, emitter, core.TaskDeadlineDuration(20*time.Second))
			So(err, ShouldBeNil)
			task.Spin()
			So(task.deadlineDuration, ShouldEqual, 20*time.Second)
			task.Option(core.TaskDeadlineDuration(20 * time.Second))

			So(core.TaskDeadlineDuration(2*time.Second), ShouldNotBeEmpty)

		})

		Convey("Tasks are created and creation of task table is checked", func() {
			sch := schedule.NewSimpleSchedule(time.Millisecond * 100)
			task, err := newTask(sch, wf, newWorkManager(), c, emitter)
			So(err, ShouldBeNil)
			task1, err := newTask(sch, wf, newWorkManager(), c, emitter)
			So(err, ShouldBeNil)
			task1.Spin()
			task.Spin()
			tC := newTaskCollection()
			tC.add(task)
			tC.add(task1)
			taskTable := tC.Table()

			So(len(taskTable), ShouldEqual, 2)

		})

		Convey("Task is created and starts to spin", func() {
			sch := schedule.NewSimpleSchedule(time.Second * 5)
			task, err := newTask(sch, wf, newWorkManager(), c, emitter)
			So(err, ShouldBeNil)
			task.Spin()
			So(task.state, ShouldEqual, core.TaskSpinning)
			Convey("Task is Stopped", func() {
				task.Stop()
				time.Sleep(time.Millisecond * 10) // it is a race so we slow down the test
				So(task.state, ShouldEqual, core.TaskStopped)
			})
		})

		Convey("task fires", func() {
			sch := schedule.NewSimpleSchedule(time.Nanosecond * 100)
			task, err := newTask(sch, wf, newWorkManager(), c, emitter)
			So(err, ShouldBeNil)
			task.Spin()
			time.Sleep(time.Millisecond * 50)
			So(task.hitCount, ShouldBeGreaterThan, 2)
			So(task.missedIntervals, ShouldBeGreaterThan, 2)
			task.Stop()
		})

		Convey("Enable a running task", func() {
			sch := schedule.NewSimpleSchedule(time.Millisecond * 10)
			task, err := newTask(sch, wf, newWorkManager(), c, emitter)
			So(err, ShouldBeNil)
			task.Spin()
			err = task.Enable()
			So(err, ShouldNotBeNil)
			So(task.State(), ShouldEqual, core.TaskSpinning)
//.........这里部分代码省略.........

func TestScheduler(t *testing.T) {
	log.SetLevel(log.FatalLevel)
	Convey("NewTask", t, func() {
		c := new(mockMetricManager)
		c.setAcceptedContentType("machine", core.ProcessorPluginType, 1, []string{"snap.*", "snap.gob", "foo.bar"})
		c.setReturnedContentType("machine", core.ProcessorPluginType, 1, []string{"snap.gob"})
		c.setAcceptedContentType("rmq", core.PublisherPluginType, -1, []string{"snap.json", "snap.gob"})
		c.setAcceptedContentType("file", core.PublisherPluginType, -1, []string{"snap.json"})
		cfg := GetDefaultConfig()
		s := New(cfg)
		s.SetMetricManager(c)
		w := wmap.NewWorkflowMap()
		// Collection node
		w.CollectNode.AddMetric("/foo/bar", 1)
		w.CollectNode.AddMetric("/foo/baz", 2)
		w.CollectNode.AddConfigItem("/foo/bar", "username", "root")
		w.CollectNode.AddConfigItem("/foo/bar", "port", 8080)
		w.CollectNode.AddConfigItem("/foo/bar", "ratio", 0.32)
		w.CollectNode.AddConfigItem("/foo/bar", "yesorno", true)

		// Add a process node
		pr1 := wmap.NewProcessNode("machine", 1)
		pr1.AddConfigItem("username", "wat")
		pr1.AddConfigItem("howmuch", 9999)

		// Add a process node
		pr12 := wmap.NewProcessNode("machine", 1)
		pr12.AddConfigItem("username", "wat2")
		pr12.AddConfigItem("howmuch", 99992)

		// Publish node for our process node
		pu1 := wmap.NewPublishNode("rmq", -1)
		pu1.AddConfigItem("birthplace", "dallas")
		pu1.AddConfigItem("monies", 2)

		// Publish node direct to collection
		pu2 := wmap.NewPublishNode("file", -1)
		pu2.AddConfigItem("color", "brown")
		pu2.AddConfigItem("purpose", 42)

		pr12.Add(pu2)
		pr1.Add(pr12)
		w.CollectNode.Add(pr1)
		w.CollectNode.Add(pu1)

		e := s.Start()
		So(e, ShouldBeNil)
		t, te := s.CreateTask(schedule.NewSimpleSchedule(time.Second*1), w, false)
		So(te.Errors(), ShouldBeEmpty)

		for _, i := range t.(*task).workflow.processNodes {
			testInboundContentType(i)
		}
		for _, i := range t.(*task).workflow.publishNodes {
			testInboundContentType(i)
		}
		So(t.(*task).workflow.processNodes[0].ProcessNodes[0].PublishNodes[0].InboundContentType, ShouldEqual, "snap.json")

		Convey("returns errors when metrics do not validate", func() {
			c.failValidatingMetrics = true
			c.failValidatingMetricsAfter = 1
			_, err := s.CreateTask(schedule.NewSimpleSchedule(time.Second*1), w, false)
			So(err, ShouldNotBeNil)
			fmt.Printf("%d", len(err.Errors()))
			So(len(err.Errors()), ShouldBeGreaterThan, 0)
			So(err.Errors()[0], ShouldResemble, serror.New(errors.New("metric validation error")))

		})

		Convey("returns an error when scheduler started and MetricManager is not set", func() {
			s1 := New(GetDefaultConfig())
			err := s1.Start()
			So(err, ShouldNotBeNil)
			fmt.Printf("%v", err)
			So(err, ShouldResemble, ErrMetricManagerNotSet)

		})

		Convey("returns an error when wrong namespace is given wo workflowmap ", func() {
			w.CollectNode.AddMetric("****/&&&", 3)
			w.CollectNode.AddConfigItem("****/&&&", "username", "user")
			_, err := s.CreateTask(schedule.NewSimpleSchedule(time.Second*1), w, false)

			So(len(err.Errors()), ShouldBeGreaterThan, 0)

		})

		// TODO NICK
		Convey("returns an error when a schedule does not validate", func() {
			s1 := New(GetDefaultConfig())
			s1.Start()
			_, err := s1.CreateTask(schedule.NewSimpleSchedule(time.Second*1), w, false)
			So(err, ShouldNotBeNil)
			So(len(err.Errors()), ShouldBeGreaterThan, 0)
			So(err.Errors()[0], ShouldResemble, serror.New(ErrSchedulerNotStarted))
			s1.metricManager = c
			s1.Start()
			_, err1 := s1.CreateTask(schedule.NewSimpleSchedule(time.Second*0), w, false)
			So(err1.Errors()[0].Error(), ShouldResemble, "Interval must be greater than 0")

//.........这里部分代码省略.........

func TestDistributedWorkflow(t *testing.T) {
	Convey("Create a scheduler with 2 controls and load plugins", t, func() {
		l, _ := net.Listen("tcp", ":0")
		l.Close()
		cfg := control.GetDefaultConfig()
		cfg.ListenPort = l.Addr().(*net.TCPAddr).Port
		c1 := control.New(cfg)
		c1.Start()
		m, _ := net.Listen("tcp", ":0")
		m.Close()
		cfg.ListenPort = m.Addr().(*net.TCPAddr).Port
		port1 := cfg.ListenPort
		c2 := control.New(cfg)
		schcfg := GetDefaultConfig()
		sch := New(schcfg)
		c2.Start()
		sch.SetMetricManager(c1)
		err := sch.Start()
		So(err, ShouldBeNil)
		// Load appropriate plugins into each control.
		mock2Path := helper.PluginFilePath("snap-plugin-collector-mock2")
		passthruPath := helper.PluginFilePath("snap-plugin-processor-passthru")
		filePath := helper.PluginFilePath("snap-plugin-publisher-mock-file")

		// mock2 and file onto c1

		rp, err := core.NewRequestedPlugin(mock2Path)
		So(err, ShouldBeNil)
		_, err = c1.Load(rp)
		So(err, ShouldBeNil)
		rp, err = core.NewRequestedPlugin(filePath)
		So(err, ShouldBeNil)
		_, err = c1.Load(rp)
		So(err, ShouldBeNil)
		// passthru on c2
		rp, err = core.NewRequestedPlugin(passthruPath)
		So(err, ShouldBeNil)
		passthru, err := c2.Load(rp)
		So(err, ShouldBeNil)

		Convey("Test task with one local and one remote node", func() {
			//Create a task
			//Create a workflowmap
			wf := dsWFMap(port1)
			t, errs := sch.CreateTask(schedule.NewSimpleSchedule(time.Second), wf, true)
			So(len(errs.Errors()), ShouldEqual, 0)
			So(t, ShouldNotBeNil)
			// stop the scheduler and control (since in nested Convey statements, the
			// statements in the outer Convey execute for each of the inner Conveys
			// independently; see https://github.com/smartystreets/goconvey/wiki/Execution-order
			// for details on execution order in Convey)
			sch.Stop()
			c2.Stop()
		})

		Convey("Test task with invalid remote port", func() {
			wf := dsWFMap(0)
			controlproxy.MAX_CONNECTION_TIMEOUT = 1 * time.Second
			t, errs := sch.CreateTask(schedule.NewSimpleSchedule(time.Second), wf, true)
			So(len(errs.Errors()), ShouldEqual, 1)
			So(t, ShouldBeNil)
			// stop the scheduler and control (since in nested Convey statements, the
			// statements in the outer Convey execute for each of the inner Conveys
			// independently; see https://github.com/smartystreets/goconvey/wiki/Execution-order
			// for details on execution order in Convey)
			sch.Stop()
			c2.Stop()
		})

		Convey("Test task without remote plugin", func() {
			_, err := c2.Unload(passthru)
			So(err, ShouldBeNil)
			wf := dsWFMap(port1)
			t, errs := sch.CreateTask(schedule.NewSimpleSchedule(time.Second), wf, true)
			So(len(errs.Errors()), ShouldEqual, 1)
			So(t, ShouldBeNil)
			// stop the scheduler and control (since in nested Convey statements, the
			// statements in the outer Convey execute for each of the inner Conveys
			// independently; see https://github.com/smartystreets/goconvey/wiki/Execution-order
			// for details on execution order in Convey)
			sch.Stop()
			c2.Stop()
		})

		Convey("Test task failing when control is stopped while task is running", func() {
			wf := dsWFMap(port1)
			// set timeout so that connection attempt through the controlproxy will fail after 1 second
			controlproxy.MAX_CONNECTION_TIMEOUT = time.Second
			// define an interval that the simple scheduler will run on every 100ms
			interval := time.Millisecond * 100
			// create our task; should be disabled after 3 failures
			t, errs := sch.CreateTask(schedule.NewSimpleSchedule(interval), wf, true)
			// ensure task was created successfully
			So(len(errs.Errors()), ShouldEqual, 0)
			So(t, ShouldNotBeNil)
			// create a channel to listen on for a response and setup an event handler
			// that will respond on that channel once the 'TaskDisabledEvent'  arrives
			respChan := make(chan struct{})
			sch.RegisterEventHandler("test", &failHandler{respChan})
			// then stop the controller
//.........这里部分代码省略.........

func TestDistributedSubscriptions(t *testing.T) {

	Convey("Load control/scheduler with a mock remote scheduler", t, func() {
		l, _ := net.Listen("tcp", ":0")
		l.Close()
		cfg := control.GetDefaultConfig()
		cfg.ListenPort = l.Addr().(*net.TCPAddr).Port
		c1 := control.New(cfg)
		c1.Start()
		m, _ := net.Listen("tcp", ":0")
		m.Close()
		cfg.ListenPort = m.Addr().(*net.TCPAddr).Port
		port1 := cfg.ListenPort
		c2 := control.New(cfg)
		schcfg := GetDefaultConfig()
		sch := New(schcfg)
		c2.Start()
		sch.SetMetricManager(c1)
		err := sch.Start()
		So(err, ShouldBeNil)
		// Load appropriate plugins into each control.
		mock2Path := helper.PluginFilePath("snap-plugin-collector-mock2")
		passthruPath := helper.PluginFilePath("snap-plugin-processor-passthru")
		filePath := helper.PluginFilePath("snap-plugin-publisher-mock-file")

		// mock2 and file onto c1

		rp, err := core.NewRequestedPlugin(mock2Path)
		So(err, ShouldBeNil)
		_, err = c1.Load(rp)
		So(err, ShouldBeNil)
		rp, err = core.NewRequestedPlugin(filePath)
		So(err, ShouldBeNil)
		_, err = c1.Load(rp)
		So(err, ShouldBeNil)
		// passthru on c2
		rp, err = core.NewRequestedPlugin(passthruPath)
		So(err, ShouldBeNil)
		_, err = c2.Load(rp)
		So(err, ShouldBeNil)

		Convey("Starting task should not succeed if remote dep fails to subscribe", func() {
			//Create a task
			//Create a workflowmap
			wf := dsWFMap(port1)
			// Create a task that is not started immediately so we can
			// validate deps correctly.
			t, errs := sch.CreateTask(schedule.NewSimpleSchedule(time.Second), wf, false)
			So(len(errs.Errors()), ShouldEqual, 0)
			So(t, ShouldNotBeNil)
			schTask := t.(*task)
			remoteMockManager := &subscriptionManager{Fail: true}
			schTask.RemoteManagers.Add(fmt.Sprintf("127.0.0.1:%v", port1), remoteMockManager)
			localMockManager := &subscriptionManager{Fail: false}
			schTask.RemoteManagers.Add("", localMockManager)
			// Start task. We expect it to fail while subscribing deps
			terrs := sch.StartTask(t.ID())
			So(terrs, ShouldNotBeNil)
			Convey("So dependencies should have been unsubscribed", func() {
				// Ensure that unsubscribe call count is equal to subscribe call count
				// i.e that every subscribe call was followed by an unsubscribe since
				// we errored
				So(remoteMockManager.UnsubscribeCallCount, ShouldEqual, remoteMockManager.SubscribeCallCount)
				So(localMockManager.UnsubscribeCallCount, ShouldEqual, localMockManager.UnsubscribeCallCount)
			})
		})

		Convey("Starting task should not succeed if missing local dep fails to subscribe", func() {
			//Create a task
			//Create a workflowmap
			wf := dsWFMap(port1)
			// Create a task that is not started immediately so we can
			// validate deps correctly.
			t, errs := sch.CreateTask(schedule.NewSimpleSchedule(time.Second), wf, false)
			So(len(errs.Errors()), ShouldEqual, 0)
			So(t, ShouldNotBeNil)
			schTask := t.(*task)
			localMockManager := &subscriptionManager{Fail: true}
			schTask.RemoteManagers.Add("", localMockManager)
			remoteMockManager := &subscriptionManager{Fail: false}
			schTask.RemoteManagers.Add(fmt.Sprintf("127.0.0.1:%v", port1), remoteMockManager)

			// Start task. We expect it to fail while subscribing deps
			terrs := sch.StartTask(t.ID())
			So(terrs, ShouldNotBeNil)
			Convey("So dependencies should have been unsubscribed", func() {
				// Ensure that unsubscribe call count is equal to subscribe call count
				// i.e that every subscribe call was followed by an unsubscribe since
				// we errored
				So(remoteMockManager.UnsubscribeCallCount, ShouldEqual, remoteMockManager.SubscribeCallCount)
				So(localMockManager.UnsubscribeCallCount, ShouldEqual, localMockManager.UnsubscribeCallCount)
			})
		})

		Convey("Starting task should suceed if all deps are available", func() {
			//Create a task
			//Create a workflowmap
			wf := dsWFMap(port1)
			// Create a task that is not started immediately so we can
			// validate deps correctly.
//.........这里部分代码省略.........

func (t *mockTask) Schedule() schedule.Schedule {
	return schedule.NewSimpleSchedule(time.Second * 1)
}

func TestDistributedWorkflow(t *testing.T) {
	Convey("Create a scheduler with 2 controls and load plugins", t, func() {
		l, _ := net.Listen("tcp", ":0")
		l.Close()
		cfg := control.GetDefaultConfig()
		cfg.ListenPort = l.Addr().(*net.TCPAddr).Port
		c1 := control.New(cfg)
		c1.Start()
		m, _ := net.Listen("tcp", ":0")
		m.Close()
		cfg.ListenPort = m.Addr().(*net.TCPAddr).Port
		port1 := cfg.ListenPort
		c2 := control.New(cfg)
		schcfg := GetDefaultConfig()
		sch := New(schcfg)
		c2.Start()
		sch.SetMetricManager(c1)
		err := sch.Start()
		So(err, ShouldBeNil)
		// Load appropriate plugins into each control.
		mock2Path := path.Join(PluginPath, "snap-collector-mock2")
		passthruPath := path.Join(PluginPath, "snap-processor-passthru")
		filePath := path.Join(PluginPath, "snap-publisher-file")

		// mock2 and file onto c1

		rp, err := core.NewRequestedPlugin(mock2Path)
		So(err, ShouldBeNil)
		_, err = c1.Load(rp)
		So(err, ShouldBeNil)
		rp, err = core.NewRequestedPlugin(filePath)
		So(err, ShouldBeNil)
		_, err = c1.Load(rp)
		So(err, ShouldBeNil)
		// passthru on c2
		rp, err = core.NewRequestedPlugin(passthruPath)
		So(err, ShouldBeNil)
		passthru, err := c2.Load(rp)
		So(err, ShouldBeNil)

		Convey("Test task with one local and one remote node", func() {
			//Create a task
			//Create a workflowmap
			wf := dsWFMap(port1)
			t, errs := sch.CreateTask(schedule.NewSimpleSchedule(time.Second), wf, true)
			So(len(errs.Errors()), ShouldEqual, 0)
			So(t, ShouldNotBeNil)
		})

		Convey("Test task with invalid remote port", func() {
			wf := dsWFMap(0)
			controlproxy.MAX_CONNECTION_TIMEOUT = 1 * time.Second
			t, errs := sch.CreateTask(schedule.NewSimpleSchedule(time.Second), wf, true)
			So(len(errs.Errors()), ShouldEqual, 1)
			So(t, ShouldBeNil)
		})

		Convey("Test task without remote plugin", func() {
			_, err := c2.Unload(passthru)
			So(err, ShouldBeNil)
			wf := dsWFMap(port1)
			t, errs := sch.CreateTask(schedule.NewSimpleSchedule(time.Second), wf, true)
			So(len(errs.Errors()), ShouldEqual, 1)
			So(t, ShouldBeNil)
		})

		Convey("Test task failing when control is stopped while task is running", func() {
			wf := dsWFMap(port1)
			controlproxy.MAX_CONNECTION_TIMEOUT = 10 * time.Second
			interval := time.Millisecond * 100
			t, errs := sch.CreateTask(schedule.NewSimpleSchedule(interval), wf, true)
			So(len(errs.Errors()), ShouldEqual, 0)
			So(t, ShouldNotBeNil)
			c2.Stop()
			// Give task time to fail
			time.Sleep(time.Second)
			tasks := sch.GetTasks()
			var task core.Task
			for _, v := range tasks {
				task = v
			}
			So(task.State(), ShouldEqual, core.TaskDisabled)
		})

	})

}