package influxdb_listener

import (
	"bytes"
	"crypto/tls"
	"crypto/x509"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"os"
	"runtime"
	"strconv"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/golang-jwt/jwt/v5"
	"github.com/stretchr/testify/require"

	"github.com/influxdata/telegraf"
	"github.com/influxdata/telegraf/config"
	"github.com/influxdata/telegraf/testutil"
)

const (
	testMsg = "cpu_load_short,host=server01 value=12.0 1422568543702900257\n"

	testMsgNoNewline = "cpu_load_short,host=server01 value=12.0 1422568543702900257"

	testMsgs = `cpu_load_short,host=server02 value=12.0 1422568543702900257
cpu_load_short,host=server03 value=12.0 1422568543702900257
cpu_load_short,host=server04 value=12.0 1422568543702900257
cpu_load_short,host=server05 value=12.0 1422568543702900257
cpu_load_short,host=server06 value=12.0 1422568543702900257
`
	testPartial = `cpu,host=a value1=1
cpu,host=b value1=1,value2=+Inf,value3=3
cpu,host=c value1=1`

	badMsg = "blahblahblah: 42\n"

	emptyMsg = ""

	basicUsername = "test-username-please-ignore"
	basicPassword = "super-secure-password!"
)

var (
	pki             = testutil.NewPKI("../../../testutil/pki")
	parserTestCases = []struct {
		parser string
	}{
		{"upstream"},
		{"internal"},
	}
)

func newTestListener() *InfluxDBListener {
	listener := &InfluxDBListener{
		Log:            testutil.Logger{},
		ServiceAddress: "localhost:0",
		timeFunc:       time.Now,
	}
	return listener
}

func newTestAuthListener() *InfluxDBListener {
	listener := newTestListener()
	listener.BasicUsername = basicUsername
	listener.BasicPassword = basicPassword
	return listener
}

func newTestSecureListener() *InfluxDBListener {
	listener := &InfluxDBListener{
		Log:            testutil.Logger{},
		ServiceAddress: "localhost:0",
		ServerConfig:   *pki.TLSServerConfig(),
		timeFunc:       time.Now,
	}

	return listener
}

func getSecureClient() *http.Client {
	tlsConfig, err := pki.TLSClientConfig().TLSConfig()
	if err != nil {
		panic(err)
	}
	return &http.Client{
		Transport: &http.Transport{
			TLSClientConfig: tlsConfig,
		},
	}
}

func createURL(listener *InfluxDBListener, scheme string, path string, rawquery string) string {
	u := url.URL{
		Scheme:   scheme,
		Host:     "localhost:" + strconv.Itoa(listener.port),
		Path:     path,
		RawQuery: rawquery,
	}
	return u.String()
}

func TestWriteSecureNoClientAuth(t *testing.T) {
	listener := newTestSecureListener()
	listener.TLSAllowedCACerts = nil

	acc := &testutil.Accumulator{}
	require.NoError(t, listener.Init())
	require.NoError(t, listener.Start(acc))
	defer listener.Stop()

	cas := x509.NewCertPool()
	cas.AppendCertsFromPEM([]byte(pki.ReadServerCert()))
	noClientAuthClient := &http.Client{
		Transport: &http.Transport{
			TLSClientConfig: &tls.Config{
				RootCAs: cas,
			},
		},
	}

	// post single message to listener
	resp, err := noClientAuthClient.Post(createURL(listener, "https", "/write", "db=mydb"), "", bytes.NewBuffer([]byte(testMsg)))
	require.NoError(t, err)
	require.NoError(t, resp.Body.Close())
	require.EqualValues(t, 204, resp.StatusCode)
}

func TestWriteSecureWithClientAuth(t *testing.T) {
	listener := newTestSecureListener()

	acc := &testutil.Accumulator{}
	require.NoError(t, listener.Init())
	require.NoError(t, listener.Start(acc))
	defer listener.Stop()

	// post single message to listener
	resp, err := getSecureClient().Post(createURL(listener, "https", "/write", "db=mydb"), "", bytes.NewBuffer([]byte(testMsg)))
	require.NoError(t, err)
	require.NoError(t, resp.Body.Close())
	require.EqualValues(t, 204, resp.StatusCode)
}

