diff --git a/spinifex/ebsprovider/memory_test.go b/spinifex/ebsprovider/memory_test.go index 053ccac68..9f78d6691 100644 --- a/spinifex/ebsprovider/memory_test.go +++ b/spinifex/ebsprovider/memory_test.go @@ -267,6 +267,163 @@ func TestMemoryProviderSeedIsStoredOnce(t *testing.T) { assert.Equal(t, seed, stored, "an existing volume must never be reseeded") } +// TestMemoryProviderEnumerationRequiresCapability pins the refusal a caller +// branches on: a provider that cannot enumerate must say so, because an empty +// page and "I do not enumerate" mean opposite things to a reconciler deciding +// whether volumes with no metadata document have been orphaned. +func TestMemoryProviderEnumerationRequiresCapability(t *testing.T) { + provider := NewMemoryProvider(Capabilities{}) + ctx := context.Background() + + _, err := provider.ListVolumes(ctx, ListVolumesRequest{Versioned: NewVersioned()}) + require.ErrorIs(t, err, ErrUnsupportedCapability) + + _, err = provider.ListSnapshots(ctx, ListSnapshotsRequest{Versioned: NewVersioned()}) + require.ErrorIs(t, err, ErrUnsupportedCapability) +} + +// TestMemoryProviderListVolumesPages walks a full enumeration one page at a +// time. The token is the ID to resume after, so the property worth pinning is +// that concatenating the pages yields every volume exactly once, in ID order, +// with the last page carrying no token. +func TestMemoryProviderListVolumesPages(t *testing.T) { + provider := NewMemoryProvider(Capabilities{VolumeEnumeration: true}) + ctx := context.Background() + + empty, err := provider.ListVolumes(ctx, ListVolumesRequest{Versioned: NewVersioned()}) + require.NoError(t, err) + assert.Empty(t, empty.Volumes) + assert.Empty(t, empty.NextToken, "holding nothing is a last page, not a page to resume") + + for _, id := range []string{"vol-c", "vol-a", "vol-e", "vol-b", "vol-d"} { + _, err := provider.CreateVolume(ctx, CreateVolumeRequest{ + Versioned: NewVersioned(), VolumeID: id, CapacityRange: CapacityRange{RequiredBytes: 1 << 30}, + }) + require.NoError(t, err) + } + + var seen []VolumeRef + var token string + for range 3 { + page, err := provider.ListVolumes(ctx, ListVolumesRequest{ + Versioned: NewVersioned(), MaxResults: 2, StartingToken: token, + }) + require.NoError(t, err) + require.LessOrEqual(t, len(page.Volumes), 2, "a page must not exceed MaxResults") + seen = append(seen, page.Volumes...) + token = page.NextToken + if token == "" { + break + } + } + + assert.Empty(t, token, "the walk must terminate") + assert.Equal(t, []VolumeRef{ + {ID: "vol-a", Handle: "memory://volume/vol-a"}, + {ID: "vol-b", Handle: "memory://volume/vol-b"}, + {ID: "vol-c", Handle: "memory://volume/vol-c"}, + {ID: "vol-d", Handle: "memory://volume/vol-d"}, + {ID: "vol-e", Handle: "memory://volume/vol-e"}, + }, seen) +} + +// TestMemoryProviderListSnapshotsCarriesSourceVolume covers the one field that +// makes snapshot enumeration actionable: a snapshot found with no metadata +// document can only be reconciled once the volume it came from is known. +func TestMemoryProviderListSnapshotsCarriesSourceVolume(t *testing.T) { + provider := NewMemoryProvider(Capabilities{SnapshotEnumeration: true}) + ctx := context.Background() + + _, err := provider.CreateVolume(ctx, CreateVolumeRequest{ + Versioned: NewVersioned(), VolumeID: "vol-src", CapacityRange: CapacityRange{RequiredBytes: 1 << 30}, + }) + require.NoError(t, err) + for _, id := range []string{"snap-b", "snap-a"} { + _, err := provider.CreateSnapshot(ctx, CreateSnapshotRequest{ + Versioned: NewVersioned(), SnapshotID: id, VolumeID: "vol-src", + }) + require.NoError(t, err) + } + + first, err := provider.ListSnapshots(ctx, ListSnapshotsRequest{Versioned: NewVersioned(), MaxResults: 1}) + require.NoError(t, err) + assert.Equal(t, []SnapshotRef{{ID: "snap-a", SourceVolumeID: "vol-src", Handle: "memory://snapshot/snap-a"}}, first.Snapshots) + require.Equal(t, "snap-a", first.NextToken) + + second, err := provider.ListSnapshots(ctx, ListSnapshotsRequest{ + Versioned: NewVersioned(), MaxResults: 1, StartingToken: first.NextToken, + }) + require.NoError(t, err) + assert.Equal(t, []SnapshotRef{{ID: "snap-b", SourceVolumeID: "vol-src", Handle: "memory://snapshot/snap-b"}}, second.Snapshots) + assert.Empty(t, second.NextToken) +} + +// TestMemoryProviderCopySnapshot covers the duplication contract: the copy is +// an independent snapshot over the same frozen data, and every refusal is a +// case where copying would either lose data or attribute it to the wrong +// volume. +func TestMemoryProviderCopySnapshot(t *testing.T) { + provider := NewMemoryProvider(Capabilities{}) + copiedAt := time.Date(2026, 8, 5, 6, 0, 0, 0, time.UTC) + provider.now = func() time.Time { return copiedAt } + ctx := context.Background() + + _, err := provider.CreateVolume(ctx, CreateVolumeRequest{ + Versioned: NewVersioned(), VolumeID: "vol-src", CapacityRange: CapacityRange{RequiredBytes: 4 << 30}, + }) + require.NoError(t, err) + source, err := provider.CreateSnapshot(ctx, CreateSnapshotRequest{ + Versioned: NewVersioned(), SnapshotID: "snap-src", VolumeID: "vol-src", + }) + require.NoError(t, err) + + copied, err := provider.CopySnapshot(ctx, CopySnapshotRequest{ + Versioned: NewVersioned(), SourceSnapshotID: "snap-src", DestinationSnapshotID: "snap-dst", VolumeID: "vol-src", + }) + require.NoError(t, err) + assert.Equal(t, "snap-dst", copied.ID) + assert.Equal(t, "memory://snapshot/snap-dst", copied.Handle, "the copy must be independently addressable") + assert.Equal(t, source.SourceVolumeID, copied.SourceVolumeID) + assert.Equal(t, source.SizeBytes, copied.SizeBytes) + assert.Equal(t, copiedAt, copied.CreatedAt) + assert.Equal(t, SnapshotStateCompleted, copied.State) + + stored, ok := provider.Snapshot("snap-dst") + require.True(t, ok, "the copy must reach the provider, not just the response") + assert.Equal(t, copied, stored) + _, ok = provider.Snapshot("snap-absent") + assert.False(t, ok) + + refusals := []struct { + name string + req CopySnapshotRequest + want error + }{ + {"missing source", CopySnapshotRequest{DestinationSnapshotID: "snap-x", VolumeID: "vol-src"}, ErrInvalidArgument}, + {"missing destination", CopySnapshotRequest{SourceSnapshotID: "snap-src", VolumeID: "vol-src"}, ErrInvalidArgument}, + {"missing volume", CopySnapshotRequest{SourceSnapshotID: "snap-src", DestinationSnapshotID: "snap-x"}, ErrInvalidArgument}, + {"destination equals source", CopySnapshotRequest{SourceSnapshotID: "snap-src", DestinationSnapshotID: "snap-src", VolumeID: "vol-src"}, ErrInvalidArgument}, + {"unknown source", CopySnapshotRequest{SourceSnapshotID: "snap-absent", DestinationSnapshotID: "snap-x", VolumeID: "vol-src"}, ErrNotFound}, + {"wrong source volume", CopySnapshotRequest{SourceSnapshotID: "snap-src", DestinationSnapshotID: "snap-x", VolumeID: "vol-other"}, ErrInvalidArgument}, + {"destination exists", CopySnapshotRequest{SourceSnapshotID: "snap-src", DestinationSnapshotID: "snap-dst", VolumeID: "vol-src"}, ErrAlreadyExists}, + } + for _, tt := range refusals { + t.Run(tt.name, func(t *testing.T) { + req := tt.req + req.Versioned = NewVersioned() + _, err := provider.CopySnapshot(ctx, req) + require.ErrorIs(t, err, tt.want) + }) + } + + t.Run("version skew", func(t *testing.T) { + _, err := provider.CopySnapshot(ctx, CopySnapshotRequest{ + SourceSnapshotID: "snap-src", DestinationSnapshotID: "snap-y", VolumeID: "vol-src", + }) + require.ErrorIs(t, err, ErrUnsupportedVersion) + }) +} + func TestValidateSeedData(t *testing.T) { require.NoError(t, ValidateSeedData(nil)) require.NoError(t, ValidateSeedData(make([]byte, MaxSeedBytes))) diff --git a/spinifex/ebsprovider/nats_test.go b/spinifex/ebsprovider/nats_test.go index 6c352e44b..36eccf722 100644 --- a/spinifex/ebsprovider/nats_test.go +++ b/spinifex/ebsprovider/nats_test.go @@ -365,6 +365,221 @@ func TestNATSProviderOwnerFirst_NonNoRespondersDoesNotFallBack(t *testing.T) { assert.Equal(t, int32(0), queueHits.Load(), "a timeout must never fall back to the queue-group subject") } +// respondWith subscribes subject with a canned reply, so a client-side guard +// can be driven against a response shape a healthy server never sends. +func respondWith(t *testing.T, conn *nats.Conn, subject string, response any) { + t.Helper() + payload, err := json.Marshal(response) + require.NoError(t, err) + sub, err := conn.Subscribe(subject, func(msg *nats.Msg) { + require.NoError(t, msg.Respond(payload)) + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + require.NoError(t, conn.Flush()) +} + +// TestNATSProviderRejectsEmptySuccessResponse covers the nil-payload guard on +// every method that returns a pointer. A reply carrying neither an error nor a +// body is a server bug, and without these guards each method would hand the +// caller a nil *Volume or *Snapshot alongside a nil error. +func TestNATSProviderRejectsEmptySuccessResponse(t *testing.T) { + _, conn := testutil.StartTestNATS(t) + provider := NewNATSProvider(conn, time.Second) + ctx := t.Context() + empty := Versioned{SchemaVersion: SchemaVersion} + + t.Run("CreateVolume", func(t *testing.T) { + respondWith(t, conn, CreateVolumeSubject, CreateVolumeResponse{Versioned: empty}) + _, err := provider.CreateVolume(ctx, CreateVolumeRequest{Versioned: NewVersioned(), VolumeID: "vol-empty"}) + require.EqualError(t, err, "ebs.create returned no volume") + }) + + t.Run("GetVolume", func(t *testing.T) { + respondWith(t, conn, GetVolumeSubject, GetVolumeResponse{Versioned: empty}) + _, err := provider.GetVolume(ctx, GetVolumeRequest{Versioned: NewVersioned(), VolumeID: "vol-empty"}) + require.EqualError(t, err, "ebs.describe returned no volume") + }) + + t.Run("ExpandVolume", func(t *testing.T) { + respondWith(t, conn, ExpandVolumeSubject, ExpandVolumeResponse{Versioned: empty}) + _, err := provider.ExpandVolume(ctx, ExpandVolumeRequest{Versioned: NewVersioned(), VolumeID: "vol-empty"}) + require.EqualError(t, err, "ebs.expand returned no volume") + }) + + t.Run("CopySnapshot", func(t *testing.T) { + respondWith(t, conn, CopySnapshotSubject, CopySnapshotResponse{Versioned: empty}) + _, err := provider.CopySnapshot(ctx, CopySnapshotRequest{ + Versioned: NewVersioned(), SourceSnapshotID: "snap-1", DestinationSnapshotID: "snap-2", VolumeID: "vol-empty", + }) + require.EqualError(t, err, "ebs.snapshot.copy returned no snapshot") + }) + + t.Run("PublishVolume", func(t *testing.T) { + subject, err := PublishSubject("node-empty") + require.NoError(t, err) + respondWith(t, conn, subject, PublishVolumeResponse{Versioned: empty}) + _, err = provider.PublishVolume(ctx, PublishVolumeRequest{Versioned: NewVersioned(), VolumeID: "vol-empty", NodeID: "node-empty"}) + require.EqualError(t, err, subject+" returned no published volume") + }) + + t.Run("CreateSnapshot", func(t *testing.T) { + subject, err := SnapshotSubject("vol-empty") + require.NoError(t, err) + respondWith(t, conn, subject, CreateSnapshotResponse{Versioned: empty}) + _, err = provider.CreateSnapshot(ctx, CreateSnapshotRequest{Versioned: NewVersioned(), SnapshotID: "snap-empty", VolumeID: "vol-empty"}) + require.EqualError(t, err, subject+" returned no operation ID") + }) +} + +// TestNATSProviderPropagatesProviderErrors covers responseError's populated +// branch on the enumeration and unpublish verbs, whose errors otherwise reach +// the caller as an empty page or a silent success. +func TestNATSProviderPropagatesProviderErrors(t *testing.T) { + t.Run("ListSnapshots refusal", func(t *testing.T) { + _, conn := testutil.StartTestNATS(t) + respondWith(t, conn, ListSnapshotsSubject, ListSnapshotsResponse{ + Versioned: NewVersioned(), Error: &ProviderError{Code: ErrorCodeUnsupportedCap, Message: "snapshot enumeration"}, + }) + _, err := NewNATSProvider(conn, time.Second).ListSnapshots(t.Context(), ListSnapshotsRequest{Versioned: NewVersioned()}) + require.ErrorIs(t, err, ErrUnsupportedCapability) + }) + + t.Run("ListSnapshots page", func(t *testing.T) { + _, conn := testutil.StartTestNATS(t) + respondWith(t, conn, ListSnapshotsSubject, ListSnapshotsResponse{ + Versioned: NewVersioned(), + Snapshots: []SnapshotRef{{ID: "snap-1", SourceVolumeID: "vol-1", Handle: "vb://snap-1"}}, + NextToken: "snap-1", + }) + page, err := NewNATSProvider(conn, time.Second).ListSnapshots(t.Context(), ListSnapshotsRequest{Versioned: NewVersioned()}) + require.NoError(t, err) + assert.Equal(t, []SnapshotRef{{ID: "snap-1", SourceVolumeID: "vol-1", Handle: "vb://snap-1"}}, page.Snapshots) + assert.Equal(t, "snap-1", page.NextToken) + }) + + t.Run("UnpublishVolume", func(t *testing.T) { + _, conn := testutil.StartTestNATS(t) + subject, err := UnpublishSubject("node-err") + require.NoError(t, err) + respondWith(t, conn, subject, UnpublishVolumeResponse{ + Versioned: NewVersioned(), Error: &ProviderError{Code: ErrorCodeNotFound, Message: "no such mount"}, + }) + err = NewNATSProvider(conn, time.Second).UnpublishVolume(t.Context(), UnpublishVolumeRequest{ + Versioned: NewVersioned(), VolumeID: "vol-1", NodeID: "node-err", + }) + require.ErrorIs(t, err, ErrNotFound) + }) +} + +// TestNATSProviderValidatesSubjectTokensBeforeTransport extends the check to +// every method that embeds a caller-supplied ID in a subject: an unsafe token +// must be refused before a request is published, or the client publishes to a +// wildcard subject and takes an unrelated node's reply as its own. +func TestNATSProviderValidatesSubjectTokensBeforeTransport(t *testing.T) { + provider := NewNATSProvider(nil, time.Second) + ctx := t.Context() + + t.Run("GetVolume", func(t *testing.T) { + _, err := provider.GetVolume(ctx, GetVolumeRequest{Versioned: NewVersioned(), VolumeID: "vol.*"}) + require.ErrorIs(t, err, ErrInvalidArgument) + }) + t.Run("ExpandVolume", func(t *testing.T) { + _, err := provider.ExpandVolume(ctx, ExpandVolumeRequest{Versioned: NewVersioned(), VolumeID: "vol.*"}) + require.ErrorIs(t, err, ErrInvalidArgument) + }) + t.Run("CopySnapshot", func(t *testing.T) { + _, err := provider.CopySnapshot(ctx, CopySnapshotRequest{ + Versioned: NewVersioned(), SourceSnapshotID: "snap-1", DestinationSnapshotID: "snap-2", VolumeID: "vol.*", + }) + require.ErrorIs(t, err, ErrInvalidArgument) + }) + t.Run("CreateSnapshot volume", func(t *testing.T) { + _, err := provider.CreateSnapshot(ctx, CreateSnapshotRequest{Versioned: NewVersioned(), SnapshotID: "snap-1", VolumeID: "vol.*"}) + require.ErrorIs(t, err, ErrInvalidArgument) + }) + t.Run("CreateSnapshot snapshot", func(t *testing.T) { + _, err := provider.CreateSnapshot(ctx, CreateSnapshotRequest{Versioned: NewVersioned(), SnapshotID: "snap.*", VolumeID: "vol-1"}) + require.ErrorIs(t, err, ErrInvalidArgument) + }) + t.Run("UnpublishVolume", func(t *testing.T) { + err := provider.UnpublishVolume(ctx, UnpublishVolumeRequest{Versioned: NewVersioned(), VolumeID: "vol-1", NodeID: "node.*"}) + require.ErrorIs(t, err, ErrInvalidArgument) + }) +} + +// TestNATSProviderCreateSnapshotRejectsMismatchedCompletion covers the two +// identity checks on the asynchronous path. Both exist because the completion +// arrives out of band: a subject or operation ID the client did not ask for +// means it is reading someone else's snapshot result. +func TestNATSProviderCreateSnapshotRejectsMismatchedCompletion(t *testing.T) { + t.Run("completion subject mismatch", func(t *testing.T) { + _, conn := testutil.StartTestNATS(t) + subject, err := SnapshotSubject("vol-mismatch") + require.NoError(t, err) + respondWith(t, conn, subject, CreateSnapshotResponse{ + Versioned: NewVersioned(), OperationID: "op-1", CompletionSubject: "ebs.provider.v1.snapshot.response.someone-else", + }) + + provider := NewNATSProvider(conn, time.Second) + _, err = provider.CreateSnapshot(t.Context(), CreateSnapshotRequest{ + Versioned: NewVersioned(), SnapshotID: "snap-mismatch", VolumeID: "vol-mismatch", + }) + require.Error(t, err) + assert.Contains(t, err.Error(), "snapshot completion subject mismatch") + }) + + t.Run("operation id mismatch", func(t *testing.T) { + _, conn := testutil.StartTestNATS(t) + subject, err := SnapshotSubject("vol-opid") + require.NoError(t, err) + completionSubject, err := SnapshotCompletionSubject("snap-opid") + require.NoError(t, err) + + sub, err := conn.Subscribe(subject, func(msg *nats.Msg) { + accepted, marshalErr := json.Marshal(CreateSnapshotResponse{ + Versioned: NewVersioned(), OperationID: "op-expected", CompletionSubject: completionSubject, + }) + require.NoError(t, marshalErr) + require.NoError(t, msg.Respond(accepted)) + + completed, marshalErr := json.Marshal(CreateSnapshotResponse{ + Versioned: NewVersioned(), OperationID: "op-other", + Snapshot: &Snapshot{ID: "snap-opid", SourceVolumeID: "vol-opid", State: SnapshotStateCompleted}, + }) + require.NoError(t, marshalErr) + require.NoError(t, conn.Publish(completionSubject, completed)) + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + require.NoError(t, conn.Flush()) + + provider := NewNATSProvider(conn, time.Second) + _, err = provider.CreateSnapshot(t.Context(), CreateSnapshotRequest{ + Versioned: NewVersioned(), SnapshotID: "snap-opid", VolumeID: "vol-opid", + }) + require.Error(t, err) + assert.Contains(t, err.Error(), "snapshot operation mismatch") + }) + + t.Run("already completed reply short-circuits the wait", func(t *testing.T) { + _, conn := testutil.StartTestNATS(t) + subject, err := SnapshotSubject("vol-sync") + require.NoError(t, err) + respondWith(t, conn, subject, CreateSnapshotResponse{ + Versioned: NewVersioned(), OperationID: "op-sync", + Snapshot: &Snapshot{ID: "snap-sync", SourceVolumeID: "vol-sync", State: SnapshotStateCompleted}, + }) + + provider := NewNATSProvider(conn, time.Second) + snapshot, err := provider.CreateSnapshot(t.Context(), CreateSnapshotRequest{ + Versioned: NewVersioned(), SnapshotID: "snap-sync", VolumeID: "vol-sync", + }) + require.NoError(t, err, "a server answering synchronously must not be made to wait for a completion it will never publish") + assert.Equal(t, "snap-sync", snapshot.ID) + }) +} + // TestResponseErrorRejectsZeroSchemaVersion covers responseError directly: a // reply carrying the zero SchemaVersion (e.g. a server that forgot to set // Versioned on an error path) must be rejected before its Error field is diff --git a/spinifex/ebsprovider/natsserve/handlers_test.go b/spinifex/ebsprovider/natsserve/handlers_test.go new file mode 100644 index 000000000..d3571e351 --- /dev/null +++ b/spinifex/ebsprovider/natsserve/handlers_test.go @@ -0,0 +1,544 @@ +package natsserve_test + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/mulgadc/spinifex/spinifex/ebsprovider" + "github.com/mulgadc/spinifex/spinifex/ebsprovider/natsserve" + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const requestTimeout = 2 * time.Second + +// wireEnvelope decodes the two fields every response in the contract carries, +// so one type reads the reply of any subject in the table below. +type wireEnvelope struct { + ebsprovider.Versioned + + Error *ebsprovider.ProviderError `json:"error"` +} + +func decodeEnvelope(t *testing.T, data []byte) wireEnvelope { + t.Helper() + var envelope wireEnvelope + require.NoError(t, json.Unmarshal(data, &envelope)) + return envelope +} + +// serveProvider starts a server over its own NATS connection and returns the +// connection, so a test can drive raw payloads the typed client would refuse +// to send. +func serveProvider(t *testing.T, provider ebsprovider.EBSProvider, opts natsserve.Options) *nats.Conn { + t.Helper() + _, conn := testutil.StartTestNATS(t) + stop, err := natsserve.Serve(t.Context(), conn, provider, opts) + require.NoError(t, err) + t.Cleanup(stop) + return conn +} + +// failingProvider fails every verb with one wrapped sentinel, so a handler's +// classifyError arm can be reached without contriving provider state. +type failingProvider struct{ err error } + +var _ ebsprovider.EBSProvider = (*failingProvider)(nil) + +func (p *failingProvider) GetCapabilities(context.Context, ebsprovider.GetCapabilitiesRequest) (*ebsprovider.GetCapabilitiesResponse, error) { + return nil, p.err +} + +func (p *failingProvider) CreateVolume(context.Context, ebsprovider.CreateVolumeRequest) (*ebsprovider.Volume, error) { + return nil, p.err +} + +func (p *failingProvider) GetVolume(context.Context, ebsprovider.GetVolumeRequest) (*ebsprovider.Volume, error) { + return nil, p.err +} + +func (p *failingProvider) ListVolumes(context.Context, ebsprovider.ListVolumesRequest) (*ebsprovider.ListVolumesResponse, error) { + return nil, p.err +} + +func (p *failingProvider) ExpandVolume(context.Context, ebsprovider.ExpandVolumeRequest) (*ebsprovider.Volume, error) { + return nil, p.err +} + +func (p *failingProvider) DeleteVolume(context.Context, ebsprovider.DeleteVolumeRequest) error { + return p.err +} + +func (p *failingProvider) CreateSnapshot(context.Context, ebsprovider.CreateSnapshotRequest) (*ebsprovider.Snapshot, error) { + return nil, p.err +} + +func (p *failingProvider) DeleteSnapshot(context.Context, ebsprovider.DeleteSnapshotRequest) error { + return p.err +} + +func (p *failingProvider) CopySnapshot(context.Context, ebsprovider.CopySnapshotRequest) (*ebsprovider.Snapshot, error) { + return nil, p.err +} + +func (p *failingProvider) ListSnapshots(context.Context, ebsprovider.ListSnapshotsRequest) (*ebsprovider.ListSnapshotsResponse, error) { + return nil, p.err +} + +func (p *failingProvider) PublishVolume(context.Context, ebsprovider.PublishVolumeRequest) (*ebsprovider.PublishedVolume, error) { + return nil, p.err +} + +func (p *failingProvider) UnpublishVolume(context.Context, ebsprovider.UnpublishVolumeRequest) error { + return p.err +} + +// wireVerb names one subject Serve subscribes, a valid request body for it, +// and — for the one asynchronous verb — the subject its provider failure is +// published to rather than returned in the reply. +type wireVerb struct { + name string + subject string + request any + completionSubject string +} + +func wireVerbs(t *testing.T) []wireVerb { + t.Helper() + snapshotCreateSubject, err := ebsprovider.SnapshotSubject("vol-1") + require.NoError(t, err) + completionSubject, err := ebsprovider.SnapshotCompletionSubject("snap-1") + require.NoError(t, err) + mountSubject, err := ebsprovider.PublishSubject("node-1") + require.NoError(t, err) + unmountSubject, err := ebsprovider.UnpublishSubject("node-1") + require.NoError(t, err) + + return []wireVerb{ + {name: "capabilities", subject: ebsprovider.CapabilitiesSubject, + request: ebsprovider.GetCapabilitiesRequest{Versioned: ebsprovider.NewVersioned()}}, + {name: "volume.create", subject: ebsprovider.CreateVolumeSubject, + request: ebsprovider.CreateVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-1", CapacityRange: ebsprovider.CapacityRange{RequiredBytes: 1 << 30}}}, + {name: "volume.describe", subject: ebsprovider.GetVolumeSubject, + request: ebsprovider.GetVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-1"}}, + {name: "volume.list", subject: ebsprovider.ListVolumesSubject, + request: ebsprovider.ListVolumesRequest{Versioned: ebsprovider.NewVersioned()}}, + {name: "volume.expand", subject: ebsprovider.ExpandVolumeSubject, + request: ebsprovider.ExpandVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-1", CapacityRange: ebsprovider.CapacityRange{RequiredBytes: 2 << 30}}}, + {name: "volume.delete", subject: ebsprovider.DeleteVolumeSubject, + request: ebsprovider.DeleteVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-1"}}, + {name: "snapshot.delete", subject: ebsprovider.DeleteSnapshotSubject, + request: ebsprovider.DeleteSnapshotRequest{Versioned: ebsprovider.NewVersioned(), SnapshotID: "snap-1"}}, + {name: "snapshot.copy", subject: ebsprovider.CopySnapshotSubject, + request: ebsprovider.CopySnapshotRequest{Versioned: ebsprovider.NewVersioned(), SourceSnapshotID: "snap-1", DestinationSnapshotID: "snap-2", VolumeID: "vol-1"}}, + {name: "snapshot.list", subject: ebsprovider.ListSnapshotsSubject, + request: ebsprovider.ListSnapshotsRequest{Versioned: ebsprovider.NewVersioned()}}, + {name: "snapshot.create", subject: snapshotCreateSubject, completionSubject: completionSubject, + request: ebsprovider.CreateSnapshotRequest{Versioned: ebsprovider.NewVersioned(), SnapshotID: "snap-1", VolumeID: "vol-1"}}, + {name: "mount", subject: mountSubject, + request: ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-1", NodeID: "node-1"}}, + {name: "unmount", subject: unmountSubject, + request: ebsprovider.UnpublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-1", NodeID: "node-1"}}, + } +} + +// TestHandlersRejectMalformedRequests pins the caller-fault arm of every +// subject: a body that is not JSON must come back as a versioned +// invalid_argument reply, not as a dropped message the caller waits out. +func TestHandlersRejectMalformedRequests(t *testing.T) { + conn := serveProvider(t, ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{}), natsserve.Options{NoQueueGroup: true}) + + for _, verb := range wireVerbs(t) { + t.Run(verb.name, func(t *testing.T) { + msg, err := conn.Request(verb.subject, []byte("{not json"), requestTimeout) + require.NoError(t, err) + + envelope := decodeEnvelope(t, msg.Data) + assert.Equal(t, ebsprovider.SchemaVersion, envelope.SchemaVersion) + require.NotNil(t, envelope.Error) + assert.Equal(t, ebsprovider.ErrorCodeInvalidArgument, envelope.Error.Code) + assert.True(t, strings.HasPrefix(envelope.Error.Message, "bad request: "), "got %q", envelope.Error.Message) + }) + } +} + +// TestHandlersRejectVersionSkew is the guard the whole Versioned embedding +// exists for: a peer speaking a different schema must be refused at the +// boundary rather than have its payload decoded into this build's types. +func TestHandlersRejectVersionSkew(t *testing.T) { + conn := serveProvider(t, ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{}), natsserve.Options{NoQueueGroup: true}) + + for _, verb := range wireVerbs(t) { + t.Run(verb.name, func(t *testing.T) { + msg, err := conn.Request(verb.subject, []byte(`{"schema_version":99}`), requestTimeout) + require.NoError(t, err) + + envelope := decodeEnvelope(t, msg.Data) + assert.Equal(t, ebsprovider.SchemaVersion, envelope.SchemaVersion, "the refusal itself must be versioned") + require.NotNil(t, envelope.Error) + assert.Equal(t, ebsprovider.ErrorCodeUnsupportedVersion, envelope.Error.Code) + assert.Equal(t, fmt.Sprintf("unsupported schema version 99, want %d", ebsprovider.SchemaVersion), envelope.Error.Message) + }) + } +} + +// TestHandlersClassifyProviderFailures proves each handler carries the +// provider's sentinel across the wire rather than flattening it to internal: +// a caller switching on ErrorCodeUnavailable to decide whether to retry gets +// the wrong answer from any subject that loses it. +func TestHandlersClassifyProviderFailures(t *testing.T) { + provider := &failingProvider{err: fmt.Errorf("vol-1: %w", ebsprovider.ErrUnavailable)} + conn := serveProvider(t, provider, natsserve.Options{NoQueueGroup: true}) + + for _, verb := range wireVerbs(t) { + t.Run(verb.name, func(t *testing.T) { + payload, err := json.Marshal(verb.request) + require.NoError(t, err) + + // The asynchronous create replies "accepted" before it calls the + // provider, so its failure arrives on the completion subject. + var completionSub *nats.Subscription + if verb.completionSubject != "" { + completionSub, err = conn.SubscribeSync(verb.completionSubject) + require.NoError(t, err) + t.Cleanup(func() { _ = completionSub.Unsubscribe() }) + require.NoError(t, conn.Flush()) + } + + msg, err := conn.Request(verb.subject, payload, requestTimeout) + require.NoError(t, err) + if completionSub != nil { + msg, err = completionSub.NextMsg(requestTimeout) + require.NoError(t, err) + } + + envelope := decodeEnvelope(t, msg.Data) + require.NotNil(t, envelope.Error) + assert.Equal(t, ebsprovider.ErrorCodeUnavailable, envelope.Error.Code) + assert.Equal(t, "vol-1: "+ebsprovider.ErrUnavailable.Error(), envelope.Error.Message, "the wrapping context must survive the wire") + assert.True(t, envelope.Error.Code.Retryable(), "a transient failure must still invite a retry after crossing the wire") + }) + } +} + +// TestServedProviderRoundTrip drives every verb through NATSProvider against a +// served MemoryProvider, so the client's encoding and the server's decoding are +// asserted against each other rather than each against a hand-written payload. +func TestServedProviderRoundTrip(t *testing.T) { + capabilities := ebsprovider.Capabilities{ + VolumeEnumeration: true, + SnapshotEnumeration: true, + OnlineExpansion: true, + } + conn := serveProvider(t, ebsprovider.NewMemoryProvider(capabilities), natsserve.Options{NoQueueGroup: true}) + client := ebsprovider.NewNATSProvider(conn, requestTimeout) + ctx := t.Context() + + caps, err := client.GetCapabilities(ctx, ebsprovider.GetCapabilitiesRequest{Versioned: ebsprovider.NewVersioned()}) + require.NoError(t, err) + assert.True(t, caps.Capabilities.VolumeEnumeration) + assert.Equal(t, ebsprovider.ExclusionScopeNode, caps.Capabilities.Exclusion.Scope) + + volume, err := client.CreateVolume(ctx, ebsprovider.CreateVolumeRequest{ + Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-roundtrip", + CapacityRange: ebsprovider.CapacityRange{RequiredBytes: 4 << 30}, AvailabilityZone: "ap-southeast-2a", + }) + require.NoError(t, err) + assert.Equal(t, "memory://volume/vol-roundtrip", volume.Handle) + assert.Equal(t, "ap-southeast-2a", volume.AvailabilityZone) + + described, err := client.GetVolume(ctx, ebsprovider.GetVolumeRequest{ + Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-roundtrip", Handle: volume.Handle, + }) + require.NoError(t, err) + assert.Equal(t, volume, described) + + listed, err := client.ListVolumes(ctx, ebsprovider.ListVolumesRequest{Versioned: ebsprovider.NewVersioned()}) + require.NoError(t, err) + assert.Equal(t, []ebsprovider.VolumeRef{{ID: "vol-roundtrip", Handle: volume.Handle}}, listed.Volumes) + assert.Empty(t, listed.NextToken) + + expanded, err := client.ExpandVolume(ctx, ebsprovider.ExpandVolumeRequest{ + Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-roundtrip", + CapacityRange: ebsprovider.CapacityRange{RequiredBytes: 8 << 30}, + }) + require.NoError(t, err) + assert.Equal(t, int64(8<<30), expanded.CapacityBytes) + + published, err := client.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{ + Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-roundtrip", NodeID: "node-1", + }) + require.NoError(t, err) + assert.Equal(t, "nbd+unix:///?socket=/memory/vol-roundtrip.sock", published.NBDURI) + + inUse, err := client.GetVolume(ctx, ebsprovider.GetVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-roundtrip"}) + require.NoError(t, err) + assert.Equal(t, ebsprovider.VolumeStateInUse, inUse.State) + + require.NoError(t, client.UnpublishVolume(ctx, ebsprovider.UnpublishVolumeRequest{ + Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-roundtrip", NodeID: "node-1", + })) + available, err := client.GetVolume(ctx, ebsprovider.GetVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-roundtrip"}) + require.NoError(t, err) + assert.Equal(t, ebsprovider.VolumeStateAvailable, available.State) + + snapshot, err := client.CreateSnapshot(ctx, ebsprovider.CreateSnapshotRequest{ + Versioned: ebsprovider.NewVersioned(), SnapshotID: "snap-roundtrip", VolumeID: "vol-roundtrip", + }) + require.NoError(t, err) + assert.Equal(t, ebsprovider.SnapshotStateCompleted, snapshot.State, "the pending reply must be superseded by the completion event") + assert.Equal(t, int64(8<<30), snapshot.SizeBytes) + + copied, err := client.CopySnapshot(ctx, ebsprovider.CopySnapshotRequest{ + Versioned: ebsprovider.NewVersioned(), SourceSnapshotID: "snap-roundtrip", + DestinationSnapshotID: "snap-copy", VolumeID: "vol-roundtrip", + }) + require.NoError(t, err) + assert.Equal(t, "vol-roundtrip", copied.SourceVolumeID) + assert.Equal(t, snapshot.SizeBytes, copied.SizeBytes) + + snapshots, err := client.ListSnapshots(ctx, ebsprovider.ListSnapshotsRequest{Versioned: ebsprovider.NewVersioned()}) + require.NoError(t, err) + assert.Equal(t, []ebsprovider.SnapshotRef{ + {ID: "snap-copy", SourceVolumeID: "vol-roundtrip", Handle: "memory://snapshot/snap-copy"}, + {ID: "snap-roundtrip", SourceVolumeID: "vol-roundtrip", Handle: "memory://snapshot/snap-roundtrip"}, + }, snapshots.Snapshots) + + require.NoError(t, client.DeleteSnapshot(ctx, ebsprovider.DeleteSnapshotRequest{Versioned: ebsprovider.NewVersioned(), SnapshotID: "snap-copy"})) + require.NoError(t, client.DeleteVolume(ctx, ebsprovider.DeleteVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-roundtrip"})) + + _, err = client.GetVolume(ctx, ebsprovider.GetVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-roundtrip"}) + require.ErrorIs(t, err, ebsprovider.ErrNotFound) +} + +// TestCreateSnapshotAcceptsBeforeCompleting pins the two-stage shape of the +// asynchronous create: the immediate reply must name a pending snapshot, an +// operation ID and the completion subject the caller is expected to already +// be subscribed to, since a reply that omitted any of them would leave the +// caller waiting on a subject nothing will ever publish to. +func TestCreateSnapshotAcceptsBeforeCompleting(t *testing.T) { + conn := serveProvider(t, ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{}), natsserve.Options{NoQueueGroup: true}) + ctx := t.Context() + + provider := ebsprovider.NewNATSProvider(conn, requestTimeout) + _, err := provider.CreateVolume(ctx, ebsprovider.CreateVolumeRequest{ + Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-async", CapacityRange: ebsprovider.CapacityRange{RequiredBytes: 1 << 30}, + }) + require.NoError(t, err) + + subject, err := ebsprovider.SnapshotSubject("vol-async") + require.NoError(t, err) + completionSubject, err := ebsprovider.SnapshotCompletionSubject("snap-async") + require.NoError(t, err) + completionSub, err := conn.SubscribeSync(completionSubject) + require.NoError(t, err) + t.Cleanup(func() { _ = completionSub.Unsubscribe() }) + require.NoError(t, conn.Flush()) + + payload, err := json.Marshal(ebsprovider.CreateSnapshotRequest{ + Versioned: ebsprovider.NewVersioned(), SnapshotID: "snap-async", VolumeID: "vol-async", + }) + require.NoError(t, err) + msg, err := conn.Request(subject, payload, requestTimeout) + require.NoError(t, err) + + var accepted ebsprovider.CreateSnapshotResponse + require.NoError(t, json.Unmarshal(msg.Data, &accepted)) + require.Nil(t, accepted.Error) + assert.Equal(t, completionSubject, accepted.CompletionSubject) + assert.NotEmpty(t, accepted.OperationID) + require.NotNil(t, accepted.Snapshot) + assert.Equal(t, ebsprovider.SnapshotStatePending, accepted.Snapshot.State) + assert.Equal(t, "vol-async", accepted.Snapshot.SourceVolumeID) + + completionMsg, err := completionSub.NextMsg(requestTimeout) + require.NoError(t, err) + var completed ebsprovider.CreateSnapshotResponse + require.NoError(t, json.Unmarshal(completionMsg.Data, &completed)) + require.Nil(t, completed.Error) + assert.Equal(t, accepted.OperationID, completed.OperationID, "the completion must be attributable to the operation it finishes") + require.NotNil(t, completed.Snapshot) + assert.Equal(t, ebsprovider.SnapshotStateCompleted, completed.Snapshot.State) +} + +// TestCreateSnapshotRejectsSubjectBodyMismatch covers the wildcard's own +// check: the subject names the volume the request is routed by, so a body +// naming a different one would snapshot a volume the router never selected. +func TestCreateSnapshotRejectsSubjectBodyMismatch(t *testing.T) { + conn := serveProvider(t, ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{}), natsserve.Options{NoQueueGroup: true}) + + subject, err := ebsprovider.SnapshotSubject("vol-subject") + require.NoError(t, err) + payload, err := json.Marshal(ebsprovider.CreateSnapshotRequest{ + Versioned: ebsprovider.NewVersioned(), SnapshotID: "snap-1", VolumeID: "vol-body", + }) + require.NoError(t, err) + + msg, err := conn.Request(subject, payload, requestTimeout) + require.NoError(t, err) + + envelope := decodeEnvelope(t, msg.Data) + require.NotNil(t, envelope.Error) + assert.Equal(t, ebsprovider.ErrorCodeInvalidArgument, envelope.Error.Code) + assert.Equal(t, `volume id "vol-subject" in subject does not match request volume id "vol-body"`, envelope.Error.Message) +} + +// TestCreateSnapshotRejectsUnroutableSnapshotID covers the completion-subject +// build failure: a snapshot ID containing a NATS wildcard has no addressable +// completion subject, so the request must be refused rather than accepted +// into a background operation whose result can never be delivered. +func TestCreateSnapshotRejectsUnroutableSnapshotID(t *testing.T) { + conn := serveProvider(t, ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{}), natsserve.Options{NoQueueGroup: true}) + + subject, err := ebsprovider.SnapshotSubject("vol-1") + require.NoError(t, err) + payload, err := json.Marshal(ebsprovider.CreateSnapshotRequest{ + Versioned: ebsprovider.NewVersioned(), SnapshotID: "snap.*", VolumeID: "vol-1", + }) + require.NoError(t, err) + + msg, err := conn.Request(subject, payload, requestTimeout) + require.NoError(t, err) + + envelope := decodeEnvelope(t, msg.Data) + require.NotNil(t, envelope.Error) + assert.Equal(t, ebsprovider.ErrorCodeInvalidArgument, envelope.Error.Code) + assert.Contains(t, envelope.Error.Message, "invalid NATS subject token") +} + +// countingProvider records how many times each subject reached a provider, so +// a subscription's delivery semantics can be asserted from the provider side. +type countingProvider struct { + ebsprovider.EBSProvider + + capabilities atomic.Int32 + publishes atomic.Int32 +} + +func (p *countingProvider) GetCapabilities(ctx context.Context, req ebsprovider.GetCapabilitiesRequest) (*ebsprovider.GetCapabilitiesResponse, error) { + p.capabilities.Add(1) + return p.EBSProvider.GetCapabilities(ctx, req) +} + +func (p *countingProvider) PublishVolume(ctx context.Context, req ebsprovider.PublishVolumeRequest) (*ebsprovider.PublishedVolume, error) { + p.publishes.Add(1) + return p.EBSProvider.PublishVolume(ctx, req) +} + +// TestServeQueueGroupBalancesSharedSubjects is the difference between two +// workers sharing a workload and both doing all of it. The mount subject is +// never queue-grouped, so once both servers have handled a mount, message +// ordering on the shared connection guarantees the earlier capabilities +// request has already been delivered to everyone that was going to get it. +func TestServeQueueGroupBalancesSharedSubjects(t *testing.T) { + for _, tc := range []struct { + name string + opts natsserve.Options + wantHandlers int32 + }{ + {name: "queue group", opts: natsserve.Options{}, wantHandlers: 1}, + {name: "no queue group", opts: natsserve.Options{NoQueueGroup: true}, wantHandlers: 2}, + } { + t.Run(tc.name, func(t *testing.T) { + _, conn := testutil.StartTestNATS(t) + workers := []*countingProvider{ + {EBSProvider: ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{})}, + {EBSProvider: ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{})}, + } + for _, worker := range workers { + stop, err := natsserve.Serve(t.Context(), conn, worker, tc.opts) + require.NoError(t, err) + t.Cleanup(stop) + } + + client := ebsprovider.NewNATSProvider(conn, requestTimeout) + _, err := client.GetCapabilities(t.Context(), ebsprovider.GetCapabilitiesRequest{Versioned: ebsprovider.NewVersioned()}) + require.NoError(t, err) + + _, err = client.PublishVolume(t.Context(), ebsprovider.PublishVolumeRequest{ + Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-absent", NodeID: "node-1", + }) + require.ErrorIs(t, err, ebsprovider.ErrNotFound) + require.Eventually(t, func() bool { + return workers[0].publishes.Load()+workers[1].publishes.Load() == 2 + }, requestTimeout, 5*time.Millisecond, "mount is never queue-grouped, so both servers must see it") + + assert.Equal(t, tc.wantHandlers, workers[0].capabilities.Load()+workers[1].capabilities.Load()) + }) + } +} + +// TestServeNodeIDScopesMountSubjects covers the production shape: a daemon that +// can only mount on its own node must not answer another node's mount subject, +// or a volume gets attached on a host that cannot see the guest asking for it. +func TestServeNodeIDScopesMountSubjects(t *testing.T) { + conn := serveProvider(t, ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{}), natsserve.Options{NodeID: "node-1"}) + + payload, err := json.Marshal(ebsprovider.PublishVolumeRequest{ + Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-1", NodeID: "node-1", + }) + require.NoError(t, err) + + ownSubject, err := ebsprovider.PublishSubject("node-1") + require.NoError(t, err) + msg, err := conn.Request(ownSubject, payload, requestTimeout) + require.NoError(t, err) + envelope := decodeEnvelope(t, msg.Data) + require.NotNil(t, envelope.Error) + assert.Equal(t, ebsprovider.ErrorCodeNotFound, envelope.Error.Code, "the server answered its own subject") + + otherSubject, err := ebsprovider.PublishSubject("node-2") + require.NoError(t, err) + _, err = conn.Request(otherSubject, payload, requestTimeout) + require.ErrorIs(t, err, nats.ErrNoResponders, "another node's mount subject must go unanswered") +} + +// TestServeRejectsUnroutableNodeID covers the subject-build failure path: an +// unsubscribable node ID must fail Serve outright rather than leave a server +// running with the shared subjects subscribed and no mount subject at all. +func TestServeRejectsUnroutableNodeID(t *testing.T) { + _, conn := testutil.StartTestNATS(t) + + stop, err := natsserve.Serve(t.Context(), conn, ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{}), natsserve.Options{NodeID: "node.*"}) + require.ErrorIs(t, err, ebsprovider.ErrInvalidArgument) + assert.Contains(t, err.Error(), "build publish subject") + assert.Nil(t, stop) + + _, err = conn.Request(ebsprovider.CapabilitiesSubject, []byte(`{"schema_version":1}`), 200*time.Millisecond) + require.ErrorIs(t, err, nats.ErrNoResponders, "a failed Serve must leave no subscription behind") +} + +// TestServeFailsOnClosedConnection covers the subscribe error path for the +// shared subjects, which is the first thing Serve does. +func TestServeFailsOnClosedConnection(t *testing.T) { + _, conn := testutil.StartTestNATS(t) + conn.Close() + + stop, err := natsserve.Serve(t.Context(), conn, ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{}), natsserve.Options{NoQueueGroup: true}) + require.ErrorIs(t, err, nats.ErrConnectionClosed) + assert.Contains(t, err.Error(), "subscribe to "+ebsprovider.CapabilitiesSubject) + assert.Nil(t, stop) +} + +// TestServeStopIsIdempotent covers the sync.Once in the returned stop: the +// documented contract is that calling it twice is safe, and a caller with both +// a defer and an explicit shutdown path does exactly that. +func TestServeStopIsIdempotent(t *testing.T) { + _, conn := testutil.StartTestNATS(t) + stop, err := natsserve.Serve(t.Context(), conn, ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{}), natsserve.Options{NoQueueGroup: true}) + require.NoError(t, err) + + stop() + stop() + require.NoError(t, conn.Flush()) + + _, err = conn.Request(ebsprovider.CapabilitiesSubject, []byte(`{"schema_version":1}`), 200*time.Millisecond) + require.ErrorIs(t, err, nats.ErrNoResponders, "stop must unsubscribe every subject Serve registered") +} diff --git a/spinifex/ebsprovider/nullprovider/nullprovider_test.go b/spinifex/ebsprovider/nullprovider/nullprovider_test.go new file mode 100644 index 000000000..541a8d25b --- /dev/null +++ b/spinifex/ebsprovider/nullprovider/nullprovider_test.go @@ -0,0 +1,64 @@ +package nullprovider_test + +import ( + "testing" + + "github.com/mulgadc/spinifex/spinifex/ebsprovider" + "github.com/mulgadc/spinifex/spinifex/ebsprovider/natsserve" + "github.com/mulgadc/spinifex/spinifex/ebsprovider/nullprovider" + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestNewAdvertisesEveryOptionalCapability is what makes the floor a floor: a +// capability left off is a verb that returns early instead of being measured, +// so a baseline taken against a partly-capable provider prices less work than +// the provider it is compared against. +func TestNewAdvertisesEveryOptionalCapability(t *testing.T) { + caps, err := nullprovider.New().GetCapabilities(t.Context(), ebsprovider.GetCapabilitiesRequest{Versioned: ebsprovider.NewVersioned()}) + require.NoError(t, err) + + assert.Equal(t, nullprovider.Capabilities, caps.Capabilities) + assert.True(t, caps.Capabilities.OnlineExpansion) + assert.True(t, caps.Capabilities.VolumeSeeding) + assert.True(t, caps.Capabilities.VolumeEnumeration) + assert.True(t, caps.Capabilities.SnapshotEnumeration) + assert.True(t, caps.Capabilities.ReadOnlyPublish) + assert.False(t, caps.Capabilities.Exclusion.SingleWriter(), "an in-process map excludes within a node, not across a cluster") +} + +// TestServeRoundTripsOverNATS covers the composition Serve exists to name: the +// returned client must actually reach the served provider, since a measurement +// taken against a client wired to nothing would report only its own timeout. +func TestServeRoundTripsOverNATS(t *testing.T) { + _, conn := testutil.StartTestNATS(t) + client, stop, err := nullprovider.Serve(t.Context(), conn, natsserve.Options{NoQueueGroup: true}) + require.NoError(t, err) + t.Cleanup(stop) + + volume, err := client.CreateVolume(t.Context(), ebsprovider.CreateVolumeRequest{ + Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-null", + CapacityRange: ebsprovider.CapacityRange{RequiredBytes: 1 << 30}, + }) + require.NoError(t, err) + assert.Equal(t, "memory://volume/vol-null", volume.Handle) + + listed, err := client.ListVolumes(t.Context(), ebsprovider.ListVolumesRequest{Versioned: ebsprovider.NewVersioned()}) + require.NoError(t, err) + assert.Equal(t, []ebsprovider.VolumeRef{{ID: "vol-null", Handle: volume.Handle}}, listed.Volumes) +} + +// TestServeFailsOnClosedConnection covers the wrapped failure path: a caller +// that cannot stand the provider up must get an error rather than a client +// pointed at a subject nothing serves. +func TestServeFailsOnClosedConnection(t *testing.T) { + _, conn := testutil.StartTestNATS(t) + conn.Close() + + client, stop, err := nullprovider.Serve(t.Context(), conn, natsserve.Options{NoQueueGroup: true}) + require.Error(t, err) + assert.Contains(t, err.Error(), "serve null provider") + assert.Nil(t, client) + assert.Nil(t, stop) +}