chore: test timeline ingress publishing and filtering separately in unit tests

backend/ingress/handler_test.go

@@ -120,10 +120,11 @@ func TestIngress(t *testing.T) {
 			// Publish the test module to the event source
 			assert.NoError(t, eventSource.PublishModuleForTest(testModule))
 
+			timeline := timelineclient.NewFakePublisher()
 			svc := &service{
 				view:           syncView(ctx, eventSource),
 				client:         fv,
-				timelineClient: timelineclient.NewClient(ctx, timelineclient.NullConfig),
+				timelineClient: timeline,
 				routeTable:     routing.New(ctx, eventSource),
 			}
 			svc.handleHTTP(time.Now(), sch, reqKey, routes, rec, req, fv)
@@ -134,6 +135,13 @@ func TestIngress(t *testing.T) {
 				return
 			}
 			assert.Equal(t, response.Body, rec.Body.Bytes())
+
+			t.Log("ingress produces a timeline event")
+			assert.Equal(t, 1, len(timeline.Events))
+			ingress, ok := timeline.Events[0].(timelineclient.Ingress)
+			assert.True(t, ok, "expected ingress event, got %T", timeline.Events[0])
+			assert.Equal(t, ingress.Verb.Module, "test")
+			assert.Equal(t, ingress.Verb.Name, "getAlias")
 		})
 	}
 }

backend/timeline/integration_test.go

@@ -40,22 +40,6 @@ func TestTimeline(t *testing.T) {
 		in.Call("publisher", "publish", in.Obj{}, func(t testing.TB, resp in.Obj) {}),
 
 		in.SubTests(
-			in.SubTest{Name: "IngressEvent", Action: in.VerifyTimeline(1000, []*timelinepb.TimelineQuery_Filter{
-				{
-					Filter: &timelinepb.TimelineQuery_Filter_EventTypes{
-						EventTypes: &timelinepb.TimelineQuery_EventTypeFilter{
-							EventTypes: []timelinepb.EventType{timelinepb.EventType_EVENT_TYPE_INGRESS},
-						},
-					},
-				},
-			}, func(ctx context.Context, t testing.TB, events []*timelinepb.Event) {
-				assert.Equal(t, 1, len(events))
-				ingress, ok := events[0].Entry.(*timelinepb.Event_Ingress)
-				assert.True(t, ok, "expected ingress event")
-
-				assert.Equal(t, ingress.Ingress.VerbRef.Module, "ingress")
-				assert.Equal(t, ingress.Ingress.VerbRef.Name, "get")
-			})},
 			in.SubTest{Name: "CallEvent", Action: in.VerifyTimeline(1000, []*timelinepb.TimelineQuery_Filter{
 				{
 					Filter: &timelinepb.TimelineQuery_Filter_EventTypes{

backend/timeline/service_test.go

@@ -177,12 +177,11 @@ func TestStreamTimeline(t *testing.T) {
 func TestQueries(t *testing.T) {
 	t.Parallel()
 
-	service := createTestService(t, callEventsFixture(50))
-
 	type testCase struct {
 		name    string
 		query   *timelinepb.TimelineQuery
 		wantIDs []int
+		fixture *timelinepb.CreateEventsRequest
 	}
 
 	testCases := []testCase{{
@@ -191,13 +190,15 @@ func TestQueries(t *testing.T) {
 			Limit: 5,
 			Order: timelinepb.TimelineQuery_ORDER_DESC,
 		},
+		fixture: callEventsFixture(50),
 		wantIDs: []int{4, 3, 2, 1, 0},
 	}, {
 		name: "events in ascending order",
 		query: &timelinepb.TimelineQuery{
 			Limit: 5,
 			Order: timelinepb.TimelineQuery_ORDER_ASC,
 		},
+		fixture: callEventsFixture(50),
 		wantIDs: []int{45, 46, 47, 48, 49},
 	}, {
 		name: "events in ascending order with a lower than id filter",
@@ -208,6 +209,7 @@ func TestQueries(t *testing.T) {
 				higherIDThanFilter(45),
 			},
 		},
+		fixture: callEventsFixture(50),
 		wantIDs: []int{46, 47, 48, 49},
 	}, {
 		name: "events in descending order with a higher than id filter",
@@ -218,11 +220,37 @@ func TestQueries(t *testing.T) {
 				lowerIDThanFilter(5),
 			},
 		},
+		fixture: callEventsFixture(50),
 		wantIDs: []int{4, 3, 2, 1, 0},
+	}, {
+		name: "events filtered by ingress type",
+		query: &timelinepb.TimelineQuery{
+			Limit: 10,
+			Order: timelinepb.TimelineQuery_ORDER_DESC,
+			Filters: []*timelinepb.TimelineQuery_Filter{
+				evetTypeFilter(timelinepb.EventType_EVENT_TYPE_INGRESS),
+			},
+		},
+		fixture: &timelinepb.CreateEventsRequest{Entries: []*timelinepb.CreateEventsRequest_EventEntry{{
+			Timestamp: timestamppb.New(time.Now()),
+			Entry: &timelinepb.CreateEventsRequest_EventEntry_Ingress{
+				Ingress: &timelinepb.IngressEvent{},
+			},
+		}, {
+			Timestamp: timestamppb.New(time.Now()),
+			Entry: &timelinepb.CreateEventsRequest_EventEntry_Call{
+				Call: &timelinepb.CallEvent{
+					Request:  "request",
+					Response: "response",
+				},
+			},
+		}}},
+		wantIDs: []int{0},
 	}}
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
+			service := createTestService(t, tc.fixture)
 			iter, err := service.streamTimelineIter(t.Context(), &timelinepb.StreamTimelineRequest{Query: tc.query})
 			assert.NoError(t, err)
 			events := readEventIDs(t, 1, iter)

internal/timelineclient/client.go

@@ -146,3 +146,16 @@ func (c *RealClient) flushEvents(ctx context.Context, entries []*timelinepb.Crea
 	}
 	metrics.Inserted(ctx, len(entries))
 }
+
+// NewFakePublisher for testing.
+func NewFakePublisher() *FakePublisher {
+	return &FakePublisher{}
+}
+
+type FakePublisher struct {
+	Events []Event
+}
+
+func (f *FakePublisher) Publish(_ context.Context, event Event) {
+	f.Events = append(f.Events, event)
+}