From 6df6be4f9601074f8c0df6cba2ba86e2c615c20b Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Tue, 6 Oct 2026 22:06:02 +0000 Subject: [PATCH] test(stovepipe): cover List end to end Summary: Build on #795 with real gRPC/MySQL coverage for ingestion, deduplication, lifecycle summaries, and acceptance-time preservation. Separate suite setup, ingest workflows, focused List scenarios, and shared helpers. Seed deterministic read-path fixtures through storage interfaces on isolated queues, add polling diagnostics, and verify processing through durable acknowledgments. Mark end-to-end coverage complete in the RFC. Revert Plan: Revert this test and documentation change; runtime behavior is unchanged. API Changes: None. --- doc/rfc/stovepipe/list-api.md | 2 +- test/e2e/stovepipe/BUILD.bazel | 6 + test/e2e/stovepipe/harness_test.go | 191 ++++++++++++++++------------- test/e2e/stovepipe/ingest_test.go | 85 +++++++++++++ test/e2e/stovepipe/list_test.go | 150 ++++++++++++++++++++++ test/e2e/stovepipe/suite_test.go | 147 ++++------------------ 6 files changed, 371 insertions(+), 210 deletions(-) create mode 100644 test/e2e/stovepipe/ingest_test.go create mode 100644 test/e2e/stovepipe/list_test.go diff --git a/doc/rfc/stovepipe/list-api.md b/doc/rfc/stovepipe/list-api.md index fef8dc003..e6138de80 100644 --- a/doc/rfc/stovepipe/list-api.md +++ b/doc/rfc/stovepipe/list-api.md @@ -1,6 +1,6 @@ # Stovepipe List API -The [protobuf contract](../../../api/stovepipe/proto/stovepipe.proto) defines the approved API. Summary fields, acceptance-mapping storage, the List controller, and RPC/server wiring are implemented; end-to-end coverage remains. +The [protobuf contract](../../../api/stovepipe/proto/stovepipe.proto) defines the approved API. Summary fields, acceptance-mapping storage, the List controller, RPC/server wiring, and end-to-end coverage are implemented. ## Proposal diff --git a/test/e2e/stovepipe/BUILD.bazel b/test/e2e/stovepipe/BUILD.bazel index 155472721..a90281887 100644 --- a/test/e2e/stovepipe/BUILD.bazel +++ b/test/e2e/stovepipe/BUILD.bazel @@ -4,6 +4,8 @@ go_test( name = "go_default_test", srcs = [ "harness_test.go", + "ingest_test.go", + "list_test.go", "suite_test.go", ], data = [ @@ -20,11 +22,15 @@ go_test( ], deps = [ "//api/stovepipe/protopb:go_default_library", + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/storage/mysql:go_default_library", "//test/testutil:go_default_library", "@com_github_go_sql_driver_mysql//:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", "@com_github_stretchr_testify//require:go_default_library", "@com_github_stretchr_testify//suite:go_default_library", + "@com_github_uber_go_tally//:go_default_library", "@org_golang_google_grpc//:go_default_library", + "@org_golang_google_protobuf//proto:go_default_library", ], ) diff --git a/test/e2e/stovepipe/harness_test.go b/test/e2e/stovepipe/harness_test.go index 81173a41a..656cdf79c 100644 --- a/test/e2e/stovepipe/harness_test.go +++ b/test/e2e/stovepipe/harness_test.go @@ -14,28 +14,19 @@ package e2e_test -// Reusable e2e helpers so tests read as intent, not plumbing. They drive the -// stack through the real Stovepipe gRPC surface (Ingest) and observe outcomes -// two ways: -// -// - the synchronous side effects of Ingest via raw SQL on the storage DB -// (the request row and its (queue, URI) mapping) and the queue DB (the -// published process message); and -// - the asynchronous completion of the process stage by polling the queue -// backend's per-consumer-group delivery state until the message is acked. -// -// The process consumer runs inside the stovepipe-service container, so there is -// no in-process signal to await. Polling continues until the condition holds or -// Bazel's test timeout terminates a genuinely stuck suite. - import ( "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" pb "github.com/uber/submitqueue/api/stovepipe/protopb" + "github.com/uber/submitqueue/stovepipe/entity" + "google.golang.org/protobuf/proto" ) +// Container boundaries require polling; Bazel owns the convergence deadline. +const convergencePollInterval = 500 * time.Millisecond + func pollUntil(interval time.Duration, condition func() bool) { ticker := time.NewTicker(interval) defer ticker.Stop() @@ -47,72 +38,42 @@ func pollUntil(interval time.Duration, condition func() bool) { } } -// The process consumer's topic and consumer group as wired in -// service/stovepipe/server/main.go (topic name "process", consumer group -// "stovepipe-process"). awaitProcessed reads the queue backend's delivery state -// keyed by this group. const ( processTopic = "process" processConsumerGroup = "stovepipe-process" ) -// ingest admits a queue's head commit into the pipeline and returns the minted -// request id. func (s *StovepipeE2ESuite) ingest(queue string) string { - t := s.T() + s.T().Helper() resp, err := s.client.Ingest(s.ctx, &pb.IngestRequest{Queue: queue}) - require.NoError(t, err, "Ingest failed for queue %s", queue) - require.NotEmpty(t, resp.Id, "Ingest returned an empty id for queue %s", queue) + require.NoError(s.T(), err, "Ingest failed for queue %s", queue) + require.NotEmpty(s.T(), resp.Id, "Ingest returned an empty id for queue %s", queue) return resp.Id } -// requestRowCount returns the number of request rows with the given queue and id (0 or 1). -func (s *StovepipeE2ESuite) requestRowCount(queue, id string) int { - t := s.T() +func (s *StovepipeE2ESuite) assertIngestPersisted(queue, id string) { + s.T().Helper() var count int - require.NoError(t, s.db.QueryRow("SELECT COUNT(*) FROM request WHERE queue = ? AND id = ?", queue, id).Scan(&count), - "failed to count request rows for queue=%s id=%s", queue, id) - return count -} + require.NoError(s.T(), s.db.QueryRowContext(s.ctx, + "SELECT COUNT(*) FROM request WHERE queue = ? AND id = ?", queue, id).Scan(&count)) + assert.Equal(s.T(), 1, count) -// uriMapping returns the request id the (queue, URI) mapping points at. -func (s *StovepipeE2ESuite) uriMapping(queue string) string { - t := s.T() var mappedID string - require.NoError(t, s.db.QueryRow("SELECT request_id FROM request_uri WHERE queue = ?", queue).Scan(&mappedID), - "failed to read request_uri mapping for queue %s", queue) - return mappedID + require.NoError(s.T(), s.db.QueryRowContext(s.ctx, + "SELECT request_id FROM request_uri WHERE queue = ?", queue).Scan(&mappedID)) + assert.Equal(s.T(), id, mappedID) } -// publishedMessageCount returns the number of process messages published for the -// given tenant and request id (0 or 1). -func (s *StovepipeE2ESuite) publishedMessageCount(tenant, id string) int { - t := s.T() - var count int - require.NoError(t, s.queueDB.QueryRow("SELECT COUNT(*) FROM queue_messages WHERE tenant = ? AND topic = ? AND id = ?", tenant, processTopic, id).Scan(&count), - "failed to count queue messages for %s", id) - return count -} - -// awaitProcessed blocks until the process consumer has acked a message on the -// given queue's partition, proving the ingest→process pipeline ran end-to-end. -// The durable signal is the consumer group's acked-offset watermark in -// queue_offsets: it starts at 0 and only advances once a message is acked (see -// offset_store.go). We poll that rather than the message's own delivery-state -// row because the queue GCs acked messages (and their delivery state) from -// queue_messages once the watermark passes them. The queue uses the queue name -// as the message's partition key (see the ingest controller), so the partition -// key here is the queue. func (s *StovepipeE2ESuite) awaitProcessed(queue string) { + s.T().Helper() + // Acked messages can be garbage-collected; their offset watermark remains durable. const query = ` - SELECT offset_acked - FROM queue_offsets - WHERE tenant = ? AND consumer_group = ? AND topic = ? AND partition_key = ?` - pollUntil(processPollInterval, func() bool { + SELECT offset_acked FROM queue_offsets + WHERE tenant = ? AND consumer_group = ? AND topic = ? AND partition_key = ?` + pollUntil(convergencePollInterval, func() bool { var ackedOffset int64 - err := s.queueDB.QueryRow(query, queue, processConsumerGroup, processTopic, queue).Scan(&ackedOffset) + err := s.queueDB.QueryRowContext(s.ctx, query, queue, processConsumerGroup, processTopic, queue).Scan(&ackedOffset) if err != nil { - // sql.ErrNoRows means the partition offset is not initialized yet. s.log.Logf("acked offset for queue %s not ready yet: %v", queue, err) return false } @@ -121,23 +82,11 @@ func (s *StovepipeE2ESuite) awaitProcessed(queue string) { }) } -// assertIngestPersisted asserts the synchronous side effects of a successful -// Ingest: the request row, the (queue, URI) mapping pointing at the minted id, -// and exactly one published process message. -func (s *StovepipeE2ESuite) assertIngestPersisted(queue, id string) { - t := s.T() - assert.Equal(t, 1, s.requestRowCount(queue, id), "request row should be persisted for queue=%s id=%s", queue, id) - assert.Equal(t, id, s.uriMapping(queue), "URI mapping should point at the minted request id") - assert.Equal(t, 1, s.publishedMessageCount(queue, id), "should have published one process message for %s", id) -} - -// awaitRequestState blocks until the request row reaches want. buildsignal projects -// the build's terminal status onto the request, so this is the durable, black-box -// signal that the whole ingest→process→build→buildsignal chain converged. func (s *StovepipeE2ESuite) awaitRequestState(queue, id, want string) { - pollUntil(processPollInterval, func() bool { + s.T().Helper() + pollUntil(convergencePollInterval, func() bool { var state string - if err := s.db.QueryRow("SELECT state FROM request WHERE queue = ? AND id = ?", queue, id).Scan(&state); err != nil { + if err := s.db.QueryRowContext(s.ctx, "SELECT state FROM request WHERE queue = ? AND id = ?", queue, id).Scan(&state); err != nil { s.log.Logf("request queue=%s id=%s state not readable yet: %v", queue, id, err) return false } @@ -146,26 +95,19 @@ func (s *StovepipeE2ESuite) awaitRequestState(queue, id, want string) { }) } -// inFlightCount returns the queue row's in_flight_count, the process concurrency -// gate's counter. func (s *StovepipeE2ESuite) inFlightCount(queue string) int32 { - t := s.T() + s.T().Helper() var count int32 - require.NoError(t, s.db.QueryRow("SELECT in_flight_count FROM queue WHERE name = ?", queue).Scan(&count), - "failed to read in_flight_count for queue %s", queue) + require.NoError(s.T(), s.db.QueryRowContext(s.ctx, "SELECT in_flight_count FROM queue WHERE name = ?", queue).Scan(&count)) return count } -// awaitBuildStatus blocks until the build row for a request reaches want. The -// pipeline runs inside the stovepipe-service container, so the build's own status -// column is the durable, black-box signal that the poll loop converged: buildsignal -// is its only writer after build creates the row. func (s *StovepipeE2ESuite) awaitBuildStatus(queue, requestID, want string) { - pollUntil(processPollInterval, func() bool { + s.T().Helper() + pollUntil(convergencePollInterval, func() bool { var status string - err := s.db.QueryRow("SELECT status FROM build WHERE queue = ? AND request_id = ?", queue, requestID).Scan(&status) + err := s.db.QueryRowContext(s.ctx, "SELECT status FROM build WHERE queue = ? AND request_id = ?", queue, requestID).Scan(&status) if err != nil { - // sql.ErrNoRows means the build row is not created yet. s.log.Logf("build for queue=%s request=%s not readable yet: %v", queue, requestID, err) return false } @@ -173,3 +115,76 @@ func (s *StovepipeE2ESuite) awaitBuildStatus(queue, requestID, want string) { return status == want }) } + +func (s *StovepipeE2ESuite) listRequests(req *pb.ListRequest) *pb.ListResponse { + s.T().Helper() + response, err := s.client.List(s.ctx, req) + require.NoError(s.T(), err, "List failed for queue %s", req.Queue) + require.NotNil(s.T(), response) + return response +} + +func (s *StovepipeE2ESuite) awaitListedRequest(queue, requestID, state string) *pb.RequestSummary { + s.T().Helper() + var found *pb.RequestSummary + pollUntil(convergencePollInterval, func() bool { + response := s.listRequests(&pb.ListRequest{Queue: queue, PageSize: 200}) + require.Empty(s.T(), response.NextPageToken) + found = nil + matches := 0 + for _, summary := range response.Requests { + require.Equal(s.T(), queue, summary.Queue) + if summary.RequestId == requestID { + found = summary + matches++ + } + } + require.LessOrEqual(s.T(), matches, 1, "duplicate listing for %s", requestID) + s.log.Logf("List queue=%s id=%s matches=%d state=%q (want %q)", + queue, requestID, matches, found.GetRequestState(), state) + return matches == 1 && (state == "" || found.RequestState == state) + }) + return found +} + +func (s *StovepipeE2ESuite) assertListSummaries(response *pb.ListResponse, want ...*pb.RequestSummary) { + s.T().Helper() + require.Len(s.T(), response.Requests, len(want)) + for i, summary := range response.Requests { + assert.True(s.T(), proto.Equal(want[i], summary), "want %v, got %v", want[i], summary) + } +} + +// Fixed read-path fixtures guarantee timestamp ties without depending on clock resolution. +// Projection behavior is covered by the real-ingest scenarios, not these seeded records. +func (s *StovepipeE2ESuite) seedListSummary(queue, requestID string, acceptedAtMs int64) *pb.RequestSummary { + s.T().Helper() + summary := entity.RequestSummary{ + Queue: queue, RequestID: requestID, URI: "git://" + queue + "/" + requestID, + BaseURI: "git://repo/base", State: entity.RequestStateSucceeded, StateTimestampMs: acceptedAtMs + 100, + AcceptedAtMs: acceptedAtMs, OutcomeReason: entity.RequestOutcomeReasonBuildSucceeded, + RequestVersion: 3, Version: 1, + } + stores, err := s.appStorage.For(queue) + require.NoError(s.T(), err) + require.NoError(s.T(), stores.GetRequestSummaryStore().Create(s.ctx, summary)) + if acceptedAtMs > 0 { + require.NoError(s.T(), stores.GetRequestAcceptanceStore().Create(s.ctx, entity.RequestAcceptance{ + Queue: queue, AcceptedAtMs: acceptedAtMs, RequestID: requestID, + })) + } + return &pb.RequestSummary{ + Queue: queue, RequestId: requestID, ChangeUri: summary.URI, BaseUri: summary.BaseURI, + RequestState: string(summary.State), StateUpdatedAtMs: summary.StateTimestampMs, + AcceptedAtMs: acceptedAtMs, OutcomeReason: string(summary.OutcomeReason), + } +} + +func (s *StovepipeE2ESuite) seedListPaginationSummaries(queue string) []*pb.RequestSummary { + s.T().Helper() + lower := s.seedListSummary(queue, "list/lower", 1000) + ten := s.seedListSummary(queue, "list/10", 1500) + z := s.seedListSummary(queue, "list/z", 1500) + nine := s.seedListSummary(queue, "list/9", 1500) + return []*pb.RequestSummary{z, nine, ten, lower} +} diff --git a/test/e2e/stovepipe/ingest_test.go b/test/e2e/stovepipe/ingest_test.go new file mode 100644 index 000000000..4fd564d5d --- /dev/null +++ b/test/e2e/stovepipe/ingest_test.go @@ -0,0 +1,85 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package e2e_test + +import ( + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + pb "github.com/uber/submitqueue/api/stovepipe/protopb" +) + +func (s *StovepipeE2ESuite) TestIngest_HappyPath_Processes() { + const queue = "monorepo/main" + id := s.ingest(queue) + s.assertIngestPersisted(queue, id) + s.awaitProcessed(queue) +} + +func (s *StovepipeE2ESuite) TestIngest_Idempotent() { + const queue = "monorepo/release" + id := s.ingest(queue) + first := s.awaitListedRequest(queue, id, "") + require.Positive(s.T(), first.AcceptedAtMs) + + id2 := s.ingest(queue) + assert.Equal(s.T(), id, id2, "re-ingest of the same head should dedup to the same id") + second := s.awaitListedRequest(queue, id2, "") + assert.Equal(s.T(), first.AcceptedAtMs, second.AcceptedAtMs) +} + +func (s *StovepipeE2ESuite) TestIngest_IndependentQueuesReuseIDs() { + queues := []string{"monorepo/main", "monorepo/release"} + ids := []string{s.ingest(queues[0]), s.ingest(queues[1])} + require.Equal(s.T(), ids[0], ids[1], "independent queues may share a resource ID") + + for i, queue := range queues { + s.Run(queue, func() { + s.assertIngestPersisted(queue, ids[i]) + s.awaitBuildStatus(queue, ids[i], "succeeded") + s.awaitRequestState(queue, ids[i], "succeeded") + summary := s.awaitListedRequest(queue, ids[i], "succeeded") + assert.Equal(s.T(), "git://"+queue+"/HEAD", summary.ChangeUri) + }) + } +} + +func (s *StovepipeE2ESuite) TestIngest_RejectsUnconfiguredTenant() { + resp, err := s.client.Ingest(s.ctx, &pb.IngestRequest{Queue: "monorepo/unconfigured"}) + require.Error(s.T(), err) + assert.Nil(s.T(), resp) +} + +func (s *StovepipeE2ESuite) TestIngest_SlowBuild_PollsToCompletion() { + const queue = "monorepo/slow?buildrunner-fake=build-slow" + id := s.ingest(queue) + s.assertIngestPersisted(queue, id) + first := s.awaitListedRequest(queue, id, "") + require.Positive(s.T(), first.AcceptedAtMs) + + s.awaitProcessed(queue) + s.awaitBuildStatus(queue, id, "succeeded") + s.awaitRequestState(queue, id, "succeeded") + summary := s.awaitListedRequest(queue, id, "succeeded") + assert.Equal(s.T(), first.AcceptedAtMs, summary.AcceptedAtMs) + assert.GreaterOrEqual(s.T(), summary.StateUpdatedAtMs, summary.AcceptedAtMs) + assert.Equal(s.T(), "build_succeeded", summary.OutcomeReason) + history, err := s.client.GetRequestHistoryByID(s.ctx, &pb.GetRequestHistoryByIDRequest{Queue: queue, RequestId: id}) + require.NoError(s.T(), err) + require.NotEmpty(s.T(), history.Events) + assert.Equal(s.T(), "accepted", history.Events[0].GetRequestState()) + assert.Equal(s.T(), history.Events[0].TimestampMs, summary.AcceptedAtMs) + assert.Equal(s.T(), int32(0), s.inFlightCount(queue), + "a terminal build should release the queue's build slot") +} diff --git a/test/e2e/stovepipe/list_test.go b/test/e2e/stovepipe/list_test.go new file mode 100644 index 000000000..071faa64d --- /dev/null +++ b/test/e2e/stovepipe/list_test.go @@ -0,0 +1,150 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package e2e_test + +import ( + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + pb "github.com/uber/submitqueue/api/stovepipe/protopb" +) + +func (s *StovepipeE2ESuite) TestList_Pagination() { + want := s.seedListPaginationSummaries(listPaginationQueue) + first := s.listRequests(&pb.ListRequest{ + Queue: listPaginationQueue, PageSize: 2, + AcceptedLowerBound: &pb.ListRequest_AcceptedAtOrAfterMs{AcceptedAtOrAfterMs: 1000}, + AcceptedUpperBound: &pb.ListRequest_AcceptedBeforeMs{AcceptedBeforeMs: 2000}, + }) + s.assertListSummaries(first, want[:2]...) + require.NotEmpty(s.T(), first.NextPageToken) + + for _, tt := range []struct { + name string + req *pb.ListRequest + }{ + {"omitted continuation bounds", &pb.ListRequest{ + Queue: listPaginationQueue, PageSize: 3, PageToken: first.NextPageToken, + }}, + {"matching continuation bounds", &pb.ListRequest{ + Queue: listPaginationQueue, PageSize: 3, PageToken: first.NextPageToken, + AcceptedLowerBound: &pb.ListRequest_AcceptedAtOrAfterMs{AcceptedAtOrAfterMs: 1000}, + AcceptedUpperBound: &pb.ListRequest_AcceptedBeforeMs{AcceptedBeforeMs: 2000}, + }}, + } { + s.Run(tt.name, func() { + next := s.listRequests(tt.req) + s.assertListSummaries(next, want[2:]...) + assert.Empty(s.T(), next.NextPageToken) + }) + } +} + +func (s *StovepipeE2ESuite) TestList_TimeBounds() { + old := s.seedListSummary(listTimeBoundsQueue, "list/old", 500) + lower := s.seedListSummary(listTimeBoundsQueue, "list/lower", 1000) + upper := s.seedListSummary(listTimeBoundsQueue, "list/upper", 2000) + s.seedListSummary(listTimeBoundsQueue, "list/unknown", 0) + futureMs := time.Now().AddDate(1, 0, 0).UnixMilli() + future := s.seedListSummary(listTimeBoundsQueue, "list/future", futureMs) + + for _, tt := range []struct { + name string + req *pb.ListRequest + want []*pb.RequestSummary + }{ + {"defaults include old requests and exclude unknown and future times", + &pb.ListRequest{Queue: listTimeBoundsQueue}, []*pb.RequestSummary{upper, lower, old}}, + {"inclusive lower and exclusive upper", + &pb.ListRequest{ + Queue: listTimeBoundsQueue, + AcceptedLowerBound: &pb.ListRequest_AcceptedAtOrAfterMs{AcceptedAtOrAfterMs: 1000}, + AcceptedUpperBound: &pb.ListRequest_AcceptedBeforeMs{AcceptedBeforeMs: 2000}, + }, []*pb.RequestSummary{lower}}, + {"explicit future window", + &pb.ListRequest{ + Queue: listTimeBoundsQueue, + AcceptedLowerBound: &pb.ListRequest_AcceptedAtOrAfterMs{AcceptedAtOrAfterMs: futureMs}, + AcceptedUpperBound: &pb.ListRequest_AcceptedBeforeMs{AcceptedBeforeMs: futureMs + 1}, + }, []*pb.RequestSummary{future}}, + {"empty window", + &pb.ListRequest{ + Queue: listTimeBoundsQueue, + AcceptedUpperBound: &pb.ListRequest_AcceptedBeforeMs{AcceptedBeforeMs: 500}, + }, nil}, + } { + s.Run(tt.name, func() { + response := s.listRequests(tt.req) + s.assertListSummaries(response, tt.want...) + assert.Empty(s.T(), response.NextPageToken) + }) + } +} + +func (s *StovepipeE2ESuite) TestList_QueueIsolation() { + primary := s.seedListSummary(listPrimaryQueue, "list/shared", 1000) + other := s.seedListSummary(listOtherQueue, "list/shared", 1000) + otherOnly := s.seedListSummary(listOtherQueue, "list/other-only", 1500) + + for _, tt := range []struct { + queue string + want []*pb.RequestSummary + }{ + {listPrimaryQueue, []*pb.RequestSummary{primary}}, + {listOtherQueue, []*pb.RequestSummary{otherOnly, other}}, + } { + s.Run(tt.queue, func() { + response := s.listRequests(&pb.ListRequest{Queue: tt.queue}) + s.assertListSummaries(response, tt.want...) + assert.Empty(s.T(), response.NextPageToken) + }) + } +} + +func (s *StovepipeE2ESuite) TestList_InvalidRequests() { + s.seedListPaginationSummaries(listValidationQueue) + first := s.listRequests(&pb.ListRequest{ + Queue: listValidationQueue, PageSize: 1, + AcceptedLowerBound: &pb.ListRequest_AcceptedAtOrAfterMs{AcceptedAtOrAfterMs: 1000}, + AcceptedUpperBound: &pb.ListRequest_AcceptedBeforeMs{AcceptedBeforeMs: 2000}, + }) + require.NotEmpty(s.T(), first.NextPageToken) + + for _, tt := range []struct { + name string + req *pb.ListRequest + }{ + {"unconfigured queue", &pb.ListRequest{Queue: "monorepo/unconfigured"}}, + {"missing queue", &pb.ListRequest{}}, + {"negative page size", &pb.ListRequest{Queue: listValidationQueue, PageSize: -1}}, + {"oversized page", &pb.ListRequest{Queue: listValidationQueue, PageSize: 201}}, + {"malformed token", &pb.ListRequest{Queue: listValidationQueue, PageToken: "!"}}, + {"token for another queue", &pb.ListRequest{Queue: listOtherQueue, PageToken: first.NextPageToken}}, + {"changed continuation bounds", &pb.ListRequest{ + Queue: listValidationQueue, PageToken: first.NextPageToken, + AcceptedUpperBound: &pb.ListRequest_AcceptedBeforeMs{AcceptedBeforeMs: 2001}, + }}, + {"explicit zero upper bound", &pb.ListRequest{ + Queue: listValidationQueue, AcceptedUpperBound: &pb.ListRequest_AcceptedBeforeMs{}, + }}, + } { + s.Run(tt.name, func() { + response, err := s.client.List(s.ctx, tt.req) + require.Error(s.T(), err) + assert.Nil(s.T(), response) + }) + } +} diff --git a/test/e2e/stovepipe/suite_test.go b/test/e2e/stovepipe/suite_test.go index 65c3d77c6..e0a9e74a9 100644 --- a/test/e2e/stovepipe/suite_test.go +++ b/test/e2e/stovepipe/suite_test.go @@ -14,59 +14,44 @@ package e2e_test -// Stovepipe end-to-end tests. -// -// These tests use docker-compose from service/stovepipe/docker-compose.yml. -// They are hermetic: the stovepipe image is built from a staged context whose -// inputs (Bazel-built Linux binary, Dockerfile) are all declared data -// dependencies of the test target. -// -// Run with: -// -// make e2e-test -// -// or only this package: -// -// bazel test //test/e2e/stovepipe:stovepipe_test -// -// The stack runs the Stovepipe gRPC service plus a storage MySQL (request, -// request_uri, queue, build) and a queue MySQL (the pipeline stages). Unlike the -// integration suite (test/integration/stovepipe), which asserts only that Ingest -// *publishes* a process message, this suite additionally drives the asynchronous -// consumers to completion — proving the ingest→process→build→buildsignal pipeline -// runs end-to-end. +// Ingest scenarios drive the complete pipeline; List scenarios use fixed read-path +// fixtures. Both exercise the real gRPC service and MySQL in a hermetic Compose stack. +// Run with: bazel test //test/e2e/stovepipe:go_default_test import ( "context" "database/sql" + "strings" "testing" - "time" _ "github.com/go-sql-driver/mysql" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" + "github.com/uber-go/tally" pb "github.com/uber/submitqueue/api/stovepipe/protopb" + storagemysql "github.com/uber/submitqueue/stovepipe/extension/storage/mysql" "github.com/uber/submitqueue/test/testutil" "google.golang.org/grpc" ) -// The process consumer runs inside the stovepipe-service container, so this -// suite can only observe its completion black-box through the queue backend's -// delivery-state table — there is no in-process signal to await across the -// container boundary. A bounded poll is the deterministic-enough analog: -// processPollInterval bounds re-query frequency; Bazel's test timeout is the -// only convergence deadline. -const processPollInterval = 500 * time.Millisecond +const ( + listPaginationQueue = "e2e-stovepipe/list-pagination" + listTimeBoundsQueue = "e2e-stovepipe/list-time-bounds" + listPrimaryQueue = "e2e-stovepipe/list-primary" + listOtherQueue = "e2e-stovepipe/list-other" + listValidationQueue = "e2e-stovepipe/list-validation" +) type StovepipeE2ESuite struct { suite.Suite - ctx context.Context - log *testutil.TestLogger - stack *testutil.ComposeStack - client pb.StovepipeClient - db *sql.DB // storage database (request, request_uri) - queueDB *sql.DB // queue database (process stage) + ctx context.Context + log *testutil.TestLogger + stack *testutil.ComposeStack + client pb.StovepipeClient + db *sql.DB // storage database (request, request_uri) + queueDB *sql.DB // queue database (process stage) + appStorage *storagemysql.Storage } func TestStovepipeE2E(t *testing.T) { @@ -77,12 +62,13 @@ func (s *StovepipeE2ESuite) SetupSuite() { t := s.T() s.ctx = context.Background() s.log = testutil.NewTestLogger(t) + t.Setenv("MQ_TENANTS", strings.Join([]string{ + "monorepo/main", "monorepo/release", "monorepo/slow?buildrunner-fake=build-slow", + listPaginationQueue, listTimeBoundsQueue, listPrimaryQueue, listOtherQueue, listValidationQueue, + }, ",")) s.log.Logf("Starting Stovepipe e2e test suite using docker-compose") - // Compose file and image build inputs come from the test runfiles; the - // stovepipe image is built from a staged build context assembled entirely - // from declared data dependencies. composeFile := testutil.Runfile("service/stovepipe/docker-compose.yml") s.stack = testutil.NewComposeStack(t, s.log, s.ctx, composeFile, "e2e-stovepipe", testutil.WithBuildContext(map[string]string{ @@ -104,6 +90,8 @@ func (s *StovepipeE2ESuite) SetupSuite() { testutil.ApplySchema(t, s.log, s.db, testutil.SchemaDir("platform/extension/counter/mysql/schema")) testutil.ApplySchema(t, s.log, s.db, testutil.SchemaDir("stovepipe/extension/storage/mysql/schema")) testutil.ApplySchema(t, s.log, s.queueDB, testutil.SchemaDir("platform/extension/messagequeue/mysql/schema")) + s.appStorage, err = storagemysql.NewStorage(s.db, tally.NoopScope) + require.NoError(t, err) var conn *grpc.ClientConn conn, err = s.stack.ConnectGRPC("stovepipe-service", 8080) @@ -127,86 +115,3 @@ func (s *StovepipeE2ESuite) TestPing() { assert.NotEmpty(t, resp.Message) assert.NotZero(t, resp.Timestamp) } - -// TestIngest_HappyPath_Processes drives a queue's head commit through the whole -// pipeline. Ingest synchronously resolves the head URI via the (fake) -// SourceControl, persists the Request and its (queue, URI) mapping, and publishes -// the request id to the process stage; the process consumer then drains that -// message. This asserts both the synchronous side effects and — the piece the -// integration suite does not cover — that the async process stage acked the -// message. -func (s *StovepipeE2ESuite) TestIngest_HappyPath_Processes() { - const queue = "monorepo/main" - - id := s.ingest(queue) - s.log.Logf("Ingest succeeded: id=%s; waiting for process stage", id) - - // Synchronous side effects of Ingest. - s.assertIngestPersisted(queue, id) - - // Asynchronous completion: the process consumer acked the message. - s.awaitProcessed(queue) -} - -// TestIngest_Idempotent verifies that re-ingesting the same queue resolves the -// same head URI and dedups to the same request id. -func (s *StovepipeE2ESuite) TestIngest_Idempotent() { - const queue = "monorepo/release" - - id := s.ingest(queue) - s.log.Logf("First ingest: id=%s", id) - - id2 := s.ingest(queue) - assert.Equal(s.T(), id, id2, "re-ingest of the same head should dedup to the same id") -} - -func (s *StovepipeE2ESuite) TestIngest_IndependentQueuesReuseIDs() { - queues := []string{"monorepo/main", "monorepo/release"} - ids := []string{s.ingest(queues[0]), s.ingest(queues[1])} - require.Equal(s.T(), ids[0], ids[1], "independent queues may share a resource ID") - - for i, queue := range queues { - s.Run(queue, func() { - assert.Equal(s.T(), 1, s.requestRowCount(queue, ids[i])) - assert.Equal(s.T(), ids[i], s.uriMapping(queue)) - s.awaitBuildStatus(queue, ids[i], "succeeded") - s.awaitRequestState(queue, ids[i], "succeeded") - }) - } -} - -func (s *StovepipeE2ESuite) TestIngest_RejectsUnconfiguredTenant() { - resp, err := s.client.Ingest(s.ctx, &pb.IngestRequest{Queue: "monorepo/unconfigured"}) - require.Error(s.T(), err) - assert.Nil(s.T(), resp) -} - -// TestIngest_SlowBuild_PollsToCompletion drives a build that is not terminal on its -// first poll, which is the only path that exercises buildsignal's poll loop. -// -// The queue name carries a fake-buildrunner marker: the fake SourceControl resolves a -// queue to "git:///HEAD", so the marker rides into the head URI and the fake -// BuildRunner reports running for a while before succeeding. Reaching a terminal build -// status therefore requires the poll loop to tick more than once. -// -// The loop is driven by holding the delivery: each non-terminal poll postpones the -// same BuildSignal message, which redelivers after the poll delay without minting new -// rows or burning retry attempts. A build that stalled the loop would sit at `running` -// forever, so reaching a terminal status proves the held message kept redelivering. -func (s *StovepipeE2ESuite) TestIngest_SlowBuild_PollsToCompletion() { - const queue = "monorepo/slow?buildrunner-fake=build-slow" - - id := s.ingest(queue) - s.log.Logf("Ingest succeeded: id=%s; waiting for the poll loop to reach a terminal build", id) - - s.assertIngestPersisted(queue, id) - - // Getting here at all means the held delivery redelivered and re-polled. - s.awaitBuildStatus(queue, id, "succeeded") - - // buildsignal projects the terminal build status onto the request and, in the - // same step, releases the build slot that reopens the process gate. - s.awaitRequestState(queue, id, "succeeded") - assert.Equal(s.T(), int32(0), s.inFlightCount(queue), - "a terminal build should release the queue's build slot") -}