diff --git a/Dockerfile b/Dockerfile
index 6ec58be..f8e9637 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -4,7 +4,8 @@ WORKDIR /tests
COPY . .
-RUN go test -c \
+RUN go test ./tests/integration/clients/... \
+ && go test -c -o quest.test ./tests/integration \
&& apt install wget \
&& wget https://github.com/grafana/k6/releases/download/v0.46.0/k6-v0.46.0-linux-amd64.deb \
&& apt install -f ./k6-v0.46.0-linux-amd64.deb \
@@ -12,6 +13,14 @@ RUN go test -c \
&& apt install -y jq \
&& wget https://github.com/mingrammer/flog/releases/download/v0.4.3/flog_0.4.3_linux_amd64.tar.gz \
&& tar -xvf flog_0.4.3_linux_amd64.tar.gz \
- && cp flog /usr/local/bin
+ && cp flog /usr/local/bin \
+ && pb_release_url=$(wget -qO- --server-response https://github.com/parseablehq/pb/releases/latest 2>&1 | awk '/^ Location: / { url=$2 } END { sub(/\r$/, "", url); print url }') \
+ && pb_version=${pb_release_url##*/v} \
+ && wget https://github.com/parseablehq/pb/releases/download/v${pb_version}/pb_${pb_version}_linux_amd64.tar.gz \
+ && wget https://github.com/parseablehq/pb/releases/download/v${pb_version}/pb_${pb_version}_checksums.txt \
+ && grep "pb_${pb_version}_linux_amd64.tar.gz" pb_${pb_version}_checksums.txt | sha256sum -c - \
+ && tar -xzf pb_${pb_version}_linux_amd64.tar.gz pb \
+ && install -m 0755 pb /usr/local/bin/pb \
+ && pb --help > /dev/null
ENTRYPOINT ["./main.sh"]
diff --git a/main.sh b/main.sh
index 76ed671..82811a0 100755
--- a/main.sh
+++ b/main.sh
@@ -39,9 +39,16 @@ ingestor_username=${13}
ingestor_password=${14}
stream_name=$(head /dev/urandom | tr -dc a-z | head -c10)
+configure_pb () {
+ export XDG_CONFIG_HOME="${XDG_CONFIG_HOME:-/tmp/quest-pb-config}"
+ pb profile add quest "$endpoint" "$username" "$password" -o json \
+ && pb profile default quest -o json
+}
+
run () {
- ./quest.test -test.v -mode="$mode" -query-url="$endpoint" -stream="$stream_name" -query-user="$username" -query-pass="$password" -minio-url="$minio_url" -minio-user="$minio_access_key" -minio-pass="$minio_secret_key" -minio-bucket="$minio_bucket" -ingestor-url="$ingestor_endpoint" -ingestor-user="$ingestor_username" -ingestor-pass="$ingestor_password"
+ ./quest.test -test.v -test.parallel=32 -mode="$mode" -query-url="$endpoint" -stream="$stream_name" -query-user="$username" -query-pass="$password" -minio-url="$minio_url" -minio-user="$minio_access_key" -minio-pass="$minio_secret_key" -minio-bucket="$minio_bucket" -ingestor-url="$ingestor_endpoint" -ingestor-user="$ingestor_username" -ingestor-pass="$ingestor_password"
return $?
}
+configure_pb || exit $?
run
diff --git a/quest_test.go b/quest_test.go
deleted file mode 100644
index 8ebf253..0000000
--- a/quest_test.go
+++ /dev/null
@@ -1,840 +0,0 @@
-// Copyright (c) 2023 Cloudnatively Services Pvt Ltd
-//
-//
-// This program is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Affero General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// This program is distributed in the hope that it will be useful
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Affero General Public License for more details.
-//
-// You should have received a copy of the GNU Affero General Public License
-// along with this program. If not, see .
-
-package main
-
-import (
- "bytes"
- "fmt"
- "io"
- "os/exec"
- "strings"
- "testing"
- "time"
-
- "github.com/stretchr/testify/require"
-)
-
-const (
- vus = "10"
- duration = "2m"
- schema_count = "10"
- events_count = "5"
-)
-
-func TestSmokeListLogStream(t *testing.T) {
- CreateStream(t, NewGlob.QueryClient, NewGlob.Stream)
- req, err := NewGlob.QueryClient.NewRequest("GET", "logstream", nil)
- require.NoErrorf(t, err, "Request failed: %s", err)
-
- response, err := NewGlob.QueryClient.Do(req)
- require.NoErrorf(t, err, "Request failed: %s", err)
-
- body := readAsString(response.Body)
- require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s and response: %s", response.Status)
- res, err := readJsonBody[[]string](bytes.NewBufferString(body))
- if err != nil {
- for _, stream := range res {
- if stream == NewGlob.Stream {
- DeleteStream(t, NewGlob.QueryClient, NewGlob.Stream)
- }
- }
- }
- DeleteStream(t, NewGlob.QueryClient, NewGlob.Stream)
-}
-
-func TestSmokeCreateStream(t *testing.T) {
- CreateStream(t, NewGlob.QueryClient, NewGlob.Stream)
- DeleteStream(t, NewGlob.QueryClient, NewGlob.Stream)
-}
-
-func TestSmokeDetectSchema(t *testing.T) {
- DetectSchema(t, NewGlob.QueryClient, SampleJson, SchemaBody)
-}
-
-func TestSmokeIngestEventsToStream(t *testing.T) {
- CreateStream(t, NewGlob.QueryClient, NewGlob.Stream)
- if NewGlob.IngestorUrl.String() == "" {
- RunFlog(t, NewGlob.QueryClient, NewGlob.Stream)
- } else {
- RunFlog(t, NewGlob.IngestorClient, NewGlob.Stream)
- }
- // Calling Sleep method
- time.Sleep(120 * time.Second)
-
- QueryLogStreamCount(t, NewGlob.QueryClient, NewGlob.Stream, 50)
- AssertStreamSchema(t, NewGlob.QueryClient, NewGlob.Stream, FlogJsonSchema)
- DeleteStream(t, NewGlob.QueryClient, NewGlob.Stream)
-}
-
-// func TestTimePartition_TimeStampMismatch(t *testing.T) {
-// historicalStream := NewGlob.Stream + "historical"
-// timeHeader := map[string]string{"X-P-Time-Partition": "source_time"}
-// CreateStreamWithHeader(t, NewGlob.QueryClient, historicalStream, timeHeader)
-// if NewGlob.IngestorUrl.String() == "" {
-// IngestOneEventWithTimePartition_TimeStampMismatch(t, NewGlob.QueryClient, historicalStream)
-// } else {
-// IngestOneEventWithTimePartition_TimeStampMismatch(t, NewGlob.IngestorClient, historicalStream)
-// }
-// DeleteStream(t, NewGlob.QueryClient, historicalStream)
-// }
-
-// func TestTimePartition_NoTimePartitionInLog(t *testing.T) {
-// historicalStream := NewGlob.Stream + "historical"
-// timeHeader := map[string]string{"X-P-Time-Partition": "source_time"}
-// CreateStreamWithHeader(t, NewGlob.QueryClient, historicalStream, timeHeader)
-// if NewGlob.IngestorUrl.String() == "" {
-// IngestOneEventWithTimePartition_NoTimePartitionInLog(t, NewGlob.QueryClient, historicalStream)
-// } else {
-// IngestOneEventWithTimePartition_NoTimePartitionInLog(t, NewGlob.IngestorClient, historicalStream)
-// }
-// DeleteStream(t, NewGlob.QueryClient, historicalStream)
-// }
-
-// func TestTimePartition_IncorrectDateTimeFormatTimePartitionInLog(t *testing.T) {
-// historicalStream := NewGlob.Stream + "historical"
-// timeHeader := map[string]string{"X-P-Time-Partition": "source_time"}
-// CreateStreamWithHeader(t, NewGlob.QueryClient, historicalStream, timeHeader)
-// if NewGlob.IngestorUrl.String() == "" {
-// IngestOneEventWithTimePartition_IncorrectDateTimeFormatTimePartitionInLog(t, NewGlob.QueryClient, historicalStream)
-// } else {
-// IngestOneEventWithTimePartition_IncorrectDateTimeFormatTimePartitionInLog(t, NewGlob.IngestorClient, historicalStream)
-// }
-// DeleteStream(t, NewGlob.QueryClient, historicalStream)
-// }
-
-func TestLoadStream_StaticSchema_EventWithSameFields(t *testing.T) {
- staticSchemaStream := NewGlob.Stream + "staticschema"
- staticSchemaFlagHeader := map[string]string{"X-P-Static-Schema-Flag": "true"}
- CreateStreamWithSchemaBody(t, NewGlob.QueryClient, staticSchemaStream, staticSchemaFlagHeader, SchemaPayload)
- if NewGlob.IngestorUrl.String() == "" {
- IngestOneEventForStaticSchemaStream_SameFieldsInLog(t, NewGlob.QueryClient, staticSchemaStream)
- } else {
- IngestOneEventForStaticSchemaStream_SameFieldsInLog(t, NewGlob.IngestorClient, staticSchemaStream)
- }
- DeleteStream(t, NewGlob.QueryClient, staticSchemaStream)
-}
-
-func TestLoadStreamBatchWithK6_StaticSchema(t *testing.T) {
- if NewGlob.Mode == "load" {
- staticSchemaStream := NewGlob.Stream + "staticschema"
- staticSchemaFlagHeader := map[string]string{"X-P-Static-Schema-Flag": "true"}
- CreateStreamWithSchemaBody(t, NewGlob.QueryClient, staticSchemaStream, staticSchemaFlagHeader, SchemaPayload)
- if NewGlob.IngestorUrl.String() == "" {
- cmd := exec.Command("k6",
- "run",
- "-e", fmt.Sprintf("P_URL=%s", &NewGlob.QueryUrl),
- "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
- "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
- "-e", fmt.Sprintf("P_STREAM=%s", staticSchemaStream),
- "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
- "-e", fmt.Sprintf("P_EVENTS_COUNT=%s", events_count),
- "./scripts/load_batch_events.js",
- "--vus=", vus,
- "--duration=", duration)
-
- cmd.Run()
- op, err := cmd.Output()
- if err != nil {
- t.Log(err)
- }
- t.Log(string(op))
- } else {
- cmd := exec.Command("k6",
- "run",
- "-e", fmt.Sprintf("P_URL=%s", &NewGlob.IngestorUrl),
- "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
- "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
- "-e", fmt.Sprintf("P_STREAM=%s", staticSchemaStream),
- "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
- "-e", fmt.Sprintf("P_EVENTS_COUNT=%s", events_count),
- "./scripts/load_batch_events.js",
- "--vus=", vus,
- "--duration=", duration)
-
- cmd.Run()
- op, err := cmd.Output()
- if err != nil {
- t.Log(err)
- }
- t.Log(string(op))
- }
-
- DeleteStream(t, NewGlob.QueryClient, staticSchemaStream)
- }
-}
-
-func TestLoadStream_StaticSchema_EventWithNewField(t *testing.T) {
- staticSchemaStream := NewGlob.Stream + "staticschema"
- staticSchemaFlagHeader := map[string]string{"X-P-Static-Schema-Flag": "true"}
- CreateStreamWithSchemaBody(t, NewGlob.QueryClient, staticSchemaStream, staticSchemaFlagHeader, SchemaPayload)
- if NewGlob.IngestorUrl.String() == "" {
- IngestOneEventForStaticSchemaStream_NewFieldInLog(t, NewGlob.QueryClient, staticSchemaStream)
- } else {
- IngestOneEventForStaticSchemaStream_NewFieldInLog(t, NewGlob.IngestorClient, staticSchemaStream)
- }
- DeleteStream(t, NewGlob.QueryClient, staticSchemaStream)
-}
-
-func TestCreateStream_WithCustomPartition_Success(t *testing.T) {
- customPartitionStream := NewGlob.Stream + "custompartition"
- customHeader := map[string]string{"X-P-Custom-Partition": "level"}
- CreateStreamWithHeader(t, NewGlob.QueryClient, customPartitionStream, customHeader)
- DeleteStream(t, NewGlob.QueryClient, customPartitionStream)
-}
-
-func TestCreateStream_WithCustomPartition_Error(t *testing.T) {
- customPartitionStream := NewGlob.Stream + "custompartition"
- customHeader := map[string]string{"X-P-Custom-Partition": "level,os"}
- CreateStreamWithCustompartitionError(t, NewGlob.QueryClient, customPartitionStream, customHeader)
-}
-
-func TestSmokeQueryTwoStreams(t *testing.T) {
- stream1 := NewGlob.Stream + "1"
- stream2 := NewGlob.Stream + "2"
- CreateStream(t, NewGlob.QueryClient, stream1)
- CreateStream(t, NewGlob.QueryClient, stream2)
- if NewGlob.IngestorUrl.String() == "" {
- RunFlog(t, NewGlob.QueryClient, stream1)
- RunFlog(t, NewGlob.QueryClient, stream2)
- } else {
- RunFlog(t, NewGlob.IngestorClient, stream1)
- RunFlog(t, NewGlob.IngestorClient, stream2)
-
- }
- time.Sleep(120 * time.Second)
- QueryTwoLogStreamCount(t, NewGlob.QueryClient, stream1, stream2, 100)
- DeleteStream(t, NewGlob.QueryClient, stream1)
- DeleteStream(t, NewGlob.QueryClient, stream2)
-}
-
-func TestSmokeRunQueries(t *testing.T) {
- CreateStream(t, NewGlob.QueryClient, NewGlob.Stream)
- if NewGlob.IngestorUrl.String() == "" {
- RunFlog(t, NewGlob.QueryClient, NewGlob.Stream)
- } else {
- RunFlog(t, NewGlob.IngestorClient, NewGlob.Stream)
- }
- time.Sleep(120 * time.Second)
- // test count
- QueryLogStreamCount(t, NewGlob.QueryClient, NewGlob.Stream, 50)
- // test yeild all values
- AssertQueryOK(t, NewGlob.QueryClient, "SELECT * FROM %s", NewGlob.Stream)
- AssertQueryOK(t, NewGlob.QueryClient, "SELECT * FROM %s OFFSET 25 LIMIT 25", NewGlob.Stream)
- // test fetch single column
- for _, item := range flogStreamFields() {
- AssertQueryOK(t, NewGlob.QueryClient, "SELECT %s FROM %s", item, NewGlob.Stream)
- }
- // test basic filter
- AssertQueryOK(t, NewGlob.QueryClient, "SELECT * FROM %s WHERE method = 'POST'", NewGlob.Stream)
- // test group by
- AssertQueryOK(t, NewGlob.QueryClient, "SELECT method, COUNT(*) FROM %s GROUP BY method", NewGlob.Stream)
- AssertQueryOK(t, NewGlob.QueryClient, `SELECT DATE_TRUNC('minute', p_timestamp) as minute, COUNT(*) FROM %s GROUP BY minute`, NewGlob.Stream)
-
- DeleteStream(t, NewGlob.QueryClient, NewGlob.Stream)
-}
-
-func TestSmokeLoadWithK6Stream(t *testing.T) {
- CreateStream(t, NewGlob.QueryClient, NewGlob.Stream)
- if NewGlob.IngestorUrl.String() == "" {
- cmd := exec.Command("k6",
- "run",
- "-e", fmt.Sprintf("P_URL=%s", NewGlob.QueryUrl.String()),
- "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
- "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
- "-e", fmt.Sprintf("P_STREAM=%s", NewGlob.Stream),
- "./scripts/smoke.js")
-
- cmd.Run()
- cmd.Output()
- } else {
- cmd := exec.Command("k6",
- "run",
- "-e", fmt.Sprintf("P_URL=%s", NewGlob.IngestorUrl.String()),
- "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
- "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
- "-e", fmt.Sprintf("P_STREAM=%s", NewGlob.Stream),
- "./scripts/smoke.js")
-
- cmd.Run()
- cmd.Output()
- }
- time.Sleep(150 * time.Second)
- QueryLogStreamCount(t, NewGlob.QueryClient, NewGlob.Stream, 20000)
- AssertStreamSchema(t, NewGlob.QueryClient, NewGlob.Stream, SchemaBody)
- DeleteStream(t, NewGlob.QueryClient, NewGlob.Stream)
-}
-
-// func TestSmokeLoad_TimePartition_WithK6Stream(t *testing.T) {
-// time_partition_stream := NewGlob.Stream + "timepartition"
-// timeHeader := map[string]string{"X-P-Time-Partition": "source_time", "X-P-Time-Partition-Limit": "365d"}
-// CreateStreamWithHeader(t, NewGlob.QueryClient, time_partition_stream, timeHeader)
-// if NewGlob.IngestorUrl.String() == "" {
-// cmd := exec.Command("k6",
-// "run",
-// "-e", fmt.Sprintf("P_URL=%s", NewGlob.QueryUrl.String()),
-// "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
-// "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
-// "-e", fmt.Sprintf("P_STREAM=%s", time_partition_stream),
-// "./scripts/smoke.js")
-
-// cmd.Run()
-// cmd.Output()
-// } else {
-// cmd := exec.Command("k6",
-// "run",
-// "-e", fmt.Sprintf("P_URL=%s", NewGlob.IngestorUrl.String()),
-// "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
-// "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
-// "-e", fmt.Sprintf("P_STREAM=%s", time_partition_stream),
-// "./scripts/smoke.js")
-
-// cmd.Run()
-// cmd.Output()
-// }
-// time.Sleep(120 * time.Second)
-// QueryLogStreamCount_Historical(t, NewGlob.QueryClient, time_partition_stream, 20000)
-// DeleteStream(t, NewGlob.QueryClient, time_partition_stream)
-// }
-
-func TestSmokeLoad_CustomPartition_WithK6Stream(t *testing.T) {
- custom_partition_stream := NewGlob.Stream + "custompartition"
- customHeader := map[string]string{"X-P-Custom-Partition": "level"}
- CreateStreamWithHeader(t, NewGlob.QueryClient, custom_partition_stream, customHeader)
- if NewGlob.IngestorUrl.String() == "" {
- cmd := exec.Command("k6",
- "run",
- "-e", fmt.Sprintf("P_URL=%s", NewGlob.QueryUrl.String()),
- "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
- "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
- "-e", fmt.Sprintf("P_STREAM=%s", custom_partition_stream),
- "./scripts/smoke.js")
-
- cmd.Run()
- cmd.Output()
- } else {
- cmd := exec.Command("k6",
- "run",
- "-e", fmt.Sprintf("P_URL=%s", NewGlob.IngestorUrl.String()),
- "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
- "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
- "-e", fmt.Sprintf("P_STREAM=%s", custom_partition_stream),
- "./scripts/smoke.js")
-
- cmd.Run()
- cmd.Output()
- }
- time.Sleep(120 * time.Second)
- QueryLogStreamCount(t, NewGlob.QueryClient, custom_partition_stream, 20000)
- DeleteStream(t, NewGlob.QueryClient, custom_partition_stream)
-}
-
-// func TestSmokeLoad_TimeAndCustomPartition_WithK6Stream(t *testing.T) {
-// custom_partition_stream := NewGlob.Stream + "timecustompartition"
-// customHeader := map[string]string{"X-P-Custom-Partition": "level", "X-P-Time-Partition": "source_time", "X-P-Time-Partition-Limit": "365d"}
-// CreateStreamWithHeader(t, NewGlob.QueryClient, custom_partition_stream, customHeader)
-// if NewGlob.IngestorUrl.String() == "" {
-// cmd := exec.Command("k6",
-// "run",
-// "-e", fmt.Sprintf("P_URL=%s", NewGlob.QueryUrl.String()),
-// "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
-// "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
-// "-e", fmt.Sprintf("P_STREAM=%s", custom_partition_stream),
-// "./scripts/smoke.js")
-
-// cmd.Run()
-// cmd.Output()
-// } else {
-// cmd := exec.Command("k6",
-// "run",
-// "-e", fmt.Sprintf("P_URL=%s", NewGlob.IngestorUrl.String()),
-// "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
-// "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
-// "-e", fmt.Sprintf("P_STREAM=%s", custom_partition_stream),
-// "./scripts/smoke.js")
-
-// cmd.Run()
-// cmd.Output()
-// }
-// time.Sleep(180 * time.Second)
-// QueryLogStreamCount_Historical(t, NewGlob.QueryClient, custom_partition_stream, 20000)
-// DeleteStream(t, NewGlob.QueryClient, custom_partition_stream)
-// }
-
-func TestSmokeSetTarget(t *testing.T) {
- body := getTargetBody()
- req, _ := NewGlob.QueryClient.NewRequest("POST", "/targets", strings.NewReader(body))
- response, err := NewGlob.QueryClient.Do(req)
- require.NoErrorf(t, err, "Request failed: %s", err)
- require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s and response: %s", response.Status, readAsString(response.Body))
-}
-
-func TestSmokeSetAlert(t *testing.T) {
- stream := NewGlob.Stream + "alert_testing"
- CreateStream(t, NewGlob.QueryClient, stream)
- if NewGlob.IngestorUrl.String() == "" {
- cmd := exec.Command("k6",
- "run",
- "-e", fmt.Sprintf("P_URL=%s", NewGlob.QueryUrl.String()),
- "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
- "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
- "-e", fmt.Sprintf("P_STREAM=%s", stream),
- "./scripts/smoke.js")
-
- cmd.Run()
- cmd.Output()
- } else {
- cmd := exec.Command("k6",
- "run",
- "-e", fmt.Sprintf("P_URL=%s", NewGlob.IngestorUrl.String()),
- "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
- "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
- "-e", fmt.Sprintf("P_STREAM=%s", stream),
- "./scripts/smoke.js")
-
- cmd.Run()
- cmd.Output()
- }
- time.Sleep(120 * time.Second)
- req, _ := NewGlob.QueryClient.NewRequest("GET", "/targets", nil)
- response, err := NewGlob.QueryClient.Do(req)
- require.NoErrorf(t, err, "Request failed: %s", err)
- bodyTargets, _ := io.ReadAll(response.Body)
- reader1 := bytes.NewReader(bodyTargets)
- targetId := getIdFromTargetResponse(reader1)
- body := getAlertBody(stream, targetId)
- req, _ = NewGlob.QueryClient.NewRequest("POST", "/alerts", strings.NewReader(body))
- response, err = NewGlob.QueryClient.Do(req)
- require.NoErrorf(t, err, "Request failed: %s", err)
- require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s and response: %s", response.Status, readAsString(response.Body))
-}
-
-func TestSmokeGetAlert(t *testing.T) {
- stream := NewGlob.Stream + "alert_testing"
- req, _ := NewGlob.QueryClient.NewRequest("GET", "/targets", nil)
- response, err := NewGlob.QueryClient.Do(req)
- require.NoErrorf(t, err, "Request failed: %s", err)
- bodyTargets, _ := io.ReadAll(response.Body)
- reader1 := bytes.NewReader(bodyTargets)
- targetId := getIdFromTargetResponse(reader1)
- req, _ = NewGlob.QueryClient.NewRequest("GET", "/alerts", nil)
- response, err = NewGlob.QueryClient.Do(req)
- require.NoErrorf(t, err, "Request failed: %s", err)
- body, _ := io.ReadAll(response.Body)
- reader1 = bytes.NewReader(body)
- reader2 := bytes.NewReader(body)
- expected := readAsString(reader1)
- id, state, created, datasets := getMetadataFromAlertResponse(reader2)
- require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s and response: %s", response.Status, body)
- res := createAlertResponse(id, state, created, datasets)
- require.JSONEq(t, expected, res, "Get alert response doesn't match with Alert config returned")
- DeleteAlert(t, NewGlob.QueryClient, id)
- DeleteTarget(t, NewGlob.QueryClient, targetId)
- DeleteStream(t, NewGlob.QueryClient, stream)
-}
-
-func TestSmokeSetRetention(t *testing.T) {
- CreateStream(t, NewGlob.QueryClient, NewGlob.Stream)
- req, _ := NewGlob.QueryClient.NewRequest("PUT", "logstream/"+NewGlob.Stream+"/retention", strings.NewReader(RetentionBody))
- response, err := NewGlob.QueryClient.Do(req)
- require.NoErrorf(t, err, "Request failed: %s", err)
- require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s and response: %s", response.Status, readAsString(response.Body))
-}
-
-func TestSmokeGetRetention(t *testing.T) {
- req, _ := NewGlob.QueryClient.NewRequest("GET", "logstream/"+NewGlob.Stream+"/retention", nil)
- response, err := NewGlob.QueryClient.Do(req)
- require.NoErrorf(t, err, "Request failed: %s", err)
- body := readAsString(response.Body)
- require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s and response: %s", response.Status, body)
- require.JSONEq(t, RetentionBody, body, "Get retention response doesn't match with retention config returned")
- DeleteStream(t, NewGlob.QueryClient, NewGlob.Stream)
-}
-
-// This test calls all the User API endpoints
-// in a sequence to check if they work as expected.
-func TestSmoke_AllUsersAPI(t *testing.T) {
- CreateRole(t, NewGlob.QueryClient, "dummyrole", dummyRole)
- AssertRole(t, NewGlob.QueryClient, "dummyrole", dummyRole)
-
- CreateUser(t, NewGlob.QueryClient, "dummyuser")
- CreateUserWithRole(t, NewGlob.QueryClient, "dummyanotheruser", []string{"dummyrole"})
- AssertUserRole(t, NewGlob.QueryClient, "dummyanotheruser", "dummyrole", dummyRole)
- RegenPassword(t, NewGlob.QueryClient, "dummyuser")
- DeleteUser(t, NewGlob.QueryClient, "dummyuser")
- DeleteUser(t, NewGlob.QueryClient, "dummyanotheruser")
- DeleteRole(t, NewGlob.QueryClient, "dummyrole")
-}
-
-// This test checks that a new user doesn't get any role by default
-// even if a default role is set.
-func TestSmoke_NewUserNoRole(t *testing.T) {
- CreateStream(t, NewGlob.QueryClient, NewGlob.Stream)
-
- CreateRole(t, NewGlob.QueryClient, "dummyrole", dummyRole)
- SetDefaultRole(t, NewGlob.QueryClient, "dummyrole")
- AssertDefaultRole(t, NewGlob.QueryClient, "\"dummyrole\"")
-
- CreateUser(t, NewGlob.QueryClient, "dummyuser")
- DeleteStream(t, NewGlob.QueryClient, NewGlob.Stream)
-}
-
-func TestSmokeRbacBasic(t *testing.T) {
- CreateStream(t, NewGlob.QueryClient, NewGlob.Stream)
- CreateRole(t, NewGlob.QueryClient, "dummy", dummyRole)
- AssertRole(t, NewGlob.QueryClient, "dummy", dummyRole)
- CreateUserWithRole(t, NewGlob.QueryClient, "dummy", []string{"dummy"})
- userClient := NewGlob.QueryClient
- userClient.Username = "dummy"
- userClient.Password = RegenPassword(t, NewGlob.QueryClient, "dummy")
- checkAPIAccess(t, userClient, NewGlob.QueryClient, NewGlob.Stream, "editor")
- DeleteUser(t, NewGlob.QueryClient, "dummy")
- DeleteRole(t, NewGlob.QueryClient, "dummy")
-}
-
-func TestSmokeRoles(t *testing.T) {
- CreateStream(t, NewGlob.QueryClient, NewGlob.Stream)
- cases := []struct {
- roleName string
- body string
- }{
- {
- roleName: "ingestor",
- body: Roleingestor(NewGlob.Stream),
- },
- {
- roleName: "reader",
- body: RoleReader(NewGlob.Stream),
- },
- {
- roleName: "writer",
- body: RoleWriter(NewGlob.Stream),
- },
- {
- roleName: "editor",
- body: RoleEditor,
- },
- }
-
- for _, tc := range cases {
- t.Run(tc.roleName, func(t *testing.T) {
- CreateRole(t, NewGlob.QueryClient, tc.roleName, tc.body)
- AssertRole(t, NewGlob.QueryClient, tc.roleName, tc.body)
- username := tc.roleName + "_user"
- password := CreateUserWithRole(t, NewGlob.QueryClient, username, []string{tc.roleName})
- var ingestClient HTTPClient
- queryClient := NewGlob.QueryClient
- queryClient.Username = username
- queryClient.Password = password
- if NewGlob.IngestorUrl.String() != "" {
- ingestClient = NewGlob.IngestorClient
- ingestClient.Username = username
- ingestClient.Password = password
- } else {
- ingestClient = NewGlob.QueryClient
- ingestClient.Username = username
- ingestClient.Password = password
- }
-
- checkAPIAccess(t, queryClient, ingestClient, NewGlob.Stream, tc.roleName)
- DeleteUser(t, NewGlob.QueryClient, username)
- DeleteRole(t, NewGlob.QueryClient, tc.roleName)
- })
- }
-}
-
-func TestLoadStreamBatchWithK6(t *testing.T) {
- if NewGlob.Mode == "load" {
- CreateStream(t, NewGlob.QueryClient, NewGlob.Stream)
- if NewGlob.IngestorUrl.String() == "" {
- cmd := exec.Command("k6",
- "run",
- "-e", fmt.Sprintf("P_URL=%s", NewGlob.QueryUrl.String()),
- "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
- "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
- "-e", fmt.Sprintf("P_STREAM=%s", NewGlob.Stream),
- "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
- "-e", fmt.Sprintf("P_EVENTS_COUNT=%s", events_count),
- "./scripts/load_batch_events.js",
- "--vus=", vus,
- "--duration=", duration)
-
- cmd.Run()
- op, err := cmd.Output()
- if err != nil {
- t.Log(err)
- }
- t.Log(string(op))
- } else {
- cmd := exec.Command("k6",
- "run",
- "-e", fmt.Sprintf("P_URL=%s", NewGlob.IngestorUrl.String()),
- "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
- "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
- "-e", fmt.Sprintf("P_STREAM=%s", NewGlob.Stream),
- "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
- "-e", fmt.Sprintf("P_EVENTS_COUNT=%s", events_count),
- "./scripts/load_batch_events.js",
- "--vus=", vus,
- "--duration=", duration)
-
- cmd.Run()
- op, err := cmd.Output()
- if err != nil {
- t.Log(err)
- }
- t.Log(string(op))
- }
- DeleteStream(t, NewGlob.QueryClient, NewGlob.Stream)
-
- }
-}
-
-// func TestLoadHistoricalStreamBatchWithK6(t *testing.T) {
-// if NewGlob.Mode == "load" {
-// historicalStream := NewGlob.Stream + "historical"
-// timeHeader := map[string]string{"X-P-Time-Partition": "source_time"}
-// CreateStreamWithHeader(t, NewGlob.QueryClient, historicalStream, timeHeader)
-// if NewGlob.IngestorUrl.String() == "" {
-// cmd := exec.Command("k6",
-// "run",
-// "-e", fmt.Sprintf("P_URL=%s", NewGlob.QueryUrl.String()),
-// "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
-// "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
-// "-e", fmt.Sprintf("P_STREAM=%s", historicalStream),
-// "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
-// "-e", fmt.Sprintf("P_EVENTS_COUNT=%s", events_count),
-// "./scripts/load_historical_batch_events.js",
-// "--vus=", vus,
-// "--duration=", duration)
-
-// cmd.Run()
-// op, err := cmd.Output()
-// if err != nil {
-// t.Log(err)
-// }
-// t.Log(string(op))
-// } else {
-// cmd := exec.Command("k6",
-// "run",
-// "-e", fmt.Sprintf("P_URL=%s", NewGlob.IngestorUrl.String()),
-// "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
-// "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
-// "-e", fmt.Sprintf("P_STREAM=%s", historicalStream),
-// "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
-// "-e", fmt.Sprintf("P_EVENTS_COUNT=%s", events_count),
-// "./scripts/load_historical_batch_events.js",
-// "--vus=", vus,
-// "--duration=", duration)
-
-// cmd.Run()
-// op, err := cmd.Output()
-// if err != nil {
-// t.Log(err)
-// }
-// t.Log(string(op))
-// }
-
-// DeleteStream(t, NewGlob.QueryClient, historicalStream)
-// }
-// }
-
-func TestLoadStreamBatchWithCustomPartitionWithK6(t *testing.T) {
- customPartitionStream := NewGlob.Stream + "custompartition"
- customHeader := map[string]string{"X-P-Custom-Partition": "level"}
- CreateStreamWithHeader(t, NewGlob.QueryClient, customPartitionStream, customHeader)
- if NewGlob.IngestorUrl.String() == "" {
- cmd := exec.Command("k6",
- "run",
- "-e", fmt.Sprintf("P_URL=%s", NewGlob.QueryUrl.String()),
- "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
- "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
- "-e", fmt.Sprintf("P_STREAM=%s", customPartitionStream),
- "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
- "-e", fmt.Sprintf("P_EVENTS_COUNT=%s", events_count),
- "./scripts/load_batch_events.js",
- "--vus=", vus,
- "--duration=", duration)
-
- cmd.Run()
- op, err := cmd.Output()
- if err != nil {
- t.Log(err)
- }
- t.Log(string(op))
- } else {
- cmd := exec.Command("k6",
- "run",
- "-e", fmt.Sprintf("P_URL=%s", NewGlob.IngestorUrl.String()),
- "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
- "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
- "-e", fmt.Sprintf("P_STREAM=%s", customPartitionStream),
- "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
- "-e", fmt.Sprintf("P_EVENTS_COUNT=%s", events_count),
- "./scripts/load_batch_events.js",
- "--vus=", vus,
- "--duration=", duration)
-
- cmd.Run()
- op, err := cmd.Output()
- if err != nil {
- t.Log(err)
- }
- t.Log(string(op))
- }
-
- DeleteStream(t, NewGlob.QueryClient, customPartitionStream)
-}
-
-func TestLoadStreamNoBatchWithK6(t *testing.T) {
- if NewGlob.Mode == "load" {
- CreateStream(t, NewGlob.QueryClient, NewGlob.Stream)
- if NewGlob.IngestorUrl.String() == "" {
- cmd := exec.Command("k6",
- "run",
- "-e", fmt.Sprintf("P_URL=%s", NewGlob.QueryUrl.String()),
- "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
- "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
- "-e", fmt.Sprintf("P_STREAM=%s", NewGlob.Stream),
- "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
- "./scripts/load_single_events.js",
- "--vus=", vus,
- "--duration=", duration)
-
- cmd.Run()
- op, err := cmd.Output()
- if err != nil {
- t.Log(err)
- }
- t.Log(string(op))
- } else {
- cmd := exec.Command("k6",
- "run",
- "-e", fmt.Sprintf("P_URL=%s", NewGlob.IngestorUrl.String()),
- "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
- "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
- "-e", fmt.Sprintf("P_STREAM=%s", NewGlob.Stream),
- "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
- "./scripts/load_single_events.js",
- "--vus=", vus,
- "--duration=", duration)
-
- cmd.Run()
- op, err := cmd.Output()
- if err != nil {
- t.Log(err)
- }
- t.Log(string(op))
- }
-
- }
-}
-
-// func TestLoadHistoricalStreamNoBatchWithK6(t *testing.T) {
-// if NewGlob.Mode == "load" {
-// historicalStream := NewGlob.Stream + "historical"
-// timeHeader := map[string]string{"X-P-Time-Partition": "source_time"}
-// CreateStreamWithHeader(t, NewGlob.QueryClient, historicalStream, timeHeader)
-// if NewGlob.IngestorUrl.String() == "" {
-// cmd := exec.Command("k6",
-// "run",
-// "-e", fmt.Sprintf("P_URL=%s", NewGlob.QueryUrl.String()),
-// "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
-// "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
-// "-e", fmt.Sprintf("P_STREAM=%s", historicalStream),
-// "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
-// "./scripts/load_single_events.js",
-// "--vus=", vus,
-// "--duration=", duration)
-
-// cmd.Run()
-// op, err := cmd.Output()
-// if err != nil {
-// t.Log(err)
-// }
-// t.Log(string(op))
-// } else {
-// cmd := exec.Command("k6",
-// "run",
-// "-e", fmt.Sprintf("P_URL=%s", NewGlob.IngestorUrl.String()),
-// "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
-// "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
-// "-e", fmt.Sprintf("P_STREAM=%s", historicalStream),
-// "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
-// "./scripts/load_single_events.js",
-// "--vus=", vus,
-// "--duration=", duration)
-
-// cmd.Run()
-// op, err := cmd.Output()
-// if err != nil {
-// t.Log(err)
-// }
-// t.Log(string(op))
-// }
-
-// DeleteStream(t, NewGlob.QueryClient, historicalStream)
-// }
-// }
-
-func TestLoadStreamNoBatchWithCustomPartitionWithK6(t *testing.T) {
- customPartitionStream := NewGlob.Stream + "custompartition"
- customHeader := map[string]string{"X-P-Custom-Partition": "level"}
- CreateStreamWithHeader(t, NewGlob.QueryClient, customPartitionStream, customHeader)
- if NewGlob.IngestorUrl.String() == "" {
- cmd := exec.Command("k6",
- "run",
- "-e", fmt.Sprintf("P_URL=%s", NewGlob.QueryUrl.String()),
- "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
- "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
- "-e", fmt.Sprintf("P_STREAM=%s", customPartitionStream),
- "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
- "./scripts/load_single_events.js",
- "--vus=", vus,
- "--duration=", duration)
-
- cmd.Run()
- op, err := cmd.Output()
- if err != nil {
- t.Log(err)
- }
- t.Log(string(op))
- } else {
- cmd := exec.Command("k6",
- "run",
- "-e", fmt.Sprintf("P_URL=%s", NewGlob.IngestorUrl.String()),
- "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
- "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
- "-e", fmt.Sprintf("P_STREAM=%s", customPartitionStream),
- "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
- "./scripts/load_single_events.js",
- "--vus=", vus,
- "--duration=", duration)
-
- cmd.Run()
- op, err := cmd.Output()
- if err != nil {
- t.Log(err)
- }
- t.Log(string(op))
- }
-
- DeleteStream(t, NewGlob.QueryClient, customPartitionStream)
-}
-
-func TestDeleteStream(t *testing.T) {
- DeleteStream(t, NewGlob.QueryClient, NewGlob.Stream)
-}
diff --git a/scripts/load_single_event.js b/scripts/load_single_event.js
index 54b7b93..9a468f1 100644
--- a/scripts/load_single_event.js
+++ b/scripts/load_single_event.js
@@ -1,5 +1,6 @@
import http from 'k6/http';
import { check, sleep } from 'k6';
+import exec from 'k6/execution';
import encoding from 'k6/encoding';
import { randomString, randomItem, randomIntBetween, uuidv4 } from 'https://jslib.k6.io/k6-utils/1.4.0/index.js'
@@ -167,5 +168,11 @@ export default function () {
}
let batch_requests = generateEvents(1).map(event => ['POST', url, event, params]);
- http.batch(batch_requests);
-}
\ No newline at end of file
+ let responses = http.batch(batch_requests);
+
+ if (!check(responses, {
+ 'status code MUST be 200': (responses) => responses.every(response => response.status == 200),
+ })) {
+ exec.test.abort("Failed to send event.. status != 200");
+ }
+}
diff --git a/tests/integration/alert_test.go b/tests/integration/alert_test.go
new file mode 100644
index 0000000..f18ab0d
--- /dev/null
+++ b/tests/integration/alert_test.go
@@ -0,0 +1,150 @@
+// Copyright (c) 2023 Cloudnatively Services Pvt Ltd
+//
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see .
+
+package main
+
+import (
+ "fmt"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+)
+
+type testTargetResponse struct {
+ Target struct {
+ ID string `json:"id"`
+ Name string `json:"name"`
+ } `json:"target"`
+}
+
+type testAlertResponse struct {
+ Severity string `json:"severity"`
+ Title string `json:"title"`
+ ID string `json:"id"`
+ State string `json:"state"`
+ AlertType string `json:"alertType"`
+ Tags []string `json:"tags"`
+ Created string `json:"created"`
+ Datasets []string `json:"datasets"`
+}
+
+func ingestAlertFixture(t *testing.T, stream string) {
+ t.Helper()
+ client := NewGlob.QueryClient
+ if NewGlob.IngestorUrl.String() != "" {
+ client = NewGlob.IngestorClient
+ }
+ req, _ := client.NewRequest("POST", "ingest", strings.NewReader(`[{"level":"info"}]`))
+ req.Header.Add("X-P-Stream", stream)
+ response, err := client.Do(req)
+ require.NoErrorf(t, err, "Request failed: %s", err)
+ require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s and response: %s", response.Status, readAsString(response.Body))
+}
+
+func createTestTarget(t *testing.T, name string) string {
+ t.Helper()
+ body := fmt.Sprintf(`{
+ "name": %q,
+ "type": "webhook",
+ "endpoint": "https://webhook.site/ec627445-d52b-44e9-948d-56671df3581e",
+ "headers": {},
+ "skipTlsCheck": false
+ }`, name)
+ req, _ := NewGlob.QueryClient.NewRequest("POST", "/targets", strings.NewReader(body))
+ response, err := NewGlob.QueryClient.Do(req)
+ require.NoErrorf(t, err, "Request failed: %s", err)
+ require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s and response: %s", response.Status, readAsString(response.Body))
+
+ req, _ = NewGlob.QueryClient.NewRequest("GET", "/targets", nil)
+ response, err = NewGlob.QueryClient.Do(req)
+ require.NoErrorf(t, err, "Request failed: %s", err)
+ targets, err := readJsonBody[[]testTargetResponse](response.Body)
+ require.NoError(t, err)
+ for _, target := range targets {
+ if target.Target.Name == name {
+ return target.Target.ID
+ }
+ }
+ t.Fatalf("target %q was not returned by GET /targets", name)
+ return ""
+}
+
+func createTestAlert(t *testing.T, stream, targetID, title string) string {
+ t.Helper()
+ body := strings.Replace(getAlertBody(stream, targetID), `"title": "AlertTitle"`, fmt.Sprintf(`"title": %q`, title), 1)
+ req, _ := NewGlob.QueryClient.NewRequest("POST", "/alerts", strings.NewReader(body))
+ response, err := NewGlob.QueryClient.Do(req)
+ require.NoErrorf(t, err, "Request failed: %s", err)
+ require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s and response: %s", response.Status, readAsString(response.Body))
+
+ alert := getTestAlert(t, title)
+ return alert.ID
+}
+
+func getTestAlert(t *testing.T, title string) testAlertResponse {
+ t.Helper()
+ req, _ := NewGlob.QueryClient.NewRequest("GET", "/alerts", nil)
+ response, err := NewGlob.QueryClient.Do(req)
+ require.NoErrorf(t, err, "Request failed: %s", err)
+ require.Equal(t, 200, response.StatusCode)
+ alerts, err := readJsonBody[[]testAlertResponse](response.Body)
+ require.NoError(t, err)
+ for _, alert := range alerts {
+ if alert.Title == title {
+ return alert
+ }
+ }
+ t.Fatalf("alert %q was not returned by GET /alerts", title)
+ return testAlertResponse{}
+}
+
+func TestSmokeSetTarget(t *testing.T) {
+ // Verifies that a webhook target can be created.
+ t.Parallel()
+ targetID := createTestTarget(t, NewGlob.Stream+"settarget")
+ t.Cleanup(func() {
+ DeleteTarget(t, NewGlob.QueryClient, targetID)
+ })
+}
+
+func TestSmokeAlertLifecycle(t *testing.T) {
+ // Verifies that an alert can be created and returns the expected details.
+ t.Parallel()
+ stream := NewGlob.Stream + "alert"
+ title := NewGlob.Stream + "alerttitle"
+ CreateStream(t, NewGlob.PBClient, stream)
+ ingestAlertFixture(t, stream)
+ time.Sleep(120 * time.Second)
+ targetID := createTestTarget(t, NewGlob.Stream+"alerttarget")
+ t.Cleanup(func() {
+ DeleteTarget(t, NewGlob.QueryClient, targetID)
+ })
+ alertID := createTestAlert(t, stream, targetID, title)
+ t.Cleanup(func() {
+ DeleteAlert(t, NewGlob.QueryClient, alertID)
+ })
+
+ alert := getTestAlert(t, title)
+ require.Equal(t, alertID, alert.ID)
+ require.Equal(t, title, alert.Title)
+ require.Equal(t, "threshold", alert.AlertType)
+ require.Equal(t, "Medium", alert.Severity)
+ require.Equal(t, []string{stream}, alert.Datasets)
+ require.NotEmpty(t, alert.State)
+ require.NotEmpty(t, alert.Created)
+}
diff --git a/client.go b/tests/integration/clients/http/client.go
similarity index 98%
rename from client.go
rename to tests/integration/clients/http/client.go
index 81de796..7f90f7a 100644
--- a/client.go
+++ b/tests/integration/clients/http/client.go
@@ -14,7 +14,7 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see .
-package main
+package httpclient
import (
"io"
diff --git a/tests/integration/clients/pb/client.go b/tests/integration/clients/pb/client.go
new file mode 100644
index 0000000..541327b
--- /dev/null
+++ b/tests/integration/clients/pb/client.go
@@ -0,0 +1,118 @@
+// Copyright (c) 2023 Cloudnatively Services Pvt Ltd
+//
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see .
+
+package pb
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os/exec"
+ "strings"
+ "time"
+)
+
+const defaultPBTimeout = 60 * time.Second
+
+type PBClient struct {
+ Binary string
+ Timeout time.Duration
+}
+
+type PBResult struct {
+ Stdout string
+ Stderr string
+ ExitCode int
+ Duration time.Duration
+}
+
+func DefaultPBClient(binary string) PBClient {
+ return PBClient{
+ Binary: binary,
+ Timeout: defaultPBTimeout,
+ }
+}
+
+func (client PBClient) Run(ctx context.Context, args ...string) (PBResult, error) {
+ if client.Binary == "" {
+ client.Binary = "pb"
+ }
+
+ if client.Timeout > 0 {
+ var cancel context.CancelFunc
+ ctx, cancel = context.WithTimeout(ctx, client.Timeout)
+ defer cancel()
+ }
+
+ var stdout bytes.Buffer
+ var stderr bytes.Buffer
+ cmd := exec.CommandContext(ctx, client.Binary, args...)
+ cmd.Stdout = &stdout
+ cmd.Stderr = &stderr
+
+ startedAt := time.Now()
+ err := cmd.Run()
+ result := PBResult{
+ Stdout: stdout.String(),
+ Stderr: stderr.String(),
+ ExitCode: 0,
+ Duration: time.Since(startedAt),
+ }
+
+ if err == nil {
+ return result, nil
+ }
+
+ result.ExitCode = -1
+ var exitError *exec.ExitError
+ if errors.As(err, &exitError) {
+ result.ExitCode = exitError.ExitCode()
+ }
+
+ if ctx.Err() != nil {
+ return result, fmt.Errorf("pb command timed out: %w", ctx.Err())
+ }
+
+ return result, err
+}
+
+func (client PBClient) RunJSON(ctx context.Context, output any, args ...string) (PBResult, error) {
+ jsonArgs := append(append([]string{}, args...), "-o", "json")
+ result, err := client.Run(ctx, jsonArgs...)
+ if err != nil {
+ return result, err
+ }
+
+ if err := json.Unmarshal([]byte(result.Stdout), output); err != nil {
+ return result, fmt.Errorf("decode pb JSON output: %w", err)
+ }
+
+ return result, nil
+}
+
+func PasswordFromUserAddOutput(output string) (string, error) {
+ for _, line := range strings.Split(output, "\n") {
+ if password, found := strings.CutPrefix(strings.TrimSpace(line), "Password is:"); found {
+ password = strings.TrimSpace(password)
+ if password != "" {
+ return password, nil
+ }
+ }
+ }
+ return "", fmt.Errorf("password not found in pb output")
+}
diff --git a/tests/integration/clients/pb/client_test.go b/tests/integration/clients/pb/client_test.go
new file mode 100644
index 0000000..647b3a4
--- /dev/null
+++ b/tests/integration/clients/pb/client_test.go
@@ -0,0 +1,142 @@
+// Copyright (c) 2023 Cloudnatively Services Pvt Ltd
+//
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see .
+
+package pb
+
+import (
+ "context"
+ "fmt"
+ "os"
+ "strings"
+ "testing"
+ "time"
+)
+
+func TestPBClientRun(t *testing.T) {
+ // Verifies PB command output, errors, JSON, and timeouts.
+ t.Setenv("QUEST_PB_HELPER_PROCESS", "1")
+
+ client := PBClient{
+ Binary: os.Args[0],
+ Timeout: time.Second,
+ }
+ command := []string{"-test.run=TestPBClientHelperProcess", "--"}
+
+ t.Run("captures output", func(t *testing.T) {
+ result, err := client.Run(context.Background(), append(command, "success")...)
+ if err != nil {
+ t.Fatalf("run command: %v", err)
+ }
+ if result.ExitCode != 0 {
+ t.Fatalf("expected exit code 0, got %d", result.ExitCode)
+ }
+ if result.Stdout != `{"name":"pstats"}` {
+ t.Fatalf("unexpected stdout: %q", result.Stdout)
+ }
+ if result.Stderr != "warning" {
+ t.Fatalf("unexpected stderr: %q", result.Stderr)
+ }
+ })
+
+ t.Run("returns exit code", func(t *testing.T) {
+ result, err := client.Run(context.Background(), append(command, "failure")...)
+ if err == nil {
+ t.Fatal("expected command to fail")
+ }
+ if result.ExitCode != 7 {
+ t.Fatalf("expected exit code 7, got %d", result.ExitCode)
+ }
+ if result.Stderr != "failed" {
+ t.Fatalf("unexpected stderr: %q", result.Stderr)
+ }
+ })
+
+ t.Run("decodes JSON", func(t *testing.T) {
+ var output struct {
+ Name string `json:"name"`
+ }
+ result, err := client.RunJSON(context.Background(), &output, append(command, "json")...)
+ if err != nil {
+ t.Fatalf("run JSON command: %v (stderr: %s)", err, result.Stderr)
+ }
+ if output.Name != "pstats" {
+ t.Fatalf("unexpected decoded name: %q", output.Name)
+ }
+ })
+
+ t.Run("enforces timeout", func(t *testing.T) {
+ timeoutClient := client
+ timeoutClient.Timeout = 50 * time.Millisecond
+
+ result, err := timeoutClient.Run(context.Background(), append(command, "timeout")...)
+ if err == nil || !strings.Contains(err.Error(), "timed out") {
+ t.Fatalf("expected timeout error, got %v", err)
+ }
+ if result.ExitCode != -1 {
+ t.Fatalf("expected exit code -1, got %d", result.ExitCode)
+ }
+ })
+}
+
+func TestPBClientHelperProcess(t *testing.T) {
+ // Provides controlled command results for PB client tests.
+ if os.Getenv("QUEST_PB_HELPER_PROCESS") != "1" {
+ return
+ }
+
+ separator := -1
+ for index, arg := range os.Args {
+ if arg == "--" {
+ separator = index
+ break
+ }
+ }
+ if separator == -1 || separator+1 >= len(os.Args) {
+ os.Exit(2)
+ }
+
+ switch os.Args[separator+1] {
+ case "success", "json":
+ fmt.Fprint(os.Stdout, `{"name":"pstats"}`)
+ if os.Args[separator+1] == "success" {
+ fmt.Fprint(os.Stderr, "warning")
+ }
+ os.Exit(0)
+ case "failure":
+ fmt.Fprint(os.Stderr, "failed")
+ os.Exit(7)
+ case "timeout":
+ time.Sleep(2 * time.Second)
+ default:
+ os.Exit(2)
+ }
+}
+
+func TestPasswordFromPBUserAddOutput(t *testing.T) {
+ // Verifies password parsing from PB user creation output.
+ output := "Added user: alice\nPassword is: generated-password\nRole(s) assigned: reader\n"
+ password, err := PasswordFromUserAddOutput(output)
+ if err != nil {
+ t.Fatalf("extract password: %v", err)
+ }
+ if password != "generated-password" {
+ t.Fatalf("unexpected password: %q", password)
+ }
+
+ if _, err := PasswordFromUserAddOutput("Added user: alice\n"); err == nil {
+ t.Fatal("expected missing password to fail")
+ }
+}
diff --git a/main.go b/tests/integration/config.go
similarity index 85%
rename from main.go
rename to tests/integration/config.go
index 1918788..10f61e0 100644
--- a/main.go
+++ b/tests/integration/config.go
@@ -20,11 +20,10 @@ import (
"flag"
"net/url"
"testing"
-)
-func main() {
- println("hello")
-}
+ httpclient "quest/tests/integration/clients/http"
+ "quest/tests/integration/clients/pb"
+)
type Glob struct {
QueryUrl url.URL
@@ -34,8 +33,9 @@ type Glob struct {
IngestorUsername string
IngestorPassword string
Stream string
- QueryClient HTTPClient
- IngestorClient HTTPClient
+ QueryClient httpclient.HTTPClient
+ IngestorClient httpclient.HTTPClient
+ PBClient pb.PBClient
Mode string
MinIoConfig
}
@@ -59,6 +59,7 @@ var NewGlob = func() Glob {
var stream string
var mode string
+ var pbBinary string
// XXX
var minioUrl string
var minioUser string
@@ -75,6 +76,7 @@ var NewGlob = func() Glob {
flag.StringVar(&stream, "stream", "app", "Specify stream. Default is app")
flag.StringVar(&mode, "mode", "smoke", "Specify mode. Default is smoke")
+ flag.StringVar(&pbBinary, "pb-bin", "pb", "Specify the pb binary path. Default is pb from PATH")
flag.StringVar(&minioUrl, "minio-url", "localhost:9000", "Specify MinIO URL. Default is localhost:9000")
flag.StringVar(&minioUser, "minio-user", "minioadmin", "Specify MinIO User. Default is `minioadmin`")
@@ -88,7 +90,8 @@ var NewGlob = func() Glob {
panic("Could not parse url")
}
- queryClient := DefaultClient(*parsedQueryTargetUrl, queryUsername, queryPassword)
+ queryClient := httpclient.DefaultClient(*parsedQueryTargetUrl, queryUsername, queryPassword)
+ pbClient := pb.DefaultPBClient(pbBinary)
if targetIngestorUrl != "" {
parsedIngestorTargetUrl, err := url.Parse(targetIngestorUrl)
@@ -96,7 +99,7 @@ var NewGlob = func() Glob {
panic("Could not parse url")
}
- ingestorClient := DefaultClient(*parsedIngestorTargetUrl, ingestorUsername, ingestorPassword)
+ ingestorClient := httpclient.DefaultClient(*parsedIngestorTargetUrl, ingestorUsername, ingestorPassword)
return Glob{
QueryUrl: *parsedQueryTargetUrl,
QueryUsername: queryUsername,
@@ -106,6 +109,7 @@ var NewGlob = func() Glob {
IngestorUsername: ingestorUsername,
IngestorPassword: ingestorPassword,
IngestorClient: ingestorClient,
+ PBClient: pbClient,
Stream: stream,
Mode: mode,
MinIoConfig: MinIoConfig{
@@ -121,6 +125,7 @@ var NewGlob = func() Glob {
QueryUsername: queryUsername,
QueryPassword: queryPassword,
QueryClient: queryClient,
+ PBClient: pbClient,
Stream: stream,
Mode: mode,
MinIoConfig: MinIoConfig{
diff --git a/tests/integration/dataset_test.go b/tests/integration/dataset_test.go
new file mode 100644
index 0000000..b0c6fe6
--- /dev/null
+++ b/tests/integration/dataset_test.go
@@ -0,0 +1,138 @@
+// Copyright (c) 2023 Cloudnatively Services Pvt Ltd
+//
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see .
+
+package main
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/require"
+)
+
+func TestSmokeListLogStream(t *testing.T) {
+ // Verifies that the dataset list includes a newly created stream.
+ t.Parallel()
+ streamName := NewGlob.Stream + "list"
+ CreateStream(t, NewGlob.PBClient, streamName)
+ t.Cleanup(func() {
+ DeleteStream(t, NewGlob.PBClient, streamName)
+ })
+ datasets := ListDatasetsWithPB(t, NewGlob.PBClient)
+ require.Contains(t, datasets, PBDataset{Title: streamName})
+}
+
+func TestSmokeCreateStream(t *testing.T) {
+ // Verifies that PB creates a logs dataset.
+ t.Parallel()
+ stream := NewGlob.Stream + "create"
+ CreateStream(t, NewGlob.PBClient, stream)
+ t.Cleanup(func() {
+ DeleteStream(t, NewGlob.PBClient, stream)
+ })
+ info := DatasetInfoWithPB(t, NewGlob.PBClient, stream)
+ require.Equal(t, "logs", info.DatasetType)
+}
+
+func TestSmokeDeleteStream(t *testing.T) {
+ // Verifies that PB deletes an empty stream.
+ t.Parallel()
+ stream := NewGlob.Stream + "delete"
+ CreateStream(t, NewGlob.PBClient, stream)
+ DeleteStream(t, NewGlob.PBClient, stream)
+ datasets := ListDatasetsWithPB(t, NewGlob.PBClient)
+ require.NotContains(t, datasets, PBDataset{Title: stream})
+}
+
+func TestSmokeDetectSchema(t *testing.T) {
+ // Verifies that schema detection returns the expected schema.
+ t.Parallel()
+ DetectSchema(t, NewGlob.QueryClient, SampleJson, SchemaBody)
+}
+
+// func TestTimePartition_TimeStampMismatch(t *testing.T) {
+// historicalStream := NewGlob.Stream + "historical"
+// timeHeader := map[string]string{"X-P-Time-Partition": "source_time"}
+// CreateStreamWithHeader(t, NewGlob.QueryClient, historicalStream, timeHeader)
+// if NewGlob.IngestorUrl.String() == "" {
+// IngestOneEventWithTimePartition_TimeStampMismatch(t, NewGlob.QueryClient, historicalStream)
+// } else {
+// IngestOneEventWithTimePartition_TimeStampMismatch(t, NewGlob.IngestorClient, historicalStream)
+// }
+// DeleteStream(t, NewGlob.PBClient, historicalStream)
+// }
+
+// func TestTimePartition_NoTimePartitionInLog(t *testing.T) {
+// historicalStream := NewGlob.Stream + "historical"
+// timeHeader := map[string]string{"X-P-Time-Partition": "source_time"}
+// CreateStreamWithHeader(t, NewGlob.QueryClient, historicalStream, timeHeader)
+// if NewGlob.IngestorUrl.String() == "" {
+// IngestOneEventWithTimePartition_NoTimePartitionInLog(t, NewGlob.QueryClient, historicalStream)
+// } else {
+// IngestOneEventWithTimePartition_NoTimePartitionInLog(t, NewGlob.IngestorClient, historicalStream)
+// }
+// DeleteStream(t, NewGlob.PBClient, historicalStream)
+// }
+
+// func TestTimePartition_IncorrectDateTimeFormatTimePartitionInLog(t *testing.T) {
+// historicalStream := NewGlob.Stream + "historical"
+// timeHeader := map[string]string{"X-P-Time-Partition": "source_time"}
+// CreateStreamWithHeader(t, NewGlob.QueryClient, historicalStream, timeHeader)
+// if NewGlob.IngestorUrl.String() == "" {
+// IngestOneEventWithTimePartition_IncorrectDateTimeFormatTimePartitionInLog(t, NewGlob.QueryClient, historicalStream)
+// } else {
+// IngestOneEventWithTimePartition_IncorrectDateTimeFormatTimePartitionInLog(t, NewGlob.IngestorClient, historicalStream)
+// }
+// DeleteStream(t, NewGlob.PBClient, historicalStream)
+// }
+
+func TestStaticSchemaIngestion(t *testing.T) {
+ // Verifies that a static schema accepts matching fields and rejects new fields.
+ t.Parallel()
+ staticSchemaStream := NewGlob.Stream + "staticschema"
+ staticSchemaFlagHeader := map[string]string{"X-P-Static-Schema-Flag": "true"}
+ CreateStreamWithSchemaBody(t, NewGlob.QueryClient, staticSchemaStream, staticSchemaFlagHeader, SchemaPayload)
+
+ client := NewGlob.QueryClient
+ if NewGlob.IngestorUrl.String() != "" {
+ client = NewGlob.IngestorClient
+ }
+
+ t.Run("AcceptMatchingFields", func(t *testing.T) {
+ IngestOneEventForStaticSchemaStream_SameFieldsInLog(t, client, staticSchemaStream)
+ })
+ t.Run("RejectNewField", func(t *testing.T) {
+ IngestOneEventForStaticSchemaStream_NewFieldInLog(t, client, staticSchemaStream)
+ })
+}
+
+func TestCreateStream_WithCustomPartition_Success(t *testing.T) {
+ // Verifies that a stream accepts one custom partition field.
+ t.Parallel()
+ customPartitionStream := NewGlob.Stream + "custompartitionsuccess"
+ customHeader := map[string]string{"X-P-Custom-Partition": "level"}
+ CreateStreamWithHeader(t, NewGlob.QueryClient, customPartitionStream, customHeader)
+ t.Cleanup(func() {
+ DeleteStream(t, NewGlob.PBClient, customPartitionStream)
+ })
+}
+
+func TestCreateStream_WithCustomPartition_Error(t *testing.T) {
+ // Verifies that multiple custom partition fields are rejected.
+ t.Parallel()
+ customPartitionStream := NewGlob.Stream + "custompartitionerror"
+ customHeader := map[string]string{"X-P-Custom-Partition": "level,os"}
+ CreateStreamWithCustompartitionError(t, NewGlob.QueryClient, customPartitionStream, customHeader)
+}
diff --git a/integrity_test.go b/tests/integration/integrity_test.go
similarity index 92%
rename from integrity_test.go
rename to tests/integration/integrity_test.go
index 6423581..7fdec71 100644
--- a/integrity_test.go
+++ b/tests/integration/integrity_test.go
@@ -79,12 +79,12 @@ func (flog *ParquetFlog) Deref() Flog {
}
}
-// - Send logs to Parseable
-// - Wait for sync
-// - Download parquet files from the store created by Parseable for the minute
-// - Compare the sent logs with the ones loaded from the downloaded parquet
func TestIntegrity(t *testing.T) {
- CreateStream(t, NewGlob.QueryClient, NewGlob.Stream)
+ // Verifies that ingested logs match the logs stored in Parquet files.
+ t.Parallel()
+ stream := NewGlob.Stream + "integrity"
+ workDir := t.TempDir()
+ CreateStream(t, NewGlob.PBClient, stream)
iterations := 1
flogsPerIteration := 100
@@ -97,7 +97,7 @@ func TestIntegrity(t *testing.T) {
flogs := make([]Flog, 0, iterations*flogsPerIteration)
for i := 0; i < iterations; i++ {
- flogsFile := fmt.Sprintf("%d.log", i)
+ flogsFile := filepath.Join(workDir, fmt.Sprintf("%d.log", i))
err := exec.Command("flog",
"--number", strconv.Itoa(flogsPerIteration),
@@ -111,7 +111,7 @@ func TestIntegrity(t *testing.T) {
loadedFlogs := loadFlogsFromFile(flogsFile)
- err = ingestFlogs(loadedFlogs, NewGlob.Stream)
+ err = ingestFlogs(loadedFlogs, stream)
if err != nil {
t.Fatal("error ingesting flogs", err)
}
@@ -127,7 +127,7 @@ func TestIntegrity(t *testing.T) {
// XXX: We don't need to sleep for the entire minute, just until the next minute boundary.
}
- parquetFiles := downloadParquetFiles(NewGlob.Stream, NewGlob.MinIoConfig)
+ parquetFiles := downloadParquetFiles(stream, NewGlob.MinIoConfig, workDir)
actualFlogs := loadFlogsFromParquetFiles(parquetFiles)
rowCount := len(actualFlogs)
@@ -140,7 +140,6 @@ func TestIntegrity(t *testing.T) {
require.Equal(t, actualFlog, expectedFlog)
}
- DeleteStream(t, NewGlob.QueryClient, NewGlob.Stream)
}
func ingestFlogs(flogs []Flog, stream string) error {
@@ -172,7 +171,7 @@ func ingestFlogs(flogs []Flog, stream string) error {
return nil
}
-func downloadParquetFiles(stream string, config MinIoConfig) []string {
+func downloadParquetFiles(stream string, config MinIoConfig, downloadDir string) []string {
client, err := minio.New(config.Url, config.User, config.Pass, false)
if err != nil {
slog.Error("couldn't create MinIO client", "error", err)
@@ -199,7 +198,7 @@ func downloadParquetFiles(stream string, config MinIoConfig) []string {
// Write the MinIO Object we got, into `downloadPath`.
- fileName := strings.ReplaceAll(key, "/", ".")
+ fileName := filepath.Join(downloadDir, strings.ReplaceAll(key, "/", "."))
f, _ := os.Create(fileName)
_, err = io.Copy(f, parquetObject)
@@ -286,6 +285,10 @@ func loadFlogsFromFile(path string) []Flog {
flogs = append(flogs, flog)
}
+ linesErr := lines.Err()
+ if linesErr != nil {
+ slog.Error("error reading lines", "error", linesErr)
+ }
return flogs
}
diff --git a/tests/integration/load_test.go b/tests/integration/load_test.go
new file mode 100644
index 0000000..77c398f
--- /dev/null
+++ b/tests/integration/load_test.go
@@ -0,0 +1,500 @@
+// Copyright (c) 2023 Cloudnatively Services Pvt Ltd
+//
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see .
+
+package main
+
+import (
+ "fmt"
+ "os/exec"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+)
+
+const (
+ vus = "10"
+ duration = "2m"
+ schema_count = "10"
+ events_count = "5"
+ parseableLoadSettleWait = 3 * time.Minute // Allows asynchronous flush and conversion to finish.
+)
+
+// loadPhaseLock lets all finite regular load tests join the active read phase,
+// while the smoke load remains exclusive from that phase.
+type loadPhaseLock struct {
+ readersMu sync.Mutex
+ resourceMu sync.Mutex
+ readers int
+}
+
+func (l *loadPhaseLock) RLock() {
+ l.readersMu.Lock()
+ l.readers++
+ if l.readers == 1 {
+ l.resourceMu.Lock()
+ }
+ l.readersMu.Unlock()
+}
+
+func (l *loadPhaseLock) RUnlock() {
+ l.readersMu.Lock()
+ l.readers--
+ if l.readers == 0 {
+ l.resourceMu.Unlock()
+ }
+ l.readersMu.Unlock()
+}
+
+func (l *loadPhaseLock) Lock() {
+ l.resourceMu.Lock()
+}
+
+func (l *loadPhaseLock) Unlock() {
+ l.resourceMu.Unlock()
+}
+
+var k6Mu loadPhaseLock
+
+func TestLoadStreamBatchWithK6_StaticSchema(t *testing.T) {
+ // Verifies batch ingestion into a static-schema stream under load.
+ if NewGlob.Mode == "load" {
+ t.Parallel()
+
+ staticSchemaStream := NewGlob.Stream + "loadbatchstaticschema"
+ staticSchemaFlagHeader := map[string]string{"X-P-Static-Schema-Flag": "true"}
+ CreateStreamWithSchemaBody(t, NewGlob.QueryClient, staticSchemaStream, staticSchemaFlagHeader, SchemaPayload)
+ if NewGlob.IngestorUrl.String() == "" {
+ cmd := exec.Command("k6",
+ "run",
+ "--address", "",
+ "--vus", vus,
+ "--duration", duration,
+ "-e", fmt.Sprintf("P_URL=%s", &NewGlob.QueryUrl),
+ "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
+ "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
+ "-e", fmt.Sprintf("P_STREAM=%s", staticSchemaStream),
+ "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
+ "-e", fmt.Sprintf("P_EVENTS_COUNT=%s", events_count),
+ "./scripts/load_batch_events.js")
+
+ runK6Load(t, cmd)
+ } else {
+ cmd := exec.Command("k6",
+ "run",
+ "--address", "",
+ "--vus", vus,
+ "--duration", duration,
+ "-e", fmt.Sprintf("P_URL=%s", &NewGlob.IngestorUrl),
+ "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
+ "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
+ "-e", fmt.Sprintf("P_STREAM=%s", staticSchemaStream),
+ "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
+ "-e", fmt.Sprintf("P_EVENTS_COUNT=%s", events_count),
+ "./scripts/load_batch_events.js")
+
+ runK6Load(t, cmd)
+ }
+ }
+}
+
+func TestSmokeLoadWithK6Streams(t *testing.T) {
+ // Verifies smoke ingestion for normal and custom-partition streams.
+ t.Parallel()
+ stream := NewGlob.Stream + "smokeload"
+ CreateStream(t, NewGlob.PBClient, stream)
+ k6Mu.Lock()
+ defer k6Mu.Unlock()
+ runK6Smoke(t, stream)
+
+ customPartitionStream := NewGlob.Stream + "smokeloadcustompartition"
+ customHeader := map[string]string{"X-P-Custom-Partition": "level"}
+ CreateStreamWithHeader(t, NewGlob.QueryClient, customPartitionStream, customHeader)
+ runK6Smoke(t, customPartitionStream)
+
+ time.Sleep(parseableLoadSettleWait)
+
+ t.Run("LoadWithK6Stream", func(t *testing.T) {
+ QueryLogStreamCount(t, NewGlob.PBClient, stream, 20000)
+ AssertStreamSchema(t, NewGlob.QueryClient, stream, SchemaBody)
+ })
+
+ t.Run("Load_CustomPartition_WithK6Stream", func(t *testing.T) {
+ QueryLogStreamCount(t, NewGlob.PBClient, customPartitionStream, 20000)
+ })
+
+}
+
+func runK6Smoke(t *testing.T, stream string) {
+ t.Helper()
+ url := NewGlob.QueryUrl.String()
+ username := NewGlob.QueryUsername
+ password := NewGlob.QueryPassword
+ if NewGlob.IngestorUrl.String() != "" {
+ url = NewGlob.IngestorUrl.String()
+ username = NewGlob.IngestorUsername
+ password = NewGlob.IngestorPassword
+ }
+
+ cmd := exec.Command("k6",
+ "run",
+ "--address", "",
+ "-e", fmt.Sprintf("P_URL=%s", url),
+ "-e", fmt.Sprintf("P_USERNAME=%s", username),
+ "-e", fmt.Sprintf("P_PASSWORD=%s", password),
+ "-e", fmt.Sprintf("P_STREAM=%s", stream),
+ "./scripts/smoke.js")
+
+ op, err := cmd.CombinedOutput()
+ require.NoErrorf(t, err, "k6 failed: %s", string(op))
+ t.Log(string(op))
+}
+
+func runK6Load(t *testing.T, cmd *exec.Cmd) {
+ t.Helper()
+ k6Mu.RLock()
+ defer k6Mu.RUnlock()
+ op, err := cmd.CombinedOutput()
+ require.NoErrorf(t, err, "k6 failed: %s", string(op))
+ t.Log(string(op))
+ time.Sleep(parseableLoadSettleWait)
+}
+
+// func TestSmokeLoad_TimePartition_WithK6Stream(t *testing.T) {
+// time_partition_stream := NewGlob.Stream + "timepartition"
+// timeHeader := map[string]string{"X-P-Time-Partition": "source_time", "X-P-Time-Partition-Limit": "365d"}
+// CreateStreamWithHeader(t, NewGlob.QueryClient, time_partition_stream, timeHeader)
+// if NewGlob.IngestorUrl.String() == "" {
+// cmd := exec.Command("k6",
+// "run",
+// "-e", fmt.Sprintf("P_URL=%s", NewGlob.QueryUrl.String()),
+// "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
+// "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
+// "-e", fmt.Sprintf("P_STREAM=%s", time_partition_stream),
+// "./scripts/smoke.js")
+
+// cmd.Run()
+// cmd.Output()
+// } else {
+// cmd := exec.Command("k6",
+// "run",
+// "-e", fmt.Sprintf("P_URL=%s", NewGlob.IngestorUrl.String()),
+// "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
+// "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
+// "-e", fmt.Sprintf("P_STREAM=%s", time_partition_stream),
+// "./scripts/smoke.js")
+
+// cmd.Run()
+// cmd.Output()
+// }
+// time.Sleep(120 * time.Second)
+// QueryLogStreamCount_Historical(t, NewGlob.PBClient, time_partition_stream, 20000)
+// DeleteStream(t, NewGlob.PBClient, time_partition_stream)
+// }
+
+// func TestSmokeLoad_TimeAndCustomPartition_WithK6Stream(t *testing.T) {
+// custom_partition_stream := NewGlob.Stream + "timecustompartition"
+// customHeader := map[string]string{"X-P-Custom-Partition": "level", "X-P-Time-Partition": "source_time", "X-P-Time-Partition-Limit": "365d"}
+// CreateStreamWithHeader(t, NewGlob.QueryClient, custom_partition_stream, customHeader)
+// if NewGlob.IngestorUrl.String() == "" {
+// cmd := exec.Command("k6",
+// "run",
+// "-e", fmt.Sprintf("P_URL=%s", NewGlob.QueryUrl.String()),
+// "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
+// "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
+// "-e", fmt.Sprintf("P_STREAM=%s", custom_partition_stream),
+// "./scripts/smoke.js")
+
+// cmd.Run()
+// cmd.Output()
+// } else {
+// cmd := exec.Command("k6",
+// "run",
+// "-e", fmt.Sprintf("P_URL=%s", NewGlob.IngestorUrl.String()),
+// "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
+// "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
+// "-e", fmt.Sprintf("P_STREAM=%s", custom_partition_stream),
+// "./scripts/smoke.js")
+
+// cmd.Run()
+// cmd.Output()
+// }
+// time.Sleep(180 * time.Second)
+// QueryLogStreamCount_Historical(t, NewGlob.PBClient, custom_partition_stream, 20000)
+// DeleteStream(t, NewGlob.PBClient, custom_partition_stream)
+// }
+
+func TestLoadStreamBatchWithK6(t *testing.T) {
+ // Verifies batch ingestion into a normal stream under load.
+ if NewGlob.Mode == "load" {
+ t.Parallel()
+
+ stream := NewGlob.Stream + "loadbatch"
+ CreateStream(t, NewGlob.PBClient, stream)
+ if NewGlob.IngestorUrl.String() == "" {
+ cmd := exec.Command("k6",
+ "run",
+ "--address", "",
+ "--vus", vus,
+ "--duration", duration,
+ "-e", fmt.Sprintf("P_URL=%s", NewGlob.QueryUrl.String()),
+ "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
+ "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
+ "-e", fmt.Sprintf("P_STREAM=%s", stream),
+ "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
+ "-e", fmt.Sprintf("P_EVENTS_COUNT=%s", events_count),
+ "./scripts/load_batch_events.js")
+
+ runK6Load(t, cmd)
+ } else {
+ cmd := exec.Command("k6",
+ "run",
+ "--address", "",
+ "--vus", vus,
+ "--duration", duration,
+ "-e", fmt.Sprintf("P_URL=%s", NewGlob.IngestorUrl.String()),
+ "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
+ "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
+ "-e", fmt.Sprintf("P_STREAM=%s", stream),
+ "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
+ "-e", fmt.Sprintf("P_EVENTS_COUNT=%s", events_count),
+ "./scripts/load_batch_events.js")
+
+ runK6Load(t, cmd)
+ }
+ }
+}
+
+// func TestLoadHistoricalStreamBatchWithK6(t *testing.T) {
+// if NewGlob.Mode == "load" {
+// historicalStream := NewGlob.Stream + "historical"
+// timeHeader := map[string]string{"X-P-Time-Partition": "source_time"}
+// CreateStreamWithHeader(t, NewGlob.QueryClient, historicalStream, timeHeader)
+// if NewGlob.IngestorUrl.String() == "" {
+// cmd := exec.Command("k6",
+// "run",
+// "-e", fmt.Sprintf("P_URL=%s", NewGlob.QueryUrl.String()),
+// "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
+// "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
+// "-e", fmt.Sprintf("P_STREAM=%s", historicalStream),
+// "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
+// "-e", fmt.Sprintf("P_EVENTS_COUNT=%s", events_count),
+// "./scripts/load_historical_batch_events.js",
+// "--vus=", vus,
+// "--duration=", duration)
+
+// cmd.Run()
+// op, err := cmd.Output()
+// if err != nil {
+// t.Log(err)
+// }
+// t.Log(string(op))
+// } else {
+// cmd := exec.Command("k6",
+// "run",
+// "-e", fmt.Sprintf("P_URL=%s", NewGlob.IngestorUrl.String()),
+// "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
+// "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
+// "-e", fmt.Sprintf("P_STREAM=%s", historicalStream),
+// "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
+// "-e", fmt.Sprintf("P_EVENTS_COUNT=%s", events_count),
+// "./scripts/load_historical_batch_events.js",
+// "--vus=", vus,
+// "--duration=", duration)
+
+// cmd.Run()
+// op, err := cmd.Output()
+// if err != nil {
+// t.Log(err)
+// }
+// t.Log(string(op))
+// }
+
+// DeleteStream(t, NewGlob.PBClient, historicalStream)
+// }
+// }
+
+func TestLoadStreamBatchWithCustomPartitionWithK6(t *testing.T) {
+ // Verifies batch ingestion into a custom-partition stream under load.
+ if NewGlob.Mode != "load" {
+ return
+ }
+ t.Parallel()
+
+ customPartitionStream := NewGlob.Stream + "loadbatchcustompartition"
+ customHeader := map[string]string{"X-P-Custom-Partition": "level"}
+ CreateStreamWithHeader(t, NewGlob.QueryClient, customPartitionStream, customHeader)
+ if NewGlob.IngestorUrl.String() == "" {
+ cmd := exec.Command("k6",
+ "run",
+ "--address", "",
+ "--vus", vus,
+ "--duration", duration,
+ "-e", fmt.Sprintf("P_URL=%s", NewGlob.QueryUrl.String()),
+ "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
+ "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
+ "-e", fmt.Sprintf("P_STREAM=%s", customPartitionStream),
+ "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
+ "-e", fmt.Sprintf("P_EVENTS_COUNT=%s", events_count),
+ "./scripts/load_batch_events.js")
+
+ runK6Load(t, cmd)
+ } else {
+ cmd := exec.Command("k6",
+ "run",
+ "--address", "",
+ "--vus", vus,
+ "--duration", duration,
+ "-e", fmt.Sprintf("P_URL=%s", NewGlob.IngestorUrl.String()),
+ "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
+ "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
+ "-e", fmt.Sprintf("P_STREAM=%s", customPartitionStream),
+ "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
+ "-e", fmt.Sprintf("P_EVENTS_COUNT=%s", events_count),
+ "./scripts/load_batch_events.js")
+
+ runK6Load(t, cmd)
+ }
+}
+
+func TestLoadStreamNoBatchWithK6(t *testing.T) {
+ // Verifies single-event ingestion into a normal stream under load.
+ if NewGlob.Mode == "load" {
+ t.Parallel()
+
+ stream := NewGlob.Stream + "loadsingle"
+ CreateStream(t, NewGlob.PBClient, stream)
+ if NewGlob.IngestorUrl.String() == "" {
+ cmd := exec.Command("k6",
+ "run",
+ "--address", "",
+ "--vus", vus,
+ "--duration", duration,
+ "-e", fmt.Sprintf("P_URL=%s", NewGlob.QueryUrl.String()),
+ "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
+ "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
+ "-e", fmt.Sprintf("P_STREAM=%s", stream),
+ "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
+ "./scripts/load_single_event.js")
+
+ runK6Load(t, cmd)
+ } else {
+ cmd := exec.Command("k6",
+ "run",
+ "--address", "",
+ "--vus", vus,
+ "--duration", duration,
+ "-e", fmt.Sprintf("P_URL=%s", NewGlob.IngestorUrl.String()),
+ "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
+ "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
+ "-e", fmt.Sprintf("P_STREAM=%s", stream),
+ "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
+ "./scripts/load_single_event.js")
+
+ runK6Load(t, cmd)
+ }
+
+ }
+}
+
+// func TestLoadHistoricalStreamNoBatchWithK6(t *testing.T) {
+// if NewGlob.Mode == "load" {
+// historicalStream := NewGlob.Stream + "historical"
+// timeHeader := map[string]string{"X-P-Time-Partition": "source_time"}
+// CreateStreamWithHeader(t, NewGlob.QueryClient, historicalStream, timeHeader)
+// if NewGlob.IngestorUrl.String() == "" {
+// cmd := exec.Command("k6",
+// "run",
+// "-e", fmt.Sprintf("P_URL=%s", NewGlob.QueryUrl.String()),
+// "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
+// "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
+// "-e", fmt.Sprintf("P_STREAM=%s", historicalStream),
+// "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
+// "./scripts/load_single_events.js",
+// "--vus=", vus,
+// "--duration=", duration)
+
+// cmd.Run()
+// op, err := cmd.Output()
+// if err != nil {
+// t.Log(err)
+// }
+// t.Log(string(op))
+// } else {
+// cmd := exec.Command("k6",
+// "run",
+// "-e", fmt.Sprintf("P_URL=%s", NewGlob.IngestorUrl.String()),
+// "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
+// "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
+// "-e", fmt.Sprintf("P_STREAM=%s", historicalStream),
+// "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
+// "./scripts/load_single_events.js",
+// "--vus=", vus,
+// "--duration=", duration)
+
+// cmd.Run()
+// op, err := cmd.Output()
+// if err != nil {
+// t.Log(err)
+// }
+// t.Log(string(op))
+// }
+
+// DeleteStream(t, NewGlob.PBClient, historicalStream)
+// }
+// }
+
+func TestLoadStreamNoBatchWithCustomPartitionWithK6(t *testing.T) {
+ // Verifies single-event ingestion into a custom-partition stream under load.
+ if NewGlob.Mode != "load" {
+ return
+ }
+ t.Parallel()
+
+ customPartitionStream := NewGlob.Stream + "loadsinglecustompartition"
+ customHeader := map[string]string{"X-P-Custom-Partition": "level"}
+ CreateStreamWithHeader(t, NewGlob.QueryClient, customPartitionStream, customHeader)
+ if NewGlob.IngestorUrl.String() == "" {
+ cmd := exec.Command("k6",
+ "run",
+ "--address", "",
+ "--vus", vus,
+ "--duration", duration,
+ "-e", fmt.Sprintf("P_URL=%s", NewGlob.QueryUrl.String()),
+ "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
+ "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
+ "-e", fmt.Sprintf("P_STREAM=%s", customPartitionStream),
+ "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
+ "./scripts/load_single_event.js")
+
+ runK6Load(t, cmd)
+ } else {
+ cmd := exec.Command("k6",
+ "run",
+ "--address", "",
+ "--vus", vus,
+ "--duration", duration,
+ "-e", fmt.Sprintf("P_URL=%s", NewGlob.IngestorUrl.String()),
+ "-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
+ "-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
+ "-e", fmt.Sprintf("P_STREAM=%s", customPartitionStream),
+ "-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
+ "./scripts/load_single_event.js")
+
+ runK6Load(t, cmd)
+ }
+}
diff --git a/model.go b/tests/integration/model.go
similarity index 100%
rename from model.go
rename to tests/integration/model.go
diff --git a/tests/integration/query_test.go b/tests/integration/query_test.go
new file mode 100644
index 0000000..46af31d
--- /dev/null
+++ b/tests/integration/query_test.go
@@ -0,0 +1,67 @@
+// Copyright (c) 2023 Cloudnatively Services Pvt Ltd
+//
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see .
+
+package main
+
+import (
+ "testing"
+ "time"
+)
+
+func TestSmokeIngestAndQuery(t *testing.T) {
+ // Verifies ingestion and SQL queries across two streams.
+ t.Parallel()
+ stream1 := NewGlob.Stream + "ingestquery1"
+ stream2 := NewGlob.Stream + "ingestquery2"
+ CreateStream(t, NewGlob.PBClient, stream1)
+ CreateStream(t, NewGlob.PBClient, stream2)
+
+ if NewGlob.IngestorUrl.String() == "" {
+ RunFlog(t, NewGlob.QueryClient, stream1)
+ RunFlog(t, NewGlob.QueryClient, stream2)
+ } else {
+ RunFlog(t, NewGlob.IngestorClient, stream1)
+ RunFlog(t, NewGlob.IngestorClient, stream2)
+ }
+
+ // Parseable persists ingested events in a two-minute batch. Both streams are
+ // populated before this wait so all ingestion and query assertions can share
+ // the same batch window.
+ time.Sleep(120 * time.Second)
+
+ t.Run("IngestEventsToStream", func(t *testing.T) {
+ QueryLogStreamCount(t, NewGlob.PBClient, stream1, 50)
+ AssertStreamSchema(t, NewGlob.QueryClient, stream1, FlogJsonSchema)
+ })
+
+ t.Run("RunQueries", func(t *testing.T) {
+ QueryLogStreamCount(t, NewGlob.PBClient, stream1, 50)
+ AssertQueryOK(t, NewGlob.PBClient, "SELECT * FROM %s", stream1)
+ AssertQueryOK(t, NewGlob.PBClient, "SELECT * FROM %s OFFSET 25 LIMIT 25", stream1)
+
+ for _, item := range flogStreamFields() {
+ AssertQueryOK(t, NewGlob.PBClient, "SELECT %s FROM %s", item, stream1)
+ }
+
+ AssertQueryOK(t, NewGlob.PBClient, "SELECT * FROM %s WHERE method = 'POST'", stream1)
+ AssertQueryOK(t, NewGlob.PBClient, "SELECT method, COUNT(*) FROM %s GROUP BY method", stream1)
+ AssertQueryOK(t, NewGlob.PBClient, `SELECT DATE_TRUNC('minute', p_timestamp) as minute, COUNT(*) FROM %s GROUP BY minute`, stream1)
+ })
+
+ t.Run("QueryTwoStreams", func(t *testing.T) {
+ QueryTwoLogStreamCount(t, NewGlob.PBClient, stream1, stream2, 100)
+ })
+}
diff --git a/tests/integration/rbac_test.go b/tests/integration/rbac_test.go
new file mode 100644
index 0000000..39ac104
--- /dev/null
+++ b/tests/integration/rbac_test.go
@@ -0,0 +1,149 @@
+// Copyright (c) 2023 Cloudnatively Services Pvt Ltd
+//
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see .
+
+package main
+
+import (
+ "strings"
+ "sync"
+ "testing"
+
+ httpclient "quest/tests/integration/clients/http"
+)
+
+// RBAC tests mutate shared server-wide user and role state. They are still
+// scheduled as parallel tests, but those mutations must not overlap.
+var rbacMu sync.Mutex
+
+func TestSmoke_AllUsersAPI(t *testing.T) {
+ // Verifies the user creation, role, password, and deletion flow.
+ t.Parallel()
+ rbacMu.Lock()
+ defer rbacMu.Unlock()
+
+ role := NewGlob.Stream + "allusersrole"
+ user := NewGlob.Stream + "allusers"
+ CreateRole(t, NewGlob.QueryClient, role, dummyRole)
+ AssertRole(t, NewGlob.QueryClient, role, dummyRole)
+
+ CreateUserWithRole(t, NewGlob.PBClient, user, []string{role})
+ AssertUserRole(t, NewGlob.QueryClient, user, role, dummyRole)
+ RegenPassword(t, NewGlob.QueryClient, user)
+ DeleteUser(t, NewGlob.PBClient, user)
+ DeleteRole(t, NewGlob.PBClient, role)
+}
+
+func TestSmoke_NewUserWithRole(t *testing.T) {
+ // Verifies that a new user can be created with a role.
+ t.Parallel()
+ rbacMu.Lock()
+ defer rbacMu.Unlock()
+
+ role := NewGlob.Stream + "newuserrole"
+ user := NewGlob.Stream + "newuser"
+
+ CreateRole(t, NewGlob.QueryClient, role, dummyRole)
+ AssertRole(t, NewGlob.QueryClient, role, dummyRole)
+ CreateUserWithRole(t, NewGlob.PBClient, user, []string{role})
+ AssertUserRole(t, NewGlob.QueryClient, user, role, dummyRole)
+ DeleteUser(t, NewGlob.PBClient, user)
+ DeleteRole(t, NewGlob.PBClient, role)
+}
+
+func TestSmokeRbacBasic(t *testing.T) {
+ // Verifies that a user's role controls basic API access.
+ t.Parallel()
+ rbacMu.Lock()
+ defer rbacMu.Unlock()
+
+ stream := NewGlob.Stream + "rbacbasic"
+ role := NewGlob.Stream + "rbacbasicrole"
+ user := NewGlob.Stream + "rbacbasicuser"
+ CreateStream(t, NewGlob.PBClient, stream)
+ CreateRole(t, NewGlob.QueryClient, role, dummyRole)
+ AssertRole(t, NewGlob.QueryClient, role, dummyRole)
+ CreateUserWithRole(t, NewGlob.PBClient, user, []string{role})
+ userClient := NewGlob.QueryClient
+ userClient.Username = user
+ userClient.Password = RegenPassword(t, NewGlob.QueryClient, user)
+ checkAPIAccess(t, userClient, NewGlob.QueryClient, stream, "editor")
+ DeleteUser(t, NewGlob.PBClient, user)
+ DeleteRole(t, NewGlob.PBClient, role)
+}
+
+func TestSmokeRoles(t *testing.T) {
+ // Verifies API access for ingestor, reader, writer, and editor roles.
+ t.Parallel()
+ rbacMu.Lock()
+ defer rbacMu.Unlock()
+
+ stream := NewGlob.Stream + "roles"
+ editorDeleteStream := NewGlob.Stream + "roleseditordelete"
+ CreateStream(t, NewGlob.PBClient, stream)
+ CreateStream(t, NewGlob.PBClient, editorDeleteStream)
+ cases := []struct {
+ roleName string
+ body string
+ }{
+ {
+ roleName: NewGlob.Stream + "ingestor",
+ body: Roleingestor(stream),
+ },
+ {
+ roleName: NewGlob.Stream + "reader",
+ body: RoleReader(stream),
+ },
+ {
+ roleName: NewGlob.Stream + "writer",
+ body: RoleWriter(stream),
+ },
+ {
+ roleName: NewGlob.Stream + "editor",
+ body: RoleEditor,
+ },
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.roleName, func(t *testing.T) {
+ CreateRole(t, NewGlob.QueryClient, tc.roleName, tc.body)
+ AssertRole(t, NewGlob.QueryClient, tc.roleName, tc.body)
+ username := tc.roleName + "_user"
+ password := CreateUserWithRole(t, NewGlob.PBClient, username, []string{tc.roleName})
+ var ingestClient httpclient.HTTPClient
+ queryClient := NewGlob.QueryClient
+ queryClient.Username = username
+ queryClient.Password = password
+ if NewGlob.IngestorUrl.String() != "" {
+ ingestClient = NewGlob.IngestorClient
+ ingestClient.Username = username
+ ingestClient.Password = password
+ } else {
+ ingestClient = NewGlob.QueryClient
+ ingestClient.Username = username
+ ingestClient.Password = password
+ }
+
+ roleKind := strings.TrimPrefix(tc.roleName, NewGlob.Stream)
+ accessStream := stream
+ if roleKind == "editor" {
+ accessStream = editorDeleteStream
+ }
+ checkAPIAccess(t, queryClient, ingestClient, accessStream, roleKind)
+ DeleteUser(t, NewGlob.PBClient, username)
+ DeleteRole(t, NewGlob.PBClient, tc.roleName)
+ })
+ }
+}
diff --git a/tests/integration/retention_test.go b/tests/integration/retention_test.go
new file mode 100644
index 0000000..196242d
--- /dev/null
+++ b/tests/integration/retention_test.go
@@ -0,0 +1,59 @@
+// Copyright (c) 2023 Cloudnatively Services Pvt Ltd
+//
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see .
+
+package main
+
+import (
+ "encoding/json"
+ "strings"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+)
+
+func TestSmokeSetRetention(t *testing.T) {
+ // Verifies that retention rules can be set on a stream.
+ t.Parallel()
+ stream := NewGlob.Stream + "setretention"
+ CreateStream(t, NewGlob.PBClient, stream)
+ t.Cleanup(func() {
+ DeleteStream(t, NewGlob.PBClient, stream)
+ })
+ req, _ := NewGlob.QueryClient.NewRequest("PUT", "logstream/"+stream+"/retention", strings.NewReader(RetentionBody))
+ response, err := NewGlob.QueryClient.Do(req)
+ require.NoErrorf(t, err, "Request failed: %s", err)
+ require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s and response: %s", response.Status, readAsString(response.Body))
+}
+
+func TestSmokeGetRetention(t *testing.T) {
+ // Verifies that PB returns the configured retention rules.
+ t.Parallel()
+ stream := NewGlob.Stream + "getretention"
+ CreateStream(t, NewGlob.PBClient, stream)
+ t.Cleanup(func() {
+ DeleteStream(t, NewGlob.PBClient, stream)
+ })
+
+ req, _ := NewGlob.QueryClient.NewRequest("PUT", "logstream/"+stream+"/retention", strings.NewReader(RetentionBody))
+ response, err := NewGlob.QueryClient.Do(req)
+ require.NoErrorf(t, err, "Request failed: %s", err)
+ require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s and response: %s", response.Status, readAsString(response.Body))
+
+ info := DatasetInfoWithPB(t, NewGlob.PBClient, stream)
+ var expected []PBRetentionRule
+ require.NoError(t, json.Unmarshal([]byte(RetentionBody), &expected))
+ require.Equal(t, expected, info.Retention, "Get retention response doesn't match with retention config returned")
+}
diff --git a/test_utils.go b/tests/integration/test_utils.go
similarity index 63%
rename from test_utils.go
rename to tests/integration/test_utils.go
index caabfc9..1c03363 100644
--- a/test_utils.go
+++ b/tests/integration/test_utils.go
@@ -18,6 +18,7 @@ package main
import (
"bytes"
+ "context"
"encoding/json"
"fmt"
"io"
@@ -27,10 +28,8 @@ import (
"time"
"github.com/stretchr/testify/require"
-)
-
-const (
- sleepDuration = 2 * time.Second
+ httpclient "quest/tests/integration/clients/http"
+ "quest/tests/integration/clients/pb"
)
func flogStreamFields() []string {
@@ -59,18 +58,50 @@ func readJsonBody[T any](body io.Reader) (res T, err error) {
return
}
-func Sleep() {
- time.Sleep(sleepDuration)
+type PBDataset struct {
+ Title string `json:"title"`
}
-func CreateStream(t *testing.T, client HTTPClient, stream string) {
- req, _ := client.NewRequest("PUT", "logstream/"+stream, nil)
- response, err := client.Do(req)
- require.NoErrorf(t, err, "Request failed: %s", err)
- require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s", response.Status)
+type PBDatasetInfo struct {
+ DatasetType string `json:"dataset_type"`
+ Retention []PBRetentionRule `json:"retention"`
+}
+
+type PBRetentionRule struct {
+ Description string `json:"description"`
+ Action string `json:"action"`
+ Duration string `json:"duration"`
+}
+
+func CreateStream(t *testing.T, client pb.PBClient, dataset string) {
+ t.Helper()
+ result, err := client.Run(context.Background(), "dataset", "add", dataset, "--type", "logs")
+ require.NoErrorf(t, err, "pb dataset add failed (exit=%d, stdout=%q, stderr=%q)", result.ExitCode, result.Stdout, result.Stderr)
+}
+
+func ListDatasetsWithPB(t *testing.T, client pb.PBClient) []PBDataset {
+ t.Helper()
+ var datasets []PBDataset
+ result, err := client.RunJSON(context.Background(), &datasets, "dataset", "list")
+ require.NoErrorf(t, err, "pb dataset list failed (exit=%d, stdout=%q, stderr=%q)", result.ExitCode, result.Stdout, result.Stderr)
+ return datasets
+}
+
+func DatasetInfoWithPB(t *testing.T, client pb.PBClient, dataset string) PBDatasetInfo {
+ t.Helper()
+ var info PBDatasetInfo
+ result, err := client.RunJSON(context.Background(), &info, "dataset", "info", dataset)
+ require.NoErrorf(t, err, "pb dataset info failed (exit=%d, stdout=%q, stderr=%q)", result.ExitCode, result.Stdout, result.Stderr)
+ return info
+}
+
+func DeleteStream(t *testing.T, client pb.PBClient, dataset string) {
+ t.Helper()
+ result, err := client.Run(context.Background(), "dataset", "remove", dataset)
+ require.NoErrorf(t, err, "pb dataset remove failed (exit=%d, stdout=%q, stderr=%q)", result.ExitCode, result.Stdout, result.Stderr)
}
-func CreateStreamWithHeader(t *testing.T, client HTTPClient, stream string, header map[string]string) {
+func CreateStreamWithHeader(t *testing.T, client httpclient.HTTPClient, stream string, header map[string]string) {
req, _ := client.NewRequest("PUT", "logstream/"+stream, nil)
for k, v := range header {
req.Header.Add(k, v)
@@ -80,7 +111,7 @@ func CreateStreamWithHeader(t *testing.T, client HTTPClient, stream string, head
require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s", response.Status)
}
-func CreateStreamWithCustompartitionError(t *testing.T, client HTTPClient, stream string, header map[string]string) {
+func CreateStreamWithCustompartitionError(t *testing.T, client httpclient.HTTPClient, stream string, header map[string]string) {
req, _ := client.NewRequest("PUT", "logstream/"+stream, nil)
for k, v := range header {
req.Header.Add(k, v)
@@ -89,7 +120,7 @@ func CreateStreamWithCustompartitionError(t *testing.T, client HTTPClient, strea
require.Equalf(t, 500, response.StatusCode, "Server returned http code: %s", response.Status)
}
-func CreateStreamWithSchemaBody(t *testing.T, client HTTPClient, stream string, header map[string]string, schema_payload string) {
+func CreateStreamWithSchemaBody(t *testing.T, client httpclient.HTTPClient, stream string, header map[string]string, schema_payload string) {
req, _ := client.NewRequest("PUT", "logstream/"+stream, bytes.NewBufferString(schema_payload))
for k, v := range header {
@@ -100,7 +131,7 @@ func CreateStreamWithSchemaBody(t *testing.T, client HTTPClient, stream string,
require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s", response.Status)
}
-func DetectSchema(t *testing.T, client HTTPClient, sampleJson string, schemaBody string) {
+func DetectSchema(t *testing.T, client httpclient.HTTPClient, sampleJson string, schemaBody string) {
req, _ := client.NewRequest("POST", "logstream/schema/detect", bytes.NewBufferString(sampleJson))
response, err := client.Do(req)
require.NoErrorf(t, err, "Request failed: %s", err)
@@ -109,28 +140,21 @@ func DetectSchema(t *testing.T, client HTTPClient, sampleJson string, schemaBody
require.JSONEq(t, schemaBody, body, "Schema detection failed")
}
-func DeleteStream(t *testing.T, client HTTPClient, stream string) {
- req, _ := client.NewRequest("DELETE", "logstream/"+stream, nil)
- response, err := client.Do(req)
- require.NoErrorf(t, err, "Request failed: %s", err)
- require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s", response.Status)
-}
-
-func DeleteAlert(t *testing.T, client HTTPClient, alert_id string) {
+func DeleteAlert(t *testing.T, client httpclient.HTTPClient, alert_id string) {
req, _ := client.NewRequest("DELETE", "alerts/"+alert_id, nil)
response, err := client.Do(req)
require.NoErrorf(t, err, "Request failed: %s", err)
require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s", response.Status)
}
-func DeleteTarget(t *testing.T, client HTTPClient, target_id string) {
+func DeleteTarget(t *testing.T, client httpclient.HTTPClient, target_id string) {
req, _ := client.NewRequest("DELETE", "targets/"+target_id, nil)
response, err := client.Do(req)
require.NoErrorf(t, err, "Request failed: %s", err)
require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s", response.Status)
}
-func RunFlog(t *testing.T, client HTTPClient, stream string) {
+func RunFlog(t *testing.T, client httpclient.HTTPClient, stream string) {
cmd := exec.Command("flog", "-f", "json", "-n", "50")
var out strings.Builder
cmd.Stdout = &out
@@ -151,7 +175,7 @@ func RunFlog(t *testing.T, client HTTPClient, stream string) {
}
}
-func IngestOneEventWithTimePartition_TimeStampMismatch(t *testing.T, client HTTPClient, stream string) {
+func IngestOneEventWithTimePartition_TimeStampMismatch(t *testing.T, client httpclient.HTTPClient, stream string) {
var test_payload string = `{"source_time":"2024-03-26T18:08:00.434Z","level":"info","message":"Application is failing","version":"1.2.0","user_id":13912,"device_id":4138,"session_id":"abc","os":"Windows","host":"112.168.1.110","location":"ngeuprqhynuvpxgp","request_body":"rnkmffyawtdcindtrdqruyxbndbjpfsptzpwtujbmkwcqastmxwbvjwphmyvpnhordwljnodxhtvpjesjldtifswqbpyuhlcytmm","status_code":300,"app_meta":"ckgpibhmlusqqfunnpxbfxbc", "new_field_added_by":"ingestor 8020"}`
req, _ := client.NewRequest("POST", "ingest", bytes.NewBufferString(test_payload))
req.Header.Add("X-P-Stream", stream)
@@ -160,7 +184,7 @@ func IngestOneEventWithTimePartition_TimeStampMismatch(t *testing.T, client HTTP
require.Equalf(t, 400, response.StatusCode, "Server returned http code: %s resp %s", response.Status, readAsString(response.Body))
}
-func IngestOneEventWithTimePartition_NoTimePartitionInLog(t *testing.T, client HTTPClient, stream string) {
+func IngestOneEventWithTimePartition_NoTimePartitionInLog(t *testing.T, client httpclient.HTTPClient, stream string) {
var test_payload string = `{"level":"info","message":"Application is failing","version":"1.2.0","user_id":13912,"device_id":4138,"session_id":"abc","os":"Windows","host":"112.168.1.110","location":"ngeuprqhynuvpxgp","request_body":"rnkmffyawtdcindtrdqruyxbndbjpfsptzpwtujbmkwcqastmxwbvjwphmyvpnhordwljnodxhtvpjesjldtifswqbpyuhlcytmm","status_code":300,"app_meta":"ckgpibhmlusqqfunnpxbfxbc", "new_field_added_by":"ingestor 8020"}`
req, _ := client.NewRequest("POST", "ingest", bytes.NewBufferString(test_payload))
req.Header.Add("X-P-Stream", stream)
@@ -169,7 +193,7 @@ func IngestOneEventWithTimePartition_NoTimePartitionInLog(t *testing.T, client H
require.Equalf(t, 400, response.StatusCode, "Server returned http code: %s resp %s", response.Status, readAsString(response.Body))
}
-func IngestOneEventWithTimePartition_IncorrectDateTimeFormatTimePartitionInLog(t *testing.T, client HTTPClient, stream string) {
+func IngestOneEventWithTimePartition_IncorrectDateTimeFormatTimePartitionInLog(t *testing.T, client httpclient.HTTPClient, stream string) {
var test_payload string = `{"source_time":"2024-03-26", "level":"info","message":"Application is failing","version":"1.2.0","user_id":13912,"device_id":4138,"session_id":"abc","os":"Windows","host":"112.168.1.110","location":"ngeuprqhynuvpxgp","request_body":"rnkmffyawtdcindtrdqruyxbndbjpfsptzpwtujbmkwcqastmxwbvjwphmyvpnhordwljnodxhtvpjesjldtifswqbpyuhlcytmm","status_code":300,"app_meta":"ckgpibhmlusqqfunnpxbfxbc", "new_field_added_by":"ingestor 8020"}`
req, _ := client.NewRequest("POST", "ingest", bytes.NewBufferString(test_payload))
req.Header.Add("X-P-Stream", stream)
@@ -178,7 +202,7 @@ func IngestOneEventWithTimePartition_IncorrectDateTimeFormatTimePartitionInLog(t
require.Equalf(t, 400, response.StatusCode, "Server returned http code: %s resp %s", response.Status, readAsString(response.Body))
}
-func IngestOneEventForStaticSchemaStream_NewFieldInLog(t *testing.T, client HTTPClient, stream string) {
+func IngestOneEventForStaticSchemaStream_NewFieldInLog(t *testing.T, client httpclient.HTTPClient, stream string) {
var test_payload string = `{"source_time":"2024-03-26", "level":"info","message":"Application is failing","version":"1.2.0","user_id":13912,"device_id":4138,"session_id":"abc","os":"Windows","host":"112.168.1.110","location":"ngeuprqhynuvpxgp","request_body":"rnkmffyawtdcindtrdqruyxbndbjpfsptzpwtujbmkwcqastmxwbvjwphmyvpnhordwljnodxhtvpjesjldtifswqbpyuhlcytmm","status_code":300,"app_meta":"ckgpibhmlusqqfunnpxbfxbc", "new_field_added_by":"ingestor 8020"}`
req, _ := client.NewRequest("POST", "ingest", bytes.NewBufferString(test_payload))
req.Header.Add("X-P-Stream", stream)
@@ -187,7 +211,7 @@ func IngestOneEventForStaticSchemaStream_NewFieldInLog(t *testing.T, client HTTP
require.Equalf(t, 400, response.StatusCode, "Server returned http code: %s resp %s", response.Status, readAsString(response.Body))
}
-func IngestOneEventForStaticSchemaStream_SameFieldsInLog(t *testing.T, client HTTPClient, stream string) {
+func IngestOneEventForStaticSchemaStream_SameFieldsInLog(t *testing.T, client httpclient.HTTPClient, stream string) {
var test_payload string = `{"source_time":"2024-03-26", "level":"info","message":"Application is failing","version":"1.2.0","user_id":13912,"device_id":4138,"session_id":"abc","os":"Windows","host":"112.168.1.110","location":"ngeuprqhynuvpxgp","request_body":"rnkmffyawtdcindtrdqruyxbndbjpfsptzpwtujbmkwcqastmxwbvjwphmyvpnhordwljnodxhtvpjesjldtifswqbpyuhlcytmm","status_code":300,"app_meta":"ckgpibhmlusqqfunnpxbfxbc"}`
req, _ := client.NewRequest("POST", "ingest", bytes.NewBufferString(test_payload))
req.Header.Add("X-P-Stream", stream)
@@ -196,68 +220,57 @@ func IngestOneEventForStaticSchemaStream_SameFieldsInLog(t *testing.T, client HT
require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s resp %s", response.Status, readAsString(response.Body))
}
-func QueryLogStreamCount(t *testing.T, client HTTPClient, stream string, count uint64) {
+func runSQLWithPB(t *testing.T, client pb.PBClient, query, startTime, endTime string, output any) {
+ t.Helper()
+ result, err := client.RunJSON(
+ context.Background(),
+ output,
+ "sql", "run", query,
+ "--from", startTime,
+ "--to", endTime,
+ )
+ require.NoErrorf(t, err, "pb sql run failed (exit=%d, stdout=%q, stderr=%q)", result.ExitCode, result.Stdout, result.Stderr)
+}
+
+type PBCountRow struct {
+ Count uint64 `json:"count"`
+}
+
+func QueryLogStreamCount(t *testing.T, client pb.PBClient, stream string, count uint64) {
// Query last 30 minutes of data only
endTime := time.Now().Add(time.Second).Format(time.RFC3339Nano)
startTime := time.Now().Add(-30 * time.Minute).Format(time.RFC3339Nano)
- query := map[string]interface{}{
- "query": "select count(*) as count from " + stream,
- "startTime": startTime,
- "endTime": endTime,
- }
- queryJSON, _ := json.Marshal(query)
- req, _ := client.NewRequest("POST", "query", bytes.NewBuffer(queryJSON))
- response, err := client.Do(req)
- require.NoErrorf(t, err, "Request failed: %s", err)
- body := readAsString(response.Body)
- require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s and response: %s", response.Status, body)
- expected := fmt.Sprintf(`[{"count":%d}]`, count)
- require.Equalf(t, expected, body, "Query count incorrect; Expected %s, Actual %s", expected, body)
+ query := "select count(*) as count from " + stream
+ var rows []PBCountRow
+ runSQLWithPB(t, client, query, startTime, endTime, &rows)
+ require.Equalf(t, []PBCountRow{{Count: count}}, rows, "Query count incorrect; Expected %d, Actual %v", count, rows)
}
-func QueryLogStreamCount_Historical(t *testing.T, client HTTPClient, stream string, count uint64) {
+func QueryLogStreamCount_Historical(t *testing.T, client pb.PBClient, stream string, count uint64) {
// Query last 30 minutes of data only
now := time.Now()
startTime := now.AddDate(0, 0, -33).Format(time.RFC3339Nano)
endTime := now.AddDate(0, 0, -27).Format(time.RFC3339Nano)
- query := map[string]interface{}{
- "query": "select count(*) as count from " + stream,
- "startTime": startTime,
- "endTime": endTime,
- }
- queryJSON, _ := json.Marshal(query)
- req, _ := client.NewRequest("POST", "query", bytes.NewBuffer(queryJSON))
- response, err := client.Do(req)
- require.NoErrorf(t, err, "Request failed: %s", err)
- body := readAsString(response.Body)
- require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s and response: %s", response.Status, body)
- expected := fmt.Sprintf(`[{"count":%d}]`, count)
- require.Equalf(t, expected, body, "Query count incorrect; Expected %s, Actual %s", expected, body)
+ query := "select count(*) as count from " + stream
+ var rows []PBCountRow
+ runSQLWithPB(t, client, query, startTime, endTime, &rows)
+ require.Equalf(t, []PBCountRow{{Count: count}}, rows, "Query count incorrect; Expected %d, Actual %v", count, rows)
}
-func QueryTwoLogStreamCount(t *testing.T, client HTTPClient, stream1 string, stream2 string, count uint64) {
+func QueryTwoLogStreamCount(t *testing.T, client pb.PBClient, stream1 string, stream2 string, count uint64) {
// Query last 30 minutes of data only
endTime := time.Now().Add(time.Second).Format(time.RFC3339Nano)
startTime := time.Now().Add(-30 * time.Minute).Format(time.RFC3339Nano)
- query := map[string]interface{}{
- "query": fmt.Sprintf("select sum(c) as count from (select count(*) as c from %s union all select count(*) as c from %s)", stream1, stream2),
- "startTime": startTime,
- "endTime": endTime,
- }
- queryJSON, _ := json.Marshal(query)
- req, _ := client.NewRequest("POST", "query", bytes.NewBuffer(queryJSON))
- response, err := client.Do(req)
- require.NoErrorf(t, err, "Request failed: %s", err)
- body := readAsString(response.Body)
- require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s and response: %s", response.Status, body)
- expected := fmt.Sprintf(`[{"count":%d}]`, count)
- require.Equalf(t, expected, body, "Query count incorrect; Expected %s, Actual %s", expected, body)
+ query := fmt.Sprintf("select sum(c) as count from (select count(*) as c from %s union all select count(*) as c from %s)", stream1, stream2)
+ var rows []PBCountRow
+ runSQLWithPB(t, client, query, startTime, endTime, &rows)
+ require.Equalf(t, []PBCountRow{{Count: count}}, rows, "Query count incorrect; Expected %d, Actual %v", count, rows)
}
-func AssertQueryOK(t *testing.T, client HTTPClient, query string, args ...any) {
+func AssertQueryOK(t *testing.T, client pb.PBClient, query string, args ...any) {
// Query last 30 minutes of data only
endTime := time.Now().Add(time.Second).Format(time.RFC3339Nano)
startTime := time.Now().Add(-30 * time.Minute).Format(time.RFC3339Nano)
@@ -269,20 +282,11 @@ func AssertQueryOK(t *testing.T, client HTTPClient, query string, args ...any) {
finalQuery = fmt.Sprintf(query, args...)
}
- queryJSON, _ := json.Marshal(map[string]interface{}{
- "query": finalQuery,
- "startTime": startTime,
- "endTime": endTime,
- })
-
- req, _ := client.NewRequest("POST", "query", bytes.NewBuffer(queryJSON))
- response, err := client.Do(req)
- require.NoErrorf(t, err, "Request failed: %s", err)
- body := readAsString(response.Body)
- require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s and response: %s", response.Status, body)
+ var rows []json.RawMessage
+ runSQLWithPB(t, client, finalQuery, startTime, endTime, &rows)
}
-func AssertStreamSchema(t *testing.T, client HTTPClient, stream string, schema string) {
+func AssertStreamSchema(t *testing.T, client httpclient.HTTPClient, stream string, schema string) {
req, _ := client.NewRequest("GET", "logstream/"+stream+"/schema", nil)
response, err := client.Do(req)
require.NoErrorf(t, err, "Request failed: %s", err)
@@ -291,14 +295,14 @@ func AssertStreamSchema(t *testing.T, client HTTPClient, stream string, schema s
require.JSONEq(t, schema, body, "Get schema response doesn't match with expected schema")
}
-func CreateRole(t *testing.T, client HTTPClient, name string, role string) {
+func CreateRole(t *testing.T, client httpclient.HTTPClient, name string, role string) {
req, _ := client.NewRequest("PUT", "role/"+name, strings.NewReader(role))
response, err := client.Do(req)
require.NoErrorf(t, err, "Request failed: %s", err)
require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s and response: %s", response.Status, readAsString(response.Body))
}
-func AssertRole(t *testing.T, client HTTPClient, name string, role string) {
+func AssertRole(t *testing.T, client httpclient.HTTPClient, name string, role string) {
req, _ := client.NewRequest("GET", "role/"+name, nil)
response, err := client.Do(req)
require.NoErrorf(t, err, "Request failed: %s", err)
@@ -307,34 +311,16 @@ func AssertRole(t *testing.T, client HTTPClient, name string, role string) {
require.JSONEq(t, role, body, "Get role response doesn't match with retention config returned")
}
-func CreateUser(t *testing.T, client HTTPClient, user string) string {
- req, _ := client.NewRequest("POST", "user/"+user, nil)
- response, err := client.Do(req)
- require.NoErrorf(t, err, "Request failed: %s", err)
- body := readAsString(response.Body)
- require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s resp %s", response.Status, readAsString(response.Body))
- return body
+func CreateUserWithRole(t *testing.T, client pb.PBClient, user string, roles []string) string {
+ t.Helper()
+ result, err := client.Run(context.Background(), "user", "add", user, "--role", strings.Join(roles, ","))
+ require.NoErrorf(t, err, "pb user add failed (exit=%d, stdout=%q, stderr=%q)", result.ExitCode, result.Stdout, result.Stderr)
+ password, err := pb.PasswordFromUserAddOutput(result.Stdout)
+ require.NoErrorf(t, err, "pb user add returned no password (stdout=%q, stderr=%q)", result.Stdout, result.Stderr)
+ return password
}
-func CreateUserWithRole(t *testing.T, client HTTPClient, user string, roles []string) string {
- payload, _ := json.Marshal(roles)
- req, _ := client.NewRequest("POST", "user/"+user, bytes.NewBuffer(payload))
- response, err := client.Do(req)
- require.NoErrorf(t, err, "Request failed: %s", err)
- body := readAsString(response.Body)
- require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s and response: %s", response.Status, body)
- return body
-}
-
-func AssignRolesToUser(t *testing.T, client HTTPClient, user string, roles []string) {
- payload, _ := json.Marshal(roles)
- req, _ := client.NewRequest("PUT", "user/"+user+"/role", bytes.NewBuffer(payload))
- response, err := client.Do(req)
- require.NoErrorf(t, err, "Request failed: %s", err)
- require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s and response: %s", response.Status, readAsString(response.Body))
-}
-
-func AssertUserRole(t *testing.T, client HTTPClient, user string, roleName, roleBody string) {
+func AssertUserRole(t *testing.T, client httpclient.HTTPClient, user string, roleName, roleBody string) {
req, _ := client.NewRequest("GET", "user/"+user+"/role", nil)
response, err := client.Do(req)
require.NoErrorf(t, err, "Request failed: %s", err)
@@ -344,7 +330,7 @@ func AssertUserRole(t *testing.T, client HTTPClient, user string, roleName, role
require.JSONEq(t, userRoleBody, expectedRoleBody, "Get user role response doesn't match with expected role")
}
-func RegenPassword(t *testing.T, client HTTPClient, user string) string {
+func RegenPassword(t *testing.T, client httpclient.HTTPClient, user string) string {
req, _ := client.NewRequest("POST", "user/"+user+"/generate-new-password", nil)
response, err := client.Do(req)
require.NoErrorf(t, err, "Request failed: %s", err)
@@ -353,68 +339,19 @@ func RegenPassword(t *testing.T, client HTTPClient, user string) string {
return body
}
-func SetUserRole(t *testing.T, client HTTPClient, user string, roles []string) {
- payload, _ := json.Marshal(roles)
- req, _ := client.NewRequest("PUT", "user/"+user+"/role", bytes.NewBuffer(payload))
- response, err := client.Do(req)
- require.NoErrorf(t, err, "Request failed: %s", err)
- require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s and response: %s", response.Status, readAsString(response.Body))
-}
-
-func DeleteUser(t *testing.T, client HTTPClient, user string) {
- req, _ := client.NewRequest("DELETE", "user/"+user, nil)
- response, err := client.Do(req)
- require.NoErrorf(t, err, "Request failed: %s", err)
- require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s and response: %s", response.Status, readAsString(response.Body))
+func DeleteUser(t *testing.T, client pb.PBClient, user string) {
+ t.Helper()
+ result, err := client.Run(context.Background(), "user", "remove", user)
+ require.NoErrorf(t, err, "pb user remove failed (exit=%d, stdout=%q, stderr=%q)", result.ExitCode, result.Stdout, result.Stderr)
}
-func DeleteRole(t *testing.T, client HTTPClient, roleName string) {
- req, _ := client.NewRequest("DELETE", "role/"+roleName, nil)
- response, err := client.Do(req)
- require.NoErrorf(t, err, "Request failed: %s", err)
- require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s and response: %s", response.Status, readAsString(response.Body))
-}
-
-func SetDefaultRole(t *testing.T, client HTTPClient, roleName string) {
- payload, _ := json.Marshal(roleName)
- req, _ := client.NewRequest("PUT", "role/default", bytes.NewBuffer(payload))
- response, err := client.Do(req)
- require.NoErrorf(t, err, "Request failed: %s", err)
- require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s and response: %s", response.Status, readAsString(response.Body))
-}
-
-func AssertDefaultRole(t *testing.T, client HTTPClient, roleName string) {
- req, _ := client.NewRequest("GET", "role/default", nil)
- response, err := client.Do(req)
- require.NoErrorf(t, err, "Request failed: %s", err)
- body := readAsString(response.Body)
- require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s and response: %s", response.Status, body)
- require.Equalf(t, roleName, body, "Get default role response doesn't match with expected role")
-}
-
-func PutSingleEventExpectErr(t *testing.T, client HTTPClient, stream string) {
- payload := `{
- "id": "id;objectId",
- "maxRunDistance": "float;1;20;1",
- "cpf": "cpf",
- "cnpj": "cnpj",
- "pretendSalary": "money",
- "age": "int;20;80",
- "gender": "gender",
- "firstName": "firstName",
- "lastName": "lastName",
- "phone": "maskInt;+55 (83) 9####-####",
- "address": "address",
- "hairColor": "color"
- }`
- req, _ := client.NewRequest("POST", "logstream/"+stream, bytes.NewBufferString(payload))
- response, err := client.Do(req)
-
- require.NoErrorf(t, err, "Request failed when expected to pass: %s", err)
- require.Equalf(t, 403, response.StatusCode, "Server returned http code: %s and response: %s", response.Status, readAsString(response.Body))
+func DeleteRole(t *testing.T, client pb.PBClient, roleName string) {
+ t.Helper()
+ result, err := client.Run(context.Background(), "role", "remove", roleName)
+ require.NoErrorf(t, err, "pb role remove failed (exit=%d, stdout=%q, stderr=%q)", result.ExitCode, result.Stdout, result.Stderr)
}
-func PutSingleEvent(t *testing.T, client HTTPClient, stream string) {
+func PutSingleEvent(t *testing.T, client httpclient.HTTPClient, stream string) {
payload := `{
"id": "id;objectId",
"maxRunDistance": "float;1;20;1",
@@ -436,7 +373,7 @@ func PutSingleEvent(t *testing.T, client HTTPClient, stream string) {
require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s and response: %s", response.Status, readAsString(response.Body))
}
-func checkAPIAccess(t *testing.T, queryClient HTTPClient, ingestClient HTTPClient, stream string, role string) {
+func checkAPIAccess(t *testing.T, queryClient httpclient.HTTPClient, ingestClient httpclient.HTTPClient, stream string, role string) {
switch role {
case "editor":
// Check access to non-protected API