mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-08-17 04:36:50 +00:00
test(s3/lifecycle): fake LifecycleDelete server (Phase 12 slice) (#9391)
* test(s3/lifecycle): fake LifecycleDelete server for component tests A reusable double for SeaweedS3LifecycleInternalServer with per-key FIFO outcome queues, a fallback Default, and recorded request capture. Tests of the worker pipeline that need to hit the proto boundary can queue up DONE/NOOP/RETRY/FATAL/SKIPPED_OBJECT_LOCK responses per (bucket, objectPath, versionId) and assert dispatch order against Recorded(). SetError flips the server into transport-failure mode without polluting the request log. * test(s3/lifecycle): use struct map key for FakeLifecycleServer queues Bucket / object path / version-id are user-supplied strings that can contain "/" or "@", which would collide if the queue map were keyed by "<bucket>/<object>@<version>". Switch to a struct key so the components stay separate. * test(s3/lifecycle): deep-copy recorded LifecycleDelete requests Tests that mutate a Recorded() entry — or a request pointer they already passed in — were able to corrupt the fake's bookkeeping because the slice carried shared pointers. Clone with proto.Clone at both record and read time so the fake holds an independent snapshot of every arriving request and hands callers an independent snapshot back. Tightened TestFake_VersionIDPartOfKey error checks while there.
This commit is contained in:
Chris Lu
@@ -0,0 +1,159 @@
|
||||
// Package lifecycletest provides reusable test doubles for the lifecycle
|
||||
// worker pipeline. The pieces here let component-level tests stand up the
|
||||
// gRPC boundary the worker dials at runtime without pulling in a real
|
||||
// S3ApiServer or filer.
|
||||
package lifecycletest
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
|
||||
"google.golang.org/protobuf/proto"
|
||||
)
|
||||
|
||||
// Outcome is what the fake returns for a single LifecycleDelete call.
|
||||
// Code is required; Reason is echoed verbatim into LifecycleDeleteResponse.
|
||||
type Outcome struct {
|
||||
Code s3_lifecycle_pb.LifecycleDeleteOutcome
|
||||
Reason string
|
||||
}
|
||||
|
||||
// Done, NoopResolved, RetryLater, Blocked, SkippedObjectLock are
|
||||
// constructors for the common outcomes; tests can also build Outcome
|
||||
// values directly when they need a specific Reason.
|
||||
func Done() Outcome {
|
||||
return Outcome{Code: s3_lifecycle_pb.LifecycleDeleteOutcome_DONE}
|
||||
}
|
||||
func NoopResolved(reason string) Outcome {
|
||||
return Outcome{Code: s3_lifecycle_pb.LifecycleDeleteOutcome_NOOP_RESOLVED, Reason: reason}
|
||||
}
|
||||
func RetryLater(reason string) Outcome {
|
||||
return Outcome{Code: s3_lifecycle_pb.LifecycleDeleteOutcome_RETRY_LATER, Reason: reason}
|
||||
}
|
||||
func Blocked(reason string) Outcome {
|
||||
return Outcome{Code: s3_lifecycle_pb.LifecycleDeleteOutcome_BLOCKED, Reason: reason}
|
||||
}
|
||||
func SkippedObjectLock(reason string) Outcome {
|
||||
return Outcome{Code: s3_lifecycle_pb.LifecycleDeleteOutcome_SKIPPED_OBJECT_LOCK, Reason: reason}
|
||||
}
|
||||
|
||||
// FakeLifecycleServer implements s3_lifecycle_pb.SeaweedS3LifecycleInternalServer.
|
||||
// It returns per-key queued outcomes (FIFO) and falls back to Default when a
|
||||
// key has no queued entry. Every received request is recorded; tests assert
|
||||
// against Recorded() in arrival order.
|
||||
//
|
||||
// A non-nil Err short-circuits everything — LifecycleDelete returns (nil, Err)
|
||||
// immediately, before the per-key lookup or the request recording. Use it to
|
||||
// simulate transport failures.
|
||||
//
|
||||
// All methods are safe for concurrent use. Outcomes/Default may be set at
|
||||
// construction or via Queue/SetDefault between calls; mid-call mutation is
|
||||
// supported but ordering across that boundary is undefined.
|
||||
type FakeLifecycleServer struct {
|
||||
s3_lifecycle_pb.UnimplementedSeaweedS3LifecycleInternalServer
|
||||
|
||||
mu sync.Mutex
|
||||
queues map[requestKey][]Outcome
|
||||
def Outcome
|
||||
err error
|
||||
received []*s3_lifecycle_pb.LifecycleDeleteRequest
|
||||
}
|
||||
|
||||
// requestKey is the map key for queues. A struct rather than a delimited
|
||||
// string so bucket/object/versionId values containing "/" or "@" can't
|
||||
// collide.
|
||||
type requestKey struct {
|
||||
bucket, objectPath, versionId string
|
||||
}
|
||||
|
||||
// NewFakeLifecycleServer returns a server whose Default outcome is DONE.
|
||||
// Most tests want a different default; call SetDefault to change it.
|
||||
func NewFakeLifecycleServer() *FakeLifecycleServer {
|
||||
return &FakeLifecycleServer{
|
||||
queues: map[requestKey][]Outcome{},
|
||||
def: Done(),
|
||||
}
|
||||
}
|
||||
|
||||
// Queue appends an outcome to the FIFO for (bucket, objectPath, versionId).
|
||||
// Subsequent calls matching the same key return outcomes in the order they
|
||||
// were queued; once the queue is drained, Default applies.
|
||||
func (f *FakeLifecycleServer) Queue(bucket, objectPath, versionId string, outcome Outcome) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
k := key(bucket, objectPath, versionId)
|
||||
f.queues[k] = append(f.queues[k], outcome)
|
||||
}
|
||||
|
||||
// SetDefault sets the outcome returned when no per-key queue entry remains.
|
||||
func (f *FakeLifecycleServer) SetDefault(o Outcome) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.def = o
|
||||
}
|
||||
|
||||
// SetError makes LifecycleDelete return (nil, err) on every call until
|
||||
// SetError(nil) clears it. The request is not recorded while Err is set —
|
||||
// transport-error tests should rely on the worker's own bookkeeping.
|
||||
func (f *FakeLifecycleServer) SetError(err error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.err = err
|
||||
}
|
||||
|
||||
// Recorded returns a deep-copied snapshot of every request the server has
|
||||
// received (excluding calls that returned a transport error). Both the
|
||||
// slice and each element are independent of the fake's internal state, so
|
||||
// callers can mutate freely without affecting later Recorded() snapshots.
|
||||
func (f *FakeLifecycleServer) Recorded() []*s3_lifecycle_pb.LifecycleDeleteRequest {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
out := make([]*s3_lifecycle_pb.LifecycleDeleteRequest, len(f.received))
|
||||
for i, r := range f.received {
|
||||
out[i] = cloneRequest(r)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// LifecycleDelete is the gRPC handler. It honors Err first, then dequeues
|
||||
// the per-key outcome, falling back to Default.
|
||||
func (f *FakeLifecycleServer) LifecycleDelete(ctx context.Context, req *s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
|
||||
f.mu.Lock()
|
||||
if f.err != nil {
|
||||
err := f.err
|
||||
f.mu.Unlock()
|
||||
return nil, err
|
||||
}
|
||||
// Record a deep copy so a caller mutating a Recorded() entry can't
|
||||
// reach back into the fake's bookkeeping (and so subsequent calls'
|
||||
// view of "what arrived" stays stable across assertions).
|
||||
f.received = append(f.received, cloneRequest(req))
|
||||
if req == nil {
|
||||
out := f.def
|
||||
f.mu.Unlock()
|
||||
return &s3_lifecycle_pb.LifecycleDeleteResponse{Outcome: out.Code, Reason: out.Reason}, nil
|
||||
}
|
||||
k := key(req.Bucket, req.ObjectPath, req.VersionId)
|
||||
q := f.queues[k]
|
||||
var out Outcome
|
||||
if len(q) > 0 {
|
||||
out = q[0]
|
||||
f.queues[k] = q[1:]
|
||||
} else {
|
||||
out = f.def
|
||||
}
|
||||
f.mu.Unlock()
|
||||
return &s3_lifecycle_pb.LifecycleDeleteResponse{Outcome: out.Code, Reason: out.Reason}, nil
|
||||
}
|
||||
|
||||
func key(bucket, objectPath, versionId string) requestKey {
|
||||
return requestKey{bucket: bucket, objectPath: objectPath, versionId: versionId}
|
||||
}
|
||||
|
||||
func cloneRequest(req *s3_lifecycle_pb.LifecycleDeleteRequest) *s3_lifecycle_pb.LifecycleDeleteRequest {
|
||||
if req == nil {
|
||||
return nil
|
||||
}
|
||||
return proto.Clone(req).(*s3_lifecycle_pb.LifecycleDeleteRequest)
|
||||
}
|
||||
@@ -0,0 +1,209 @@
|
||||
package lifecycletest
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestFake_DefaultIsDoneOutOfTheBox(t *testing.T) {
|
||||
// A test that doesn't queue anything should still get a non-error
|
||||
// response so it can exercise the worker's happy path.
|
||||
f := NewFakeLifecycleServer()
|
||||
resp, err := f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{
|
||||
Bucket: "b",
|
||||
ObjectPath: "k",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, resp)
|
||||
assert.Equal(t, s3_lifecycle_pb.LifecycleDeleteOutcome_DONE, resp.Outcome)
|
||||
assert.Equal(t, "", resp.Reason)
|
||||
}
|
||||
|
||||
func TestFake_QueuedOutcomesPopFIFO(t *testing.T) {
|
||||
// Per-key queue is FIFO and one-shot per entry; after the queue
|
||||
// drains, Default kicks in.
|
||||
f := NewFakeLifecycleServer()
|
||||
f.SetDefault(NoopResolved("nothing more queued"))
|
||||
f.Queue("b", "k", "", RetryLater("first"))
|
||||
f.Queue("b", "k", "", Blocked("second"))
|
||||
|
||||
got := []s3_lifecycle_pb.LifecycleDeleteOutcome{}
|
||||
reasons := []string{}
|
||||
for i := 0; i < 3; i++ {
|
||||
resp, err := f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{
|
||||
Bucket: "b", ObjectPath: "k",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
got = append(got, resp.Outcome)
|
||||
reasons = append(reasons, resp.Reason)
|
||||
}
|
||||
assert.Equal(t, []s3_lifecycle_pb.LifecycleDeleteOutcome{
|
||||
s3_lifecycle_pb.LifecycleDeleteOutcome_RETRY_LATER,
|
||||
s3_lifecycle_pb.LifecycleDeleteOutcome_BLOCKED,
|
||||
s3_lifecycle_pb.LifecycleDeleteOutcome_NOOP_RESOLVED,
|
||||
}, got)
|
||||
assert.Equal(t, []string{"first", "second", "nothing more queued"}, reasons)
|
||||
}
|
||||
|
||||
func TestFake_QueuesIsolatedByKey(t *testing.T) {
|
||||
// Queues are partitioned by (bucket, objectPath, versionId); a queued
|
||||
// outcome for one key must not bleed into another's lookup.
|
||||
f := NewFakeLifecycleServer()
|
||||
f.Queue("b", "a", "", Blocked("a-only"))
|
||||
f.Queue("b", "b", "", RetryLater("b-only"))
|
||||
|
||||
respA, err := f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{Bucket: "b", ObjectPath: "a"})
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, s3_lifecycle_pb.LifecycleDeleteOutcome_BLOCKED, respA.Outcome)
|
||||
|
||||
respB, err := f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{Bucket: "b", ObjectPath: "b"})
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, s3_lifecycle_pb.LifecycleDeleteOutcome_RETRY_LATER, respB.Outcome)
|
||||
}
|
||||
|
||||
func TestFake_VersionIDPartOfKey(t *testing.T) {
|
||||
// Two requests for the same bucket/objectPath but different
|
||||
// versionIds must address different queues.
|
||||
f := NewFakeLifecycleServer()
|
||||
f.Queue("b", "k", "v1", SkippedObjectLock("v1-locked"))
|
||||
f.Queue("b", "k", "v2", Done())
|
||||
|
||||
respV1, err := f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{Bucket: "b", ObjectPath: "k", VersionId: "v1"})
|
||||
require.NoError(t, err)
|
||||
respV2, err := f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{Bucket: "b", ObjectPath: "k", VersionId: "v2"})
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, s3_lifecycle_pb.LifecycleDeleteOutcome_SKIPPED_OBJECT_LOCK, respV1.Outcome)
|
||||
assert.Equal(t, s3_lifecycle_pb.LifecycleDeleteOutcome_DONE, respV2.Outcome)
|
||||
}
|
||||
|
||||
func TestFake_KeyComponentsWithDelimitersDoNotCollide(t *testing.T) {
|
||||
// String-concatenation keys would have made these two requests
|
||||
// indistinguishable. The struct key keeps them separate.
|
||||
f := NewFakeLifecycleServer()
|
||||
f.Queue("b/k", "", "", Blocked("variant-a"))
|
||||
f.Queue("b", "k", "", Done())
|
||||
|
||||
respA, err := f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{Bucket: "b/k", ObjectPath: ""})
|
||||
require.NoError(t, err)
|
||||
respB, err := f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{Bucket: "b", ObjectPath: "k"})
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, s3_lifecycle_pb.LifecycleDeleteOutcome_BLOCKED, respA.Outcome)
|
||||
assert.Equal(t, s3_lifecycle_pb.LifecycleDeleteOutcome_DONE, respB.Outcome)
|
||||
}
|
||||
|
||||
func TestFake_ErrShortCircuitsBeforeRecording(t *testing.T) {
|
||||
// Err makes LifecycleDelete return (nil, err) without recording the
|
||||
// request — transport-error tests rely on the worker's own
|
||||
// bookkeeping, not the fake's.
|
||||
f := NewFakeLifecycleServer()
|
||||
transportErr := errors.New("connection refused")
|
||||
f.SetError(transportErr)
|
||||
|
||||
resp, err := f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{Bucket: "b", ObjectPath: "k"})
|
||||
assert.Nil(t, resp)
|
||||
assert.ErrorIs(t, err, transportErr)
|
||||
assert.Empty(t, f.Recorded(), "transport-error calls must not be recorded")
|
||||
|
||||
// Clearing the error returns the server to normal behavior.
|
||||
f.SetError(nil)
|
||||
resp2, err2 := f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{Bucket: "b", ObjectPath: "k"})
|
||||
require.NoError(t, err2)
|
||||
require.NotNil(t, resp2)
|
||||
assert.Len(t, f.Recorded(), 1)
|
||||
}
|
||||
|
||||
func TestFake_RecordsRequestsInOrder(t *testing.T) {
|
||||
// Recorded() preserves arrival order so tests can assert that
|
||||
// dispatch happened in the expected sequence.
|
||||
f := NewFakeLifecycleServer()
|
||||
for _, key := range []string{"k1", "k2", "k3"} {
|
||||
_, err := f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{
|
||||
Bucket: "b", ObjectPath: key,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
}
|
||||
rec := f.Recorded()
|
||||
require.Len(t, rec, 3)
|
||||
assert.Equal(t, "k1", rec[0].ObjectPath)
|
||||
assert.Equal(t, "k2", rec[1].ObjectPath)
|
||||
assert.Equal(t, "k3", rec[2].ObjectPath)
|
||||
}
|
||||
|
||||
func TestFake_RecordedIsSnapshot(t *testing.T) {
|
||||
// Mutating the slice the caller got back must not bleed into the
|
||||
// fake's internal state — otherwise a flaky test could corrupt
|
||||
// bookkeeping for later assertions.
|
||||
f := NewFakeLifecycleServer()
|
||||
_, err := f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{Bucket: "b", ObjectPath: "k"})
|
||||
require.NoError(t, err)
|
||||
|
||||
snap := f.Recorded()
|
||||
require.Len(t, snap, 1)
|
||||
snap[0] = nil
|
||||
again := f.Recorded()
|
||||
require.Len(t, again, 1)
|
||||
assert.NotNil(t, again[0], "internal record must survive caller-side mutation of the snapshot")
|
||||
}
|
||||
|
||||
func TestFake_RecordedRequestsAreDeepCopies(t *testing.T) {
|
||||
// A caller mutating fields on a Recorded() entry must not bleed
|
||||
// into a later Recorded() snapshot. We also confirm the original
|
||||
// request the caller passed in stays decoupled from internal state
|
||||
// — proto.Clone runs at record time, so the caller's pointer is
|
||||
// no longer the one the fake holds.
|
||||
f := NewFakeLifecycleServer()
|
||||
orig := &s3_lifecycle_pb.LifecycleDeleteRequest{Bucket: "b", ObjectPath: "k", VersionId: "v"}
|
||||
_, err := f.LifecycleDelete(context.Background(), orig)
|
||||
require.NoError(t, err)
|
||||
|
||||
snap := f.Recorded()
|
||||
require.Len(t, snap, 1)
|
||||
snap[0].Bucket = "mutated-by-caller"
|
||||
snap[0].ObjectPath = "also-mutated"
|
||||
|
||||
again := f.Recorded()
|
||||
require.Len(t, again, 1)
|
||||
assert.Equal(t, "b", again[0].Bucket, "field mutations on a snapshot must not bleed back")
|
||||
assert.Equal(t, "k", again[0].ObjectPath)
|
||||
assert.Equal(t, "v", again[0].VersionId)
|
||||
|
||||
// The caller's own pointer is independent of the fake's record.
|
||||
orig.Bucket = "caller-mutates-original"
|
||||
yetAgain := f.Recorded()
|
||||
assert.Equal(t, "b", yetAgain[0].Bucket, "caller mutating the input pointer must not bleed in either")
|
||||
}
|
||||
|
||||
func TestFake_NilRequestUsesDefault(t *testing.T) {
|
||||
// gRPC won't deliver a nil request in practice, but defensive code
|
||||
// in the fake should still produce a deterministic response so a
|
||||
// regressing client doesn't panic the test process.
|
||||
f := NewFakeLifecycleServer()
|
||||
f.SetDefault(Blocked("no request"))
|
||||
resp, err := f.LifecycleDelete(context.Background(), nil)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, resp)
|
||||
assert.Equal(t, s3_lifecycle_pb.LifecycleDeleteOutcome_BLOCKED, resp.Outcome)
|
||||
}
|
||||
|
||||
func TestFake_ConcurrentCallsSerializeWithoutDeadlock(t *testing.T) {
|
||||
// The dispatcher fans dispatch across many goroutines; the fake
|
||||
// must not livelock or drop records under concurrent load.
|
||||
f := NewFakeLifecycleServer()
|
||||
const N = 64
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(N)
|
||||
for i := 0; i < N; i++ {
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
_, _ = f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{Bucket: "b", ObjectPath: "k"})
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
assert.Len(t, f.Recorded(), N)
|
||||
}
|
||||
Reference in New Issue
Block a user