[WIP] Add basic heartbeat job tests

heartbeat/monitors/active/http/http_test.go

@@ -0,0 +1,88 @@
+package http
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/elastic/beats/heartbeat/monitors"
+	"github.com/elastic/beats/heartbeat/skima"
+	"github.com/elastic/beats/heartbeat/testutil"
+	"github.com/elastic/beats/libbeat/beat"
+	"github.com/elastic/beats/libbeat/common"
+)
+
+func executeHTTPMonitorHostJob(t *testing.T, handlerFunc http.HandlerFunc) (*httptest.Server, beat.Event) {
+	server := httptest.NewServer(handlerFunc)
+	defer server.Close()
+
+	config := common.NewConfig()
+	config.SetString("urls", 0, server.URL)
+
+	jobs, err := create(monitors.Info{}, config)
+	if err != nil {
+		t.FailNow()
+	}
+	job := jobs[0]
+
+	event, _, err := job.Run()
+
+	return server, event
+}
+
+func httpChecks(urlStr string, statusCode int) skima.Validator {
+	return skima.Schema(skima.Map{
+		"http": skima.Map{
+			"url": urlStr,
+			"response.status_code":   statusCode,
+			"rtt.content.us":         skima.IsDuration,
+			"rtt.response_header.us": skima.IsDuration,
+			"rtt.total.us":           skima.IsDuration,
+			"rtt.validate.us":        skima.IsDuration,
+			"rtt.write_request.us":   skima.IsDuration,
+		},
+	})
+}
+
+func httpErrorChecks(urlStr string, statusCode int) skima.Validator {
+	return skima.Schema(skima.Map{
+		"error": skima.Map{
+			"message": "502 Bad Gateway",
+			"type":    "validate",
+		},
+		"http": skima.Map{
+			"url": urlStr,
+			// TODO: This should work in the future "response.status_code":   statusCode,
+			"rtt.content.us":         skima.IsDuration,
+			"rtt.response_header.us": skima.IsDuration,
+			"rtt.validate.us":        skima.IsDuration,
+			"rtt.write_request.us":   skima.IsDuration,
+		},
+	})
+}
+
+func TestOKJob(t *testing.T) {
+	server, event := executeHTTPMonitorHostJob(t, testutil.HelloWorldHandler)
+	port, err := testutil.ServerPort(server)
+	assert.Nil(t, err)
+
+	skima.Strict(skima.Compose(
+		testutil.MonitorChecks("http@"+server.URL, "127.0.0.1", "http", "up"),
+		testutil.TcpChecks(port),
+		httpChecks(server.URL, http.StatusOK),
+	))(t, event.Fields)
+}
+
+func TestBadGatewayJob(t *testing.T) {
+	server, event := executeHTTPMonitorHostJob(t, testutil.BadGatewayHandler)
+	port, err := testutil.ServerPort(server)
+	assert.Nil(t, err)
+
+	skima.Strict(skima.Compose(
+		testutil.MonitorChecks("http@"+server.URL, "127.0.0.1", "http", "down"),
+		testutil.TcpChecks(port),
+		httpErrorChecks(server.URL, http.StatusBadGateway),
+	))(t, event.Fields)
+}

heartbeat/monitors/active/http/task.go

@@ -218,41 +218,62 @@ func execPing(
 	body []byte,
 	timeout time.Duration,
 	validator func(*http.Response) error,
-) (time.Time, time.Time, common.MapStr, reason.Reason) {
+) (start time.Time, end time.Time, event common.MapStr, errReason reason.Reason) {
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()
 
-	req = req.WithContext(ctx)
+	req = attachRequestBody(&ctx, req, body)
+	start, end, resp, errReason := execRequest(client, req, validator)
+
+	if errReason != nil {
+		return start, end, nil, errReason
+	}
+
+	event = makeEvent(end.Sub(start), resp)
+
+	return start, end, event, nil
+}
+
+func attachRequestBody(ctx *context.Context, req *http.Request, body []byte) *http.Request {
+	req = req.WithContext(*ctx)
 	if len(body) > 0 {
 		req.Body = ioutil.NopCloser(bytes.NewBuffer(body))
 		req.ContentLength = int64(len(body))
 	}
 
-	start := time.Now()
+	return req
+}
+
+func execRequest(client *http.Client, req *http.Request, validator func(*http.Response) error) (start time.Time, end time.Time, resp *http.Response, errReason reason.Reason) {
+	start = time.Now()
 	resp, err := client.Do(req)
-	end := time.Now()
+	if resp != nil { // If above errors, the response will be nil
+		defer resp.Body.Close()
+	}
+	end = time.Now()
+
 	if err != nil {
 		return start, end, nil, reason.IOFailed(err)
 	}
-	defer resp.Body.Close()
 
 	err = validator(resp)
 	end = time.Now()
+	if err != nil {
+		return start, end, resp, reason.ValidateFailed(err)
+	}
+
+	return start, end, resp, nil
+}
 
