package e2e import ( "context" "fmt" "net/http" "strings" "testing" "time" "github.com/brianvoe/gofakeit/v6" cloudevents "github.com/cloudevents/sdk-go/v2/event" "github.com/oklog/ulid/v2" "github.com/samber/lo" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" api "github.com/openmeterio/openmeter/api/client/go" v3sdk "github.com/openmeterio/openmeter/api/v3/client" "github.com/openmeterio/openmeter/openmeter/meter" "github.com/openmeterio/openmeter/openmeter/testutils" "github.com/openmeterio/openmeter/pkg/convert" ) type customerIOTA int func (c customerIOTA) Key() string { return fmt.Sprintf("customer-%d", c) } const ( customer1 customerIOTA = iota customer2 customer3 ) func TestIngest(t *testing.T) { client := initClient(t) // ensure subject exists { resp, err := client.UpsertSubjectWithResponse(context.Background(), api.UpsertSubjectJSONRequestBody{ api.SubjectUpsert{ Key: "customer-1", }, }) require.NoError(t, err) require.Equal(t, http.StatusOK, resp.StatusCode()) } // Make clickhouse's job easier by sending events within a fix time range now := time.Now() var sum int for i := 0; i < 1000; i++ { // Make clickhouse's job easier by sending events within a fix time range timestamp := gofakeit.DateRange(now.Add(-30*24*time.Hour), now.Add(30*24*time.Hour)) duration := gofakeit.Number(1, 100) sum += duration ev := cloudevents.New() ev.SetID(gofakeit.UUID()) ev.SetSource("my-app") ev.SetType("ingest") ev.SetSubject("customer-1") ev.SetTime(timestamp) _ = ev.SetData("application/json", map[string]string{ "duration_ms": fmt.Sprintf("%d", duration), }) resp, err := client.IngestEventWithResponse(context.Background(), ev) require.NoError(t, err) require.Equal(t, http.StatusNoContent, resp.StatusCode()) } // Wait for events to be processed assert.EventuallyWithT(t, func(t *assert.CollectT) { resp, err := client.QueryMeterWithResponse(context.Background(), "ingest", nil) require.NoError(t, err) require.Equal(t, http.StatusOK, resp.StatusCode()) require.Len(t, resp.JSON200.Data, 1) assert.Equal(t, float64(sum), resp.JSON200.Data[0].Value) }, time.Minute, time.Second) } // Send an event with content type application/json. // We treat request as application/cloudevents+json if it's a single event // and treat application/cloudevents-batch+json if it's an array of events. func TestIngestContentTypeApplicationJSON(t *testing.T) { client := initClient(t) // ensure subject exists { resp, err := client.UpsertSubjectWithResponse(context.Background(), api.UpsertSubjectJSONRequestBody{ api.SubjectUpsert{ Key: "customer-1", }, }) require.NoError(t, err) require.Equal(t, http.StatusOK, resp.StatusCode()) } tm := time.Now().Add(-time.Hour).Format(time.RFC3339) eventType := "ingest_content_type_application_json" // Send a single event { payload := fmt.Sprintf(`{ "specversion" : "1.0", "id": "%s", "source": "my-app", "type": "%s", "subject": "customer-1", "time": "%s", "data": { "duration_ms": "100" } }`, ulid.Make().String(), eventType, tm) resp, err := client.IngestEventsWithBody(context.Background(), "application/json", strings.NewReader(payload)) require.NoError(t, err) require.Equal(t, http.StatusNoContent, resp.StatusCode) resp.Body.Close() } // Send an array of events { payload := fmt.Sprintf(`[ { "specversion" : "1.0", "id": "%s", "source": "my-app", "type": "%s", "subject": "customer-1", "time": "%s", "data": { "duration_ms": "100" } }, { "specversion" : "1.0", "id": "%s", "source": "my-app", "type": "%s", "subject": "customer-1", "time": "%s", "data": { "duration_ms": "100" } } ]`, ulid.Make().String(), eventType, tm, ulid.Make().String(), eventType, tm, ) resp, err := client.IngestEventsWithBody(context.Background(), "application/json", strings.NewReader(payload)) require.NoError(t, err) require.Equal(t, http.StatusNoContent, resp.StatusCode) resp.Body.Close() } // Wait for events to be processed assert.EventuallyWithT(t, func(t *assert.CollectT) { resp, err := client.QueryMeterWithResponse(context.Background(), eventType, nil) require.NoError(t, err) require.Equal(t, http.StatusOK, resp.StatusCode()) require.Len(t, resp.JSON200.Data, 1) assert.Equal(t, 300.0, resp.JSON200.Data[0].Value) }, time.Minute, time.Second) } func TestBatchIngest(t *testing.T) { client := initClient(t) // ensure subject exists { resp, err := client.UpsertSubjectWithResponse(context.Background(), api.UpsertSubjectJSONRequestBody{ api.SubjectUpsert{ Key: "customer-1", }, }) require.NoError(t, err) require.Equal(t, http.StatusOK, resp.StatusCode()) } // Make clickhouse's job easier by sending events within a fix time range now := time.Now() var sum int var events []cloudevents.Event for i := 0; i < 1000; i++ { // Make clickhouse's job easier by sending events within a fix time range timestamp := gofakeit.DateRange(now.Add(-30*24*time.Hour), now.Add(30*24*time.Hour)) duration := gofakeit.Number(1, 1000) sum += duration ev := cloudevents.New() ev.SetID(gofakeit.UUID()) ev.SetSource("my-app") ev.SetType("batchingest") ev.SetSubject("customer-1") ev.SetTime(timestamp) _ = ev.SetData("application/json", map[string]string{ "duration_ms": fmt.Sprintf("%d", duration), }) events = append(events, ev) } resp, err := client.IngestEventBatchWithResponse(context.Background(), events) require.NoError(t, err) require.Equal(t, http.StatusNoContent, resp.StatusCode()) // Wait for events to be processed assert.EventuallyWithT(t, func(t *assert.CollectT) { resp, err := client.QueryMeterWithResponse(context.Background(), "batchingest", nil) require.NoError(t, err) require.Equal(t, http.StatusOK, resp.StatusCode()) require.Len(t, resp.JSON200.Data, 1) assert.Equal(t, float64(sum), resp.JSON200.Data[0].Value) }, time.Minute, time.Second) } func TestInvalidIngest(t *testing.T) { client := initClient(t) // Make clickhouse's job easier by sending events within a fix time range timeBase := time.Now().Add(-time.Hour) timeIdx := 0 eventType := "ingest_invalid" subject := eventType meterKey := eventType // ensure subject exists { resp, err := client.UpsertSubjectWithResponse(context.Background(), api.UpsertSubjectJSONRequestBody{ api.SubjectUpsert{ Key: subject, }, }) require.NoError(t, err) require.Equal(t, http.StatusOK, resp.StatusCode()) } getTime := func() time.Time { timeIdx++ // Event list is in reverse order by time return timeBase.Add(time.Duration(-timeIdx) * time.Minute) } // Send an event with unsupported data content type: xml { ev := cloudevents.New() ev.SetID(ulid.Make().String()) ev.SetSource("my-app") ev.SetType(eventType) ev.SetSubject(subject) ev.SetTime(getTime()) _ = ev.SetData(cloudevents.ApplicationXML, []byte("")) resp, err := client.IngestEventWithResponse(context.Background(), ev) require.NoError(t, err) require.Equal(t, http.StatusBadRequest, resp.StatusCode()) } // Send an event where data is a string { payload := fmt.Sprintf(`{ "specversion" : "1.0", "id": "%s", "source": "my-app", "type": "%s", "subject": "%s", "time": "%s", "data": "string" }`, ulid.Make().String(), eventType, subject, getTime().Format(time.RFC3339)) resp, err := client.IngestEventsWithBody(context.Background(), "application/cloudevents+json", strings.NewReader(payload)) require.NoError(t, err) require.Equal(t, http.StatusBadRequest, resp.StatusCode) resp.Body.Close() } // Send an event where data is null // null data should be treated as no data in CloudEvents and should be accepted { payload := fmt.Sprintf(`{ "specversion" : "1.0", "id": "%s", "source": "my-app", "type": "%s", "subject": "%s", "time": "%s", "data": null }`, ulid.Make().String(), eventType, subject, getTime().Format(time.RFC3339)) resp, err := client.IngestEventsWithBody(context.Background(), "application/cloudevents+json", strings.NewReader(payload)) require.NoError(t, err) require.Equal(t, http.StatusNoContent, resp.StatusCode) resp.Body.Close() } // Send an event without data { ev := cloudevents.New() ev.SetID(ulid.Make().String()) ev.SetSource("my-app") ev.SetType(eventType) ev.SetSubject(subject) ev.SetTime(getTime()) resp, err := client.IngestEventWithResponse(context.Background(), ev) require.NoError(t, err) require.Equal(t, http.StatusNoContent, resp.StatusCode()) } // Send an event with empty data { ev := cloudevents.New() ev.SetID(ulid.Make().String()) ev.SetSource("my-app") ev.SetType(eventType) ev.SetSubject(subject) ev.SetTime(getTime()) err := ev.SetData(cloudevents.ApplicationJSON, map[string]string{}) require.NoError(t, err) resp, err := client.IngestEventWithResponse(context.Background(), ev) require.NoError(t, err) require.Equal(t, http.StatusNoContent, resp.StatusCode()) } // Send an event with a NaN value (will skip from aggregation) { ev := cloudevents.New() ev.SetID(ulid.Make().String()) ev.SetSource("my-app") ev.SetType(eventType) ev.SetSubject(subject) ev.SetTime(getTime()) _ = ev.SetData(cloudevents.ApplicationJSON, map[string]string{ "duration_ms": "NaN", }) resp, err := client.IngestEventWithResponse(context.Background(), ev) require.NoError(t, err) require.Equal(t, http.StatusNoContent, resp.StatusCode()) } // Send an event with a Inf value (will skip from aggregation) { ev := cloudevents.New() ev.SetID(ulid.Make().String()) ev.SetSource("my-app") ev.SetType(eventType) ev.SetSubject(subject) ev.SetTime(getTime()) _ = ev.SetData(cloudevents.ApplicationJSON, map[string]string{ "duration_ms": "Inf", }) resp, err := client.IngestEventWithResponse(context.Background(), ev) require.NoError(t, err) require.Equal(t, http.StatusNoContent, resp.StatusCode()) } // Send a valid event, this is what we should get back { ev := cloudevents.New() ev.SetID(ulid.Make().String()) ev.SetSource("my-app") ev.SetType(eventType) ev.SetSubject(subject) ev.SetTime(getTime()) _ = ev.SetData(cloudevents.ApplicationJSON, map[string]string{ "duration_ms": "100", }) resp, err := client.IngestEventWithResponse(context.Background(), ev) require.NoError(t, err) require.Equal(t, http.StatusNoContent, resp.StatusCode()) } // Wait for events to be processed assert.EventuallyWithT(t, func(t *assert.CollectT) { resp, err := client.QueryMeterWithResponse(context.Background(), meterKey, nil) require.NoError(t, err) require.Equal(t, http.StatusOK, resp.StatusCode()) require.Len(t, resp.JSON200.Data, 1) assert.Equal(t, 100.0, resp.JSON200.Data[0].Value) }, time.Minute, time.Second) // List events with has error should return the invalid events resp, err := client.ListEventsWithResponse(context.Background(), &api.ListEventsParams{ Subject: &subject, }) require.NoError(t, err) require.Equal(t, http.StatusOK, resp.StatusCode()) require.NotNil(t, resp.JSON200) events := *resp.JSON200 require.Len(t, events, 6) // unsupported data content gets rejected with a bad request so it should not be in the list // non json data should be rejected with a bad request so it should not be in the list // null data should have processing error require.NotNil(t, events[0].ValidationError) require.Contains(t, *events[0].ValidationError, `invalid event: null and missing value property`) // missing data should have processing error // we only validate events against meters in the processing pipeline so this is an async error require.NotNil(t, events[1].ValidationError) require.Contains(t, *events[1].ValidationError, `invalid event: null and missing value property`) // empty data should have processing error as it does not have the required value property require.NotNil(t, events[2].ValidationError) require.Contains(t, *events[2].ValidationError, `invalid event: missing value property: "$.duration_ms"`) // nan data should have processing error as it does not have the required value property require.NotNil(t, events[3].ValidationError) require.Contains(t, *events[3].ValidationError, `invalid event: value cannot be NaN`) // inf data should have processing error as it does not have the required value property require.NotNil(t, events[4].ValidationError) require.Contains(t, *events[4].ValidationError, `invalid event: value cannot be infinity`) } func TestDedupe(t *testing.T) { client := initClient(t) // ensure subject exists { resp, err := client.UpsertSubjectWithResponse(context.Background(), api.UpsertSubjectJSONRequestBody{ api.SubjectUpsert{ Key: "customer-1", }, }) require.NoError(t, err) require.Equal(t, http.StatusOK, resp.StatusCode()) } // Make clickhouse's job easier by sending events within a fix time range now := time.Now() for i := 0; i < 1000; i++ { // Make clickhouse's job easier by sending events within a fix time range timestamp := gofakeit.DateRange(now.Add(-30*24*time.Hour), now.Add(30*24*time.Hour)) ev := cloudevents.New() ev.SetID("52f44f66-020f-4fa9-a733-102a8ef6f515") ev.SetSource("my-app") ev.SetType("dedupe") ev.SetSubject("customer-1") ev.SetTime(timestamp) _ = ev.SetData("application/json", map[string]string{}) resp, err := client.IngestEventWithResponse(context.Background(), ev) require.NoError(t, err) require.Equal(t, http.StatusNoContent, resp.StatusCode()) } // Wait for events to be processed assert.EventuallyWithT(t, func(t *assert.CollectT) { resp, err := client.QueryMeterWithResponse(context.Background(), "dedupe", nil) require.NoError(t, err) require.Equal(t, http.StatusOK, resp.StatusCode()) require.Len(t, resp.JSON200.Data, 1) assert.Equal(t, 1.0, resp.JSON200.Data[0].Value) }, time.Minute, time.Second) } func TestQuery(t *testing.T) { startOfDay := func(t time.Time) time.Time { return time.Date(t.Year(), t.Month(), t.Day(), 0, 0, 0, 0, t.Location()) } client := initClient(t) meterID := GetMeterIDBySlug(t, client, "query") v3Numeric := func(value int) v3sdk.Numeric { return fmt.Sprintf("%d", value) } // Reproducible random data const customerCount = 5 // ensure subjects exist for i := 1; i <= customerCount; i++ { resp, err := client.UpsertSubjectWithResponse(context.Background(), api.UpsertSubjectJSONRequestBody{ api.SubjectUpsert{Key: fmt.Sprintf("customer-%d", i)}, }) require.NoError(t, err) require.Equal(t, http.StatusOK, resp.StatusCode()) } paths := []string{"/", "/about", "/users", "/contact"} faker := gofakeit.New(8675309) randTime := faker.DateRange(time.Date(2023, time.May, 6, 0, 0, 0, 0, time.UTC), faker.FutureDate().UTC()) timestamp := startOfDay(randTime.UTC()).UTC().Truncate(time.Second) t.Run("Total", func(t *testing.T) { var events []cloudevents.Event newEvents := func(fn func(ev *cloudevents.Event)) []cloudevents.Event { var events []cloudevents.Event for i := 0; i < customerCount; i++ { ev := cloudevents.New() ev.SetID(faker.UUID()) ev.SetSource("my-app") ev.SetType("query") ev.SetSubject(fmt.Sprintf("customer-%d", i+1)) fn(&ev) events = append(events, ev) } return events } newTimedEvents := func(timestamp time.Time) []cloudevents.Event { method := faker.HTTPMethod() path := paths[faker.Number(0, len(paths)-1)] return newEvents(func(ev *cloudevents.Event) { ev.SetTime(timestamp) _ = ev.SetData("application/json", map[string]string{ "duration_ms": "100", "method": method, "path": path, }) }) } // First event { events = append(events, newTimedEvents(timestamp)...) } // Plus one minute { events = append(events, newTimedEvents(timestamp.Add(time.Minute))...) } // Plus one hour { events = append(events, newTimedEvents(timestamp.Add(time.Hour))...) } // Plus one day { events = append(events, newTimedEvents(timestamp.AddDate(0, 0, 1))...) } { resp, err := client.IngestEventBatchWithResponse(context.Background(), events) require.NoError(t, err) require.Equal(t, http.StatusNoContent, resp.StatusCode()) } // Wait for events to be processed testT := t assert.EventuallyWithT(t, func(t *assert.CollectT) { resp, err := client.QueryMeterWithResponse(context.Background(), "query", nil) require.NoError(t, err) require.Equal(t, http.StatusOK, resp.StatusCode()) expected := &api.MeterQueryResult{ Data: []api.MeterQueryRow{ { Value: customerCount * 4 * 100, WindowStart: timestamp.Truncate(time.Minute), WindowEnd: timestamp.Add(24 * time.Hour).Truncate(time.Minute).Add(time.Minute), GroupBy: map[string]*string{}, }, }, } assert.Equal(t, expected, resp.JSON200) v3Resp, err := queryMeterV3(testT, meterID, v3sdk.MeterQueryRequest{}) require.NoError(t, err) expectedV3 := &v3sdk.MeterQueryResult{ Data: []v3sdk.MeterQueryRow{ { Value: v3Numeric(customerCount * 4 * 100), From: timestamp.Truncate(time.Minute), To: timestamp.Add(24 * time.Hour).Truncate(time.Minute).Add(time.Minute), Dimensions: map[string]string{}, }, }, } assert.Equal(t, expectedV3, v3Resp) }, time.Minute, time.Second) }) t.Run("WindowSize", func(t *testing.T) { t.Parallel() t.Run("Minute", func(t *testing.T) { t.Parallel() windowSize := meter.WindowSizeMinute granularity := v3sdk.MeterQueryGranularity("PT1M") // Wait for events to be processed testT := t assert.EventuallyWithT(t, func(t *assert.CollectT) { resp, err := client.QueryMeterWithResponse(context.Background(), "query", &api.QueryMeterParams{ WindowSize: lo.ToPtr(api.WindowSize(windowSize)), }) require.NoError(t, err) require.Equal(t, http.StatusOK, resp.StatusCode()) expected := &api.MeterQueryResult{ Data: []api.MeterQueryRow{ { Value: customerCount * 100, WindowStart: timestamp.Truncate(time.Minute), WindowEnd: timestamp.Truncate(time.Minute).Add(time.Minute), GroupBy: map[string]*string{}, }, { Value: customerCount * 100, WindowStart: timestamp.Add(time.Minute).Truncate(time.Minute), WindowEnd: timestamp.Add(time.Minute).Truncate(time.Minute).Add(time.Minute), GroupBy: map[string]*string{}, }, { Value: customerCount * 100, WindowStart: timestamp.Add(time.Hour).Truncate(time.Minute), WindowEnd: timestamp.Add(time.Hour).Truncate(time.Minute).Add(time.Minute), GroupBy: map[string]*string{}, }, { Value: customerCount * 100, WindowStart: timestamp.Add(24 * time.Hour).Truncate(time.Minute), WindowEnd: timestamp.Add(24 * time.Hour).Truncate(time.Minute).Add(time.Minute), GroupBy: map[string]*string{}, }, }, WindowSize: lo.ToPtr(api.WindowSize(windowSize)), } assert.Equal(t, expected, resp.JSON200) v3Resp, err := queryMeterV3(testT, meterID, v3sdk.MeterQueryRequest{ Granularity: &granularity, }) require.NoError(t, err) expectedV3 := &v3sdk.MeterQueryResult{ Data: []v3sdk.MeterQueryRow{ { Value: v3Numeric(customerCount * 100), From: timestamp.Truncate(time.Minute), To: timestamp.Truncate(time.Minute).Add(time.Minute), Dimensions: map[string]string{}, }, { Value: v3Numeric(customerCount * 100), From: timestamp.Add(time.Minute).Truncate(time.Minute), To: timestamp.Add(time.Minute).Truncate(time.Minute).Add(time.Minute), Dimensions: map[string]string{}, }, { Value: v3Numeric(customerCount * 100), From: timestamp.Add(time.Hour).Truncate(time.Minute), To: timestamp.Add(time.Hour).Truncate(time.Minute).Add(time.Minute), Dimensions: map[string]string{}, }, { Value: v3Numeric(customerCount * 100), From: timestamp.Add(24 * time.Hour).Truncate(time.Minute), To: timestamp.Add(24 * time.Hour).Truncate(time.Minute).Add(time.Minute), Dimensions: map[string]string{}, }, }, } assert.Equal(t, expectedV3, v3Resp) }, time.Minute, time.Second) }) t.Run("Hour", func(t *testing.T) { t.Parallel() windowSize := meter.WindowSizeHour granularity := v3sdk.MeterQueryGranularity("PT1H") // Wait for events to be processed testT := t assert.EventuallyWithT(t, func(t *assert.CollectT) { resp, err := client.QueryMeterWithResponse(context.Background(), "query", &api.QueryMeterParams{ WindowSize: lo.ToPtr(api.WindowSize(windowSize)), }) require.NoError(t, err) require.Equal(t, http.StatusOK, resp.StatusCode()) expected := &api.MeterQueryResult{ Data: []api.MeterQueryRow{ { Value: customerCount * 2 * 100, WindowStart: timestamp.Truncate(time.Hour), WindowEnd: timestamp.Truncate(time.Hour).Add(time.Hour), GroupBy: map[string]*string{}, }, { Value: customerCount * 100, WindowStart: timestamp.Add(time.Hour).Truncate(time.Hour), WindowEnd: timestamp.Add(time.Hour).Truncate(time.Hour).Add(time.Hour), GroupBy: map[string]*string{}, }, { Value: customerCount * 100, WindowStart: timestamp.Add(24 * time.Hour).Truncate(time.Hour), WindowEnd: timestamp.Add(24 * time.Hour).Truncate(time.Hour).Add(time.Hour), GroupBy: map[string]*string{}, }, }, WindowSize: lo.ToPtr(api.WindowSize(windowSize)), } assert.Equal(t, expected, resp.JSON200) v3Resp, err := queryMeterV3(testT, meterID, v3sdk.MeterQueryRequest{ Granularity: &granularity, }) require.NoError(t, err) expectedV3 := &v3sdk.MeterQueryResult{ Data: []v3sdk.MeterQueryRow{ { Value: v3Numeric(customerCount * 2 * 100), From: timestamp.Truncate(time.Hour), To: timestamp.Truncate(time.Hour).Add(time.Hour), Dimensions: map[string]string{}, }, { Value: v3Numeric(customerCount * 100), From: timestamp.Add(time.Hour).Truncate(time.Hour), To: timestamp.Add(time.Hour).Truncate(time.Hour).Add(time.Hour), Dimensions: map[string]string{}, }, { Value: v3Numeric(customerCount * 100), From: timestamp.Add(24 * time.Hour).Truncate(time.Hour), To: timestamp.Add(24 * time.Hour).Truncate(time.Hour).Add(time.Hour), Dimensions: map[string]string{}, }, }, } assert.Equal(t, expectedV3, v3Resp) }, time.Minute, time.Second) }) t.Run("Day", func(t *testing.T) { t.Parallel() windowSize := meter.WindowSizeDay granularity := v3sdk.MeterQueryGranularity("P1D") // Wait for events to be processed testT := t assert.EventuallyWithT(t, func(t *assert.CollectT) { resp, err := client.QueryMeterWithResponse(context.Background(), "query", &api.QueryMeterParams{ WindowSize: lo.ToPtr(api.WindowSize(windowSize)), }) require.NoError(t, err) require.Equal(t, http.StatusOK, resp.StatusCode()) expected := &api.MeterQueryResult{ Data: []api.MeterQueryRow{ { Value: customerCount * 3 * 100, WindowStart: timestamp.Truncate(24 * time.Hour), WindowEnd: timestamp.Truncate(24 * time.Hour).Add(24 * time.Hour), GroupBy: map[string]*string{}, }, { Value: customerCount * 100, WindowStart: timestamp.Add(24 * time.Hour).Truncate(24 * time.Hour), WindowEnd: timestamp.Add(24 * time.Hour).Truncate(24 * time.Hour).Add(24 * time.Hour), GroupBy: map[string]*string{}, }, }, WindowSize: lo.ToPtr(api.WindowSize(windowSize)), } assert.Equal(t, expected, resp.JSON200) v3Resp, err := queryMeterV3(testT, meterID, v3sdk.MeterQueryRequest{ Granularity: &granularity, }) require.NoError(t, err) expectedV3 := &v3sdk.MeterQueryResult{ Data: []v3sdk.MeterQueryRow{ { Value: v3Numeric(customerCount * 3 * 100), From: timestamp.Truncate(24 * time.Hour), To: timestamp.Truncate(24 * time.Hour).Add(24 * time.Hour), Dimensions: map[string]string{}, }, { Value: v3Numeric(customerCount * 100), From: timestamp.Add(24 * time.Hour).Truncate(24 * time.Hour), To: timestamp.Add(24 * time.Hour).Truncate(24 * time.Hour).Add(24 * time.Hour), Dimensions: map[string]string{}, }, }, } assert.Equal(t, expectedV3, v3Resp) }, time.Minute, time.Second) }) }) t.Run("Subject", func(t *testing.T) { t.Parallel() // TODO: randomize? // TODO: make sure we have enough subject subject := []string{"customer-1", "customer-2"} // Wait for events to be processed testT := t assert.EventuallyWithT(t, func(t *assert.CollectT) { resp, err := client.QueryMeterWithResponse(context.Background(), "query", &api.QueryMeterParams{ Subject: &subject, }) require.NoError(t, err) require.Equal(t, http.StatusOK, resp.StatusCode()) expected := &api.MeterQueryResult{ Data: []api.MeterQueryRow{ { Value: 4 * 100, WindowStart: timestamp.Truncate(time.Minute), WindowEnd: timestamp.Truncate(time.Minute).Add(24*time.Hour + time.Minute), Subject: &subject[1], GroupBy: map[string]*string{}, }, { Value: 4 * 100, WindowStart: timestamp.Truncate(time.Minute), WindowEnd: timestamp.Truncate(time.Minute).Add(24*time.Hour + time.Minute), Subject: &subject[0], GroupBy: map[string]*string{}, }, }, } assert.Equal(t, expected, resp.JSON200) v3Resp, err := queryMeterV3(testT, meterID, v3sdk.MeterQueryRequest{ Filters: &v3sdk.MeterQueryFilters{ Dimensions: &map[string]v3sdk.QueryFilterStringMapItemInput{ "subject": {In: &subject}, }, }, }) require.NoError(t, err) expectedV3 := &v3sdk.MeterQueryResult{ Data: []v3sdk.MeterQueryRow{ { Value: v3Numeric(4 * 100), From: timestamp.Truncate(time.Minute), To: timestamp.Truncate(time.Minute).Add(24*time.Hour + time.Minute), Dimensions: map[string]string{ "subject": subject[1], }, }, { Value: v3Numeric(4 * 100), From: timestamp.Truncate(time.Minute), To: timestamp.Truncate(time.Minute).Add(24*time.Hour + time.Minute), Dimensions: map[string]string{ "subject": subject[0], }, }, }, } assert.Equal(t, expectedV3, v3Resp) }, time.Minute, time.Second) }) t.Run("PostJSON", func(t *testing.T) { t.Parallel() expected := &api.MeterQueryResult{ Data: []api.MeterQueryRow{ { Value: customerCount * 4 * 100, WindowStart: timestamp.Truncate(time.Minute), WindowEnd: timestamp.Add(24 * time.Hour).Truncate(time.Minute).Add(time.Minute), GroupBy: map[string]*string{}, }, }, } t.Run("DefaultAccept", func(t *testing.T) { t.Parallel() // Wait for events to be processed assert.EventuallyWithT(t, func(t *assert.CollectT) { // No accept header — should default to JSON resp, err := client.QueryMeterPostWithResponse(context.Background(), "query", api.QueryMeterPostJSONRequestBody{}) require.NoError(t, err) require.Equal(t, http.StatusOK, resp.StatusCode()) require.NotNil(t, resp.JSON200) assert.Equal(t, expected, resp.JSON200) }, time.Minute, time.Second) }) t.Run("ExplicitAccept", func(t *testing.T) { t.Parallel() // Wait for events to be processed assert.EventuallyWithT(t, func(t *assert.CollectT) { resp, err := client.QueryMeterPostWithResponse(context.Background(), "query", api.QueryMeterPostJSONRequestBody{}, func(_ context.Context, req *http.Request) error { req.Header.Set("Accept", "application/json") return nil }) require.NoError(t, err) require.Equal(t, http.StatusOK, resp.StatusCode()) require.NotNil(t, resp.JSON200) assert.Equal(t, expected, resp.JSON200) }, time.Minute, time.Second) }) }) t.Run("PostCSV", func(t *testing.T) { t.Parallel() // Wait for events to be processed assert.EventuallyWithT(t, func(t *assert.CollectT) { resp, err := client.QueryMeterPostWithResponse(context.Background(), "query", api.QueryMeterPostJSONRequestBody{}, func(_ context.Context, req *http.Request) error { req.Header.Set("Accept", "text/csv") return nil }) require.NoError(t, err) require.Equal(t, http.StatusOK, resp.StatusCode()) // JSON200 should be nil for CSV responses require.Nil(t, resp.JSON200) // Verify Content-Type header require.Equal(t, "text/csv", resp.HTTPResponse.Header.Get("Content-Type")) // Verify exact CSV body as string windowStart := timestamp.Truncate(time.Minute).Format(time.RFC3339) windowEnd := timestamp.Add(24 * time.Hour).Truncate(time.Minute).Add(time.Minute).Format(time.RFC3339) value := fmt.Sprintf("%f", float64(customerCount*4*100)) expectedCSV := "window_start,window_end,value\n" + windowStart + "," + windowEnd + "," + value + "\n" assert.Equal(t, expectedCSV, string(resp.Body)) }, time.Minute, time.Second) }) } func TestQueryDSTTransition(t *testing.T) { client := initClient(t) meterID := GetMeterIDBySlug(t, client, "query") // Test DST transitions with DAY window size in America/Los_Angeles timezone // Verify that window boundaries are continuous across DST changes // DST start: March 9, 2025 at 2am -> 3am (spring forward) // DST end: November 2, 2025 at 2am -> 1am (fall back) losAngeles, err := time.LoadLocation("America/Los_Angeles") require.NoError(t, err) eventType := "query" subject := "customer-dst" t.Run("DST_Spring_Forward", func(t *testing.T) { // Test March 9, 2025 DST transition (spring forward: 2am -> 3am) // Test 3 days before and 3 days after the transition var events []cloudevents.Event // Generate events for March 6-12, 2025 (3 days before, transition day, 3 days after) for day := 6; day <= 12; day++ { baseTime := time.Date(2025, time.March, day, 10, 0, 0, 0, losAngeles) ev := cloudevents.New() ev.SetID(ulid.Make().String()) ev.SetSource("dst-test") ev.SetType(eventType) ev.SetSubject(subject) ev.SetTime(baseTime) _ = ev.SetData("application/json", map[string]interface{}{ "duration_ms": "100", }) events = append(events, ev) } // Ingest events { resp, err := client.IngestEventBatchWithResponse(context.Background(), events) require.NoError(t, err) require.Equal(t, http.StatusNoContent, resp.StatusCode()) } // Query with DAY window size in Los Angeles timezone testT := t assert.EventuallyWithT(t, func(t *assert.CollectT) { from := time.Date(2025, time.March, 6, 0, 0, 0, 0, losAngeles) to := time.Date(2025, time.March, 13, 0, 0, 0, 0, losAngeles) resp, err := client.QueryMeterWithResponse(context.Background(), eventType, &api.QueryMeterParams{ Subject: &[]string{subject}, WindowSize: lo.ToPtr(api.WindowSizeDay), WindowTimeZone: lo.ToPtr("America/Los_Angeles"), From: lo.ToPtr(from), To: lo.ToPtr(to), }) require.NoError(t, err) require.Equal(t, http.StatusOK, resp.StatusCode()) require.NotNil(t, resp.JSON200) require.Len(t, resp.JSON200.Data, 7) // March 6-12 // Verify window continuity - each window's end should equal the next window's start for i := 0; i < len(resp.JSON200.Data)-1; i++ { currentWindowEnd := resp.JSON200.Data[i].WindowEnd.In(losAngeles) nextWindowStart := resp.JSON200.Data[i+1].WindowStart.In(losAngeles) assert.Equal(t, currentWindowEnd, nextWindowStart, "Gap detected: window %d ends at %v but window %d starts at %v", i, currentWindowEnd, i+1, nextWindowStart) } // Verify all windows start at midnight in LA timezone for i, row := range resp.JSON200.Data { windowStart := row.WindowStart.In(losAngeles) assert.Equal(t, 0, windowStart.Hour(), "Day %d window doesn't start at midnight", i) assert.Equal(t, 0, windowStart.Minute(), "Day %d window doesn't start at midnight", i) assert.Equal(t, 0, windowStart.Second(), "Day %d window doesn't start at midnight", i) } // Verify the DST transition day (March 9) windows are correct dstDay := resp.JSON200.Data[3] // March 9 is index 3 (6,7,8,9) assert.Equal(t, time.Date(2025, time.March, 9, 0, 0, 0, 0, losAngeles), dstDay.WindowStart.In(losAngeles)) assert.Equal(t, time.Date(2025, time.March, 10, 0, 0, 0, 0, losAngeles), dstDay.WindowEnd.In(losAngeles)) granularity := v3sdk.MeterQueryGranularity("P1D") timeZone := "America/Los_Angeles" v3Resp, err := queryMeterV3(testT, meterID, v3sdk.MeterQueryRequest{ From: &from, To: &to, Granularity: &granularity, TimeZone: &timeZone, Filters: &v3sdk.MeterQueryFilters{ Dimensions: &map[string]v3sdk.QueryFilterStringMapItemInput{ "subject": {Eq: lo.ToPtr(subject)}, }, }, }) require.NoError(t, err) require.NotNil(t, v3Resp) require.Len(t, v3Resp.Data, 7) for i := 0; i < len(v3Resp.Data)-1; i++ { currentWindowEnd := v3Resp.Data[i].To.In(losAngeles) nextWindowStart := v3Resp.Data[i+1].From.In(losAngeles) assert.Equal(t, currentWindowEnd, nextWindowStart, "Gap detected: window %d ends at %v but window %d starts at %v", i, currentWindowEnd, i+1, nextWindowStart) } for i, row := range v3Resp.Data { windowStart := row.From.In(losAngeles) assert.Equal(t, 0, windowStart.Hour(), "Day %d window doesn't start at midnight", i) assert.Equal(t, 0, windowStart.Minute(), "Day %d window doesn't start at midnight", i) assert.Equal(t, 0, windowStart.Second(), "Day %d window doesn't start at midnight", i) } v3DSTDay := v3Resp.Data[3] assert.Equal(t, time.Date(2025, time.March, 9, 0, 0, 0, 0, losAngeles), v3DSTDay.From.In(losAngeles)) assert.Equal(t, time.Date(2025, time.March, 10, 0, 0, 0, 0, losAngeles), v3DSTDay.To.In(losAngeles)) }, time.Minute, time.Second) }) t.Run("DST_Fall_Back", func(t *testing.T) { // Test November 2, 2025 DST transition (fall back: 2am -> 1am) // Test 3 days before and 3 days after the transition var events []cloudevents.Event // Generate events for October 30 - November 5, 2025 (3 days before, transition day, 3 days after) days := []struct { month int day int }{ {int(time.October), 30}, {int(time.October), 31}, {int(time.November), 1}, {int(time.November), 2}, {int(time.November), 3}, {int(time.November), 4}, {int(time.November), 5}, } for _, d := range days { baseTime := time.Date(2025, time.Month(d.month), d.day, 10, 0, 0, 0, losAngeles) ev := cloudevents.New() ev.SetID(ulid.Make().String()) ev.SetSource("dst-test") ev.SetType(eventType) ev.SetSubject(subject) ev.SetTime(baseTime) _ = ev.SetData("application/json", map[string]interface{}{ "duration_ms": "100", }) events = append(events, ev) } // Ingest events { resp, err := client.IngestEventBatchWithResponse(context.Background(), events) require.NoError(t, err) require.Equal(t, http.StatusNoContent, resp.StatusCode()) } // Query with DAY window size in Los Angeles timezone testT := t assert.EventuallyWithT(t, func(t *assert.CollectT) { from := time.Date(2025, time.October, 30, 0, 0, 0, 0, losAngeles) to := time.Date(2025, time.November, 6, 0, 0, 0, 0, losAngeles) resp, err := client.QueryMeterWithResponse(context.Background(), eventType, &api.QueryMeterParams{ Subject: &[]string{subject}, WindowSize: lo.ToPtr(api.WindowSizeDay), WindowTimeZone: lo.ToPtr("America/Los_Angeles"), From: lo.ToPtr(from), To: lo.ToPtr(to), }) require.NoError(t, err) require.Equal(t, http.StatusOK, resp.StatusCode()) require.NotNil(t, resp.JSON200) require.Len(t, resp.JSON200.Data, 7) // October 30 - November 5 // Verify window continuity - each window's end should equal the next window's start for i := 0; i < len(resp.JSON200.Data)-1; i++ { currentWindowEnd := resp.JSON200.Data[i].WindowEnd.In(losAngeles) nextWindowStart := resp.JSON200.Data[i+1].WindowStart.In(losAngeles) assert.Equal(t, currentWindowEnd, nextWindowStart, "Gap detected: window %d ends at %v but window %d starts at %v", i, currentWindowEnd, i+1, nextWindowStart) } // Verify all windows start at midnight in LA timezone for i, row := range resp.JSON200.Data { windowStart := row.WindowStart.In(losAngeles) assert.Equal(t, 0, windowStart.Hour(), "Day %d window doesn't start at midnight", i) assert.Equal(t, 0, windowStart.Minute(), "Day %d window doesn't start at midnight", i) assert.Equal(t, 0, windowStart.Second(), "Day %d window doesn't start at midnight", i) } // Verify the DST transition day (November 2) windows are correct dstDay := resp.JSON200.Data[3] // November 2 is index 3 (Oct 30, 31, Nov 1, 2) assert.Equal(t, time.Date(2025, time.November, 2, 0, 0, 0, 0, losAngeles), dstDay.WindowStart.In(losAngeles)) assert.Equal(t, time.Date(2025, time.November, 3, 0, 0, 0, 0, losAngeles), dstDay.WindowEnd.In(losAngeles)) granularity := v3sdk.MeterQueryGranularity("P1D") timeZone := "America/Los_Angeles" v3Resp, err := queryMeterV3(testT, meterID, v3sdk.MeterQueryRequest{ From: &from, To: &to, Granularity: &granularity, TimeZone: &timeZone, Filters: &v3sdk.MeterQueryFilters{ Dimensions: &map[string]v3sdk.QueryFilterStringMapItemInput{ "subject": {Eq: lo.ToPtr(subject)}, }, }, }) require.NoError(t, err) require.NotNil(t, v3Resp) require.Len(t, v3Resp.Data, 7) for i := 0; i < len(v3Resp.Data)-1; i++ { currentWindowEnd := v3Resp.Data[i].To.In(losAngeles) nextWindowStart := v3Resp.Data[i+1].From.In(losAngeles) assert.Equal(t, currentWindowEnd, nextWindowStart, "Gap detected: window %d ends at %v but window %d starts at %v", i, currentWindowEnd, i+1, nextWindowStart) } for i, row := range v3Resp.Data { windowStart := row.From.In(losAngeles) assert.Equal(t, 0, windowStart.Hour(), "Day %d window doesn't start at midnight", i) assert.Equal(t, 0, windowStart.Minute(), "Day %d window doesn't start at midnight", i) assert.Equal(t, 0, windowStart.Second(), "Day %d window doesn't start at midnight", i) } v3DSTDay := v3Resp.Data[3] assert.Equal(t, time.Date(2025, time.November, 2, 0, 0, 0, 0, losAngeles), v3DSTDay.From.In(losAngeles)) assert.Equal(t, time.Date(2025, time.November, 3, 0, 0, 0, 0, losAngeles), v3DSTDay.To.In(losAngeles)) }, time.Minute, time.Second) }) } func TestCredit(t *testing.T) { client := initClient(t) meterSlug := "credit_test_meter" var featureId string var featureKey string const waitTime = time.Second * 30 apiMONTH := &api.RecurringPeriodInterval{} require.NoError(t, apiMONTH.FromRecurringPeriodIntervalEnum(api.RecurringPeriodIntervalEnumMONTH)) apiYEAR := &api.RecurringPeriodInterval{} require.NoError(t, apiYEAR.FromRecurringPeriodIntervalEnum(api.RecurringPeriodIntervalEnumYEAR)) customerKey := customer1.Key() subjectKey := customerKey // Let's create a customer with a subject CreateCustomerWithSubject(t, client, customerKey, subjectKey) t.Run("Create Feature", func(t *testing.T) { randKey := fmt.Sprintf("credit_test_feature_%d", time.Now().Unix()) resp, err := client.CreateFeatureWithResponse(context.Background(), api.CreateFeatureJSONRequestBody{ Name: "Credit Test Feature", MeterSlug: convert.ToPointer("credit_test_meter"), Key: randKey, MeterGroupByFilters: &map[string]string{ "model": "gpt-4", }, }) require.NoError(t, err) require.Equal(t, http.StatusCreated, resp.StatusCode(), "Invalid status code [response_body=%s]", string(resp.Body)) featureId = resp.JSON201.Id featureKey = resp.JSON201.Key expected := &api.Feature{ Id: featureId, Name: "Credit Test Feature", Key: randKey, MeterSlug: convert.ToPointer("credit_test_meter"), MeterGroupByFilters: &map[string]string{ "model": "gpt-4", }, AdvancedMeterGroupByFilters: &map[string]api.FilterString{ "model": { Eq: convert.ToPointer("gpt-4"), }, }, } require.NotEmpty(t, resp.JSON201.CreatedAt) require.NotEmpty(t, resp.JSON201.UpdatedAt) // Exclude created_at and update_at from comparison resp.JSON201.CreatedAt = time.Time{} resp.JSON201.UpdatedAt = time.Time{} require.Equal(t, expected, resp.JSON201) }) var entitlementId string var eCreatedAt time.Time t.Run("Create a Entitlement", func(t *testing.T) { meteredEntitlement := api.EntitlementMeteredCreateInputs{ Type: "metered", FeatureId: &featureId, UsagePeriod: api.RecurringPeriodCreateInput{ Anchor: convert.ToPointer(time.Date(2024, time.January, 1, 0, 0, 0, 0, time.UTC)), Interval: *apiMONTH, }, } body := &api.CreateEntitlementJSONRequestBody{} err := body.FromEntitlementMeteredCreateInputs(meteredEntitlement) require.NoError(t, err) resp, err := client.CreateEntitlementWithResponse(context.Background(), subjectKey, *body) require.NoError(t, err) require.Equal(t, http.StatusCreated, resp.StatusCode(), "Invalid status code [response_body=%s]", string(resp.Body)) metered, err := resp.JSON201.AsEntitlementMetered() require.NoError(t, err) require.Equal(t, metered.SubjectKey, subjectKey) entitlementId = metered.Id eCreatedAt = metered.CreatedAt }) t.Run("Create for same subject and feature", func(t *testing.T) { meteredEntitlement := api.EntitlementMeteredCreateInputs{ Type: "metered", FeatureId: &featureId, UsagePeriod: api.RecurringPeriodCreateInput{ Anchor: convert.ToPointer(time.Date(2024, time.January, 1, 0, 0, 0, 0, time.UTC)), Interval: *apiMONTH, }, } body := &api.CreateEntitlementJSONRequestBody{} err := body.FromEntitlementMeteredCreateInputs(meteredEntitlement) require.NoError(t, err) resp, err := client.CreateEntitlementWithResponse(context.Background(), subjectKey, *body) require.NoError(t, err) require.Equal(t, http.StatusConflict, resp.StatusCode(), "Invalid status code [response_body=%s]", string(resp.Body)) require.NotNil(t, resp.ApplicationproblemJSON409) require.NotEmpty(t, resp.ApplicationproblemJSON409.Extensions["conflictingEntityId"]) require.Equal(t, entitlementId, resp.ApplicationproblemJSON409.Extensions["conflictingEntityId"]) }) t.Run("Create a Entitlement With Default Grants", func(t *testing.T) { randCustomerKey := ulid.Make().String() randSubjectKey := ulid.Make().String() CreateCustomerWithSubject(t, client, randCustomerKey, randSubjectKey) measureUsageFrom := time.Date(2024, time.January, 1, 0, 0, 0, 0, time.UTC) muf := &api.MeasureUsageFrom{} err := muf.FromMeasureUsageFromTime(measureUsageFrom) require.NoError(t, err) meteredEntitlement := api.EntitlementMeteredCreateInputs{ Type: "metered", FeatureId: &featureId, UsagePeriod: api.RecurringPeriodCreateInput{ Anchor: convert.ToPointer(time.Date(2024, time.January, 1, 0, 0, 0, 0, time.UTC)), Interval: *apiMONTH, }, MeasureUsageFrom: muf, IssueAfterReset: convert.ToPointer(100.0), IssueAfterResetPriority: convert.ToPointer[uint8](6), } body := &api.CreateEntitlementJSONRequestBody{} err = body.FromEntitlementMeteredCreateInputs(meteredEntitlement) require.NoError(t, err) resp, err := client.CreateEntitlementWithResponse(context.Background(), randSubjectKey, *body) require.NoError(t, err) require.Equal(t, http.StatusCreated, resp.StatusCode(), "Invalid status code [response_body=%s]", string(resp.Body)) metered, err := resp.JSON201.AsEntitlementMetered() require.NoError(t, err) require.Equal(t, randSubjectKey, metered.SubjectKey) require.Equal(t, measureUsageFrom, metered.MeasureUsageFrom) // fetch grants for entitlement grantListResp, err := client.ListEntitlementGrantsWithResponse(context.Background(), randSubjectKey, metered.Id, nil) require.NoError(t, err) require.Equal(t, http.StatusOK, grantListResp.StatusCode()) require.NotNil(t, grantListResp.JSON200) require.Len(t, *grantListResp.JSON200, 1) require.Equal(t, *metered.IssueAfterReset, (*grantListResp.JSON200)[0].Amount) require.Equal(t, metered.IssueAfterResetPriority, (*grantListResp.JSON200)[0].Priority) require.Equal(t, metered.Id, (*grantListResp.JSON200)[0].EntitlementId) require.Equal(t, api.Annotations{ "issueAfterReset": true, }, *(*grantListResp.JSON200)[0].Annotations) }) t.Run("Create a Entitlement With MeasureUsageFrom enum", func(t *testing.T) { randSubject := ulid.Make().String() randCustomerKey := ulid.Make().String() CreateCustomerWithSubject(t, client, randCustomerKey, randSubject) periodAnchor := time.Now().Truncate(time.Minute).Add(-time.Hour).In(time.UTC) muf := &api.MeasureUsageFrom{} err := muf.FromMeasureUsageFromPreset(api.MeasureUsageFromPresetCurrentPeriodStart) require.NoError(t, err) meteredEntitlement := api.EntitlementMeteredCreateInputs{ Type: "metered", FeatureId: &featureId, UsagePeriod: api.RecurringPeriodCreateInput{ Anchor: &periodAnchor, Interval: *apiMONTH, }, MeasureUsageFrom: muf, } body := &api.CreateEntitlementJSONRequestBody{} err = body.FromEntitlementMeteredCreateInputs(meteredEntitlement) require.NoError(t, err) resp, err := client.CreateEntitlementWithResponse(context.Background(), randSubject, *body) require.NoError(t, err) require.Equal(t, http.StatusCreated, resp.StatusCode(), "Invalid status code [response_body=%s]", string(resp.Body)) metered, err := resp.JSON201.AsEntitlementMetered() require.NoError(t, err) require.Equal(t, randSubject, metered.SubjectKey) require.Equal(t, periodAnchor, metered.MeasureUsageFrom) }) t.Run("Create Grant", func(t *testing.T) { effectiveAt := time.Now().Truncate(time.Minute) priority := uint8(1) maxRolloverAmount := 100.0 minRolloverAmount := 0.0 // Create grant resp, err := client.CreateGrantWithResponse(context.Background(), subjectKey, entitlementId, api.EntitlementGrantCreateInput{ Amount: 100, EffectiveAt: effectiveAt, Expiration: api.ExpirationPeriod{ Duration: "MONTH", Count: 1, }, Priority: &priority, MaxRolloverAmount: &maxRolloverAmount, MinRolloverAmount: &minRolloverAmount, Recurrence: &api.RecurringPeriodCreateInput{ Anchor: convert.ToPointer(time.Date(2024, time.January, 1, 0, 0, 0, 0, time.UTC)), Interval: *apiYEAR, }, Metadata: &api.Metadata{ "some_key": "some_value", }, }) require.NoError(t, err) require.Equal(t, http.StatusCreated, resp.StatusCode(), "Invalid status code [response_body=%s]", resp.Body) require.NotEmpty(t, resp.JSON201.UpdatedAt) require.NotEmpty(t, resp.JSON201.ExpiresAt) require.NotEmpty(t, resp.JSON201.CreatedAt) require.NotEmpty(t, resp.JSON201.NextRecurrence) expected := &api.EntitlementGrant{ Id: resp.JSON201.Id, Amount: 100, EntitlementId: entitlementId, Priority: &priority, EffectiveAt: effectiveAt.UTC(), MaxRolloverAmount: &maxRolloverAmount, MinRolloverAmount: &minRolloverAmount, Expiration: api.ExpirationPeriod{ Duration: "MONTH", Count: 1, }, Recurrence: &api.RecurringPeriod{ Anchor: time.Date(2024, time.January, 1, 0, 0, 0, 0, time.UTC), Interval: *apiYEAR, IntervalISO: "P1Y", }, Metadata: &api.Metadata{ "some_key": "some_value", }, } // Exclude timestamps from comparison resp.JSON201.CreatedAt = time.Time{} resp.JSON201.UpdatedAt = time.Time{} resp.JSON201.ExpiresAt = nil resp.JSON201.NextRecurrence = nil require.Equal(t, expected, resp.JSON201) }) t.Run("Ingest usage", func(t *testing.T) { // Reproducible random data faker := gofakeit.New(8675309) var events []cloudevents.Event newEvent := func(timestamp string, model string) cloudevents.Event { ts, err := time.Parse(time.RFC3339, timestamp) if err != nil { t.Fatal(err) } ev := cloudevents.New() ev.SetID(faker.UUID()) ev.SetSource("credit-test") ev.SetType("credit_event") ev.SetTime(ts) ev.SetSubject(subjectKey) _ = ev.SetData("application/json", map[string]string{ "model": model, }) return ev } et := eCreatedAt.Add(time.Second * 15) // First event { events = append(events, newEvent(et.Format(time.RFC3339), "gpt-4")) } // Irrelevant event (does not affect credit because of model mismatch) { events = append(events, newEvent(et.Format(time.RFC3339), "gpt-3")) } // Ingore events { resp, err := client.IngestEventBatchWithResponse(context.Background(), events) require.NoError(t, err) require.Equal(t, http.StatusNoContent, resp.StatusCode()) } // NOTE: This is a temporary workaround to avoid data race condition in (assert|require).EventuallyWithTf function which // was not triggered before `testify` v1.11.1. Which includes the change (https://github.com/stretchr/testify/pull/1427) // that makes the EventuallyWithTf return early triggering the race condition in our case where the result of the condition // from the previous tick is used to decide the result of the test. // Remove this when the race condition is fixed in `testify`. // Bare in mind his still can fail if sink-worker is is not running or the ingestion of the events takes longer than the sleep time. time.Sleep(5 * time.Second) // Wait for events to be processed, fail on network errors testutils.EventuallyWithTf(t, func(c *assert.CollectT, saveErr func(err any)) { resp, err := client.QueryMeterWithResponse(context.Background(), meterSlug, &api.QueryMeterParams{ Subject: &[]string{subjectKey}, }) saveErr(err) require.NoError(c, err) require.Equal(c, http.StatusOK, resp.StatusCode()) require.GreaterOrEqual(t, len(resp.JSON200.Data), 1) // As we invested two events with a count meter assert.NotNil(c, resp.JSON200) if resp.JSON200 != nil { assert.Len(c, resp.JSON200.Data, 1) if len(resp.JSON200.Data) > 0 { assert.Equal(c, 2.0, resp.JSON200.Data[0].Value) } } }, waitTime, time.Second) }) t.Run("Entitlement Value", func(t *testing.T) { // Get grants grantListResp, err := client.ListEntitlementGrantsWithResponse(context.Background(), subjectKey, entitlementId, nil) require.NoError(t, err) require.Equal(t, http.StatusOK, grantListResp.StatusCode()) require.NotNil(t, grantListResp.JSON200) require.Len(t, *grantListResp.JSON200, 1) // Get feature featureListResp, err := client.ListFeaturesWithResponse(t.Context(), &api.ListFeaturesParams{ MeterSlug: &[]string{meterSlug}, }) require.NoError(t, err) require.Equal(t, http.StatusOK, featureListResp.StatusCode()) require.NotNil(t, featureListResp.JSON200) features, err := featureListResp.JSON200.AsListFeaturesResult0() require.NoError(t, err) require.Len(t, features, 1) resp, err := client.GetEntitlementValueWithResponse(context.Background(), subjectKey, entitlementId, &api.GetEntitlementValueParams{ Time: convert.ToPointer(eCreatedAt.Add(time.Minute * 2)), }) require.NoError(t, err) require.Equal(t, http.StatusOK, resp.StatusCode()) expected := &api.EntitlementValue{ Balance: convert.ToPointer(99.0), HasAccess: true, Overage: convert.ToPointer(0.0), Usage: convert.ToPointer(1.0), TotalAvailableGrantAmount: convert.ToPointer(100.0), } assert.Equal(t, expected, resp.JSON200) }) t.Run("Slow: Reset testing", func(t *testing.T) { // Entitlements are one minute rounded, so we need to wait for a minute to pass so we can reset, if we // really want we can run this on main. (if we have a second resolution we can just enable it) if !shouldRunSlowTests(t) { t.Skip("Skipping slow test, please reenable when we have a second resolution for entitlements") } t.Run("Reset", func(t *testing.T) { // we have to wait for a minute to pass so we can reset time.Sleep(time.Minute) effectiveAt := time.Now().Truncate(time.Minute) // Get grants grantListResp, err := client.ListEntitlementGrantsWithResponse(context.Background(), subjectKey, entitlementId, nil) require.NoError(t, err) require.Equal(t, http.StatusOK, grantListResp.StatusCode()) require.NotNil(t, grantListResp.JSON200) require.Len(t, *grantListResp.JSON200, 1) // Get feature featureListResp, err := client.ListFeaturesWithResponse(context.Background(), nil) require.NoError(t, err) require.Equal(t, http.StatusOK, featureListResp.StatusCode()) require.NotNil(t, featureListResp.JSON200) features, err := featureListResp.JSON200.AsListFeaturesResult0() require.NoError(t, err) require.Len(t, features, 1) // Reset usage resetResp, err := client.ResetEntitlementUsageWithResponse(context.Background(), subjectKey, entitlementId, api.ResetEntitlementUsageJSONRequestBody{ EffectiveAt: &effectiveAt, }) require.NoError(t, err) require.Equal(t, http.StatusNoContent, resetResp.StatusCode()) }) customer2Key := customer2.Key() subject2 := customer2Key // we have to wait after the reset time.Sleep(time.Minute) time.Sleep(time.Second * 10) t.Run("Create entitlement with automatic grant issuing", func(t *testing.T) { CreateCustomerWithSubject(t, client, customer2Key, subject2) meteredEntitlement := api.EntitlementMeteredCreateInputs{ Type: "metered", FeatureId: &featureId, UsagePeriod: api.RecurringPeriodCreateInput{ Anchor: convert.ToPointer(time.Date(2024, time.January, 1, 0, 0, 0, 0, time.UTC)), Interval: *apiMONTH, }, IssueAfterReset: convert.ToPointer(50.0), } body := &api.CreateEntitlementJSONRequestBody{} err := body.FromEntitlementMeteredCreateInputs(meteredEntitlement) require.NoError(t, err) resp, err := client.CreateEntitlementWithResponse(context.Background(), subject2, *body) require.NoError(t, err) require.Equal(t, http.StatusCreated, resp.StatusCode(), "Invalid status code [response_body=%s]", string(resp.Body)) metered, err := resp.JSON201.AsEntitlementMetered() require.NoError(t, err) require.Equal(t, metered.SubjectKey, subject2) // fetch grants for entitlement grantListResp, err := client.ListEntitlementGrantsWithResponse(context.Background(), subject2, metered.Id, nil) require.NoError(t, err) require.Equal(t, http.StatusOK, grantListResp.StatusCode()) require.NotNil(t, grantListResp.JSON200) require.Len(t, *grantListResp.JSON200, 1) }) }) t.Run("Override previous entitlement", func(t *testing.T) { meteredEntitlement := api.EntitlementMeteredCreateInputs{ Type: "metered", FeatureId: &featureId, UsagePeriod: api.RecurringPeriodCreateInput{ Anchor: convert.ToPointer(time.Date(2024, time.January, 1, 0, 0, 0, 0, time.UTC)), Interval: *apiMONTH, }, } body := &api.CreateEntitlementJSONRequestBody{} err := body.FromEntitlementMeteredCreateInputs(meteredEntitlement) require.NoError(t, err) subject := customer3.Key() CreateCustomerWithSubject(t, client, subject, subject) // create an entitlement resp, err := client.CreateEntitlementWithResponse(context.Background(), subject, *body) require.NoError(t, err) require.Equal(t, http.StatusCreated, resp.StatusCode(), "Invalid status code [response_body=%s]", string(resp.Body)) metered, err := resp.JSON201.AsEntitlementMetered() require.NoError(t, err) entId := metered.Id // Override entitlement resp2, err := client.OverrideEntitlementWithResponse(context.Background(), subject, entId, *body) require.NoError(t, err) require.Equal(t, http.StatusCreated, resp.StatusCode(), "Invalid status code [response_body=%s]", string(resp.Body)) require.NotNil(t, resp2) require.NotNil(t, resp2.JSON201) metered, err = resp2.JSON201.AsEntitlementMetered() require.NoError(t, err) require.Equal(t, metered.SubjectKey, subject) require.NotEqual(t, metered.Id, entId) }) t.Run("List entitlements", func(t *testing.T) { // should return 2 entitlements for subject for feature resp, err := client.ListEntitlementsWithResponse(context.Background(), &api.ListEntitlementsParams{ EntitlementType: &[]string{"metered"}, Subject: &[]string{subjectKey}, Feature: &[]string{featureKey}, Page: convert.ToPointer(1), PageSize: convert.ToPointer(10), }) require.NoError(t, err) require.Equal(t, http.StatusOK, resp.StatusCode()) require.NotNil(t, resp.JSON200) // should be a paginated response resBody, err := resp.JSON200.AsEntitlementPaginatedResponse() require.NoError(t, err) require.Equal(t, 1, resBody.Page) require.Equal(t, 1, resBody.TotalCount) require.Equal(t, 1, len(resBody.Items)) // should return 0 entitlements due to unused types resp, err = client.ListEntitlementsWithResponse(context.Background(), &api.ListEntitlementsParams{ EntitlementType: &[]string{"static", "boolean"}, Subject: &[]string{subjectKey}, Feature: &[]string{featureKey}, Page: convert.ToPointer(1), PageSize: convert.ToPointer(10), }) require.NoError(t, err) require.Equal(t, http.StatusOK, resp.StatusCode()) require.NotNil(t, resp.JSON200) // should be a paginated response resBody, err = resp.JSON200.AsEntitlementPaginatedResponse() require.NoError(t, err) require.Equal(t, 1, resBody.Page) require.Equal(t, 0, resBody.TotalCount) require.Equal(t, 0, len(resBody.Items)) // should return 400 for invalid type resp, err = client.ListEntitlementsWithResponse(context.Background(), &api.ListEntitlementsParams{ EntitlementType: &[]string{"INVALID_STR"}, Subject: &[]string{subjectKey}, Feature: &[]string{featureKey}, Page: convert.ToPointer(1), PageSize: convert.ToPointer(10), }) require.NoError(t, err) require.Equal(t, http.StatusBadRequest, resp.StatusCode()) }) }