func TestWriteBasicAuth(t *testing.T) {
	listener := newTestAuthListener()

	acc := &testutil.Accumulator{}
	require.NoError(t, listener.Init())
	require.NoError(t, listener.Start(acc))
	defer listener.Stop()

	client := &http.Client{}

	req, err := http.NewRequest("POST", createURL(listener, "http", "/write", "db=mydb"), bytes.NewBuffer([]byte(testMsg)))
	require.NoError(t, err)
	req.SetBasicAuth(basicUsername, basicPassword)
	resp, err := client.Do(req)
	require.NoError(t, err)
	require.NoError(t, resp.Body.Close())
	require.EqualValues(t, http.StatusNoContent, resp.StatusCode)
}

func TestWriteToken(t *testing.T) {
	plugin := &InfluxDBListener{
		ServiceAddress:    "localhost:0",
		TokenSharedSecret: "a S3cr3T $sTr1ng",
		TokenUsername:     "John Doe",
		Log:               testutil.Logger{},
		timeFunc:          time.Now,
	}
	require.NoError(t, plugin.Init())

	// Create a valid token
	token, err := jwt.NewWithClaims(jwt.SigningMethodHS512, jwt.MapClaims{
		"username": plugin.TokenUsername,
		"exp":      time.Now().Add(5 * time.Minute).Unix(),
	}).SignedString([]byte(plugin.TokenSharedSecret))
	require.NoError(t, err)

	var acc testutil.Accumulator
	require.NoError(t, plugin.Start(&acc))
	defer plugin.Stop()

	client := &http.Client{}
	req, err := http.NewRequest("POST", createURL(plugin, "http", "/write", "db=mydb"), bytes.NewBuffer([]byte(testMsg)))
	require.NoError(t, err)
	req.Header.Add("Authentication", "Bearer "+token)
	resp, err := client.Do(req)
	require.NoError(t, err)
	require.NoError(t, resp.Body.Close())
	require.EqualValues(t, http.StatusNoContent, resp.StatusCode)
}

func TestWriteTokenInvalidUser(t *testing.T) {
	plugin := &InfluxDBListener{
		ServiceAddress:    "localhost:0",
		TokenSharedSecret: "a S3cr3T $sTr1ng",
		TokenUsername:     "John Doe",
		Log:               testutil.Logger{},
		timeFunc:          time.Now,
	}
	require.NoError(t, plugin.Init())

	// Create a valid token
	token, err := jwt.NewWithClaims(jwt.SigningMethodHS512, jwt.MapClaims{
		"username": "peter",
		"exp":      time.Now().Add(5 * time.Minute).Unix(),
	}).SignedString([]byte(plugin.TokenSharedSecret))
	require.NoError(t, err)

	var acc testutil.Accumulator
	require.NoError(t, plugin.Start(&acc))
	defer plugin.Stop()

	client := &http.Client{}
	req, err := http.NewRequest("POST", createURL(plugin, "http", "/write", "db=mydb"), bytes.NewBuffer([]byte(testMsg)))
	require.NoError(t, err)
	req.Header.Add("Authentication", "Bearer "+token)
	resp, err := client.Do(req)
	require.NoError(t, err)
	require.NoError(t, resp.Body.Close())
	require.EqualValues(t, http.StatusUnauthorized, resp.StatusCode)
}

