From 62e04623cecd75fb2a8c5188a997ca5e90180e05 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 9 May 2026 19:46:57 -0700 Subject: [PATCH] test(s3/lifecycle): cover worker handler Detect + helpers (#9396) * test(s3/lifecycle): cover worker handler Detect + helpers 13 tests pin the worker-handler surface that runs without a live filer or S3 server. Pure helpers: clusterS3Endpoints (nil context, empty list, filter empty entries while preserving order, all-valid passthrough); readString (missing key, nil ConfigValue, wrong kind falls back, string returned). Capability advertises jobType with single-job concurrency caps. Detect: nil request / nil sender / wrong JobType all error; no S3 endpoints emits a 'skipped' activity and completes with success; no filer addresses behaves the same; the happy path proposes one job parameterized with the first filer address; empty JobType is accepted (broadcast detect); a SendProposals failure propagates without firing complete. * test(s3/lifecycle): cover SendComplete error propagation in worker Detect The recordingSender already supported forcing an err on SendComplete via errOn, but no case exercised it. A SendComplete failure must propagate so the admin learns the completion signal never landed; proposals went out before the failure so they remain recorded. --- .../worker/tasks/s3_lifecycle/handler_test.go | 268 ++++++++++++++++++ 1 file changed, 268 insertions(+) create mode 100644 weed/worker/tasks/s3_lifecycle/handler_test.go diff --git a/weed/worker/tasks/s3_lifecycle/handler_test.go b/weed/worker/tasks/s3_lifecycle/handler_test.go new file mode 100644 index 000000000..bc27b72bf --- /dev/null +++ b/weed/worker/tasks/s3_lifecycle/handler_test.go @@ -0,0 +1,268 @@ +package s3_lifecycle + +import ( + "context" + "errors" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Tests cover the worker-handler surface that runs without a live filer +// or S3 server: pure helpers (clusterS3Endpoints, readString), +// Capability/Descriptor sanity, and the Detect input-validation / +// skip-activity / proposal paths driven by a recorder sender. + +// recordingSender captures everything the handler sends so each Detect +// case can assert on activities, proposals, and completion. +type recordingSender struct { + proposals []*plugin_pb.DetectionProposals + completes []*plugin_pb.DetectionComplete + activities []*plugin_pb.ActivityEvent + // errOn forces the named send to fail; lets tests cover the + // SendComplete error-propagation path without a full transport stub. + errOn map[string]error +} + +func (r *recordingSender) SendProposals(p *plugin_pb.DetectionProposals) error { + if err := r.errOn["proposals"]; err != nil { + return err + } + r.proposals = append(r.proposals, p) + return nil +} +func (r *recordingSender) SendComplete(c *plugin_pb.DetectionComplete) error { + if err := r.errOn["complete"]; err != nil { + return err + } + r.completes = append(r.completes, c) + return nil +} +func (r *recordingSender) SendActivity(a *plugin_pb.ActivityEvent) error { + if err := r.errOn["activity"]; err != nil { + return err + } + r.activities = append(r.activities, a) + return nil +} + +// ---------- clusterS3Endpoints ---------- + +func TestClusterS3Endpoints_NilContext(t *testing.T) { + // A nil ClusterContext occurs in early-bootstrap detect calls; the + // handler must return an empty slice rather than panic. + assert.Nil(t, clusterS3Endpoints(nil)) +} + +func TestClusterS3Endpoints_EmptyList(t *testing.T) { + assert.Empty(t, clusterS3Endpoints(&plugin_pb.ClusterContext{})) +} + +func TestClusterS3Endpoints_FiltersEmptyEntries(t *testing.T) { + // An empty-string address represents a stale registry entry the + // master is about to evict; never dial that. Order of the surviving + // entries must be preserved so the handler dials a deterministic + // host across detect runs. + cc := &plugin_pb.ClusterContext{S3GrpcAddresses: []string{"s3a:8333", "", "s3b:8333", ""}} + assert.Equal(t, []string{"s3a:8333", "s3b:8333"}, clusterS3Endpoints(cc)) +} + +func TestClusterS3Endpoints_AllValid(t *testing.T) { + cc := &plugin_pb.ClusterContext{S3GrpcAddresses: []string{"s3a:8333", "s3b:8333"}} + assert.Equal(t, []string{"s3a:8333", "s3b:8333"}, clusterS3Endpoints(cc)) +} + +// ---------- readString ---------- + +func TestReadString_MissingKeyReturnsFallback(t *testing.T) { + got := readString(map[string]*plugin_pb.ConfigValue{}, "missing", "fallback") + assert.Equal(t, "fallback", got) +} + +func TestReadString_NilValueReturnsFallback(t *testing.T) { + got := readString(map[string]*plugin_pb.ConfigValue{"k": nil}, "k", "fallback") + assert.Equal(t, "fallback", got) +} + +func TestReadString_WrongKindReturnsFallback(t *testing.T) { + // Configs are typed; an Int64 in a string slot is a writer-side bug + // the handler must tolerate rather than panic on a type assertion. + values := map[string]*plugin_pb.ConfigValue{ + "k": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 42}}, + } + assert.Equal(t, "fallback", readString(values, "k", "fallback")) +} + +func TestReadString_StringValueReturned(t *testing.T) { + values := map[string]*plugin_pb.ConfigValue{ + "k": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: "expected"}}, + } + assert.Equal(t, "expected", readString(values, "k", "fallback")) +} + +// ---------- Capability ---------- + +func TestCapability_AdvertisesJobType(t *testing.T) { + h := NewHandler(nil) + cap := h.Capability() + require.NotNil(t, cap) + assert.Equal(t, jobType, cap.JobType) + assert.True(t, cap.CanDetect) + assert.True(t, cap.CanExecute) + // Admin caps concurrency at one job per worker; a rebuild of the + // scheduler must wait for the prior one to exit. + assert.Equal(t, int32(1), cap.MaxDetectionConcurrency) + assert.Equal(t, int32(1), cap.MaxExecutionConcurrency) +} + +// ---------- Detect ---------- + +func TestDetect_NilRequestErrors(t *testing.T) { + h := NewHandler(nil) + r := &recordingSender{} + err := h.Detect(context.Background(), nil, r) + require.Error(t, err) + assert.Empty(t, r.proposals) + assert.Empty(t, r.completes) +} + +func TestDetect_NilSenderErrors(t *testing.T) { + h := NewHandler(nil) + err := h.Detect(context.Background(), &plugin_pb.RunDetectionRequest{}, nil) + require.Error(t, err) +} + +func TestDetect_WrongJobTypeErrors(t *testing.T) { + // A request routed to this handler with a foreign JobType is the + // admin's bug; surface as an error so it's visible rather than + // silently emitting a bogus proposal. + h := NewHandler(nil) + r := &recordingSender{} + err := h.Detect(context.Background(), &plugin_pb.RunDetectionRequest{JobType: "different_job"}, r) + require.Error(t, err) + assert.Contains(t, err.Error(), "different_job") +} + +func TestDetect_NoS3EndpointsCompletesWithSkipActivity(t *testing.T) { + // A cluster with no S3 servers registered yet must not spawn the + // scheduler; emit a "skipped" activity for operator visibility and + // complete with success so the admin doesn't classify as a failure. + h := NewHandler(nil) + r := &recordingSender{} + err := h.Detect(context.Background(), &plugin_pb.RunDetectionRequest{ + JobType: jobType, + ClusterContext: &plugin_pb.ClusterContext{}, + }, r) + require.NoError(t, err) + assert.Empty(t, r.proposals, "no S3 endpoints must not yield a proposal") + require.Len(t, r.activities, 1) + assert.Equal(t, "skipped", r.activities[0].Stage) + assert.Contains(t, r.activities[0].Message, "no s3 servers") + require.Len(t, r.completes, 1) + assert.True(t, r.completes[0].Success, "skip is success, not failure") + assert.Equal(t, jobType, r.completes[0].JobType) +} + +func TestDetect_NoFilerAddressesCompletesWithSkipActivity(t *testing.T) { + h := NewHandler(nil) + r := &recordingSender{} + err := h.Detect(context.Background(), &plugin_pb.RunDetectionRequest{ + JobType: jobType, + ClusterContext: &plugin_pb.ClusterContext{ + S3GrpcAddresses: []string{"s3a:8333"}, + // no FilerGrpcAddresses + }, + }, r) + require.NoError(t, err) + assert.Empty(t, r.proposals) + require.Len(t, r.activities, 1) + assert.Equal(t, "skipped", r.activities[0].Stage) + assert.Contains(t, r.activities[0].Message, "no filer addresses") + require.Len(t, r.completes, 1) + assert.True(t, r.completes[0].Success) +} + +func TestDetect_HappyPathProposesOneJobWithFirstFilerAddress(t *testing.T) { + // Detect must propose exactly one job that targets the first filer + // address in the cluster context; the master refreshes the list so + // a stale entry self-heals on the next run. + h := NewHandler(nil) + r := &recordingSender{} + err := h.Detect(context.Background(), &plugin_pb.RunDetectionRequest{ + JobType: jobType, + ClusterContext: &plugin_pb.ClusterContext{ + S3GrpcAddresses: []string{"s3a:8333"}, + FilerGrpcAddresses: []string{"filer-a:18888", "filer-b:18888"}, + }, + }, r) + require.NoError(t, err) + assert.Empty(t, r.activities, "happy path emits no skip activity") + + require.Len(t, r.proposals, 1) + require.Len(t, r.proposals[0].Proposals, 1) + prop := r.proposals[0].Proposals[0] + assert.Equal(t, jobType, prop.JobType) + assert.NotEmpty(t, prop.ProposalId, "proposal id must be unique-per-run") + require.Contains(t, prop.Parameters, "filer_grpc_address") + val := prop.Parameters["filer_grpc_address"].GetStringValue() + assert.Equal(t, "filer-a:18888", val, "first reachable filer is dialed") + + require.Len(t, r.completes, 1) + assert.True(t, r.completes[0].Success) + assert.Equal(t, int32(1), r.completes[0].TotalProposals) +} + +func TestDetect_EmptyJobTypeAccepted(t *testing.T) { + // Detect is sometimes invoked with an unset JobType (broadcast + // detect); the handler must accept and behave as if it matched. + h := NewHandler(nil) + r := &recordingSender{} + err := h.Detect(context.Background(), &plugin_pb.RunDetectionRequest{ + ClusterContext: &plugin_pb.ClusterContext{ + S3GrpcAddresses: []string{"s3a:8333"}, + FilerGrpcAddresses: []string{"f:18888"}, + }, + }, r) + require.NoError(t, err) + require.Len(t, r.proposals, 1) +} + +func TestDetect_PropagatesProposalsSendError(t *testing.T) { + // SendProposals failing must propagate; otherwise the worker would + // silently report success despite never delivering the proposal. + h := NewHandler(nil) + want := errors.New("transport down") + r := &recordingSender{errOn: map[string]error{"proposals": want}} + err := h.Detect(context.Background(), &plugin_pb.RunDetectionRequest{ + JobType: jobType, + ClusterContext: &plugin_pb.ClusterContext{ + S3GrpcAddresses: []string{"s3a:8333"}, + FilerGrpcAddresses: []string{"f:18888"}, + }, + }, r) + assert.ErrorIs(t, err, want) + assert.Empty(t, r.proposals) + assert.Empty(t, r.completes, "complete must not fire when proposals fail") +} + +func TestDetect_PropagatesCompleteSendError(t *testing.T) { + // SendComplete failing must propagate; otherwise the worker would + // report success to the admin despite the completion signal never + // landing. Proposals went out before the failure, so they remain in + // the recorder. + h := NewHandler(nil) + want := errors.New("transport down") + r := &recordingSender{errOn: map[string]error{"complete": want}} + err := h.Detect(context.Background(), &plugin_pb.RunDetectionRequest{ + JobType: jobType, + ClusterContext: &plugin_pb.ClusterContext{ + S3GrpcAddresses: []string{"s3a:8333"}, + FilerGrpcAddresses: []string{"f:18888"}, + }, + }, r) + assert.ErrorIs(t, err, want) + assert.Len(t, r.proposals, 1, "proposals send before complete and remain recorded") + assert.Empty(t, r.completes) +}