func connectTestEs(t *testing.T, cfg interface{}) (outputs.BulkOutputer, *Client) {
	config, err := common.NewConfigFrom(map[string]interface{}{
		"hosts":            GetEsHost(),
		"username":         os.Getenv("ES_USER"),
		"password":         os.Getenv("ES_PASS"),
		"template.enabled": false,
	})
	if err != nil {
		t.Fatal(err)
	}

	tmp, err := common.NewConfigFrom(cfg)
	if err != nil {
		t.Fatal(err)
	}

	err = config.Merge(tmp)
	if err != nil {
		t.Fatal(err)
	}

	output, err := New("libbeat", config, 0)
	if err != nil {
		t.Fatal(err)
	}

	es := output.(*elasticsearchOutput)
	client := es.randomClient()
	// Load version number
	client.Connect(3 * time.Second)

	return es, client
}

func getRedisModuleConfig(pass string) (*common.Config, error) {
	return common.NewConfigFrom(RedisModuleConfig{
		Module:   "redis",
		Hosts:    []string{LOCAL_REDIS},
		Password: pass,
	})
}

func TestModuleRunner(t *testing.T) {
	pubClient, factory := newPubClientFactory()

	config, err := common.NewConfigFrom(map[string]interface{}{
		"module":     moduleName,
		"metricsets": []string{metricSetName},
	})
	if err != nil {
		t.Fatal(err)
	}

	// Create a new ModuleWrapper based on the configuration.
	module, err := metricbeat.NewModuleWrapper(config, mb.Registry)
	if err != nil {
		t.Fatal(err)
	}

	// Create the ModuleRunner facade.
	runner := metricbeat.NewModuleRunner(factory, module)

	// Start the module and have it publish to a new publisher.Client.
	runner.Start()

	assert.NotNil(t, <-pubClient.Channel)

	// Stop the module. This blocks until all MetricSets in the Module have
	// stopped and the publisher.Client is closed.
	runner.Stop()
}

func newConfig(t testing.TB, moduleConfig interface{}) *common.Config {
	config, err := common.NewConfigFrom(moduleConfig)
	if err != nil {
		t.Fatal(err)
	}
	return config
}

func createWatchUpdater(monitor *Monitor) func(content []byte) {
	return func(content []byte) {
		defer logp.Recover("Failed applying monitor watch")

		// read multiple json objects from content
		dec := json.NewDecoder(bytes.NewBuffer(content))
		var configs []*common.Config
		for dec.More() {
			var obj map[string]interface{}
			err := dec.Decode(&obj)
			if err != nil {
				logp.Err("Failed parsing json object: %v", err)
				return
			}

			logp.Info("load watch object: %v", obj)

			cfg, err := common.NewConfigFrom(obj)
			if err != nil {
				logp.Err("Failed normalizing json input: %v", err)
				return
			}

			configs = append(configs, cfg)
		}

		// apply read configurations
		if err := monitor.Update(configs); err != nil {
			logp.Err("Failed applying configuration: %v", err)
		}
	}
}

func newTestLumberjackOutput(
	t *testing.T,
	test string,
	config map[string]interface{},
) outputs.BulkOutputer {
	if config == nil {
		config = map[string]interface{}{
			"hosts": []string{getLogstashHost()},
			"index": testLogstashIndex(test),
		}
	}

	plugin := outputs.FindOutputPlugin("logstash")
	if plugin == nil {
		t.Fatalf("No logstash output plugin found")
	}

	cfg, _ := common.NewConfigFrom(config)
	output, err := plugin("", cfg, 0)
	if err != nil {
		t.Fatalf("init logstash output plugin failed: %v", err)
	}

	return output.(outputs.BulkOutputer)
}

func TestNamespace(t *testing.T) {
	tests := []struct {
		name string
	}{
		{"test"},
		{"test.test"},
		{"abc.def.test"},
	}

	for i, test := range tests {
		t.Logf("run (%v): %v", i, test.name)

		ns := NewNamespace()
		err := ns.Register(test.name, newTestFilterRule)
		fatalError(t, err)

		cfg, _ := common.NewConfigFrom(map[string]interface{}{
			test.name: nil,
		})

		filter, err := ns.Plugin()(*cfg)

		assert.NoError(t, err)
		assert.NotNil(t, filter)
	}
}

func createElasticsearchConnection(flushInterval int, bulkSize int) *elasticsearchOutput {
	index := fmt.Sprintf("packetbeat-int-test-%d", os.Getpid())

	esPort, err := strconv.Atoi(GetEsPort())

	if err != nil {
		logp.Err("Invalid port. Cannot be converted to in: %s", GetEsPort())
	}

	config, _ := common.NewConfigFrom(map[string]interface{}{
		"save_topology":    true,
		"hosts":            []string{GetEsHost()},
		"port":             esPort,
		"username":         os.Getenv("ES_USER"),
		"password":         os.Getenv("ES_PASS"),
		"path":             "",
		"index":            fmt.Sprintf("%v-%%{+yyyy.MM.dd}", index),
		"protocol":         "http",
		"flush_interval":   flushInterval,
		"bulk_max_size":    bulkSize,
		"template.enabled": false,
	})

	output := &elasticsearchOutput{beatName: "test"}
	output.init(config, 10)
	return output
}