-	rtt := end.Sub(start)
-	event := common.MapStr{"http": common.MapStr{
+func makeEvent(rtt time.Duration, resp *http.Response) common.MapStr {
+	return common.MapStr{"http": common.MapStr{
 		"response": common.MapStr{
 			"status_code": resp.StatusCode,
 		},
 		"rtt": common.MapStr{
 			"total": look.RTT(rtt),
 		},
 	}}
-
-	if err != nil {
-		return start, end, event, reason.ValidateFailed(err)
-	}
-	return start, end, event, nil
 }
 
 func splitHostnamePort(requ *http.Request) (string, uint16, error) {

heartbeat/monitors/active/http/task_test.go

@@ -23,59 +23,69 @@ import (
 	"net/url"
 	"reflect"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestSplitHostnamePort(t *testing.T) {
 	var urlTests = []struct {
+		name          string
 		scheme        string
 		host          string
 		expectedHost  string
 		expectedPort  uint16
 		expectedError error
 	}{
 		{
+			"plain",
 			"http",
 			"foo",
 			"foo",
 			80,
 			nil,
 		},
 		{
+			"dotted domain",
 			"http",
 			"www.foo.com",
 			"www.foo.com",
 			80,
 			nil,
 		},
 		{
+			"dotted domain, custom port",
 			"http",
 			"www.foo.com:8080",
 			"www.foo.com",
 			8080,
 			nil,
 		},
 		{
+			"https plain",
 			"https",
 			"foo",
 			"foo",
 			443,
 			nil,
 		},
 		{
+			"custom port",
 			"http",
 			"foo:81",
 			"foo",
 			81,
 			nil,
 		},
 		{
+			"https custom port",
 			"https",
 			"foo:444",
 			"foo",
 			444,
 			nil,
 		},
 		{
+			"bad scheme",
 			"httpz",
 			"foo",
 			"foo",
@@ -84,27 +94,63 @@ func TestSplitHostnamePort(t *testing.T) {
 		},
 	}
 	for _, test := range urlTests {
-		url := &url.URL{
-			Scheme: test.scheme,
-			Host:   test.host,
-		}
-		request := &http.Request{
-			URL: url,
-		}
-		host, port, err := splitHostnamePort(request)
-		if err != nil {
-			if test.expectedError == nil {
-				t.Error(err)
-			} else if reflect.TypeOf(err) != reflect.TypeOf(test.expectedError) {
-				t.Errorf("Expected %T but got %T", err, test.expectedError)
+		test := test
+
+		t.Run(test.name, func(t *testing.T) {
+			url := &url.URL{
+				Scheme: test.scheme,
+				Host:   test.host,
+			}
+			request := &http.Request{
+				URL: url,
+			}
+			host, port, err := splitHostnamePort(request)
+
+			if err != nil {
+				if test.expectedError == nil {
+					t.Error(err)
+				} else if reflect.TypeOf(err) != reflect.TypeOf(test.expectedError) {
+					t.Errorf("Expected %T but got %T", err, test.expectedError)
+				}
+			} else {
+				if host != test.expectedHost {
+					t.Errorf("Unexpected host for %#v: expected %q, got %q", request, test.expectedHost, host)
+				}
+				if port != test.expectedPort {
+					t.Errorf("Unexpected port for %#v: expected %q, got %q", request, test.expectedPort, port)
+				}
 			}
-			continue
-		}
-		if host != test.expectedHost {
-			t.Errorf("Unexpected host for %#v: expected %q, got %q", request, test.expectedHost, host)
-		}
-		if port != test.expectedPort {
-			t.Errorf("Unexpected port for %#v: expected %q, got %q", request, test.expectedPort, port)
-		}
+
+		})
+	}
+}
+
+func makeTestHTTPRequest(t *testing.T) *http.Request {
+	req, err := http.NewRequest("GET", "http://example.net", nil)
+	assert.Nil(t, err)
+	return req
+}
+
+func TestZeroMaxRedirectShouldError(t *testing.T) {
+	checker := makeCheckRedirect(0)
+	req := makeTestHTTPRequest(t)
+
+	res := checker(req, nil)
+	assert.Equal(t, http.ErrUseLastResponse, res)
+}
+
+func TestNonZeroRedirect(t *testing.T) {
+	limit := 5
+	checker := makeCheckRedirect(limit)
+
+	var via []*http.Request
+	// Test requests within the limit
+	for i := 0; i < limit; i++ {
+		req := makeTestHTTPRequest(t)
+		assert.Nil(t, checker(req, via))
+		via = append(via, req)
 	}
+
+	// We are now at the limit, this request should fail
+	assert.Equal(t, http.ErrUseLastResponse, checker(makeTestHTTPRequest(t), via))
 }

heartbeat/monitors/active/tcp/tcp_test.go

@@ -0,0 +1,47 @@
+package tcp
+
+import (
+	"fmt"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/elastic/beats/heartbeat/monitors"
+	"github.com/elastic/beats/heartbeat/skima"
+	"github.com/elastic/beats/heartbeat/testutil"
+	"github.com/elastic/beats/libbeat/common"
+)
+
+func TestUpEndpoint(t *testing.T) {
+	server := httptest.NewServer(testutil.HelloWorldHandler)
+	defer server.Close()
+
+	port, err := testutil.ServerPort(server)
+	if err != nil {
+		t.FailNow()
+	}
+
+	config := common.NewConfig()
+	config.SetString("hosts", 0, "localhost")
+	config.SetInt("ports", 0, int64(port))
+
+	jobs, err := create(monitors.Info{}, config)
+	if err != nil {
+		t.FailNow()
+	}
+	job := jobs[0]
+
+	event, _, err := job.Run()
+	if err != nil {
+		t.FailNow()
+	}
+
+	skima.Strict(skima.Compose(
+		testutil.MonitorChecks(
+			fmt.Sprintf("tcp-tcp@localhost:%d", port),
+			"127.0.0.1",
+			"tcp",
+			"up",
+		),
+		testutil.TcpChecks(port),
+	))(t, event.Fields)
+}