func TestWriteTokenExpired(t *testing.T) {
	plugin := &InfluxDBListener{
		ServiceAddress:    "localhost:0",
		TokenSharedSecret: "a S3cr3T $sTr1ng",
		TokenUsername:     "John Doe",
		Log:               testutil.Logger{},
		timeFunc:          time.Now,
	}
	require.NoError(t, plugin.Init())

	// Create a valid token
	token, err := jwt.NewWithClaims(jwt.SigningMethodHS512, jwt.MapClaims{
		"username": plugin.TokenUsername,
		"exp":      time.Now().Add(-5 * time.Minute).Unix(),
	}).SignedString([]byte(plugin.TokenSharedSecret))
	require.NoError(t, err)

	var acc testutil.Accumulator
	require.NoError(t, plugin.Start(&acc))
	defer plugin.Stop()

	client := &http.Client{}
	req, err := http.NewRequest("POST", createURL(plugin, "http", "/write", "db=mydb"), bytes.NewBuffer([]byte(testMsg)))
	require.NoError(t, err)
	req.Header.Add("Authentication", "Bearer "+token)
	resp, err := client.Do(req)
	require.NoError(t, err)
	body, err := io.ReadAll(resp.Body)
	require.NoError(t, err)
	require.NoError(t, resp.Body.Close())
	require.EqualValues(t, http.StatusUnauthorized, resp.StatusCode)
	require.EqualValues(t, "token expired", strings.TrimSpace(string(body)))
}

func TestWriteKeepDatabase(t *testing.T) {
	testMsgWithDB := "cpu_load_short,host=server01,database=wrongdb value=12.0 1422568543702900257\n"

	for _, tc := range parserTestCases {
		t.Run(fmt.Sprintf("parser %s", tc.parser), func(t *testing.T) {
			listener := newTestListener()
			listener.ParserType = tc.parser
			listener.DatabaseTag = "database"

			acc := &testutil.Accumulator{}
			require.NoError(t, listener.Init())
			require.NoError(t, listener.Start(acc))
			defer listener.Stop()

			// post single message to listener
			resp, err := http.Post(createURL(listener, "http", "/write", "db=mydb"), "", bytes.NewBuffer([]byte(testMsg)))
			require.NoError(t, err)
			require.NoError(t, resp.Body.Close())
			require.EqualValues(t, 204, resp.StatusCode)

			acc.Wait(1)
			acc.AssertContainsTaggedFields(t, "cpu_load_short",
				map[string]interface{}{"value": float64(12)},
				map[string]string{"host": "server01", "database": "mydb"},
			)

			// post single message to listener with a database tag in it already. It should be clobbered.
			resp, err = http.Post(createURL(listener, "http", "/write", "db=mydb"), "", bytes.NewBuffer([]byte(testMsgWithDB)))
			require.NoError(t, err)
			require.NoError(t, resp.Body.Close())
			require.EqualValues(t, 204, resp.StatusCode)

			acc.Wait(1)
			acc.AssertContainsTaggedFields(t, "cpu_load_short",
				map[string]interface{}{"value": float64(12)},
				map[string]string{"host": "server01", "database": "mydb"},
			)

			// post multiple message to listener
			resp, err = http.Post(createURL(listener, "http", "/write", "db=mydb"), "", bytes.NewBuffer([]byte(testMsgs)))
			require.NoError(t, err)
			require.NoError(t, resp.Body.Close())
			require.EqualValues(t, 204, resp.StatusCode)

			acc.Wait(2)
			hostTags := []string{"server02", "server03",
				"server04", "server05", "server06"}
			for _, hostTag := range hostTags {
				acc.AssertContainsTaggedFields(t, "cpu_load_short",
					map[string]interface{}{"value": float64(12)},
					map[string]string{"host": hostTag, "database": "mydb"},
				)
			}
		})
	}
}

func TestWriteRetentionPolicyTag(t *testing.T) {
	listener := newTestListener()
	listener.RetentionPolicyTag = "rp"

	acc := &testutil.Accumulator{}
	require.NoError(t, listener.Init())
	require.NoError(t, listener.Start(acc))
	defer listener.Stop()

	resp, err := http.Post(createURL(listener, "http", "/write", "rp=myrp"), "", bytes.NewBuffer([]byte("cpu time_idle=42")))
	require.NoError(t, err)
	require.NoError(t, resp.Body.Close())
	require.Equal(t, 204, resp.StatusCode)

	expected := []telegraf.Metric{
		testutil.MustMetric(
			"cpu",
			map[string]string{
				"rp": "myrp",
			},
			map[string]interface{}{
				"time_idle": 42.0,
			},
			time.Unix(0, 0),
		),
	}

	acc.Wait(1)
	testutil.RequireMetricsEqual(t, expected, acc.GetTelegrafMetrics(), testutil.IgnoreTime())
}

