func TestSubscriptionGroups_GetSpecifiedDynamic(t *testing.T) {
	log.SetLevel(log.DebugLevel)
	c := New(getTestSGConfig())

	lpe := newLstnToPluginEvents()
	c.eventManager.RegisterHandler("TestSubscriptionGroups_AddRemove", lpe)
	c.Start()

	Convey("Loading a mock collector plugn", t, func() {
		_, err := loadPlg(c, helper.PluginFilePath("snap-plugin-collector-mock1"))
		So(err, ShouldBeNil)
		<-lpe.load

		Convey("Subscription group created for requested metric with specified instance of dynamic element", func() {
			requested := mockRequestedMetric{namespace: core.NewNamespace("intel", "mock").AddDynamicElement("host", "name of the host").AddStaticElement("baz")}
			// specified dynamic element
			requested.Namespace()[2].Value = "host0"
			subsPlugin := mockSubscribedPlugin{
				typeName: core.CollectorPluginType,
				name:     "mock",
				version:  1,
				config:   cdata.NewNode(),
			}
			subsPluginKey := key(subsPlugin)

			sg := newSubscriptionGroups(c)
			So(sg, ShouldNotBeNil)
			sg.Add("task-id", []core.RequestedMetric{requested}, cdata.NewTree(), []core.SubscribedPlugin{subsPlugin})
			<-lpe.sub
			So(len(sg.subscriptionMap), ShouldEqual, 1)
			val, ok := sg.subscriptionMap["task-id"]
			So(ok, ShouldBeTrue)
			So(val, ShouldNotBeNil)

			pluginToMetricMap, serrs, err := sg.Get("task-id")
			So(len(serrs), ShouldEqual, 0)
			So(err, ShouldBeNil)
			So(len(pluginToMetricMap), ShouldEqual, 1)
			So(pluginToMetricMap, ShouldContainKey, subsPluginKey)
			metrics := pluginToMetricMap[subsPluginKey].Metrics()
			So(len(metrics), ShouldEqual, 1)

			pluginToMetricMap, serrs, err = sg.Get("task-fake-id")
			So(len(serrs), ShouldEqual, 0)
			So(err, ShouldNotBeNil)
			So(err, ShouldResemble, ErrSubscriptionGroupDoesNotExist)
			So(pluginToMetricMap, ShouldBeEmpty)
		})
	})
}

func TestValidateSignature(t *testing.T) {
	keyringFile := []string{"pubring.gpg"}
	signedFile := "snap-plugin-collector-mock1"
	signatureFile := signedFile + ".asc"
	unsignedFile := helper.PluginFilePath("snap-plugin-collector-mock2")
	s := SigningManager{}

	signature, _ := ioutil.ReadFile(signatureFile)
	Convey("Valid files and good signature", t, func() {
		err := s.ValidateSignature(keyringFile, signedFile, signature)
		So(err, ShouldBeNil)
	})

	Convey("Valid files and good signature. Multiple keyrings", t, func() {
		keyringFiles := []string{"pubkeys.gpg", "pubring.gpg"}
		err := s.ValidateSignature(keyringFiles, signedFile, signature)
		So(err, ShouldBeNil)
	})

	Convey("Validate unsigned file with signature", t, func() {
		err := s.ValidateSignature(keyringFile, unsignedFile, signature)
		So(err, ShouldNotBeNil)
		So(err.Error(), ShouldContainSubstring, "Error checking signature")
	})

	Convey("Invalid keyring", t, func() {
		err := s.ValidateSignature([]string{""}, signedFile, signature)
		So(err, ShouldNotBeNil)
		So(err.Error(), ShouldContainSubstring, "Keyring file (.gpg) not found")
	})

	Convey("Invalid signed file", t, func() {
		err := s.ValidateSignature(keyringFile, "", signature)
		So(err, ShouldNotBeNil)
		So(err.Error(), ShouldContainSubstring, "Signed file not found")
	})

	Convey("Invalid signature file", t, func() {
		err := s.ValidateSignature(keyringFile, signedFile, nil)
		So(err, ShouldNotBeNil)
		So(err.Error(), ShouldContainSubstring, "Error checking signature")
	})
}

