From 27530ba27056f35cbfeef4a29e6fc35373a3fa02 Mon Sep 17 00:00:00 2001 From: Julian Sommer <115516918+juliansommer@users.noreply.github.com> Date: Wed, 11 Feb 2026 12:48:21 +1000 Subject: [PATCH] chore: daemon tests --- hive/daemon/daemon.go | 4 + hive/daemon/daemon_handlers.go | 2 +- hive/daemon/daemon_handlers_test.go | 657 ++++++++++++++++++++++++++++ hive/daemon/daemon_test.go | 250 +++++++---- hive/daemon/jetstream_test.go | 70 +-- hive/daemon/main_test.go | 73 ++++ 6 files changed, 914 insertions(+), 142 deletions(-) create mode 100644 hive/daemon/daemon_handlers_test.go create mode 100644 hive/daemon/main_test.go diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index b48ff0355..a1724cd7a 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -104,6 +104,9 @@ type Daemon struct { // JetStream manager for KV state storage (nil if JetStream disabled) jsManager *JetStreamManager + // Delay after QMP device_del before blockdev-del (default 1s, 0 in tests) + detachDelay time.Duration + mu sync.Mutex } @@ -425,6 +428,7 @@ func NewDaemon(cfg *config.ClusterConfig) *Daemon { Instances: vm.Instances{VMS: make(map[string]*vm.VM)}, natsSubscriptions: make(map[string]*nats.Subscription), startTime: time.Now(), + detachDelay: 1 * time.Second, } } diff --git a/hive/daemon/daemon_handlers.go b/hive/daemon/daemon_handlers.go index a141eef83..78bb0d333 100644 --- a/hive/daemon/daemon_handlers.go +++ b/hive/daemon/daemon_handlers.go @@ -392,7 +392,7 @@ func (d *Daemon) handleDetachVolume(msg *nats.Msg, command qmp.Command, instance } // Brief pause for guest to acknowledge PCI removal - time.Sleep(1 * time.Second) + time.Sleep(d.detachDelay) // Phase 2: QMP blockdev-del (remove block node) _, blockdevErr := d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{ diff --git a/hive/daemon/daemon_handlers_test.go b/hive/daemon/daemon_handlers_test.go new file mode 100644 index 000000000..8c32c6f1e --- /dev/null +++ b/hive/daemon/daemon_handlers_test.go @@ -0,0 +1,657 @@ +package daemon + +import ( + "encoding/json" + "fmt" + "testing" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/hive/hive/config" + handlers_ec2_account "github.com/mulgadc/hive/hive/handlers/ec2/account" + handlers_ec2_eigw "github.com/mulgadc/hive/hive/handlers/ec2/eigw" + handlers_ec2_image "github.com/mulgadc/hive/hive/handlers/ec2/image" + handlers_ec2_key "github.com/mulgadc/hive/hive/handlers/ec2/key" + handlers_ec2_snapshot "github.com/mulgadc/hive/hive/handlers/ec2/snapshot" + handlers_ec2_tags "github.com/mulgadc/hive/hive/handlers/ec2/tags" + handlers_ec2_volume "github.com/mulgadc/hive/hive/handlers/ec2/volume" + "github.com/mulgadc/hive/hive/objectstore" + "github.com/mulgadc/hive/hive/qmp" + "github.com/mulgadc/hive/hive/vm" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// createFullTestDaemon creates a test daemon with ALL services initialized (including +// key, image, snapshot, tags, eigw, account) using in-memory object stores. +func createFullTestDaemon(t *testing.T, natsURL string) *Daemon { + daemon := createTestDaemon(t, natsURL) + + memStore := objectstore.NewMemoryObjectStore() + cfg := daemon.config + + daemon.keyService = handlers_ec2_key.NewKeyServiceImplWithStore(memStore, cfg.Predastore.Bucket, "123456789") + daemon.imageService = handlers_ec2_image.NewImageServiceImplWithStore(memStore, cfg.Predastore.Bucket, "123456789") + daemon.volumeService = handlers_ec2_volume.NewVolumeServiceImplWithStore(cfg, memStore, daemon.natsConn) + daemon.snapshotService = handlers_ec2_snapshot.NewSnapshotServiceImplWithStore(cfg, memStore, daemon.natsConn) + daemon.tagsService = handlers_ec2_tags.NewTagsServiceImplWithStore(cfg, memStore) + daemon.eigwService = handlers_ec2_eigw.NewEgressOnlyIGWServiceImpl(cfg) + daemon.accountService = handlers_ec2_account.NewAccountSettingsServiceImpl(cfg) + + return daemon +} + +// createFullTestDaemonWithJetStream creates a test daemon with JetStream KV enabled, +// needed for tests involving state transitions (TransitionState calls WriteState). +func createFullTestDaemonWithJetStream(t *testing.T, natsURL string) *Daemon { + daemon := createFullTestDaemon(t, natsURL) + + var err error + daemon.jsManager, err = NewJetStreamManager(daemon.natsConn, 1) + require.NoError(t, err) + err = daemon.jsManager.InitKVBucket() + require.NoError(t, err) + + return daemon +} + +// --- handleNATSRequest generic tests --- + +type testInput struct { + Name string `json:"name"` +} + +type testOutput struct { + Greeting string `json:"greeting"` +} + +func TestHandleNATSRequest_ValidRequest(t *testing.T) { + natsURL := sharedNATSURL + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + serviceFn := func(in *testInput) (*testOutput, error) { + return &testOutput{Greeting: "hello " + in.Name}, nil + } + + sub, err := nc.Subscribe("test.greet", func(msg *nats.Msg) { + handleNATSRequest(msg, serviceFn) + }) + require.NoError(t, err) + defer sub.Unsubscribe() + + reqData, _ := json.Marshal(testInput{Name: "world"}) + reply, err := nc.Request("test.greet", reqData, 5*time.Second) + require.NoError(t, err) + + var out testOutput + err = json.Unmarshal(reply.Data, &out) + require.NoError(t, err) + assert.Equal(t, "hello world", out.Greeting) +} + +func TestHandleNATSRequest_MalformedJSON(t *testing.T) { + natsURL := sharedNATSURL + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + serviceFn := func(in *testInput) (*testOutput, error) { + return &testOutput{Greeting: "hello"}, nil + } + + sub, err := nc.Subscribe("test.malformed", func(msg *nats.Msg) { + handleNATSRequest(msg, serviceFn) + }) + require.NoError(t, err) + defer sub.Unsubscribe() + + reply, err := nc.Request("test.malformed", []byte(`{not valid json}`), 5*time.Second) + require.NoError(t, err) + + var errResp map[string]any + err = json.Unmarshal(reply.Data, &errResp) + require.NoError(t, err) + assert.Contains(t, errResp, "Code", "Should contain error code") +} + +func TestHandleNATSRequest_ServiceError(t *testing.T) { + natsURL := sharedNATSURL + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + serviceFn := func(in *testInput) (*testOutput, error) { + return nil, fmt.Errorf("something went wrong") + } + + sub, err := nc.Subscribe("test.err", func(msg *nats.Msg) { + handleNATSRequest(msg, serviceFn) + }) + require.NoError(t, err) + defer sub.Unsubscribe() + + reqData, _ := json.Marshal(testInput{Name: "world"}) + reply, err := nc.Request("test.err", reqData, 5*time.Second) + require.NoError(t, err) + + var errResp map[string]any + err = json.Unmarshal(reply.Data, &errResp) + require.NoError(t, err) + assert.Contains(t, errResp, "Code", "Should contain error code") +} + +// --- Handler wrapper tests (representative set via NATS round-trip) --- + +func TestHandleEC2CreateKeyPair_RoundTrip(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createFullTestDaemon(t, natsURL) + + sub, err := daemon.natsConn.QueueSubscribe("ec2.CreateKeyPair", "hive-workers", daemon.handleEC2CreateKeyPair) + require.NoError(t, err) + defer sub.Unsubscribe() + + input := &ec2.CreateKeyPairInput{ + KeyName: aws.String("test-key-001"), + } + reqData, _ := json.Marshal(input) + reply, err := daemon.natsConn.Request("ec2.CreateKeyPair", reqData, 5*time.Second) + require.NoError(t, err) + require.NotNil(t, reply) + + var output ec2.CreateKeyPairOutput + err = json.Unmarshal(reply.Data, &output) + require.NoError(t, err) + assert.Equal(t, "test-key-001", *output.KeyName) + assert.NotEmpty(t, *output.KeyFingerprint) + assert.NotEmpty(t, *output.KeyMaterial) +} + +func TestHandleEC2CreateTags_RoundTrip(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createFullTestDaemon(t, natsURL) + + sub, err := daemon.natsConn.QueueSubscribe("ec2.CreateTags", "hive-workers", daemon.handleEC2CreateTags) + require.NoError(t, err) + defer sub.Unsubscribe() + + input := &ec2.CreateTagsInput{ + Resources: []*string{aws.String("i-12345678")}, + Tags: []*ec2.Tag{ + {Key: aws.String("Name"), Value: aws.String("test-instance")}, + }, + } + reqData, _ := json.Marshal(input) + reply, err := daemon.natsConn.Request("ec2.CreateTags", reqData, 5*time.Second) + require.NoError(t, err) + require.NotNil(t, reply) + + var output ec2.CreateTagsOutput + err = json.Unmarshal(reply.Data, &output) + require.NoError(t, err) +} + +func TestHandleEC2DescribeImages_RoundTrip(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createFullTestDaemon(t, natsURL) + + sub, err := daemon.natsConn.QueueSubscribe("ec2.DescribeImages", "hive-workers", daemon.handleEC2DescribeImages) + require.NoError(t, err) + defer sub.Unsubscribe() + + input := &ec2.DescribeImagesInput{} + reqData, _ := json.Marshal(input) + reply, err := daemon.natsConn.Request("ec2.DescribeImages", reqData, 5*time.Second) + require.NoError(t, err) + require.NotNil(t, reply) + + var output ec2.DescribeImagesOutput + err = json.Unmarshal(reply.Data, &output) + require.NoError(t, err) +} + +func TestHandleEC2DescribeVolumes_RoundTrip(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createFullTestDaemon(t, natsURL) + + sub, err := daemon.natsConn.QueueSubscribe("ec2.DescribeVolumes", "hive-workers", daemon.handleEC2DescribeVolumes) + require.NoError(t, err) + defer sub.Unsubscribe() + + input := &ec2.DescribeVolumesInput{} + reqData, _ := json.Marshal(input) + reply, err := daemon.natsConn.Request("ec2.DescribeVolumes", reqData, 5*time.Second) + require.NoError(t, err) + require.NotNil(t, reply) + + var output ec2.DescribeVolumesOutput + err = json.Unmarshal(reply.Data, &output) + require.NoError(t, err) +} + +func TestHandleEC2DescribeKeyPairs_RoundTrip(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createFullTestDaemon(t, natsURL) + + sub, err := daemon.natsConn.QueueSubscribe("ec2.DescribeKeyPairs", "hive-workers", daemon.handleEC2DescribeKeyPairs) + require.NoError(t, err) + defer sub.Unsubscribe() + + input := &ec2.DescribeKeyPairsInput{} + reqData, _ := json.Marshal(input) + reply, err := daemon.natsConn.Request("ec2.DescribeKeyPairs", reqData, 5*time.Second) + require.NoError(t, err) + require.NotNil(t, reply) + + var output ec2.DescribeKeyPairsOutput + err = json.Unmarshal(reply.Data, &output) + require.NoError(t, err) +} + +// --- handleHealthCheck tests --- + +func TestHandleHealthCheck(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createTestDaemon(t, natsURL) + + topic := fmt.Sprintf("hive.admin.%s.health", daemon.node) + sub, err := daemon.natsConn.Subscribe(topic, daemon.handleHealthCheck) + require.NoError(t, err) + defer sub.Unsubscribe() + + reply, err := daemon.natsConn.Request(topic, nil, 5*time.Second) + require.NoError(t, err) + require.NotNil(t, reply) + + var resp config.NodeHealthResponse + err = json.Unmarshal(reply.Data, &resp) + require.NoError(t, err) + + assert.Equal(t, daemon.node, resp.Node) + assert.Equal(t, "running", resp.Status) + assert.NotEmpty(t, resp.ConfigHash) + assert.GreaterOrEqual(t, resp.Uptime, int64(0)) +} + +// --- handleNodeDiscover tests --- + +func TestHandleNodeDiscover(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createTestDaemon(t, natsURL) + + sub, err := daemon.natsConn.Subscribe("hive.nodes.discover", daemon.handleNodeDiscover) + require.NoError(t, err) + defer sub.Unsubscribe() + + reply, err := daemon.natsConn.Request("hive.nodes.discover", nil, 5*time.Second) + require.NoError(t, err) + require.NotNil(t, reply) + + var resp NodeDiscoverResponse + err = json.Unmarshal(reply.Data, &resp) + require.NoError(t, err) + + assert.Equal(t, daemon.node, resp.Node) +} + +// --- handleStopOrTerminateInstance tests (JetStream required for TransitionState) --- + +func TestHandleEC2Events_StopInstance(t *testing.T) { + natsURL := sharedJSNATSURL + + daemon := createFullTestDaemonWithJetStream(t, natsURL) + + instanceID := "i-test-stop-001" + daemon.Instances.VMS[instanceID] = &vm.VM{ + ID: instanceID, + InstanceType: getTestInstanceType(), + Status: vm.StateRunning, + Instance: &ec2.Instance{}, + QMPClient: &qmp.QMPClient{}, + } + + sub, err := daemon.natsConn.QueueSubscribe( + fmt.Sprintf("ec2.cmd.%s", instanceID), + "hive-events", + daemon.handleEC2Events, + ) + require.NoError(t, err) + defer sub.Unsubscribe() + + cmd := qmp.Command{ + ID: instanceID, + Attributes: qmp.Attributes{StopInstance: true}, + } + cmdData, _ := json.Marshal(cmd) + + reply, err := daemon.natsConn.Request( + fmt.Sprintf("ec2.cmd.%s", instanceID), + cmdData, + 5*time.Second, + ) + require.NoError(t, err) + require.NotNil(t, reply) + + // Should get immediate {} response + assert.Equal(t, `{}`, string(reply.Data)) + + // State should transition to stopping + daemon.Instances.Mu.Lock() + status := daemon.Instances.VMS[instanceID].Status + daemon.Instances.Mu.Unlock() + assert.Equal(t, vm.StateStopping, status) +} + +func TestHandleEC2Events_TerminateInstance(t *testing.T) { + natsURL := sharedJSNATSURL + + daemon := createFullTestDaemonWithJetStream(t, natsURL) + + instanceID := "i-test-term-001" + daemon.Instances.VMS[instanceID] = &vm.VM{ + ID: instanceID, + InstanceType: getTestInstanceType(), + Status: vm.StateRunning, + Instance: &ec2.Instance{}, + QMPClient: &qmp.QMPClient{}, + } + + sub, err := daemon.natsConn.QueueSubscribe( + fmt.Sprintf("ec2.cmd.%s", instanceID), + "hive-events", + daemon.handleEC2Events, + ) + require.NoError(t, err) + defer sub.Unsubscribe() + + cmd := qmp.Command{ + ID: instanceID, + Attributes: qmp.Attributes{TerminateInstance: true}, + } + cmdData, _ := json.Marshal(cmd) + + reply, err := daemon.natsConn.Request( + fmt.Sprintf("ec2.cmd.%s", instanceID), + cmdData, + 5*time.Second, + ) + require.NoError(t, err) + require.NotNil(t, reply) + + assert.Equal(t, `{}`, string(reply.Data)) + + daemon.Instances.Mu.Lock() + status := daemon.Instances.VMS[instanceID].Status + daemon.Instances.Mu.Unlock() + assert.Equal(t, vm.StateShuttingDown, status) +} + +func TestHandleEC2Events_InstanceNotFound(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createTestDaemon(t, natsURL) + + sub, err := daemon.natsConn.QueueSubscribe("ec2.cmd.i-nonexistent", "hive-events", daemon.handleEC2Events) + require.NoError(t, err) + defer sub.Unsubscribe() + + cmd := qmp.Command{ + ID: "i-nonexistent", + Attributes: qmp.Attributes{StopInstance: true}, + } + cmdData, _ := json.Marshal(cmd) + + reply, err := daemon.natsConn.Request("ec2.cmd.i-nonexistent", cmdData, 5*time.Second) + require.NoError(t, err) + + var errResp map[string]any + err = json.Unmarshal(reply.Data, &errResp) + require.NoError(t, err) + assert.Contains(t, errResp, "Code") +} + +func TestHandleEC2Events_MalformedJSON(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createTestDaemon(t, natsURL) + + sub, err := daemon.natsConn.QueueSubscribe("ec2.cmd.test", "hive-events", daemon.handleEC2Events) + require.NoError(t, err) + defer sub.Unsubscribe() + + reply, err := daemon.natsConn.Request("ec2.cmd.test", []byte(`{bad json}`), 5*time.Second) + require.NoError(t, err) + + var errResp map[string]any + err = json.Unmarshal(reply.Data, &errResp) + require.NoError(t, err) + assert.Contains(t, errResp, "Code") +} + +// --- respondWithVolumeAttachment tests --- + +func TestRespondWithVolumeAttachment(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createTestDaemon(t, natsURL) + + sub, err := daemon.natsConn.Subscribe("test.volume.attach", func(msg *nats.Msg) { + respondWithError := func(errCode string) { + msg.Respond([]byte(fmt.Sprintf(`{"Code":"%s"}`, errCode))) + } + daemon.respondWithVolumeAttachment(msg, respondWithError, "vol-123", "i-456", "/dev/sdf", "attached") + }) + require.NoError(t, err) + defer sub.Unsubscribe() + + reply, err := daemon.natsConn.Request("test.volume.attach", nil, 5*time.Second) + require.NoError(t, err) + + var attachment ec2.VolumeAttachment + err = json.Unmarshal(reply.Data, &attachment) + require.NoError(t, err) + + assert.Equal(t, "vol-123", *attachment.VolumeId) + assert.Equal(t, "i-456", *attachment.InstanceId) + assert.Equal(t, "/dev/sdf", *attachment.Device) + assert.Equal(t, "attached", *attachment.State) + assert.NotNil(t, attachment.AttachTime) + assert.Equal(t, false, *attachment.DeleteOnTermination) +} + +// --- handleEC2ModifyVolume tests --- + +func TestHandleEC2ModifyVolume_MalformedInput(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createFullTestDaemon(t, natsURL) + + sub, err := daemon.natsConn.QueueSubscribe("ec2.ModifyVolume", "hive-workers", daemon.handleEC2ModifyVolume) + require.NoError(t, err) + defer sub.Unsubscribe() + + reply, err := daemon.natsConn.Request("ec2.ModifyVolume", []byte(`{bad}`), 5*time.Second) + require.NoError(t, err) + + var errResp map[string]any + err = json.Unmarshal(reply.Data, &errResp) + require.NoError(t, err) + assert.Contains(t, errResp, "Code") +} + +func TestHandleEC2ModifyVolume_VolumeNotFound(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createFullTestDaemon(t, natsURL) + + sub, err := daemon.natsConn.QueueSubscribe("ec2.ModifyVolume", "hive-workers", daemon.handleEC2ModifyVolume) + require.NoError(t, err) + defer sub.Unsubscribe() + + input := &ec2.ModifyVolumeInput{ + VolumeId: aws.String("vol-nonexistent"), + Size: aws.Int64(16), + } + reqData, _ := json.Marshal(input) + reply, err := daemon.natsConn.Request("ec2.ModifyVolume", reqData, 5*time.Second) + require.NoError(t, err) + + // Should return an error since the volume doesn't exist + var errResp map[string]any + err = json.Unmarshal(reply.Data, &errResp) + require.NoError(t, err) + assert.Contains(t, errResp, "Code") +} + +// --- Account settings handler tests --- + +func TestHandleEC2GetEbsEncryptionByDefault(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createFullTestDaemon(t, natsURL) + + sub, err := daemon.natsConn.QueueSubscribe("ec2.GetEbsEncryptionByDefault", "hive-workers", daemon.handleEC2GetEbsEncryptionByDefault) + require.NoError(t, err) + defer sub.Unsubscribe() + + input := &ec2.GetEbsEncryptionByDefaultInput{} + reqData, _ := json.Marshal(input) + reply, err := daemon.natsConn.Request("ec2.GetEbsEncryptionByDefault", reqData, 5*time.Second) + require.NoError(t, err) + + var output ec2.GetEbsEncryptionByDefaultOutput + err = json.Unmarshal(reply.Data, &output) + require.NoError(t, err) + assert.NotNil(t, output.EbsEncryptionByDefault) +} + +func TestHandleEC2GetSerialConsoleAccessStatus(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createFullTestDaemon(t, natsURL) + + sub, err := daemon.natsConn.QueueSubscribe("ec2.GetSerialConsoleAccessStatus", "hive-workers", daemon.handleEC2GetSerialConsoleAccessStatus) + require.NoError(t, err) + defer sub.Unsubscribe() + + input := &ec2.GetSerialConsoleAccessStatusInput{} + reqData, _ := json.Marshal(input) + reply, err := daemon.natsConn.Request("ec2.GetSerialConsoleAccessStatus", reqData, 5*time.Second) + require.NoError(t, err) + + var output ec2.GetSerialConsoleAccessStatusOutput + err = json.Unmarshal(reply.Data, &output) + require.NoError(t, err) + assert.NotNil(t, output.SerialConsoleAccessEnabled) +} + +func TestHandleEC2EnableEbsEncryptionByDefault(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createFullTestDaemon(t, natsURL) + + sub, err := daemon.natsConn.QueueSubscribe("ec2.EnableEbsEncryptionByDefault", "hive-workers", daemon.handleEC2EnableEbsEncryptionByDefault) + require.NoError(t, err) + defer sub.Unsubscribe() + + input := &ec2.EnableEbsEncryptionByDefaultInput{} + reqData, _ := json.Marshal(input) + reply, err := daemon.natsConn.Request("ec2.EnableEbsEncryptionByDefault", reqData, 5*time.Second) + require.NoError(t, err) + + var output ec2.EnableEbsEncryptionByDefaultOutput + err = json.Unmarshal(reply.Data, &output) + require.NoError(t, err) + assert.NotNil(t, output.EbsEncryptionByDefault) + assert.True(t, *output.EbsEncryptionByDefault) +} + +func TestHandleEC2DisableEbsEncryptionByDefault(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createFullTestDaemon(t, natsURL) + + sub, err := daemon.natsConn.QueueSubscribe("ec2.DisableEbsEncryptionByDefault", "hive-workers", daemon.handleEC2DisableEbsEncryptionByDefault) + require.NoError(t, err) + defer sub.Unsubscribe() + + input := &ec2.DisableEbsEncryptionByDefaultInput{} + reqData, _ := json.Marshal(input) + reply, err := daemon.natsConn.Request("ec2.DisableEbsEncryptionByDefault", reqData, 5*time.Second) + require.NoError(t, err) + + var output ec2.DisableEbsEncryptionByDefaultOutput + err = json.Unmarshal(reply.Data, &output) + require.NoError(t, err) + assert.NotNil(t, output.EbsEncryptionByDefault) + assert.False(t, *output.EbsEncryptionByDefault) +} + +func TestHandleEC2EnableSerialConsoleAccess(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createFullTestDaemon(t, natsURL) + + sub, err := daemon.natsConn.QueueSubscribe("ec2.EnableSerialConsoleAccess", "hive-workers", daemon.handleEC2EnableSerialConsoleAccess) + require.NoError(t, err) + defer sub.Unsubscribe() + + input := &ec2.EnableSerialConsoleAccessInput{} + reqData, _ := json.Marshal(input) + reply, err := daemon.natsConn.Request("ec2.EnableSerialConsoleAccess", reqData, 5*time.Second) + require.NoError(t, err) + + var output ec2.EnableSerialConsoleAccessOutput + err = json.Unmarshal(reply.Data, &output) + require.NoError(t, err) + assert.NotNil(t, output.SerialConsoleAccessEnabled) + assert.True(t, *output.SerialConsoleAccessEnabled) +} + +func TestHandleEC2DisableSerialConsoleAccess(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createFullTestDaemon(t, natsURL) + + sub, err := daemon.natsConn.QueueSubscribe("ec2.DisableSerialConsoleAccess", "hive-workers", daemon.handleEC2DisableSerialConsoleAccess) + require.NoError(t, err) + defer sub.Unsubscribe() + + input := &ec2.DisableSerialConsoleAccessInput{} + reqData, _ := json.Marshal(input) + reply, err := daemon.natsConn.Request("ec2.DisableSerialConsoleAccess", reqData, 5*time.Second) + require.NoError(t, err) + + var output ec2.DisableSerialConsoleAccessOutput + err = json.Unmarshal(reply.Data, &output) + require.NoError(t, err) + assert.NotNil(t, output.SerialConsoleAccessEnabled) + assert.False(t, *output.SerialConsoleAccessEnabled) +} + +// --- SetConfigPath test --- + +func TestSetConfigPath(t *testing.T) { + clusterCfg := &config.ClusterConfig{ + Node: "node-1", + Nodes: map[string]config.Config{"node-1": {BaseDir: "/tmp"}}, + } + daemon := NewDaemon(clusterCfg) + + assert.Empty(t, daemon.configPath) + daemon.SetConfigPath("/etc/hive/config.toml") + assert.Equal(t, "/etc/hive/config.toml", daemon.configPath) +} diff --git a/hive/daemon/daemon_test.go b/hive/daemon/daemon_test.go index 42f7bfde4..56ce1456b 100644 --- a/hive/daemon/daemon_test.go +++ b/hive/daemon/daemon_test.go @@ -20,42 +20,12 @@ import ( "github.com/mulgadc/hive/hive/objectstore" "github.com/mulgadc/hive/hive/qmp" "github.com/mulgadc/hive/hive/vm" - "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" "github.com/pelletier/go-toml/v2" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) -// startTestNATSServer starts an embedded NATS server for testing -// Using port -1 allows NATS to automatically allocate an available port -func startTestNATSServer(t *testing.T) (*server.Server, string) { - opts := &server.Options{ - Host: "127.0.0.1", - Port: -1, // Let NATS auto-allocate an available port - JetStream: false, - NoLog: true, - NoSigs: true, - } - - ns, err := server.NewServer(opts) - require.NoError(t, err, "Failed to create NATS server") - - // Start server in a goroutine - go ns.Start() - - // Wait for server to be ready - if !ns.ReadyForConnections(5 * time.Second) { - t.Fatal("NATS server failed to start") - } - - // Get the actual URL that was assigned - url := ns.ClientURL() - t.Logf("Test NATS server started at: %s", url) - - return ns, url -} - // createTestDaemon creates a test daemon instance with minimal configuration func createTestDaemon(t *testing.T, natsURL string) *Daemon { // Create a temporary directory for test data @@ -102,6 +72,7 @@ func createTestDaemon(t *testing.T, natsURL string) *Daemon { require.NoError(t, err, "Failed to connect to NATS") daemon.natsConn = nc + daemon.detachDelay = 0 // Skip sleep in tests // Initialize services (needed for handler tests) daemon.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(cfg, daemon.resourceMgr.instanceTypes, nc, &daemon.Instances, objectstore.NewMemoryObjectStore()) @@ -236,8 +207,7 @@ func TestHandleEC2RunInstances_MessageParsing(t *testing.T) { } // Start test NATS server - ns, natsURL := startTestNATSServer(t) - defer ns.Shutdown() + natsURL := sharedNATSURL // Create test daemon daemon := createTestDaemon(t, natsURL) @@ -318,8 +288,7 @@ func TestHandleEC2RunInstances_ResourceManagement(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - ns, natsURL := startTestNATSServer(t) - defer ns.Shutdown() + natsURL := sharedNATSURL daemon := createTestDaemon(t, natsURL) @@ -609,8 +578,7 @@ func TestGetAvailableInstanceTypeInfos_ResourceFiltering(t *testing.T) { // TestHandleEC2DescribeInstanceTypes tests the DescribeInstanceTypes handler func TestHandleEC2DescribeInstanceTypes(t *testing.T) { - ns, natsURL := startTestNATSServer(t) - defer ns.Shutdown() + natsURL := sharedNATSURL daemon := createTestDaemon(t, natsURL) @@ -916,27 +884,7 @@ func TestHandleEC2DescribeInstanceTypes(t *testing.T) { // TestDaemon_BootAllocation verifies that resources are correctly reconstructed on startup func TestDaemon_BootAllocation(t *testing.T) { - // Start NATS server with JetStream - jsTmpDir, err := os.MkdirTemp("", "nats-js-boot-test-*") - require.NoError(t, err) - defer os.RemoveAll(jsTmpDir) - - opts := &server.Options{ - Host: "127.0.0.1", - Port: -1, - JetStream: true, - StoreDir: jsTmpDir, - NoLog: true, - NoSigs: true, - } - ns, err := server.NewServer(opts) - require.NoError(t, err) - go ns.Start() - if !ns.ReadyForConnections(5 * time.Second) { - t.Fatal("NATS server with JetStream failed to start") - } - defer ns.Shutdown() - natsURL := ns.ClientURL() + natsURL := sharedJSNATSURL // Create daemon temp directory tmpDir, err := os.MkdirTemp("", "hive-daemon-boot-test-*") @@ -1193,8 +1141,7 @@ func TestCanAllocate_CountEdgeCases(t *testing.T) { // TestDescribeInstances_ReservationGrouping tests that instances are grouped by reservation ID func TestDescribeInstances_ReservationGrouping(t *testing.T) { - ns, natsURL := startTestNATSServer(t) - defer ns.Shutdown() + natsURL := sharedNATSURL daemon := createTestDaemon(t, natsURL) @@ -1382,8 +1329,7 @@ func TestDescribeInstances_ReservationGrouping(t *testing.T) { // TestRunInstances_CountValidation tests MinCount/MaxCount validation scenarios func TestRunInstances_CountValidation(t *testing.T) { - ns, natsURL := startTestNATSServer(t) - defer ns.Shutdown() + natsURL := sharedNATSURL daemon := createTestDaemon(t, natsURL) @@ -1621,8 +1567,7 @@ func TestGenerateVolumes_DeleteOnTermination_FromBlockDeviceMapping(t *testing.T // correctly handles DeleteOnTermination for each volume type. // Uses embedded NATS with mock subscribers to verify NATS messages. func TestStopInstance_DeleteOnTermination_VolumeDeletion(t *testing.T) { - ns, natsURL := startTestNATSServer(t) - defer ns.Shutdown() + natsURL := sharedNATSURL daemon := createTestDaemon(t, natsURL) @@ -1720,8 +1665,7 @@ func TestStopInstance_DeleteOnTermination_VolumeDeletion(t *testing.T) { // TestStopInstance_DeleteOnTermination_False_SkipsVolumeDeletion verifies that // volumes with DeleteOnTermination=false are NOT deleted during termination. func TestStopInstance_DeleteOnTermination_False_SkipsVolumeDeletion(t *testing.T) { - ns, natsURL := startTestNATSServer(t) - defer ns.Shutdown() + natsURL := sharedNATSURL daemon := createTestDaemon(t, natsURL) @@ -1819,8 +1763,7 @@ func TestStopInstance_DeleteOnTermination_False_SkipsVolumeDeletion(t *testing.T // TestStopInstance_NoDelete_OnStop verifies that volumes are NOT deleted during // a regular stop (non-termination), regardless of DeleteOnTermination flag. func TestStopInstance_NoDelete_OnStop(t *testing.T) { - ns, natsURL := startTestNATSServer(t) - defer ns.Shutdown() + natsURL := sharedNATSURL daemon := createTestDaemon(t, natsURL) @@ -1976,8 +1919,7 @@ func TestNextAvailableDevice(t *testing.T) { // TestHandleEC2Events_AttachVolume tests the attach-volume handler in handleEC2Events func TestHandleEC2Events_AttachVolume(t *testing.T) { - ns, natsURL := startTestNATSServer(t) - defer ns.Shutdown() + natsURL := sharedNATSURL daemon := createTestDaemon(t, natsURL) @@ -2106,8 +2048,7 @@ func TestEBSRequest_JSON_Serialization(t *testing.T) { // TestHandleEC2Events_DetachVolume tests the detach-volume handler in handleEC2Events func TestHandleEC2Events_DetachVolume(t *testing.T) { - ns, natsURL := startTestNATSServer(t) - defer ns.Shutdown() + natsURL := sharedNATSURL daemon := createTestDaemon(t, natsURL) @@ -2433,8 +2374,7 @@ func newMockQMPClient(t *testing.T, responseFunc func(cmd qmp.QMPCommand) map[st // TestDetachVolume_SuccessPath tests the full happy-path detach including QMP commands // and state cleanup using a mock QMP server. func TestDetachVolume_SuccessPath(t *testing.T) { - ns, natsURL := startTestNATSServer(t) - defer ns.Shutdown() + natsURL := sharedNATSURL daemon := createTestDaemon(t, natsURL) @@ -2575,8 +2515,7 @@ func TestDetachVolume_SuccessPath(t *testing.T) { // TestDetachVolume_ForceFlag tests that force=true continues past device_del failure func TestDetachVolume_ForceFlag(t *testing.T) { - ns, natsURL := startTestNATSServer(t) - defer ns.Shutdown() + natsURL := sharedNATSURL daemon := createTestDaemon(t, natsURL) @@ -2692,8 +2631,7 @@ func TestDetachVolume_ForceFlag(t *testing.T) { // TestDetachVolume_BlockdevDelFailure tests that when blockdev-del fails, // state is left intact to prevent double-attach and VM crashes. func TestDetachVolume_BlockdevDelFailure(t *testing.T) { - ns, natsURL := startTestNATSServer(t) - defer ns.Shutdown() + natsURL := sharedNATSURL daemon := createTestDaemon(t, natsURL) @@ -2805,8 +2743,7 @@ func TestDetachVolume_BlockdevDelFailure(t *testing.T) { // TestDetachVolume_SuccessWithDeviceMatch tests detach with correct --device cross-check func TestDetachVolume_SuccessWithDeviceMatch(t *testing.T) { - ns, natsURL := startTestNATSServer(t) - defer ns.Shutdown() + natsURL := sharedNATSURL daemon := createTestDaemon(t, natsURL) @@ -2890,8 +2827,7 @@ func TestDetachVolume_SuccessWithDeviceMatch(t *testing.T) { // that already has a stale EBSRequest entry (e.g. from a stop/start cycle) // replaces it rather than creating a duplicate. func TestAttachVolume_ReplacesStaleEBSRequest(t *testing.T) { - ns, natsURL := startTestNATSServer(t) - defer ns.Shutdown() + natsURL := sharedNATSURL daemon := createTestDaemon(t, natsURL) @@ -3334,3 +3270,155 @@ func TestNewDaemon_WalDirPreservedIfSet(t *testing.T) { d := NewDaemon(cfg) assert.Equal(t, "/fast-ssd/wal", d.config.WalDir) } + +// TestMarkInstanceFailed verifies that markInstanceFailed sets the StateReason and +// transitions the instance to StateShuttingDown. +func TestMarkInstanceFailed(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createTestDaemon(t, natsURL) + + instanceID := "i-test-mark-failed" + ec2Instance := &ec2.Instance{} + ec2Instance.SetInstanceId(instanceID) + + instance := &vm.VM{ + ID: instanceID, + InstanceType: getTestInstanceType(), + Status: vm.StatePending, + Instance: ec2Instance, + QMPClient: &qmp.QMPClient{}, + } + daemon.Instances.VMS[instanceID] = instance + + daemon.markInstanceFailed(instance, "volume_preparation_failed") + + daemon.Instances.Mu.Lock() + defer daemon.Instances.Mu.Unlock() + + // Verify state transitioned to shutting-down + assert.Equal(t, vm.StateShuttingDown, instance.Status) + + // Verify StateReason was set + require.NotNil(t, instance.Instance.StateReason) + assert.Equal(t, "Server.InternalError", *instance.Instance.StateReason.Code) + assert.Equal(t, "volume_preparation_failed", *instance.Instance.StateReason.Message) +} + +// TestMarkInstanceFailed_NilInstance verifies that markInstanceFailed handles +// a VM with no ec2.Instance (Instance == nil) gracefully. +func TestMarkInstanceFailed_NilInstance(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createTestDaemon(t, natsURL) + + instanceID := "i-test-mark-failed-nil" + instance := &vm.VM{ + ID: instanceID, + InstanceType: getTestInstanceType(), + Status: vm.StatePending, + Instance: nil, // no ec2.Instance + QMPClient: &qmp.QMPClient{}, + } + daemon.Instances.VMS[instanceID] = instance + + // Should not panic + daemon.markInstanceFailed(instance, "test_failure") + + daemon.Instances.Mu.Lock() + defer daemon.Instances.Mu.Unlock() + assert.Equal(t, vm.StateShuttingDown, instance.Status) +} + +// TestRollbackEBSMount_Success verifies that rollbackEBSMount sends an ebs.unmount +// request and handles a successful unmount response. +func TestRollbackEBSMount_Success(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createTestDaemon(t, natsURL) + + unmountCalled := make(chan string, 1) + + // Mock ebs.unmount subscriber that returns success + sub, err := daemon.natsConn.Subscribe("ebs.unmount", func(msg *nats.Msg) { + var req config.EBSRequest + json.Unmarshal(msg.Data, &req) + unmountCalled <- req.Name + resp := config.EBSUnMountResponse{Volume: req.Name, Mounted: false} + data, _ := json.Marshal(resp) + msg.Respond(data) + }) + require.NoError(t, err) + defer sub.Unsubscribe() + + ebsReq := config.EBSRequest{ + Name: "vol-rollback-test", + DeviceName: "/dev/sdf", + } + + daemon.rollbackEBSMount(ebsReq) + + select { + case volName := <-unmountCalled: + assert.Equal(t, "vol-rollback-test", volName) + case <-time.After(2 * time.Second): + t.Fatal("ebs.unmount was not called") + } +} + +// TestRollbackEBSMount_UnmountError verifies that rollbackEBSMount handles +// an error response from ebs.unmount gracefully (no panic, just logs). +func TestRollbackEBSMount_UnmountError(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createTestDaemon(t, natsURL) + + // Mock ebs.unmount subscriber that returns an error + sub, err := daemon.natsConn.Subscribe("ebs.unmount", func(msg *nats.Msg) { + resp := config.EBSUnMountResponse{Error: "unmount failed: device busy"} + data, _ := json.Marshal(resp) + msg.Respond(data) + }) + require.NoError(t, err) + defer sub.Unsubscribe() + + ebsReq := config.EBSRequest{Name: "vol-rollback-err"} + + // Should not panic — errors are logged but not propagated + daemon.rollbackEBSMount(ebsReq) +} + +// TestRollbackEBSMount_StillMounted verifies that rollbackEBSMount handles +// the case where the unmount response says the volume is still mounted. +func TestRollbackEBSMount_StillMounted(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createTestDaemon(t, natsURL) + + sub, err := daemon.natsConn.Subscribe("ebs.unmount", func(msg *nats.Msg) { + resp := config.EBSUnMountResponse{Mounted: true} // still mounted + data, _ := json.Marshal(resp) + msg.Respond(data) + }) + require.NoError(t, err) + defer sub.Unsubscribe() + + ebsReq := config.EBSRequest{Name: "vol-still-mounted"} + + // Should not panic + daemon.rollbackEBSMount(ebsReq) +} + +// TestRollbackEBSMount_NATSTimeout verifies that rollbackEBSMount handles +// NATS request timeout gracefully (no subscriber on ebs.unmount). +func TestRollbackEBSMount_NATSTimeout(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createTestDaemon(t, natsURL) + + // No ebs.unmount subscriber — will timeout + ebsReq := config.EBSRequest{Name: "vol-timeout"} + + // Should not panic, just log the timeout + daemon.rollbackEBSMount(ebsReq) +} diff --git a/hive/daemon/jetstream_test.go b/hive/daemon/jetstream_test.go index 6aab5d842..58d1d17c6 100644 --- a/hive/daemon/jetstream_test.go +++ b/hive/daemon/jetstream_test.go @@ -1,58 +1,17 @@ package daemon import ( - "os" "testing" - "time" "github.com/mulgadc/hive/hive/vm" - "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) -// startTestNATSServerWithJetStream starts an embedded NATS server with JetStream enabled for testing -func startTestNATSServerWithJetStream(t *testing.T) (*server.Server, string) { - // Create a temporary directory for JetStream storage - tmpDir, err := os.MkdirTemp("", "nats-js-test-*") - require.NoError(t, err, "Failed to create temp directory for JetStream") - - t.Cleanup(func() { - os.RemoveAll(tmpDir) - }) - - opts := &server.Options{ - Host: "127.0.0.1", - Port: -1, // Let NATS auto-allocate an available port - JetStream: true, - StoreDir: tmpDir, - NoLog: true, - NoSigs: true, - } - - ns, err := server.NewServer(opts) - require.NoError(t, err, "Failed to create NATS server with JetStream") - - // Start server in a goroutine - go ns.Start() - - // Wait for server to be ready - if !ns.ReadyForConnections(5 * time.Second) { - t.Fatal("NATS server with JetStream failed to start") - } - - // Get the actual URL that was assigned - url := ns.ClientURL() - t.Logf("Test NATS server with JetStream started at: %s", url) - - return ns, url -} - // TestJetStreamManager_WriteAndLoadState tests round-trip write and load of instance state func TestJetStreamManager_WriteAndLoadState(t *testing.T) { - ns, natsURL := startTestNATSServerWithJetStream(t) - defer ns.Shutdown() + natsURL := sharedJSNATSURL // Connect to NATS nc, err := nats.Connect(natsURL) @@ -105,8 +64,7 @@ func TestJetStreamManager_WriteAndLoadState(t *testing.T) { // TestJetStreamManager_LoadState_KeyNotFound tests that LoadState returns empty state when key doesn't exist func TestJetStreamManager_LoadState_KeyNotFound(t *testing.T) { - ns, natsURL := startTestNATSServerWithJetStream(t) - defer ns.Shutdown() + natsURL := sharedJSNATSURL nc, err := nats.Connect(natsURL) require.NoError(t, err) @@ -127,8 +85,7 @@ func TestJetStreamManager_LoadState_KeyNotFound(t *testing.T) { // TestJetStreamManager_BucketCreation tests that InitKVBucket creates the bucket when it doesn't exist func TestJetStreamManager_BucketCreation(t *testing.T) { - ns, natsURL := startTestNATSServerWithJetStream(t) - defer ns.Shutdown() + natsURL := sharedJSNATSURL nc, err := nats.Connect(natsURL) require.NoError(t, err) @@ -147,8 +104,7 @@ func TestJetStreamManager_BucketCreation(t *testing.T) { // TestJetStreamManager_BucketReconnection tests that InitKVBucket connects to existing bucket func TestJetStreamManager_BucketReconnection(t *testing.T) { - ns, natsURL := startTestNATSServerWithJetStream(t) - defer ns.Shutdown() + natsURL := sharedJSNATSURL // First connection - create the bucket nc1, err := nats.Connect(natsURL) @@ -195,8 +151,7 @@ func TestJetStreamManager_BucketReconnection(t *testing.T) { // TestJetStreamManager_DeleteState tests deleting state from the KV store func TestJetStreamManager_DeleteState(t *testing.T) { - ns, natsURL := startTestNATSServerWithJetStream(t) - defer ns.Shutdown() + natsURL := sharedJSNATSURL nc, err := nats.Connect(natsURL) require.NoError(t, err) @@ -238,8 +193,7 @@ func TestJetStreamManager_DeleteState(t *testing.T) { // TestJetStreamManager_DeleteState_NonExistent tests deleting state that doesn't exist func TestJetStreamManager_DeleteState_NonExistent(t *testing.T) { - ns, natsURL := startTestNATSServerWithJetStream(t) - defer ns.Shutdown() + natsURL := sharedJSNATSURL nc, err := nats.Connect(natsURL) require.NoError(t, err) @@ -258,8 +212,7 @@ func TestJetStreamManager_DeleteState_NonExistent(t *testing.T) { // TestJetStreamManager_WriteState_UpdateExisting tests that writing state updates existing entry func TestJetStreamManager_WriteState_UpdateExisting(t *testing.T) { - ns, natsURL := startTestNATSServerWithJetStream(t) - defer ns.Shutdown() + natsURL := sharedJSNATSURL nc, err := nats.Connect(natsURL) require.NoError(t, err) @@ -311,8 +264,7 @@ func TestJetStreamManager_WriteState_UpdateExisting(t *testing.T) { // TestJetStreamManager_MultipleNodes tests storing state for multiple nodes func TestJetStreamManager_MultipleNodes(t *testing.T) { - ns, natsURL := startTestNATSServerWithJetStream(t) - defer ns.Shutdown() + natsURL := sharedJSNATSURL nc, err := nats.Connect(natsURL) require.NoError(t, err) @@ -363,8 +315,7 @@ func TestJetStreamManager_MultipleNodes(t *testing.T) { // TestJetStreamManager_KVNotInitialized tests error handling when KV is not initialized func TestJetStreamManager_KVNotInitialized(t *testing.T) { - ns, natsURL := startTestNATSServerWithJetStream(t) - defer ns.Shutdown() + natsURL := sharedJSNATSURL nc, err := nats.Connect(natsURL) require.NoError(t, err) @@ -387,8 +338,7 @@ func TestJetStreamManager_KVNotInitialized(t *testing.T) { // TestJetStreamManager_UpdateReplicas tests updating replica count for the KV bucket func TestJetStreamManager_UpdateReplicas(t *testing.T) { - ns, natsURL := startTestNATSServerWithJetStream(t) - defer ns.Shutdown() + natsURL := sharedJSNATSURL nc, err := nats.Connect(natsURL) require.NoError(t, err) diff --git a/hive/daemon/main_test.go b/hive/daemon/main_test.go new file mode 100644 index 000000000..538b165d8 --- /dev/null +++ b/hive/daemon/main_test.go @@ -0,0 +1,73 @@ +package daemon + +import ( + "fmt" + "os" + "testing" + "time" + + "github.com/nats-io/nats-server/v2/server" +) + +// Shared NATS servers for all daemon tests — started once in TestMain. +var ( + sharedNATSServer *server.Server + sharedNATSURL string + sharedJSServer *server.Server + sharedJSNATSURL string +) + +func TestMain(m *testing.M) { + // Start a plain NATS server (no JetStream) + ns, err := server.NewServer(&server.Options{ + Host: "127.0.0.1", + Port: -1, + NoLog: true, + NoSigs: true, + }) + if err != nil { + fmt.Fprintf(os.Stderr, "Failed to create shared NATS server: %v\n", err) + os.Exit(1) + } + go ns.Start() + if !ns.ReadyForConnections(5 * time.Second) { + fmt.Fprintln(os.Stderr, "Shared NATS server failed to start") + os.Exit(1) + } + sharedNATSServer = ns + sharedNATSURL = ns.ClientURL() + + // Start a JetStream-enabled NATS server + jsTmpDir, err := os.MkdirTemp("", "nats-js-shared-*") + if err != nil { + fmt.Fprintf(os.Stderr, "Failed to create JetStream temp dir: %v\n", err) + os.Exit(1) + } + jsNS, err := server.NewServer(&server.Options{ + Host: "127.0.0.1", + Port: -1, + JetStream: true, + StoreDir: jsTmpDir, + NoLog: true, + NoSigs: true, + }) + if err != nil { + fmt.Fprintf(os.Stderr, "Failed to create shared JetStream NATS server: %v\n", err) + os.Exit(1) + } + go jsNS.Start() + if !jsNS.ReadyForConnections(5 * time.Second) { + fmt.Fprintln(os.Stderr, "Shared JetStream NATS server failed to start") + os.Exit(1) + } + sharedJSServer = jsNS + sharedJSNATSURL = jsNS.ClientURL() + + code := m.Run() + + ns.Shutdown() + jsNS.Shutdown() + os.RemoveAll(jsTmpDir) + + os.Exit(code) +}