// http listener should add a newline at the end of the buffer if it's not there
func TestWriteNoNewline(t *testing.T) {
	for _, tc := range parserTestCases {
		t.Run(fmt.Sprintf("parser %s", tc.parser), func(t *testing.T) {
			listener := newTestListener()
			listener.ParserType = tc.parser

			acc := &testutil.Accumulator{}
			require.NoError(t, listener.Init())
			require.NoError(t, listener.Start(acc))
			defer listener.Stop()

			// post single message to listener
			resp, err := http.Post(createURL(listener, "http", "/write", "db=mydb"), "", bytes.NewBuffer([]byte(testMsgNoNewline)))
			require.NoError(t, err)
			require.NoError(t, resp.Body.Close())
			require.EqualValues(t, 204, resp.StatusCode)

			acc.Wait(1)
			acc.AssertContainsTaggedFields(t, "cpu_load_short",
				map[string]interface{}{"value": float64(12)},
				map[string]string{"host": "server01"},
			)
		})
	}
}

func TestPartialWrite(t *testing.T) {
	for _, tc := range parserTestCases {
		t.Run(fmt.Sprintf("parser %s", tc.parser), func(t *testing.T) {
			listener := newTestListener()
			listener.ParserType = tc.parser

			acc := &testutil.Accumulator{}
			require.NoError(t, listener.Init())
			require.NoError(t, listener.Start(acc))
			defer listener.Stop()

			// post single message to listener
			resp, err := http.Post(createURL(listener, "http", "/write", "db=mydb"), "", bytes.NewBuffer([]byte(testPartial)))
			require.NoError(t, err)
			require.NoError(t, resp.Body.Close())
			require.EqualValues(t, 400, resp.StatusCode)

			acc.Wait(1)
			acc.AssertContainsTaggedFields(t, "cpu",
				map[string]interface{}{"value1": float64(1)},
				map[string]string{"host": "a"},
			)
			acc.AssertContainsTaggedFields(t, "cpu",
				map[string]interface{}{"value1": float64(1)},
				map[string]string{"host": "c"},
			)
		})
	}
}

func TestWriteMaxLineSizeIncrease(t *testing.T) {
	// The term 'master_repl' used here is archaic language from redis
	hugeMetric, err := os.ReadFile("./testdata/huge_metric")
	require.NoError(t, err)

	for _, tc := range parserTestCases {
		t.Run(fmt.Sprintf("parser %s", tc.parser), func(t *testing.T) {
			listener := &InfluxDBListener{
				Log:            testutil.Logger{},
				ServiceAddress: "localhost:0",
				ParserType:     tc.parser,
				timeFunc:       time.Now,
			}

			acc := &testutil.Accumulator{}
			require.NoError(t, listener.Init())
			require.NoError(t, listener.Start(acc))
			defer listener.Stop()

			// Post a gigantic metric to the listener and verify that it writes OK this time:
			resp, err := http.Post(createURL(listener, "http", "/write", "db=mydb"), "", bytes.NewBuffer(hugeMetric))
			require.NoError(t, err)
			require.NoError(t, resp.Body.Close())
			require.EqualValues(t, 204, resp.StatusCode)
		})
	}
}

func TestWriteVerySmallMaxBody(t *testing.T) {
	// The term 'master_repl' used here is archaic language from redis
	hugeMetric, err := os.ReadFile("./testdata/huge_metric")
	require.NoError(t, err)

	for _, tc := range parserTestCases {
		t.Run(fmt.Sprintf("parser %s", tc.parser), func(t *testing.T) {
			listener := &InfluxDBListener{
				Log:            testutil.Logger{},
				ServiceAddress: "localhost:0",
				MaxBodySize:    config.Size(4096),
				ParserType:     tc.parser,
				timeFunc:       time.Now,
			}

			acc := &testutil.Accumulator{}
			require.NoError(t, listener.Init())
			require.NoError(t, listener.Start(acc))
			defer listener.Stop()

			for _, parser := range []string{"internal", "upstream"} {
				listener.ParserType = parser

				resp, err := http.Post(createURL(listener, "http", "/write", ""), "", bytes.NewBuffer(hugeMetric))
				require.NoError(t, err)
				require.NoError(t, resp.Body.Close())
				require.EqualValues(t, 413, resp.StatusCode)
			}
		})
	}
}

