pdbmux/aggregate_test.go

package main

import (
	"encoding/json"
	"reflect"
	"slices"
	"testing"
)

func rows(raws ...string) []record {
	out := make([]record, 0, len(raws))
	for _, r := range raws {
		out = append(out, record{Raw: json.RawMessage(r)})
	}
	return out
}

// decodeRows turns a merged result set into comparable maps.
func decodeRows(t *testing.T, raws []json.RawMessage) []map[string]any {
	t.Helper()
	out := make([]map[string]any, 0, len(raws))
	for _, raw := range raws {
		var m map[string]any
		if err := json.Unmarshal(raw, &m); err != nil {
			t.Fatalf("unmarshal %s: %v", raw, err)
		}
		out = append(out, m)
	}
	return out
}

func TestParseAggregate_ExtractWithGroupBy(t *testing.T) {
	spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
	if spec == nil {
		t.Fatal("expected an aggregate spec")
	}
	if !slices.Equal(spec.keys, []string{"status"}) {
		t.Errorf("keys = %v, want [status]", spec.keys)
	}
	if !slices.Equal(spec.sums, []string{"count"}) {
		t.Errorf("sums = %v, want [count]", spec.sums)
	}
}

func TestParseAggregate_GroupByAddsUnextractedField(t *testing.T) {
	spec := parseAggregate(`["extract",[["function","count","certname"]],["~","certname",".*"],["group_by","status"]]`)
	if spec == nil {
		t.Fatal("expected an aggregate spec")
	}
	if !slices.Equal(spec.keys, []string{"status"}) {
		t.Errorf("keys = %v, want [status] from the group_by clause", spec.keys)
	}
	if !slices.Equal(spec.sums, []string{"count"}) {
		t.Errorf("sums = %v, want [count]", spec.sums)
	}
}

func TestParseAggregate_NoFunctionIsNotAggregate(t *testing.T) {
	for _, q := range []string{
		``,
		`["=","certname","h1"]`,
		`["extract",["certname","hash"],["=","certname","h1"]]`, // projection, still real reports
		`not json`,
		`["extract"]`,
		`{"not":"an array"}`,
	} {
		if spec := parseAggregate(q); spec != nil {
			t.Errorf("parseAggregate(%q) = %+v, want nil", q, spec)
		}
	}
}

func TestSumRows_SharedKeysAreAdded(t *testing.T) {
	spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
	merged := sumRows([]backendResult{
		{name: "a", records: rows(`{"count":3,"status":"changed"}`, `{"count":1,"status":"failed"}`)},
		{name: "b", records: rows(`{"count":4,"status":"changed"}`, `{"count":2,"status":"failed"}`)},
	}, spec.columns)

	want := []map[string]any{
		{"count": float64(7), "status": "changed"},
		{"count": float64(3), "status": "failed"},
	}
	if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) {
		t.Errorf("merged = %v, want %v", got, want)
	}
}

func TestSumRows_DisjointKeysAreKept(t *testing.T) {
	spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
	merged := sumRows([]backendResult{
		{name: "a", records: rows(`{"count":3,"status":"changed"}`)},
		{name: "b", records: rows(`{"count":2,"status":"skipped"}`)},
	}, spec.columns)

	want := []map[string]any{
		{"count": float64(3), "status": "changed"},
		{"count": float64(2), "status": "skipped"},
	}
	if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) {
		t.Errorf("merged = %v, want %v", got, want)
	}
}

func TestSumRows_SingleBackendRowKeepsUpstreamBytes(t *testing.T) {
	spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
	const raw = `{"count":3,"status":"changed","extra":{"kept":true}}`
	merged := sumRows([]backendResult{
		{name: "a", records: rows(raw)},
		{name: "b", records: nil},
	}, spec.columns)

	if len(merged) != 1 || string(merged[0]) != raw {
		t.Errorf("merged = %s, want the row verbatim %s", merged, raw)
	}
}

