Function TestAcquistionSocket

pkg/acquisition/modules/http/http_test.go:343–389

(t *testing.T)

}

func TestAcquistionSocket(t *testing.T) {
	tempDir := t.TempDir()
	socketFile := filepath.Join(tempDir, "test.sock")

	ctx := t.Context()
	h := &Source{}
	out, reg, tomb := SetupAndRunHTTPSource(t, h, []byte(`
source: http
listen_socket: `+socketFile+`
path: /test
auth_type: headers
headers:
  key: test`), 2)

	time.Sleep(1 * time.Second)
	rawEvt := `{"test": "test"}`
	errChan := make(chan error)
	go assertEvents(out, []string{rawEvt}, errChan)

	dialer := &net.Dialer{}
	client := &http.Client{
		Transport: &http.Transport{
			DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
				return dialer.DialContext(ctx, "unix", socketFile)
			},
		},
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, fmt.Sprintf("%s/test", testHTTPServerAddr), strings.NewReader(rawEvt))
	require.NoError(t, err)

	req.Header.Add("Key", "test")
	resp, err := client.Do(req)
	require.NoError(t, err)

	assert.Equal(t, http.StatusOK, resp.StatusCode)
	closeBody(t, resp)

	err = <-errChan
	require.NoError(t, err)

	assertMetrics(t, reg, h.GetMetrics(), 1)

	h.Server.Close()
	tomb.Kill(nil)
	err = tomb.Wait()
	require.NoError(t, err)
}

type slowReader struct {
	delay time.Duration