func TestWriteLargeLine(t *testing.T) {
	// The term 'master_repl' used here is archaic language from redis
	hugeMetric, err := os.ReadFile("./testdata/huge_metric")
	require.NoError(t, err)
	hugeMetricString := string(hugeMetric)

	for _, tc := range parserTestCases {
		t.Run(fmt.Sprintf("parser %s", tc.parser), func(t *testing.T) {
			listener := &InfluxDBListener{
				Log:            testutil.Logger{},
				ServiceAddress: "localhost:0",
				ParserType:     tc.parser,
				timeFunc: func() time.Time {
					return time.Unix(123456789, 0)
				},
			}

			acc := &testutil.Accumulator{}
			require.NoError(t, listener.Init())
			require.NoError(t, listener.Start(acc))
			defer listener.Stop()

			resp, err := http.Post(createURL(listener, "http", "/write", ""), "", bytes.NewBuffer([]byte(hugeMetricString+testMsgs)))
			require.NoError(t, err)
			require.NoError(t, resp.Body.Close())
			//todo: with the new parser, long lines aren't a problem.  Do we need to skip them?
			//require.EqualValues(t, 400, resp.StatusCode)

			expected := testutil.MustMetric(
				"super_long_metric",
				map[string]string{"foo": "bar"},
				map[string]interface{}{
					"clients":                     42,
					"connected_followers":         43,
					"evicted_keys":                44,
					"expired_keys":                45,
					"instantaneous_ops_per_sec":   46,
					"keyspace_hitrate":            47.0,
					"keyspace_hits":               48,
					"keyspace_misses":             49,
					"latest_fork_usec":            50,
					"master_repl_offset":          51,
					"mem_fragmentation_ratio":     52.58,
					"pubsub_channels":             53,
					"pubsub_patterns":             54,
					"rdb_changes_since_last_save": 55,
					"repl_backlog_active":         56,
					"repl_backlog_histlen":        57,
					"repl_backlog_size":           58,
					"sync_full":                   59,
					"sync_partial_err":            60,
					"sync_partial_ok":             61,
					"total_commands_processed":    62,
					"total_connections_received":  63,
					"uptime":                      64,
					"used_cpu_sys":                65.07,
					"used_cpu_sys_children":       66.0,
					"used_cpu_user":               67.1,
					"used_cpu_user_children":      68.0,
					"used_memory":                 692048,
					"used_memory_lua":             70792,
					"used_memory_peak":            711128,
					"used_memory_rss":             7298144,
				},
				time.Unix(123456789, 0),
			)

			m, ok := acc.Get("super_long_metric")
			require.True(t, ok)
			testutil.RequireMetricEqual(t, expected, testutil.FromTestMetric(m))

			hostTags := []string{"server02", "server03",
				"server04", "server05", "server06"}
			acc.Wait(len(hostTags))
			for _, hostTag := range hostTags {
				acc.AssertContainsTaggedFields(t, "cpu_load_short",
					map[string]interface{}{"value": float64(12)},
					map[string]string{"host": hostTag},
				)
			}
		})
	}
}