// newMetricSet instantiates a new MetricSet using the given configuration.
// The ModuleFactory and MetricSetFactory are obtained from the global
// Registry.
func newMetricSet(t testing.TB, config interface{}) mb.MetricSet {
	c, err := common.NewConfigFrom(config)
	if err != nil {
		t.Fatal(err)
	}
	m, err := mb.NewModules([]*common.Config{c}, mb.Registry)
	if err != nil {
		t.Fatal(err)
	}
	if !assert.Len(t, m, 1) {
		t.FailNow()
	}

	var metricSet mb.MetricSet
	for _, v := range m {
		if !assert.Len(t, v, 1) {
			t.FailNow()
		}

		metricSet = v[0]
		break
	}

	if !assert.NotNil(t, metricSet) {
		t.FailNow()
	}
	return metricSet
}

func newTestElasticsearchOutput(t *testing.T, test string) *testOutputer {
	plugin := outputs.FindOutputPlugin("elasticsearch")
	if plugin == nil {
		t.Fatalf("No elasticsearch output plugin found")
	}

	index := testElasticsearchIndex(test)
	connection := esConnect(t, index)

	flushInterval := 0
	bulkSize := 0
	config, _ := common.NewConfigFrom(map[string]interface{}{
		"hosts":          []string{getElasticsearchHost()},
		"index":          index,
		"flush_interval": &flushInterval,
		"bulk_max_size":  &bulkSize,
		"username":       os.Getenv("ES_USER"),
		"password":       os.Getenv("ES_PASS"),
	})

	output, err := plugin(config, 10)
	if err != nil {
		t.Fatalf("init elasticsearch output plugin failed: %v", err)
	}

	es := &testOutputer{}
	es.BulkOutputer = output.(outputs.BulkOutputer)
	es.esConnection = connection
	return es
}

func TestValidJSONDepthTwo(t *testing.T) {
	input := common.MapStr{
		"msg":      "{\"log\":\"{\\\"level\\\":\\\"info\\\"}\",\"stream\":\"stderr\",\"count\":3}",
		"pipeline": "us1",
	}

	testConfig, _ = common.NewConfigFrom(map[string]interface{}{
		"fields":       fields,
		"processArray": false,
		"maxDepth":     2,
	})

	actual := getActualValue(t, testConfig, input)

	expected := common.MapStr{
		"msg": map[string]interface{}{
			"log": map[string]interface{}{
				"level": "info",
			},
			"stream": "stderr",
			"count":  3,
		},
		"pipeline": "us1",
	}

	assert.Equal(t, expected.String(), actual.String())

}

func newRedisTestingOutput(t *testing.T, cfg map[string]interface{}) *redisOut {
	params := struct {
		Expire int `config:"topology_expire"`
	}{15}

	config, err := common.NewConfigFrom(cfg)
	if err != nil {
		t.Fatalf("Error reading config: %v", err)
	}

	plugin := outputs.FindOutputPlugin("redis")
	if plugin == nil {
		t.Fatalf("redis output module not registered")
	}

	if err := config.Unpack(&params); err != nil {
		t.Fatalf("Failed to unpack topology_expire: %v", err)
	}

	out, err := plugin("libbeat", config, params.Expire)
	if err != nil {
		t.Fatalf("Failed to initialize redis output: %v", err)
	}

	return out.(*redisOut)
}

func TestModuleConfig(t *testing.T) {
	tests := []struct {
		in  map[string]interface{}
		out ModuleConfig
		err string
	}{
		{
			in:  map[string]interface{}{},
			out: defaultModuleConfig,
		},
	}

	for _, test := range tests {
		c, err := common.NewConfigFrom(test.in)
		if err != nil {
			t.Fatal(err)
		}

		mc := defaultModuleConfig
		err = c.Unpack(&mc)
		if test.err != "" {
			assert.Error(t, err)
			assert.Contains(t, err.Error(), test.err)
			continue
		}

		assert.Equal(t, test.out, mc)
	}
}