func TestSubscriptionGroups_AddRemoveDynamic(t *testing.T) {
	log.SetLevel(log.DebugLevel)
	c := New(getTestSGConfig())

	lpe := newLstnToPluginEvents()
	c.eventManager.RegisterHandler("TestSubscriptionGroups_AddRemove", lpe)
	c.Start()

	Convey("Loading a mock collector plugn", t, func() {
		_, err := loadPlg(c, helper.PluginFilePath("snap-plugin-collector-mock1"))
		So(err, ShouldBeNil)
		<-lpe.load

		Convey("Subscription group created for requested metric with wildcards", func() {
			requested := mockRequestedMetric{
				namespace: core.NewNamespace("intel", "mock").AddDynamicElement("wild", "wild description"),
				version:   -1,
			}
			subsPlugin := mockSubscribedPlugin{
				typeName: core.CollectorPluginType,
				name:     "mock",
				version:  1,
				config:   cdata.NewNode(),
			}

			sg := newSubscriptionGroups(c)
			So(sg, ShouldNotBeNil)
			sg.Add("task-id", []core.RequestedMetric{requested}, cdata.NewTree(), []core.SubscribedPlugin{subsPlugin})
			<-lpe.sub
			So(len(sg.subscriptionMap), ShouldEqual, 1)
			val, ok := sg.subscriptionMap["task-id"]
			So(ok, ShouldBeTrue)
			So(val, ShouldNotBeNil)

			serrs := sg.Remove("task-id")
			<-lpe.unsub
			So(len(serrs), ShouldEqual, 0)
			So(len(sg.subscriptionMap), ShouldEqual, 0)
		})
	})
}