// test that writing gzipped data works
func TestWriteGzippedData(t *testing.T) {
	for _, tc := range parserTestCases {
		t.Run(fmt.Sprintf("parser %s", tc.parser), func(t *testing.T) {
			listener := newTestListener()
			listener.ParserType = tc.parser

			acc := &testutil.Accumulator{}
			require.NoError(t, listener.Init())
			require.NoError(t, listener.Start(acc))
			defer listener.Stop()

			data, err := os.ReadFile("./testdata/testmsgs.gz")
			require.NoError(t, err)

			req, err := http.NewRequest("POST", createURL(listener, "http", "/write", ""), bytes.NewBuffer(data))
			require.NoError(t, err)
			req.Header.Set("Content-Encoding", "gzip")

			client := &http.Client{}
			resp, err := client.Do(req)
			require.NoError(t, err)
			require.NoError(t, resp.Body.Close())
			require.EqualValues(t, 204, resp.StatusCode)

			hostTags := []string{"server02", "server03",
				"server04", "server05", "server06"}
			acc.Wait(len(hostTags))
			for _, hostTag := range hostTags {
				acc.AssertContainsTaggedFields(t, "cpu_load_short",
					map[string]interface{}{"value": float64(12)},
					map[string]string{"host": hostTag},
				)
			}
		})
	}
}

// writes 25,000 metrics to the listener with 10 different writers
func TestWriteHighTraffic(t *testing.T) {
	if runtime.GOOS == "darwin" {
		t.Skip("Skipping due to hang on darwin")
	}
	// resource intensive, large test
	if testing.Short() {
		t.Skip("Skipping long test in short mode")
	}
	listener := newTestListener()

	acc := &testutil.Accumulator{}
	require.NoError(t, listener.Init())
	require.NoError(t, listener.Start(acc))
	defer listener.Stop()

	// post many messages to listener
	var wg sync.WaitGroup
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func(innerwg *sync.WaitGroup) {
			defer innerwg.Done()
			for i := 0; i < 500; i++ {
				resp, err := http.Post(createURL(listener, "http", "/write", "db=mydb"), "", bytes.NewBuffer([]byte(testMsgs)))
				if err != nil {
					return
				}
				if err := resp.Body.Close(); err != nil {
					return
				}
				if resp.StatusCode != 204 {
					return
				}
			}
		}(&wg)
	}

	wg.Wait()
	require.NoError(t, listener.Gather(acc))

	acc.Wait(25000)
	require.Equal(t, int64(25000), int64(acc.NMetrics()))
}

func TestReceive404ForInvalidEndpoint(t *testing.T) {
	for _, tc := range parserTestCases {
		t.Run(fmt.Sprintf("parser %s", tc.parser), func(t *testing.T) {
			listener := newTestListener()
			listener.ParserType = tc.parser

			acc := &testutil.Accumulator{}
			require.NoError(t, listener.Init())
			require.NoError(t, listener.Start(acc))
			defer listener.Stop()

			// post single message to listener
			resp, err := http.Post(createURL(listener, "http", "/foobar", ""), "", bytes.NewBuffer([]byte(testMsg)))
			require.NoError(t, err)
			require.NoError(t, resp.Body.Close())
			require.EqualValues(t, 404, resp.StatusCode)
		})
	}
}

func TestWriteInvalid(t *testing.T) {
	for _, tc := range parserTestCases {
		t.Run(fmt.Sprintf("parser %s", tc.parser), func(t *testing.T) {
			listener := newTestListener()
			listener.ParserType = tc.parser

			acc := &testutil.Accumulator{}
			require.NoError(t, listener.Init())
			require.NoError(t, listener.Start(acc))
			defer listener.Stop()

			// post single message to listener
			resp, err := http.Post(createURL(listener, "http", "/write", "db=mydb"), "", bytes.NewBuffer([]byte(badMsg)))
			require.NoError(t, err)
			require.NoError(t, resp.Body.Close())
			require.EqualValues(t, 400, resp.StatusCode)
		})
	}
}

func TestWriteEmpty(t *testing.T) {
	for _, tc := range parserTestCases {
		t.Run(fmt.Sprintf("parser %s", tc.parser), func(t *testing.T) {
			listener := newTestListener()
			listener.ParserType = tc.parser

			acc := &testutil.Accumulator{}
			require.NoError(t, listener.Init())
			require.NoError(t, listener.Start(acc))
			defer listener.Stop()

			// post single message to listener
			resp, err := http.Post(createURL(listener, "http", "/write", "db=mydb"), "", bytes.NewBuffer([]byte(emptyMsg)))
			require.NoError(t, err)
			require.NoError(t, resp.Body.Close())
			require.EqualValues(t, 204, resp.StatusCode)
		})
	}
}