func TestSumRows_NonNumericAggregateColumnIsNotZeroed(t *testing.T) {
	spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
	merged := sumRows([]backendResult{
		{name: "a", records: rows(`{"count":5,"status":"changed"}`)},
		{name: "b", records: rows(`{"count":null,"status":"changed"}`)},
	}, spec.columns)

	want := []map[string]any{{"count": float64(5), "status": "changed"}}
	if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) {
		t.Errorf("merged = %v, want the numeric value preserved %v", got, want)
	}
}

func TestSumRows_MissingAggregateColumnStartsFromTheNumericRow(t *testing.T) {
	spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
	merged := sumRows([]backendResult{
		{name: "a", records: rows(`{"status":"changed"}`)},
		{name: "b", records: rows(`{"count":6,"status":"changed"}`)},
	}, spec.columns)

	want := []map[string]any{{"count": float64(6), "status": "changed"}}
	if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) {
		t.Errorf("merged = %v, want %v", got, want)
	}
}

func TestSumRows_NonObjectRowsPassThrough(t *testing.T) {
	spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
	merged := sumRows([]backendResult{
		{name: "a", records: rows(`"surprise"`)},
		{name: "b", records: rows(`{"count":1,"status":"changed"}`)},
	}, spec.columns)

	if len(merged) != 2 || string(merged[0]) != `"surprise"` {
		t.Fatalf("merged = %s, want the non-object row kept as-is", merged)
	}
}

func TestSumRows_NoKeyColumnsCollapseToOneRow(t *testing.T) {
	// ["extract",[["function","count"]],...] is a whole-estate count: one row
	// per backend, and the merged answer is their sum.
	spec := parseAggregate(`["extract",[["function","count"]],["=","certname","h1"]]`)
	if spec == nil {
		t.Fatal("expected an aggregate spec")
	}
	merged := sumRows([]backendResult{
		{name: "a", records: rows(`{"count":10}`)},
		{name: "b", records: rows(`{"count":32}`)},
	}, spec.columns)

	want := []map[string]any{{"count": float64(42)}}
	if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) {
		t.Errorf("merged = %v, want %v", got, want)
	}
}

func TestInferredColumns_SplitsCountsFromIdentity(t *testing.T) {
	var row map[string]json.RawMessage
	if err := json.Unmarshal([]byte(`{"subject_type":"certname","subject":{"title":"h1"},"failures":1,"successes":2,"skips":null}`), &row); err != nil {
		t.Fatal(err)
	}
	keys, sums := inferredColumns(row)
	if !slices.Equal(keys, []string{"subject", "subject_type"}) {
		t.Errorf("keys = %v, want [subject subject_type]", keys)
	}
	// A null count is an empty aggregate, not part of the row's identity.
	if !slices.Equal(sums, []string{"failures", "skips", "successes"}) {
		t.Errorf("sums = %v, want [failures skips successes]", sums)
	}
}

func TestSumRows_EventCountsPerSubject(t *testing.T) {
	merged := sumRows([]backendResult{
		{name: "a", records: rows(
			`{"subject_type":"certname","subject":{"title":"h1"},"failures":1,"successes":2,"noops":0,"skips":0}`,
			`{"subject_type":"certname","subject":{"title":"h2"},"failures":0,"successes":5,"noops":0,"skips":0}`,
		)},
		{name: "b", records: rows(
			`{"subject_type":"certname","subject":{"title":"h1"},"failures":3,"successes":4,"noops":1,"skips":0}`,
		)},
	}, inferredColumns)

	got := decodeRows(t, merged)
	want := []map[string]any{
		{"subject_type": "certname", "subject": map[string]any{"title": "h1"},
			"failures": float64(4), "successes": float64(6), "noops": float64(1), "skips": float64(0)},
		{"subject_type": "certname", "subject": map[string]any{"title": "h2"},
			"failures": float64(0), "successes": float64(5), "noops": float64(0), "skips": float64(0)},
	}
	if !reflect.DeepEqual(got, want) {
		t.Errorf("merged = %v, want %v", got, want)
	}
}