// This test is meant to cover the grpc implementation of the subset of control
// features that scheduler uses. It is not intended to test the control features
// themselves, only that we are correctly passing data over grpc and correctly
// passing success/errors.
func TestGRPCServerScheduler(t *testing.T) {
	l, _ := net.Listen("tcp", ":0")
	l.Close()
	cfg := GetDefaultConfig()
	cfg.ListenPort = l.Addr().(*net.TCPAddr).Port
	c := New(cfg)
	err := c.Start()

	lpe := newFoo()
	c.eventManager.RegisterHandler("Control.PluginLoaded", lpe)

	Convey("Starting control_proxy server/client", t, func() {
		Convey("So err should be nil", func() {
			So(err, ShouldBeNil)
		})
	})
	// Load 3 plugins
	// collector -- mock
	// processor -- passthru
	// publisher -- file
	mock, err := core.NewRequestedPlugin(fixtures.JSONRPCPluginPath)
	if err != nil {
		log.Fatal(err)
	}
	_, serr := c.Load(mock)
	Convey("Loading mock collector", t, func() {
		Convey("should not error", func() {
			So(serr, ShouldBeNil)
		})
	})
	<-lpe.done
	passthru, err := core.NewRequestedPlugin(helper.PluginFilePath("snap-plugin-processor-passthru"))
	if err != nil {
		log.Fatal(err)
	}
	catalogedPassthru, serr := c.Load(passthru)
	Convey("Loading passthru processor", t, func() {
		Convey("should not error", func() {
			So(serr, ShouldBeNil)
		})
	})
	subscribedPassThruPlugin := subscribedPlugin{
		name:     catalogedPassthru.Name(),
		version:  catalogedPassthru.Version(),
		typeName: catalogedPassthru.TypeName(),
	}
	<-lpe.done
	filepub, err := core.NewRequestedPlugin(helper.PluginFilePath("snap-plugin-publisher-mock-file"))
	if err != nil {
		log.Fatal(err)
	}
	catalogedFile, serr := c.Load(filepub)
	Convey("Loading file publisher", t, func() {
		Convey("should not error", func() {
			So(serr, ShouldBeNil)
		})
	})
	subscribedFilePlugin := subscribedPlugin{
		name:     catalogedFile.Name(),
		version:  catalogedFile.Version(),
		typeName: catalogedFile.TypeName(),
	}

	<-lpe.done
	conn, err := rpcutil.GetClientConnection(c.Config.ListenAddr, c.Config.ListenPort)

	Convey("Creating an rpc connection", t, func() {
		Convey("Should not error", func() {
			So(err, ShouldBeNil)
		})
	})

	client := rpc.NewMetricManagerClient(conn)

	Convey("Creating an RPC client to control RPC server", t, func() {
		Convey("And a client should exist", func() {
			So(client, ShouldNotBeNil)
		})
	})

	// Verify that validate deps is properly passing through errors
	Convey("Validating Deps", t, func() {
		Convey("Should Fail if given invalid info", func() {
			req := &rpc.ValidateDepsRequest{
				Metrics: common.NewMetrics([]core.Metric{fixtures.InvalidMetric}),
				Plugins: common.ToSubPluginsMsg([]core.SubscribedPlugin{}),
			}
			reply, err := client.ValidateDeps(context.Background(), req)
			// we don't expect rpc errors
			So(err, ShouldBeNil)
			So(len(reply.Errors), ShouldNotEqual, 0)
		})
		Convey("with valid metrics", func() {
			req := &rpc.ValidateDepsRequest{
				Metrics: common.NewMetrics([]core.Metric{fixtures.ValidMetric}),
				Plugins: common.ToSubPluginsMsg([]core.SubscribedPlugin{}),
//.........这里部分代码省略.........

	log "github.com/Sirupsen/logrus"
	"github.com/pborman/uuid"

	"github.com/intelsdi-x/snap/control"
	"github.com/intelsdi-x/snap/mgmt/rest"
	"github.com/intelsdi-x/snap/plugin/helper"
	"github.com/intelsdi-x/snap/scheduler"
	"github.com/intelsdi-x/snap/scheduler/wmap"
	. "github.com/smartystreets/goconvey/convey"
)

var (
	// Change to set the REST API logging to debug
	LOG_LEVEL = log.FatalLevel

	MOCK_PLUGIN_PATH1       = []string{helper.PluginFilePath("snap-plugin-collector-mock1")}
	MOCK_PLUGIN_PATH2       = []string{helper.PluginFilePath("snap-plugin-collector-mock2")}
	ANOTHERMOCK_PLUGIN_PATH = []string{helper.PluginFilePath("snap-plugin-collector-anothermock1")}
	FILE_PLUGIN_PATH        = []string{helper.PluginFilePath("snap-plugin-publisher-mock-file")}
	DIRECTORY_PATH          = []string{helper.PluginPath()}

	NextPort = 45000

	p1 *LoadPluginResult
	p2 *LoadPluginResult
	p3 *LoadPluginResult
)

func getWMFromSample(sample string) *wmap.WorkflowMap {
	jsonP, err := ioutil.ReadFile("../wmap_sample/" + sample)
	if err != nil {

package core

import (
	"crypto/sha256"
	"io/ioutil"
	"path"
	"testing"

	"github.com/intelsdi-x/snap/plugin/helper"
	. "github.com/smartystreets/goconvey/convey"
)

var (
	PluginName    = "snap-plugin-collector-mock1"
	SnapPath      = helper.BuildPath
	PluginPath    = helper.PluginFilePath(PluginName)
	SignatureFile = path.Join(SnapPath, "../pkg/psigning", "snap-plugin-collector-mock1.asc")
)

func TestRequestedPlugin(t *testing.T) {
	// Creating a plugin request

	Convey("Creating a plugin request from a valid path", t, func() {
		rp, err := NewRequestedPlugin(PluginPath)
		Convey("Should not return an error", func() {
			So(err, ShouldBeNil)

			Convey("Should return a RequestedPlugin type", func() {
				So(rp, ShouldHaveSameTypeAs, &RequestedPlugin{})
			})

	"os"
	"strings"
	"testing"

	. "github.com/smartystreets/goconvey/convey"

	log "github.com/Sirupsen/logrus"
	"github.com/intelsdi-x/snap/core/cdata"
	"github.com/intelsdi-x/snap/core/ctypes"
	"github.com/intelsdi-x/snap/mgmt/rest/fixtures"
	"github.com/intelsdi-x/snap/plugin/helper"
)

var (
	LOG_LEVEL         = log.WarnLevel
	MOCK_PLUGIN_PATH1 = helper.PluginFilePath("snap-plugin-collector-mock1")
)

type restAPIInstance struct {
	port   int
	server *Server
}

func startV1API(cfg *mockConfig, testType string) *restAPIInstance {
	log.SetLevel(LOG_LEVEL)
	r, _ := New(cfg.RestAPI)
	switch testType {
	case "tribe":
		mockTribeManager := &fixtures.MockTribeManager{}
		r.BindTribeManager(mockTribeManager)
	case "plugin":

func TestSubscriptionGroups_ProcessSpecifiedDynamicNegative(t *testing.T) {
	log.SetLevel(log.DebugLevel)
	c := New(getTestSGConfig())

	lpe := newLstnToPluginEvents()
	c.eventManager.RegisterHandler("TestSubscriptionGroups_Process", lpe)
	c.Start()

	Convey("Loading a mock collector plugin", t, func() {
		_, err := loadPlg(c, helper.PluginFilePath("snap-plugin-collector-mock1"))
		So(err, ShouldBeNil)
		<-lpe.load

		Convey("Subscription group created for requested metric with specified instance of dynamic element", func() {
			requested := mockRequestedMetric{namespace: core.NewNamespace("intel", "mock").AddDynamicElement("host", "name of the host").AddStaticElement("baz")}
			// specified dynamic element
			requested.Namespace()[2].Value = "host0"
			mock1 := mockSubscribedPlugin{
				typeName: core.CollectorPluginType,
				name:     "mock",
				version:  1,
				config:   cdata.NewNode(),
			}

			sg := newSubscriptionGroups(c)
			So(sg, ShouldNotBeNil)
			sg.Add("task-id", []core.RequestedMetric{requested}, cdata.NewTree(), []core.SubscribedPlugin{mock1})
			<-lpe.sub
			So(len(sg.subscriptionMap), ShouldEqual, 1)
			group, ok := sg.subscriptionMap["task-id"]
			So(ok, ShouldBeTrue)
			So(group, ShouldNotBeNil)
			So(len(group.plugins), ShouldEqual, 1)
			So(subscribedPluginsContain(group.plugins, mock1), ShouldBeTrue)
			plgKey := key(group.plugins[0])
			So(group.metrics, ShouldContainKey, plgKey)
			metrics := group.metrics[plgKey].Metrics()
			// expected 1 subscribed metrics:`/intel/mock/host0/baz`
			So(len(metrics), ShouldEqual, 1)
			So(len(group.requestedMetrics), ShouldEqual, 1)
			So(metrics[0].Namespace().String(), ShouldEqual, group.requestedMetrics[0].Namespace().String())
			mts1 := len(metrics)

			Convey("loading another mock", func() {
				anotherMock1 := mockSubscribedPlugin{
					typeName: core.CollectorPluginType,
					name:     "anothermock",
					version:  1,
					config:   cdata.NewNode(),
				}
				_, err := loadPlg(c, helper.PluginFilePath("snap-plugin-collector-anothermock1"))
				So(err, ShouldBeNil)
				<-lpe.load
				serrs := sg.Process()
				So(len(serrs), ShouldEqual, 0)

				So(len(sg.subscriptionMap), ShouldEqual, 1)
				group, ok := sg.subscriptionMap["task-id"]
				So(ok, ShouldBeTrue)
				So(group, ShouldNotBeNil)
				So(len(group.plugins), ShouldEqual, 1)
				So(subscribedPluginsContain(group.plugins, mock1), ShouldBeTrue)
				So(subscribedPluginsContain(group.plugins, anotherMock1), ShouldBeFalse)
				plgKey := key(group.plugins[0])
				So(group.metrics, ShouldContainKey, plgKey)
				metrics := group.metrics[plgKey].Metrics()
				// expected 1 subscribed metrics:`/intel/mock/host0/baz`
				So(len(metrics), ShouldEqual, 1)
				So(len(group.requestedMetrics), ShouldEqual, 1)
				So(metrics[0].Namespace().String(), ShouldEqual, group.requestedMetrics[0].Namespace().String())
				mts2 := len(metrics)
				So(mts1, ShouldEqual, mts2)
			})
		})
	})
}

func TestSubscriptionGroups_ProcessDynamicPositive(t *testing.T) {
	log.SetLevel(log.DebugLevel)
	c := New(getTestSGConfig())

	lpe := newLstnToPluginEvents()
	c.eventManager.RegisterHandler("TestSubscriptionGroups_Process", lpe)
	c.Start()

	Convey("Loading a mock collector plugin", t, func() {
		_, err := loadPlg(c, helper.PluginFilePath("snap-plugin-collector-mock1"))
		So(err, ShouldBeNil)

		<-lpe.load

		Convey("ValidateDeps", func() {
			requested := mockRequestedMetric{namespace: core.NewNamespace("intel").AddDynamicElement("wild", "wild description")}
			mock1 := mockSubscribedPlugin{
				typeName: core.CollectorPluginType,
				name:     "mock",
				version:  1,
				config:   cdata.NewNode(),
			}
			cnode := cdata.NewNode()
			cnode.AddItem("password", ctypes.ConfigValueStr{Value: "secret"})
			ctree := cdata.NewTree()
			ctree.Add([]string{"intel", "mock"}, cnode)
			sg := newSubscriptionGroups(c)
			So(sg, ShouldNotBeNil)
			errs := sg.ValidateDeps([]core.RequestedMetric{requested}, []core.SubscribedPlugin{mock1}, ctree)
			So(errs, ShouldBeNil)
			Convey("Subscription group created for requested metric with wildcards", func() {
				sg.Add("task-id", []core.RequestedMetric{requested}, ctree, []core.SubscribedPlugin{mock1})
				<-lpe.sub
				So(len(sg.subscriptionMap), ShouldEqual, 1)
				group, ok := sg.subscriptionMap["task-id"]
				So(ok, ShouldBeTrue)
				So(group, ShouldNotBeNil)
				So(len(group.plugins), ShouldEqual, 1)
				So(subscribedPluginsContain(group.plugins, mock1), ShouldBeTrue)
				plgKey := key(group.plugins[0])
				So(group.metrics, ShouldContainKey, plgKey)
				metrics := group.metrics[plgKey].Metrics()
				So(len(metrics), ShouldBeGreaterThan, 1)
				So(len(group.requestedMetrics), ShouldEqual, 1)
				mts1 := len(metrics)

				Convey("loading another mock", func() {
					anotherMock1 := mockSubscribedPlugin{
						typeName: core.CollectorPluginType,
						name:     "anothermock",
						version:  1,
						config:   cdata.NewNode(),
					}
					_, err := loadPlg(c, path.Join(helper.PluginFilePath("snap-plugin-collector-anothermock1")))
					So(err, ShouldBeNil)
					<-lpe.load
					serrs := sg.Process()
					So(len(serrs), ShouldEqual, 0)

					So(len(sg.subscriptionMap), ShouldEqual, 1)
					group, ok := sg.subscriptionMap["task-id"]
					So(ok, ShouldBeTrue)
					So(group, ShouldNotBeNil)
					So(len(group.plugins), ShouldEqual, 2)
					So(subscribedPluginsContain(group.plugins, mock1), ShouldBeTrue)
					So(subscribedPluginsContain(group.plugins, anotherMock1), ShouldBeTrue)
					plgKey1 := key(group.plugins[0])
					plgKey2 := key(group.plugins[1])
					So(group.metrics, ShouldContainKey, plgKey1)
					So(group.metrics, ShouldContainKey, plgKey2)
					metricsPlg1 := group.metrics[plgKey1].Metrics()
					metricsPlg2 := group.metrics[plgKey2].Metrics()
					So(len(metricsPlg1), ShouldBeGreaterThan, 1)
					So(len(metricsPlg2), ShouldBeGreaterThan, 1)
					So(len(group.requestedMetrics), ShouldEqual, 1)
					mts2 := len(metricsPlg1) + len(metricsPlg2)
					So(mts2, ShouldBeGreaterThan, mts1)
				})
			})
		})
	})
}

func TestSubscriptionGroups_ProcessStaticPositive(t *testing.T) {
	log.SetLevel(log.DebugLevel)
	c := New(getTestSGConfig())

	lpe := newLstnToPluginEvents()
	c.eventManager.RegisterHandler("TestSubscriptionGroups_Process", lpe)
	c.Start()

	Convey("Loading a mock collector plugin", t, func() {
		_, err := loadPlg(c, helper.PluginFilePath("snap-plugin-collector-mock1"))
		So(err, ShouldBeNil)
		<-lpe.load

		Convey("Subscription group created for requested metric with no wildcards", func() {
			requested := mockRequestedMetric{namespace: core.NewNamespace("intel", "mock", "foo")}
			mock1 := mockSubscribedPlugin{
				typeName: core.CollectorPluginType,
				name:     "mock",
				version:  1,
				config:   cdata.NewNode(),
			}

			sg := newSubscriptionGroups(c)
			So(sg, ShouldNotBeNil)
			sg.Add("task-id", []core.RequestedMetric{requested}, cdata.NewTree(), []core.SubscribedPlugin{mock1})
			<-lpe.sub
			So(len(sg.subscriptionMap), ShouldEqual, 1)
			group, ok := sg.subscriptionMap["task-id"]
			So(ok, ShouldBeTrue)
			So(group, ShouldNotBeNil)
			So(len(group.plugins), ShouldEqual, 1)
			So(subscribedPluginsContain(group.plugins, mock1), ShouldBeTrue)
			plgKey := key(group.plugins[0])
			So(group.metrics, ShouldContainKey, plgKey)
			metrics := group.metrics[plgKey].Metrics()
			So(len(metrics), ShouldEqual, 1)
			So(len(group.requestedMetrics), ShouldEqual, 1)
			So(metrics[0].Namespace().String(), ShouldEqual, group.requestedMetrics[0].Namespace().String())

			Convey("loading another mock", func() {
				mock2 := mockSubscribedPlugin{
					typeName: core.CollectorPluginType,
					name:     "mock",
					version:  2,
					config:   cdata.NewNode(),
				}
				_, err := loadPlg(c, helper.PluginFilePath("snap-plugin-collector-mock2"))
				So(err, ShouldBeNil)
				<-lpe.load
				serrs := sg.Process()
				So(len(serrs), ShouldEqual, 0)

				So(len(sg.subscriptionMap), ShouldEqual, 1)
				group, ok := sg.subscriptionMap["task-id"]
				So(ok, ShouldBeTrue)
				So(group, ShouldNotBeNil)
				So(len(group.plugins), ShouldEqual, 1)
				So(subscribedPluginsContain(group.plugins, mock1), ShouldBeFalse)
				So(subscribedPluginsContain(group.plugins, mock2), ShouldBeTrue)
				plgKey := key(group.plugins[0])
				So(group.metrics, ShouldContainKey, plgKey)
				metrics := group.metrics[plgKey].Metrics()
				So(len(metrics), ShouldEqual, 1)
				So(len(group.requestedMetrics), ShouldEqual, 1)
				So(metrics[0].Namespace().String(), ShouldEqual, group.requestedMetrics[0].Namespace().String())
			})
		})
	})
}

	"github.com/intelsdi-x/snap/core/ctypes"
	"github.com/intelsdi-x/snap/mgmt/rest/rbody"
	"github.com/intelsdi-x/snap/pkg/cfgfile"
	"github.com/intelsdi-x/snap/plugin/helper"
	"github.com/intelsdi-x/snap/scheduler"
	"github.com/intelsdi-x/snap/scheduler/wmap"
	. "github.com/smartystreets/goconvey/convey"
)

var (
	// Switching this turns on logging for all the REST API calls
	LOG_LEVEL = log.WarnLevel

	SNAP_PATH              = helper.BuildPath
	SNAP_AUTODISCOVER_PATH = os.Getenv("SNAP_AUTODISCOVER_PATH")
	MOCK_PLUGIN_PATH1      = helper.PluginFilePath("snap-plugin-collector-mock1")
	MOCK_PLUGIN_PATH2      = helper.PluginFilePath("snap-plugin-collector-mock2")
	FILE_PLUGIN_PATH       = helper.PluginFilePath("snap-plugin-publisher-mock-file")

	CompressedUpload = true
	TotalUploadSize  = 0
	UploadCount      = 0
)

const (
	MOCK_CONSTRAINTS = `{
		"$schema": "http://json-schema.org/draft-04/schema#",
		"title": "snapd global config schema",
		"type": ["object", "null"],
		"properties": {
			"control": { "$ref": "#/definitions/control" },

package fixtures

import (
	"encoding/json"
	"time"

	"github.com/intelsdi-x/snap/core"
	"github.com/intelsdi-x/snap/core/cdata"
	"github.com/intelsdi-x/snap/plugin/helper"
)

var (
	PluginName = "snap-plugin-collector-mock2"
	SnapPath   = helper.BuildPath
	PluginPath = helper.PluginFilePath(PluginName)

	JSONRPCPluginName = "snap-plugin-collector-mock1"
	JSONRPCPluginPath = helper.PluginFilePath(JSONRPCPluginName)
)

// mocks a metric type
type MockMetricType struct {
	Namespace_ core.Namespace
	Cfg        *cdata.ConfigDataNode
	Ver        int
}

func (m MockMetricType) MarshalJSON() ([]byte, error) {
	return json.Marshal(&struct {
		Namespace core.Namespace        `json:"namespace"`

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.
//.........这里部分代码省略.........

package fixtures

import (
	"encoding/json"
	"time"

	"github.com/intelsdi-x/snap/core"
	"github.com/intelsdi-x/snap/core/cdata"
	"github.com/intelsdi-x/snap/plugin/helper"
)

var (
	SnapPath = helper.BuildPath

	PluginNameMock1 = "snap-plugin-collector-mock1"
	PluginPathMock1 = helper.PluginFilePath(PluginNameMock1)

	PluginNameMock2 = "snap-plugin-collector-mock2"
	PluginPathMock2 = helper.PluginFilePath(PluginNameMock2)
)

// mocks a metric type
type MockMetricType struct {
	Namespace_ core.Namespace
	Cfg        *cdata.ConfigDataNode
	Ver        int
}

func (m MockMetricType) MarshalJSON() ([]byte, error) {
	return json.Marshal(&struct {
		Namespace core.Namespace        `json:"namespace"`