func TestQuery(t *testing.T) {
	for _, tc := range parserTestCases {
		t.Run(fmt.Sprintf("parser %s", tc.parser), func(t *testing.T) {
			listener := newTestListener()
			listener.ParserType = tc.parser

			acc := &testutil.Accumulator{}
			require.NoError(t, listener.Init())
			require.NoError(t, listener.Start(acc))
			defer listener.Stop()

			// post query to listener
			resp, err := http.Post(
				createURL(listener, "http", "/query", "db=&q=CREATE+DATABASE+IF+NOT+EXISTS+%22mydb%22"), "", nil)
			require.NoError(t, err)
			require.NoError(t, resp.Body.Close())
			require.EqualValues(t, 200, resp.StatusCode)
		})
	}
}

func TestPing(t *testing.T) {
	for _, tc := range parserTestCases {
		t.Run(fmt.Sprintf("parser %s", tc.parser), func(t *testing.T) {
			listener := newTestListener()
			listener.ParserType = tc.parser

			acc := &testutil.Accumulator{}
			require.NoError(t, listener.Init())
			require.NoError(t, listener.Start(acc))
			defer listener.Stop()

			// post ping to listener
			resp, err := http.Post(createURL(listener, "http", "/ping", ""), "", nil)
			require.NoError(t, err)
			require.Equal(t, "1.0", resp.Header["X-Influxdb-Version"][0])
			require.Empty(t, resp.Header["Content-Type"])
			require.NoError(t, resp.Body.Close())
			require.EqualValues(t, 204, resp.StatusCode)
		})
	}
}

func TestPingVerbose(t *testing.T) {
	for _, tc := range parserTestCases {
		t.Run(fmt.Sprintf("parser %s", tc.parser), func(t *testing.T) {
			listener := newTestListener()
			listener.ParserType = tc.parser

			acc := &testutil.Accumulator{}
			require.NoError(t, listener.Init())
			require.NoError(t, listener.Start(acc))
			defer listener.Stop()

			// post ping to listener
			resp, err := http.Post(createURL(listener, "http", "/ping", "verbose=1"), "", nil)
			require.NoError(t, err)
			require.Equal(t, "1.0", resp.Header["X-Influxdb-Version"][0])
			require.Equal(t, "application/json", resp.Header["Content-Type"][0])
			require.NoError(t, resp.Body.Close())
			require.EqualValues(t, 200, resp.StatusCode)
		})
	}
}

func TestWriteWithPrecision(t *testing.T) {
	for _, tc := range parserTestCases {
		t.Run(fmt.Sprintf("parser %s", tc.parser), func(t *testing.T) {
			listener := newTestListener()
			listener.ParserType = tc.parser

			acc := &testutil.Accumulator{}
			require.NoError(t, listener.Init())
			require.NoError(t, listener.Start(acc))
			defer listener.Stop()

			msg := "xyzzy value=42 1422568543\n"
			resp, err := http.Post(
				createURL(listener, "http", "/write", "precision=s"), "", bytes.NewBuffer([]byte(msg)))
			require.NoError(t, err)
			require.NoError(t, resp.Body.Close())
			require.EqualValues(t, 204, resp.StatusCode)

			acc.Wait(1)
			require.Len(t, acc.Metrics, 1)
			// When timestamp is provided, the precision parameter is
			// overloaded to specify the timestamp's unit
			require.Equal(t, time.Unix(0, 1422568543000000000), acc.Metrics[0].Time)
		})
	}
}

