chore(querybuilder): group the request validation test imports

Assisted-by: Claude Opus 5.5

pkg/alertmanager/alertmanagerserver/distpatcher_test.go

@@ -28,7 +28,6 @@ import (
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/common/promslog"
 
-	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
@@ -111,44 +110,42 @@ func TestAggrGroup(t *testing.T) {
 		}
 	)
 
-	type notification struct {
-		alerts     alertmanagertypes.AlertSlice
-		notifiedAt time.Time
-	}
-	alertsCh := make(chan notification)
+	var (
+		last       = time.Now()
+		current    = time.Now()
+		lastCurMtx = &sync.Mutex{}
+		alertsCh   = make(chan alertmanagertypes.AlertSlice)
+	)
 
 	ntfy := func(ctx context.Context, alerts ...*alertmanagertypes.Alert) bool {
 		// Validate that the context is properly populated.
-		notifiedAt, ok := notify.Now(ctx)
-		assert.True(t, ok, "now missing")
-		_, ok = notify.GroupKey(ctx)
-		assert.True(t, ok, "group key missing")
-		lbls, ok := notify.GroupLabels(ctx)
-		if assert.True(t, ok, "group labels missing") {
-			assert.Equal(t, lset, lbls, "wrong group labels")
+		if _, ok := notify.Now(ctx); !ok {
+			t.Errorf("now missing")
 		}
-		rcv, ok := notify.ReceiverName(ctx)
-		if assert.True(t, ok, "receiver missing") {
-			assert.Equal(t, opts.Receiver, rcv, "wrong receiver")
+		if _, ok := notify.GroupKey(ctx); !ok {
+			t.Errorf("group key missing")
 		}
-		ri, ok := notify.RepeatInterval(ctx)
-		if assert.True(t, ok, "repeat interval missing") {
-			assert.Equal(t, notificationConfig.Renotify.RenotifyInterval, ri, "wrong repeat interval")
+		if lbls, ok := notify.GroupLabels(ctx); !ok || !reflect.DeepEqual(lbls, lset) {
+			t.Errorf("wrong group labels: %q", lbls)
+		}
+		if rcv, ok := notify.ReceiverName(ctx); !ok || rcv != opts.Receiver {
+			t.Errorf("wrong receiver: %q", rcv)
+		}
+		if ri, ok := notify.RepeatInterval(ctx); !ok || ri != notificationConfig.Renotify.RenotifyInterval {
+			t.Errorf("wrong repeat interval: %q", ri)
 		}
 
-		alertsCh <- notification{
-			alerts:     alertmanagertypes.AlertSlice(alerts),
-			notifiedAt: notifiedAt,
-		}
+		lastCurMtx.Lock()
+		last = current
+		// Subtract a millisecond to allow for races.
+		current = time.Now().Add(-time.Millisecond)
+		lastCurMtx.Unlock()
+
+		alertsCh <- alertmanagertypes.AlertSlice(alerts)
 
 		return true
 	}
 
-	assertNotifiedAfter := func(previous, current time.Time, interval time.Duration) {
-		t.Helper()
-		require.GreaterOrEqual(t, current.Sub(previous), interval, "received batch too early")
-	}
-
 	removeEndsAt := func(as alertmanagertypes.AlertSlice) alertmanagertypes.AlertSlice {
 		for i, a := range as {
 			ac := *a
@@ -159,26 +156,29 @@ func TestAggrGroup(t *testing.T) {
 	}
 
 	// Test regular situation where we wait for group_wait to send out alerts.
-	groupStartedAt := time.Now()
 	ag := newAggrGroup(context.Background(), lset, route, nil, promslog.NewNopLogger(), notificationConfig.Renotify.RenotifyInterval)
 
 	go ag.run(ntfy)
 
 	ag.insert(a1)
-	var lastNotificationAt time.Time
 
 	select {
 	case <-time.After(2 * opts.GroupWait):
-		require.FailNow(t, "expected initial batch after group_wait")
+		t.Fatalf("expected initial batch after group_wait")
 
-	case notification := <-alertsCh:
-		assertNotifiedAfter(groupStartedAt, notification.notifiedAt, opts.GroupWait)
-		lastNotificationAt = notification.notifiedAt
-		batch := notification.alerts
+	case batch := <-alertsCh:
+		lastCurMtx.Lock()
+		s := time.Since(last)
+		lastCurMtx.Unlock()
+		if s < opts.GroupWait {
+			t.Fatalf("received batch too early after %v", s)
+		}
 		exp := removeEndsAt(alertmanagertypes.AlertSlice{a1})
 		sort.Sort(batch)
 
-		require.Equal(t, exp, batch)
+		if !reflect.DeepEqual(batch, exp) {
+			t.Fatalf("expected alerts %v but got %v", exp, batch)
+		}
 	}
 
 	for i := 0; i < 3; i++ {
@@ -187,16 +187,21 @@ func TestAggrGroup(t *testing.T) {
 
 		select {
 		case <-time.After(2 * opts.GroupInterval):
-			require.FailNow(t, "expected new batch after group interval but received none")
+			t.Fatalf("expected new batch after group interval but received none")
 
-		case notification := <-alertsCh:
-			assertNotifiedAfter(lastNotificationAt, notification.notifiedAt, opts.GroupInterval)
-			lastNotificationAt = notification.notifiedAt
-			batch := notification.alerts
+		case batch := <-alertsCh:
+			lastCurMtx.Lock()
+			s := time.Since(last)
+			lastCurMtx.Unlock()
+			if s < opts.GroupInterval {
+				t.Fatalf("received batch too early after %v", s)
+			}
 			exp := removeEndsAt(alertmanagertypes.AlertSlice{a1, a3})
 			sort.Sort(batch)
 
-			require.Equal(t, exp, batch)
+			if !reflect.DeepEqual(batch, exp) {
+				t.Fatalf("expected alerts %v but got %v", exp, batch)
+			}
 		}
 	}
 
@@ -215,15 +220,15 @@ func TestAggrGroup(t *testing.T) {
 	// a2 lies way in the past so the initial group_wait should be skipped.
 	select {
 	case <-time.After(opts.GroupWait / 2):
-		require.FailNow(t, "expected immediate alert but received none")
+		t.Fatalf("expected immediate alert but received none")
 
-	case notification := <-alertsCh:
-		lastNotificationAt = notification.notifiedAt
-		batch := notification.alerts
+	case batch := <-alertsCh:
 		exp := removeEndsAt(alertmanagertypes.AlertSlice{a1, a2})
 		sort.Sort(batch)
 
-		require.Equal(t, exp, batch)
+		if !reflect.DeepEqual(batch, exp) {
+			t.Fatalf("expected alerts %v but got %v", exp, batch)
+		}
 	}
 
 	for i := 0; i < 3; i++ {
@@ -232,16 +237,21 @@ func TestAggrGroup(t *testing.T) {
 
 		select {
 		case <-time.After(2 * opts.GroupInterval):
-			require.FailNow(t, "expected new batch after group interval but received none")
+			t.Fatalf("expected new batch after group interval but received none")
 
-		case notification := <-alertsCh:
-			assertNotifiedAfter(lastNotificationAt, notification.notifiedAt, opts.GroupInterval)
-			lastNotificationAt = notification.notifiedAt
-			batch := notification.alerts
+		case batch := <-alertsCh:
+			lastCurMtx.Lock()
+			s := time.Since(last)
+			lastCurMtx.Unlock()
+			if s < opts.GroupInterval {
+				t.Fatalf("received batch too early after %v", s)
+			}
 			exp := removeEndsAt(alertmanagertypes.AlertSlice{a1, a2, a3})
 			sort.Sort(batch)
 
-			require.Equal(t, exp, batch)
+			if !reflect.DeepEqual(batch, exp) {
+				t.Fatalf("expected alerts %v but got %v", exp, batch)
+			}
 		}
 	}
 
@@ -253,14 +263,19 @@ func TestAggrGroup(t *testing.T) {
 
 	select {
 	case <-time.After(2 * opts.GroupInterval):
-		require.FailNow(t, "expected new batch after group interval but received none")
-	case notification := <-alertsCh:
-		assertNotifiedAfter(lastNotificationAt, notification.notifiedAt, opts.GroupInterval)
-		lastNotificationAt = notification.notifiedAt
-		batch := notification.alerts
+		t.Fatalf("expected new batch after group interval but received none")
+	case batch := <-alertsCh:
+		lastCurMtx.Lock()
+		s := time.Since(last)
+		lastCurMtx.Unlock()
+		if s < opts.GroupInterval {
+			t.Fatalf("received batch too early after %v", s)
+		}
 		sort.Sort(batch)
 
-		require.Equal(t, exp, batch)
+		if !reflect.DeepEqual(batch, exp) {
+			t.Fatalf("expected alerts %v but got %v", exp, batch)
+		}
 	}
 
 	// Resolve all remaining alerts, they should be removed after the next batch was sent.
@@ -274,16 +289,24 @@ func TestAggrGroup(t *testing.T) {
 
 	select {
 	case <-time.After(2 * opts.GroupInterval):
-		require.FailNow(t, "expected new batch after group interval but received none")
+		t.Fatalf("expected new batch after group interval but received none")
 
-	case notification := <-alertsCh:
-		assertNotifiedAfter(lastNotificationAt, notification.notifiedAt, opts.GroupInterval)
-		batch := notification.alerts
+	case batch := <-alertsCh:
+		lastCurMtx.Lock()
+		s := time.Since(last)
+		lastCurMtx.Unlock()
+		if s < opts.GroupInterval {
+			t.Fatalf("received batch too early after %v", s)
+		}
 		sort.Sort(batch)
 
-		require.Equal(t, resolved, batch)
+		if !reflect.DeepEqual(batch, resolved) {
+			t.Fatalf("expected alerts %v but got %v", resolved, batch)
+		}
 
-		require.Eventually(t, ag.empty, 2*opts.GroupInterval, 10*time.Millisecond, "expected aggregation group to be empty after resolving alerts: %v", ag)
+		if !ag.empty() {
+			t.Fatalf("Expected aggregation group to be empty after resolving alerts: %v", ag)
+		}
 	}
 
 	ag.stop()

pkg/errors/suggestions.go

@@ -9,12 +9,13 @@ import (
 // NewValidReferences / NewSuggestionsOnLevenshteinDistance to phrase the
 // "valid <noun> are ..." list consistently across the codebase.
 const (
-	NounFields     = "fields"
-	NounKeys       = "keys"
-	NounServices   = "services"
-	NounQueryTypes = "query types"
-	NounSignals    = "signals"
-	NounReferences = "references"
+	NounFields       = "fields"
+	NounKeys         = "keys"
+	NounServices     = "services"
+	NounQueryTypes   = "query types"
+	NounRequestTypes = "request types"
+	NounSignals      = "signals"
+	NounReferences   = "references"
 )
 
 const (

pkg/types/querybuildertypes/querybuildertypesv5/formula.go

@@ -118,6 +118,10 @@ func (f QueryBuilderFormula) Validate() error {
 		)
 	}
 
+	if f.Limit < 0 {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "limit must be non-negative, got %d", f.Limit)
+	}
+
 	// Validate functions if present
 	for i, fn := range f.Functions {
 		if err := fn.Validate(); err != nil {

pkg/types/querybuildertypes/querybuildertypesv5/request_type.go

@@ -23,7 +23,8 @@ func (r *RequestType) UnmarshalJSON(data []byte) error {
 		*r = v
 		return nil
 	default:
-		return errors.NewInvalidInputf(errors.CodeInvalidInput, "unknown request type %q; allowed values: %s", s, "`scalar`, `time_series`, `raw`, `raw_stream`, `trace`, `distribution`, `heatmap`")
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "unknown request type %q", s).
+			WithSuggestions(errors.NewValidReferences(errors.NounRequestTypes, RequestType{}.Enum()...))
 	}
 }
 

pkg/types/querybuildertypes/querybuildertypesv5/validation.go

@@ -499,6 +499,18 @@ func (q *QueryBuilderQuery[T]) validateOrderBy(cfg validationConfig) error {
 		return q.validateOrderByForAggregation()
 	}
 
+	// a raw order key resolves like a select field, so only an empty one is
+	// invalid before resolution
+	for i, order := range q.Order {
+		if order.Key.Name == "" {
+			orderID := fmt.Sprintf("order by clause #%d", i+1)
+			if q.Name != "" {
+				orderID = fmt.Sprintf("order by clause #%d in query '%s'", i+1, q.Name)
+			}
+			return errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid empty key name for %s", orderID)
+		}
+	}
+
 	return nil
 }
 
@@ -586,19 +598,10 @@ func (r *QueryRangeRequest) Validate(opts ...ValidationOption) error {
 		return err
 	}
 
-	// Validate request type
-	switch r.RequestType {
-	case RequestTypeRaw, RequestTypeRawStream, RequestTypeTrace, RequestTypeTimeSeries, RequestTypeScalar, RequestTypeHeatmap, RequestTypeDistribution:
-		opts = append(opts, GetValidationOptions(r.RequestType)...)
-	default:
-		return errors.NewInvalidInputf(
-			errors.CodeInvalidInput,
-			"invalid request type: %s",
-			r.RequestType,
-		).WithAdditional(
-			"Valid request types are: raw, timeseries, scalar, heatmap, distribution",
-		)
+	if err := r.validateRequestType(); err != nil {
+		return err
 	}
+	opts = append(opts, GetValidationOptions(r.RequestType)...)
 
 	if err := r.validateBucketCounting(); err != nil {
 		return err
@@ -639,14 +642,10 @@ func (r *QueryRangeRequest) ValidateRequestScope() ([]ValidationOption, error) {
 		return nil, err
 	}
 
-	var opts []ValidationOption
-	switch r.RequestType {
-	case RequestTypeRaw, RequestTypeRawStream, RequestTypeTrace, RequestTypeTimeSeries, RequestTypeScalar, RequestTypeHeatmap, RequestTypeDistribution:
-		opts = GetValidationOptions(r.RequestType)
-	default:
-		return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid request type: %s", r.RequestType).
-			WithAdditional("Valid request types are: raw, timeseries, scalar, heatmap, distribution")
+	if err := r.validateRequestType(); err != nil {
+		return nil, err
 	}
+	opts := GetValidationOptions(r.RequestType)
 
 	if err := r.validateBucketCounting(); err != nil {
 		return nil, err
@@ -703,6 +702,20 @@ func (e QueryEnvelope) Validate(opts ...ValidationOption) error {
 	return validateQueryEnvelope(e, opts...)
 }
 
+// Enum is the one list of request types a query can use, so the spec and the
+// suggestions agree with what is accepted.
+func (r *QueryRangeRequest) validateRequestType() error {
+	if slices.Contains(RequestType{}.Enum(), any(r.RequestType)) {
+		return nil
+	}
+	if r.RequestType == RequestTypeUnknown {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "requestType is required").
+			WithSuggestions(errors.NewValidReferences(errors.NounRequestTypes, RequestType{}.Enum()...))
+	}
+	return errors.NewInvalidInputf(errors.CodeInvalidInput, "request type %q is not supported", r.RequestType.StringValue()).
+		WithSuggestions(errors.NewValidReferences(errors.NounRequestTypes, RequestType{}.Enum()...))
+}
+
 func (r *QueryRangeRequest) validateTimeRange() error {
 	if r.RequestType == RequestTypeRawStream {
 		return nil

pkg/types/querybuildertypes/querybuildertypesv5/validation_test.go

@@ -1,9 +1,15 @@
 package querybuildertypesv5
 
 import (
+	"encoding/json"
+	"fmt"
 	"strings"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/types/metrictypes"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 	"github.com/SigNoz/signoz/pkg/valuer"
@@ -521,6 +527,28 @@ func TestQueryRangeRequest_ValidateCompositeQuery(t *testing.T) {
 			wantErr: true,
 			errMsg:  "expression cannot be blank",
 		},
+		{
+			name: "Formula_NegativeLimit_Rejected",
+			request: QueryRangeRequest{
+				Start:       1640995200000,
+				End:         1640998800000,
+				RequestType: RequestTypeTimeSeries,
+				CompositeQuery: CompositeQuery{
+					Queries: []QueryEnvelope{
+						{
+							Type: QueryTypeFormula,
+							Spec: QueryBuilderFormula{
+								Name:       "F1",
+								Expression: "A * 2",
+								Limit:      -5,
+							},
+						},
+					},
+				},
+			},
+			wantErr: true,
+			errMsg:  "invalid formula 'F1': limit must be non-negative, got -5",
+		},
 		{
 			name: "promql with empty query should return error",
 			request: QueryRangeRequest{
@@ -696,6 +724,31 @@ func TestQueryRangeRequest_ValidateCompositeQuery(t *testing.T) {
 			wantErr: true,
 			errMsg:  "raw request type is not supported for metric queries",
 		},
+		{
+			name: "Raw_EmptyOrderKey_Rejected",
+			request: QueryRangeRequest{
+				Start:       1640995200000,
+				End:         1640998800000,
+				RequestType: RequestTypeRaw,
+				CompositeQuery: CompositeQuery{
+					Queries: []QueryEnvelope{
+						{
+							Type: QueryTypeBuilder,
+							Spec: QueryBuilderQuery[LogAggregation]{
+								Name:   "B",
+								Signal: telemetrytypes.SignalLogs,
+								Order: []OrderBy{
+									{Key: OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "timestamp"}}, Direction: OrderDirectionDesc},
+									{Key: OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: ""}}, Direction: OrderDirectionAsc},
+								},
+							},
+						},
+					},
+				},
+			},
+			wantErr: true,
+			errMsg:  "invalid query 'B': invalid empty key name for order by clause #2 in query 'B'",
+		},
 		{
 			name: "timeseries request with group by timestamp should return error",
 			request: QueryRangeRequest{
@@ -1407,6 +1460,44 @@ func TestRequestType_IsAggregation(t *testing.T) {
 	}
 }
 
+func TestQueryRangeRequest_ValidateRequestType(t *testing.T) {
+	validRequestTypes := []string{"valid request types are `scalar`, `time_series`, `raw`, `raw_stream`, `trace`, `heatmap`, `distribution`"}
+
+	testCases := []struct {
+		name        string
+		requestType string
+		message     string
+	}{
+		{
+			name:        "Missing_Required",
+			requestType: "",
+			message:     "requestType is required",
+		},
+		{
+			name:        "Timeseries_Unknown",
+			requestType: `"requestType": "timeseries",`,
+			message:     `unknown request type "timeseries"`,
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			body := fmt.Sprintf(`{"schemaVersion": "v1", "start": 1640995200000, "end": 1640998800000, %s "compositeQuery": {"queries": [{"type": "builder_query", "spec": {"name": "A", "signal": "logs"}}]}}`, testCase.requestType)
+
+			var request QueryRangeRequest
+			err := json.Unmarshal([]byte(body), &request)
+			if err == nil {
+				err = request.Validate()
+			}
+			require.Error(t, err)
+
+			got := errors.AsJSON(err)
+			assert.Equal(t, testCase.message, got.Message)
+			assert.Equal(t, validRequestTypes, got.Suggestions)
+		})
+	}
+}
+
 func TestNonAggregationFieldsSkipped(t *testing.T) {
 	// Fields that only apply to aggregation queries (groupBy, having, aggregations)
 	// should be silently skipped for non-aggregation request types.

tests/integration/tests/queriercommon/13_cache_running_diff.py

@@ -1,69 +0,0 @@
-from collections.abc import Callable
-from datetime import UTC, datetime, timedelta
-from http import HTTPStatus
-from uuid import uuid4
-
-from fixtures import types
-from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
-from fixtures.metrics import Metrics
-from fixtures.querier import (
-    build_builder_query,
-    build_function,
-    get_all_series,
-    index_series_by_label,
-    make_query_request,
-)
-
-MINUTE = timedelta(minutes=1)
-
-
-def test_running_diff_full_hit_matches_no_cache(
-    signoz: types.SigNoz,
-    create_user_admin: None,  # pylint: disable=unused-argument
-    get_token: Callable[[str, str], str],
-    insert_metrics: Callable[[list[Metrics]], None],
-) -> None:
-    """
-    Setup:
-    A gauge that grows by 10 every minute over four minutes.
-
-    Tests:
-    runningDiff over the last three of those minutes, requested twice. The
-    metrics statement reads one step before the window so the first
-    interval has a difference. The second request, a full cache hit, must
-    return the same three differences as the request with noCache.
-    """
-    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
-    # the cache outlives the run, so a fixed name would serve the previous
-    # run's points back to this one
-    metric_name = f"cache_running_diff_{uuid4().hex[:8]}"
-    insert_metrics(
-        [
-            Metrics(
-                metric_name=metric_name,
-                labels={"service": "svc-a"},
-                timestamp=now - (14 - i) * MINUTE,
-                value=100.0 + 10 * i,
-                type_="Gauge",
-                is_monotonic=False,
-            )
-            for i in range(4)
-        ]
-    )
-    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
-    query = build_builder_query("A", metric_name, "avg", "avg", group_by=["service"], functions=[build_function("runningDiff")])
-    start_ms = int((now - 13 * MINUTE).timestamp() * 1000)
-    end_ms = int((now - 10 * MINUTE).timestamp() * 1000)
-
-    warm = make_query_request(signoz, token, start_ms, end_ms, [query], no_cache=False)
-    assert warm.status_code == HTTPStatus.OK, warm.text
-    cached = make_query_request(signoz, token, start_ms, end_ms, [query], no_cache=False)
-    assert cached.status_code == HTTPStatus.OK, cached.text
-    fresh = make_query_request(signoz, token, start_ms, end_ms, [query], no_cache=True)
-    assert fresh.status_code == HTTPStatus.OK, fresh.text
-
-    fresh_points = {point["timestamp"]: point["value"] for point in index_series_by_label(get_all_series(fresh.json(), "A"), "service")["svc-a"]["values"]}
-    assert sorted(fresh_points.values()) == [10, 10, 10], fresh_points
-
-    cached_points = {point["timestamp"]: point["value"] for point in index_series_by_label(get_all_series(cached.json(), "A"), "service")["svc-a"]["values"]}
-    assert cached_points == fresh_points