// ExampleModuleRunner demonstrates how to use ModuleRunner to start and stop
// a module.
func ExampleModuleRunner() {
	// A *beat.Beat is injected into a Beater when it runs and contains the
	// Publisher used to publish events. This Beat pointer is created here only
	// for demonstration purposes.
	var b *beat.Beat

	config, err := common.NewConfigFrom(map[string]interface{}{
		"module":     moduleName,
		"metricsets": []string{metricSetName},
	})
	if err != nil {
		return
	}

	// Create a new ModuleWrapper based on the configuration.
	module, err := metricbeat.NewModuleWrapper(config, mb.Registry)
	if err != nil {
		return
	}

	// Create the ModuleRunner facade.
	runner := metricbeat.NewModuleRunner(b.Publisher.Connect, module)

	// Start the module and have it publish to a new publisher.Client.
	runner.Start()

	// Stop the module. This blocks until all MetricSets in the Module have
	// stopped and the publisher.Client is closed.
	runner.Stop()
}

func mustNewConfigFrom(from interface{}) *common.Config {
	cfg, err := common.NewConfigFrom(from)
	if err != nil {
		panic(err)
	}
	return cfg
}

func makeConfig(t *testing.T, in map[string]interface{}) *common.Config {
	cfg, err := common.NewConfigFrom(in)
	if err != nil {
		t.Fatal(err)
	}
	return cfg
}

func TestMissingFields(t *testing.T) {

	if testing.Verbose() {
		logp.LogInit(logp.LOG_DEBUG, "", false, true, []string{"*"})
	}

	yml := []map[string]interface{}{
		map[string]interface{}{
			"include_fields": map[string]interface{}{
				"equals": map[string]string{
					"type": "process",
				},
			},
		},
	}

	config := filter.FilterPluginConfig{}

	for _, rule := range yml {
		c := map[string]common.Config{}

		for name, ruleYml := range rule {
			ruleConfig, err := common.NewConfigFrom(ruleYml)
			assert.Nil(t, err)

			c[name] = *ruleConfig
		}
		config = append(config, c)
	}

	_, err := filter.New(config)
	assert.NotNil(t, err)

}

// TestOutputLoadTemplate checks that the template is inserted before
// the first event is published.
func TestOutputLoadTemplate(t *testing.T) {

	client := GetTestingElasticsearch()
	err := client.Connect(5 * time.Second)
	if err != nil {
		t.Fatal(err)
	}

	// delete template if it exists
	client.request("DELETE", "/_template/libbeat", "", nil, nil)

	// Make sure template is not yet there
	assert.False(t, client.CheckTemplate("libbeat"))

	templatePath := "../../../packetbeat/packetbeat.template.json"

	if strings.HasPrefix(client.Connection.version, "2.") {
		templatePath = "../../../packetbeat/packetbeat.template-es2x.json"
	}

	tPath, err := filepath.Abs(templatePath)
	if err != nil {
		t.Fatal(err)
	}
	config := map[string]interface{}{
		"hosts": GetEsHost(),
		"template": map[string]interface{}{
			"name":                "libbeat",
			"path":                tPath,
			"versions.2x.enabled": false,
		},
	}

	cfg, err := common.NewConfigFrom(config)
	if err != nil {
		t.Fatal(err)
	}

	output, err := New("libbeat", cfg, 0)
	if err != nil {
		t.Fatal(err)
	}
	event := outputs.Data{Event: common.MapStr{
		"@timestamp": common.Time(time.Now()),
		"host":       "test-host",
		"type":       "libbeat",
		"message":    "Test message from libbeat",
	}}

	err = output.PublishEvent(nil, outputs.Options{Guaranteed: true}, event)
	if err != nil {
		t.Fatal(err)
	}

	// Guaranteed publish, so the template should be there

	assert.True(t, client.CheckTemplate("libbeat"))

}

func NewTestModule(t testing.TB, config interface{}) *TestModule {
	c, err := common.NewConfigFrom(config)
	if err != nil {
		t.Fatal(err)
	}

	return &TestModule{RawConfig: c}
}

func TestWhenProcessor(t *testing.T) {
	type config map[string]interface{}

	tests := []struct {
		title    string
		filter   config
		events   []common.MapStr
		expected int
	}{
		{
			"condition_matches",
			config{"when.equals.i": 10},
			[]common.MapStr{{"i": 10}},
			1,
		},
		{
			"condition_fails",
			config{"when.equals.i": 11},
			[]common.MapStr{{"i": 10}},
			0,
		},
		{
			"no_condition",
			config{},
			[]common.MapStr{{"i": 10}},
			1,
		},
	}

	for i, test := range tests {
		t.Logf("run test (%v): %v", i, test.title)

		config, err := common.NewConfigFrom(test.filter)
		if err != nil {
			t.Error(err)
			continue
		}

		cf := &countFilter{}
		filter, err := NewConditional(func(_ common.Config) (Processor, error) {
			return cf, nil
		})(*config)
		if err != nil {
			t.Error(err)
			continue
		}

		for _, event := range test.events {
			_, err := filter.Run(event)
			if err != nil {
				t.Error(err)
			}
		}

		assert.Equal(t, test.expected, cf.N)
	}
}