func TestWriteWithPrecisionNoTimestamp(t *testing.T) {
	for _, tc := range parserTestCases {
		t.Run(fmt.Sprintf("parser %s", tc.parser), func(t *testing.T) {
			listener := newTestListener()
			listener.ParserType = tc.parser
			listener.timeFunc = func() time.Time {
				return time.Unix(42, 123456789)
			}

			acc := &testutil.Accumulator{}
			require.NoError(t, listener.Init())
			require.NoError(t, listener.Start(acc))
			defer listener.Stop()

			msg := "xyzzy value=42\n"
			resp, err := http.Post(
				createURL(listener, "http", "/write", "precision=s"), "", bytes.NewBuffer([]byte(msg)))
			require.NoError(t, err)
			require.NoError(t, resp.Body.Close())
			require.EqualValues(t, 204, resp.StatusCode)

			acc.Wait(1)
			require.Len(t, acc.Metrics, 1)
			// When timestamp is omitted, the precision parameter actually
			// specifies the precision.  The timestamp is set to the greatest
			// integer unit less than the provided timestamp (floor).
			require.Equal(t, time.Unix(42, 0), acc.Metrics[0].Time)
		})
	}
}

func TestWriteUpstreamParseErrors(t *testing.T) {
	var tests = []struct {
		name     string
		input    string
		expected string
	}{
		{
			name:     "one parse error",
			input:    "foo value=1.0\nfoo value=2asdf2.0\nfoo value=3.0\nfoo value=4.0",
			expected: `metric parse error: cannot parse value for field key "value": invalid float value syntax at 2:11`,
		},
		{
			name:     "two parse errors",
			input:    "foo value=1asdf2.0\nfoo value=2.0\nfoo value=3asdf2.0\nfoo value=4.0",
			expected: `metric parse error: cannot parse value for field key "value": invalid float value syntax at 1:11 (and 1 other parse error)`,
		},
		{
			name:     "three or more parse errors",
			input:    "foo value=1asdf2.0\nfoo value=2.0\nfoo value=3asdf2.0\nfoo value=4asdf2.0",
			expected: `metric parse error: cannot parse value for field key "value": invalid float value syntax at 1:11 (and 2 other parse errors)`,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			listener := newTestListener()
			listener.ParserType = "upstream"

			acc := &testutil.NopAccumulator{}
			require.NoError(t, listener.Init())
			require.NoError(t, listener.Start(acc))
			defer listener.Stop()

			// post single message to listener
			resp, err := http.Post(createURL(listener, "http", "/write", ""), "", bytes.NewBuffer([]byte(tt.input)))
			require.NoError(t, err)
			require.NoError(t, resp.Body.Close())
			require.EqualValues(t, 400, resp.StatusCode)
			require.Equal(t, tt.expected, resp.Header["X-Influxdb-Error"][0])
		})
	}
}

func TestWriteParseErrors(t *testing.T) {
	var tests = []struct {
		name     string
		input    string
		expected string
	}{
		{
			name:     "one parse error",
			input:    "foo value=1.0\nfoo value=2asdf2.0\nfoo value=3.0\nfoo value=4.0",
			expected: `metric parse error: expected field at 2:12: "foo value=2"`,
		},
		{
			name:     "two parse errors",
			input:    "foo value=1asdf2.0\nfoo value=2.0\nfoo value=3asdf2.0\nfoo value=4.0",
			expected: `metric parse error: expected field at 1:12: "foo value=1" (and 1 other parse error)`,
		},
		{
			name:     "three or more parse errors",
			input:    "foo value=1asdf2.0\nfoo value=2.0\nfoo value=3asdf2.0\nfoo value=4asdf2.0",
			expected: `metric parse error: expected field at 1:12: "foo value=1" (and 2 other parse errors)`,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			listener := newTestListener()

			acc := &testutil.NopAccumulator{}
			require.NoError(t, listener.Init())
			require.NoError(t, listener.Start(acc))
			defer listener.Stop()

			// post single message to listener
			resp, err := http.Post(createURL(listener, "http", "/write", ""), "", bytes.NewBuffer([]byte(tt.input)))
			require.NoError(t, err)
			require.NoError(t, resp.Body.Close())
			require.EqualValues(t, 400, resp.StatusCode)
			require.Equal(t, tt.expected, resp.Header["X-Influxdb-Error"][0])
		})
	}
}
