diff --git a/hive/admin/admin_test.go b/hive/admin/admin_test.go new file mode 100644 index 000000000..70a40b214 --- /dev/null +++ b/hive/admin/admin_test.go @@ -0,0 +1,598 @@ +package admin + +import ( + "crypto/x509" + "encoding/base64" + "encoding/pem" + "fmt" + "net" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// --- Key / Token generation --- + +func TestGenerateAWSAccessKey_Format(t *testing.T) { + key := GenerateAWSAccessKey() + assert.Len(t, key, 20) + assert.True(t, strings.HasPrefix(key, "AKIA")) + for _, c := range key[4:] { + assert.True(t, (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9'), + "unexpected character %c in access key suffix", c) + } +} + +func TestGenerateAWSAccessKey_Uniqueness(t *testing.T) { + assert.NotEqual(t, GenerateAWSAccessKey(), GenerateAWSAccessKey()) +} + +func TestGenerateAWSSecretKey_Format(t *testing.T) { + key := GenerateAWSSecretKey() + assert.Len(t, key, 40) + _, err := base64.StdEncoding.DecodeString(key) + assert.NoError(t, err, "secret key should be valid base64") +} + +func TestGenerateAWSSecretKey_Uniqueness(t *testing.T) { + assert.NotEqual(t, GenerateAWSSecretKey(), GenerateAWSSecretKey()) +} + +func TestGenerateAccountID_Format(t *testing.T) { + id := GenerateAccountID() + assert.Len(t, id, 12) + for _, c := range id { + assert.True(t, c >= '0' && c <= '9', "expected digit, got %c", c) + } + // First digit should not be 0 + assert.NotEqual(t, byte('0'), id[0], "12-digit account ID should not start with 0") +} + +func TestGenerateAccountID_Uniqueness(t *testing.T) { + assert.NotEqual(t, GenerateAccountID(), GenerateAccountID()) +} + +func TestGenerateNATSToken_Format(t *testing.T) { + token := GenerateNATSToken() + assert.True(t, strings.HasPrefix(token, "nats_")) + assert.Len(t, token, 37) // 5 prefix + 32 random + // URL-safe base64: no '+' or '/' + assert.NotContains(t, token, "+") + assert.NotContains(t, token, "/") +} + +func TestGenerateNATSToken_Uniqueness(t *testing.T) { + assert.NotEqual(t, GenerateNATSToken(), GenerateNATSToken()) +} + +// --- Config file generation --- + +func TestGenerateConfigFile_Success(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "test.conf") + + tmpl := "region = {{.Region}}\nnode = {{.Node}}" + settings := ConfigSettings{Region: "us-east-1", Node: "node1"} + + err := GenerateConfigFile(path, tmpl, settings) + require.NoError(t, err) + + data, err := os.ReadFile(path) + require.NoError(t, err) + assert.Contains(t, string(data), "region = us-east-1") + assert.Contains(t, string(data), "node = node1") + + info, _ := os.Stat(path) + assert.Equal(t, os.FileMode(0600), info.Mode().Perm()) +} + +func TestGenerateConfigFile_InvalidTemplate(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "bad.conf") + err := GenerateConfigFile(path, "{{.Unclosed", ConfigSettings{}) + assert.Error(t, err) + assert.Contains(t, err.Error(), "failed to parse template") +} + +func TestGenerateConfigFile_InvalidPath(t *testing.T) { + err := GenerateConfigFile("/nonexistent/dir/file.conf", "ok", ConfigSettings{}) + assert.Error(t, err) + assert.Contains(t, err.Error(), "failed to create config file") +} + +func TestGenerateConfigFile_OverwritesExisting(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "overwrite.conf") + + require.NoError(t, GenerateConfigFile(path, "old={{.Region}}", ConfigSettings{Region: "old"})) + require.NoError(t, GenerateConfigFile(path, "new={{.Region}}", ConfigSettings{Region: "new"})) + + data, _ := os.ReadFile(path) + assert.Contains(t, string(data), "new=new") + assert.NotContains(t, string(data), "old") +} + +func TestGenerateConfigFiles_AllSucceed(t *testing.T) { + dir := t.TempDir() + configs := []ConfigFile{ + {Name: "a", Path: filepath.Join(dir, "a.conf"), Template: "a={{.Region}}"}, + {Name: "b", Path: filepath.Join(dir, "b.conf"), Template: "b={{.Node}}"}, + } + err := GenerateConfigFiles(configs, ConfigSettings{Region: "us-west-2", Node: "n1"}) + require.NoError(t, err) + + for _, cfg := range configs { + assert.True(t, FileExists(cfg.Path)) + } +} + +func TestGenerateConfigFiles_StopsOnFirstError(t *testing.T) { + dir := t.TempDir() + configs := []ConfigFile{ + {Name: "ok", Path: filepath.Join(dir, "ok.conf"), Template: "ok"}, + {Name: "bad", Path: "/nonexistent/dir/bad.conf", Template: "bad"}, + {Name: "never", Path: filepath.Join(dir, "never.conf"), Template: "never"}, + } + err := GenerateConfigFiles(configs, ConfigSettings{}) + assert.Error(t, err) + assert.Contains(t, err.Error(), "bad") + assert.False(t, FileExists(filepath.Join(dir, "never.conf"))) +} + +// --- FileExists --- + +func TestFileExists(t *testing.T) { + dir := t.TempDir() + existing := filepath.Join(dir, "exists.txt") + require.NoError(t, os.WriteFile(existing, []byte("hi"), 0644)) + + assert.True(t, FileExists(existing)) + assert.False(t, FileExists(filepath.Join(dir, "nope.txt"))) + assert.True(t, FileExists(dir)) // directory also returns true +} + +// --- UpdateAWSINIFile --- + +func TestUpdateAWSINIFile_CreateNew(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "credentials") + + err := UpdateAWSINIFile(path, "hive", map[string]string{ + "aws_access_key_id": "AKIATEST", + "aws_secret_access_key": "secrettest", + }) + require.NoError(t, err) + + data, _ := os.ReadFile(path) + content := string(data) + assert.Contains(t, content, "[hive]") + assert.Contains(t, content, "AKIATEST") + assert.Contains(t, content, "secrettest") +} + +func TestUpdateAWSINIFile_UpdateExisting(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "credentials") + + // Create with initial value + require.NoError(t, UpdateAWSINIFile(path, "hive", map[string]string{"key": "old"})) + // Update + require.NoError(t, UpdateAWSINIFile(path, "hive", map[string]string{"key": "new"})) + + data, _ := os.ReadFile(path) + content := string(data) + assert.Contains(t, content, "new") +} + +func TestUpdateAWSINIFile_AddNewSection(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "credentials") + + require.NoError(t, UpdateAWSINIFile(path, "default", map[string]string{"key": "default-val"})) + require.NoError(t, UpdateAWSINIFile(path, "hive", map[string]string{"key": "hive-val"})) + + data, _ := os.ReadFile(path) + content := string(data) + assert.Contains(t, content, "[default]") + assert.Contains(t, content, "[hive]") + assert.Contains(t, content, "default-val") + assert.Contains(t, content, "hive-val") +} + +// --- SetupAWSCredentials --- + +func TestSetupAWSCredentials_CreatesFiles(t *testing.T) { + dir := t.TempDir() + t.Setenv("HOME", dir) + + err := SetupAWSCredentials("AKIATEST123", "secret123", "us-east-1", "/path/to/ca.pem") + require.NoError(t, err) + + credData, _ := os.ReadFile(filepath.Join(dir, ".aws", "credentials")) + configData, _ := os.ReadFile(filepath.Join(dir, ".aws", "config")) + + assert.Contains(t, string(credData), "AKIATEST123") + assert.Contains(t, string(credData), "secret123") + assert.Contains(t, string(configData), "us-east-1") + assert.Contains(t, string(configData), "https://localhost:9999") + assert.Contains(t, string(configData), "/path/to/ca.pem") +} + +func TestSetupAWSCredentials_PreservesExistingProfiles(t *testing.T) { + dir := t.TempDir() + t.Setenv("HOME", dir) + + awsDir := filepath.Join(dir, ".aws") + require.NoError(t, os.MkdirAll(awsDir, 0700)) + require.NoError(t, UpdateAWSINIFile(filepath.Join(awsDir, "credentials"), "default", map[string]string{ + "aws_access_key_id": "EXISTING_KEY", + })) + + err := SetupAWSCredentials("NEWAKEY", "NEWSECRET", "us-west-2", "/ca.pem") + require.NoError(t, err) + + data, _ := os.ReadFile(filepath.Join(awsDir, "credentials")) + content := string(data) + assert.Contains(t, content, "EXISTING_KEY") + assert.Contains(t, content, "NEWAKEY") +} + +// --- Certificate generation --- + +func TestGenerateCACert_CreatesValidCA(t *testing.T) { + t.Parallel() + dir := t.TempDir() + certPath := filepath.Join(dir, "ca.pem") + keyPath := filepath.Join(dir, "ca.key") + + err := GenerateCACert(certPath, keyPath) + require.NoError(t, err) + + // Parse certificate + certPEM, _ := os.ReadFile(certPath) + block, _ := pem.Decode(certPEM) + require.NotNil(t, block) + assert.Equal(t, "CERTIFICATE", block.Type) + + cert, err := x509.ParseCertificate(block.Bytes) + require.NoError(t, err) + assert.True(t, cert.IsCA) + assert.Equal(t, "Hive Local CA", cert.Subject.CommonName) + assert.NotZero(t, cert.KeyUsage&x509.KeyUsageCertSign) + + // Parse key + keyPEM, _ := os.ReadFile(keyPath) + keyBlock, _ := pem.Decode(keyPEM) + require.NotNil(t, keyBlock) + assert.Equal(t, "PRIVATE KEY", keyBlock.Type) + + // Verify key file permissions + info, _ := os.Stat(keyPath) + assert.Equal(t, os.FileMode(0600), info.Mode().Perm()) +} + +// TestGenerateSignedCert uses subtests sharing a single CA to avoid repeated 4096-bit key generation. +func TestGenerateSignedCert(t *testing.T) { + t.Parallel() + // Generate CA once for all subtests (~0.7s instead of ~0.7s x 3) + caDir := t.TempDir() + caCertPath := filepath.Join(caDir, "ca.pem") + caKeyPath := filepath.Join(caDir, "ca.key") + require.NoError(t, GenerateCACert(caCertPath, caKeyPath)) + + t.Run("CreatesValidCert", func(t *testing.T) { + t.Parallel() + dir := t.TempDir() + certPath := filepath.Join(dir, "server.pem") + keyPath := filepath.Join(dir, "server.key") + + require.NoError(t, GenerateSignedCert(certPath, keyPath, caCertPath, caKeyPath)) + + certPEM, _ := os.ReadFile(certPath) + block, _ := pem.Decode(certPEM) + require.NotNil(t, block) + + cert, err := x509.ParseCertificate(block.Bytes) + require.NoError(t, err) + assert.False(t, cert.IsCA) + assert.Equal(t, "localhost", cert.Subject.CommonName) + assert.Contains(t, cert.DNSNames, "localhost") + + hasLoopback := false + for _, ip := range cert.IPAddresses { + if ip.Equal(net.ParseIP("127.0.0.1")) { + hasLoopback = true + } + } + assert.True(t, hasLoopback) + + // Verify against CA + caCertPEM, _ := os.ReadFile(caCertPath) + caBlock, _ := pem.Decode(caCertPEM) + caCert, _ := x509.ParseCertificate(caBlock.Bytes) + pool := x509.NewCertPool() + pool.AddCert(caCert) + _, err = cert.Verify(x509.VerifyOptions{Roots: pool}) + assert.NoError(t, err) + + info, _ := os.Stat(keyPath) + assert.Equal(t, os.FileMode(0600), info.Mode().Perm()) + }) + + t.Run("ExtraIPs", func(t *testing.T) { + t.Parallel() + dir := t.TempDir() + certPath := filepath.Join(dir, "server.pem") + keyPath := filepath.Join(dir, "server.key") + + require.NoError(t, GenerateSignedCert(certPath, keyPath, caCertPath, caKeyPath, "192.168.1.100")) + + certPEM, _ := os.ReadFile(certPath) + block, _ := pem.Decode(certPEM) + cert, _ := x509.ParseCertificate(block.Bytes) + + hasExtraIP := false + for _, ip := range cert.IPAddresses { + if ip.Equal(net.ParseIP("192.168.1.100")) { + hasExtraIP = true + } + } + assert.True(t, hasExtraIP, "cert should contain extra IP 192.168.1.100") + }) + + t.Run("SkipsDuplicateAndSpecialIPs", func(t *testing.T) { + t.Parallel() + dir := t.TempDir() + certPath := filepath.Join(dir, "server.pem") + keyPath := filepath.Join(dir, "server.key") + + require.NoError(t, GenerateSignedCert(certPath, keyPath, caCertPath, caKeyPath, "127.0.0.1", "::1", "0.0.0.0", "")) + + certPEM, _ := os.ReadFile(certPath) + block, _ := pem.Decode(certPEM) + cert, _ := x509.ParseCertificate(block.Bytes) + + assert.Len(t, cert.IPAddresses, 2) + }) + + t.Run("InvalidCACert", func(t *testing.T) { + t.Parallel() + dir := t.TempDir() + badCACert := filepath.Join(dir, "bad-ca.pem") + require.NoError(t, os.WriteFile(badCACert, []byte("not-a-cert"), 0600)) + + err := GenerateSignedCert(filepath.Join(dir, "s.pem"), filepath.Join(dir, "s.key"), badCACert, caKeyPath) + assert.Error(t, err) + }) +} + +func TestGenerateSelfSignedCert_CreatesValidCert(t *testing.T) { + t.Parallel() + dir := t.TempDir() + certPath := filepath.Join(dir, "self.pem") + keyPath := filepath.Join(dir, "self.key") + + require.NoError(t, GenerateSelfSignedCert(certPath, keyPath)) + + certPEM, _ := os.ReadFile(certPath) + block, _ := pem.Decode(certPEM) + require.NotNil(t, block) + + cert, err := x509.ParseCertificate(block.Bytes) + require.NoError(t, err) + assert.Contains(t, cert.DNSNames, "localhost") + + info, _ := os.Stat(keyPath) + assert.Equal(t, os.FileMode(0600), info.Mode().Perm()) +} + +// --- Certificate orchestrator --- + +// TestGenerateCertificatesIfNeeded uses subtests to share the initial generation. +func TestGenerateCertificatesIfNeeded(t *testing.T) { + t.Parallel() + dir := t.TempDir() + + // First call generates all certs + caCertPath := GenerateCertificatesIfNeeded(dir, false, "") + assert.Equal(t, filepath.Join(dir, "ca.pem"), caCertPath) + assert.True(t, FileExists(filepath.Join(dir, "ca.pem"))) + assert.True(t, FileExists(filepath.Join(dir, "ca.key"))) + assert.True(t, FileExists(filepath.Join(dir, "server.pem"))) + assert.True(t, FileExists(filepath.Join(dir, "server.key"))) + + t.Run("SkipsWhenAllExist", func(t *testing.T) { + caInfo, _ := os.Stat(filepath.Join(dir, "ca.pem")) + origModTime := caInfo.ModTime() + + GenerateCertificatesIfNeeded(dir, false, "") + caInfo2, _ := os.Stat(filepath.Join(dir, "ca.pem")) + assert.Equal(t, origModTime, caInfo2.ModTime()) + }) + + t.Run("ForceRegenerates", func(t *testing.T) { + origCA, _ := os.ReadFile(filepath.Join(dir, "ca.pem")) + + GenerateCertificatesIfNeeded(dir, true, "") + newCA, _ := os.ReadFile(filepath.Join(dir, "ca.pem")) + assert.NotEqual(t, origCA, newCA) + }) +} + +func TestGenerateServerCertOnly(t *testing.T) { + t.Parallel() + // Generate CA once, reuse for subtests + caDir := t.TempDir() + require.NoError(t, GenerateCACert(filepath.Join(caDir, "ca.pem"), filepath.Join(caDir, "ca.key"))) + + t.Run("Success", func(t *testing.T) { + dir := t.TempDir() + // Copy CA files into test dir + caCert, _ := os.ReadFile(filepath.Join(caDir, "ca.pem")) + caKey, _ := os.ReadFile(filepath.Join(caDir, "ca.key")) + require.NoError(t, os.WriteFile(filepath.Join(dir, "ca.pem"), caCert, 0600)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "ca.key"), caKey, 0600)) + + err := GenerateServerCertOnly(dir, "10.0.0.5") + require.NoError(t, err) + assert.True(t, FileExists(filepath.Join(dir, "server.pem"))) + assert.True(t, FileExists(filepath.Join(dir, "server.key"))) + + certPEM, _ := os.ReadFile(filepath.Join(dir, "server.pem")) + block, _ := pem.Decode(certPEM) + cert, _ := x509.ParseCertificate(block.Bytes) + + hasBindIP := false + for _, ip := range cert.IPAddresses { + if ip.Equal(net.ParseIP("10.0.0.5")) { + hasBindIP = true + } + } + assert.True(t, hasBindIP) + }) + + t.Run("MissingCA", func(t *testing.T) { + dir := t.TempDir() + err := GenerateServerCertOnly(dir, "10.0.0.5") + assert.Error(t, err) + assert.Contains(t, err.Error(), "CA files not found") + }) +} + +// --- Directory creation --- + +func TestCreateServiceDirectories_CreatesAll(t *testing.T) { + dir := t.TempDir() + CreateServiceDirectories(dir) + + expected := []string{"images", "amis", "volumes", "state", "logs", "nats", "predastore", "viperblock", "hive"} + for _, name := range expected { + path := filepath.Join(dir, name) + info, err := os.Stat(path) + assert.NoError(t, err, "directory %s should exist", name) + if err == nil { + assert.True(t, info.IsDir()) + } + } +} + +func TestCreateServiceDirectories_Idempotent(t *testing.T) { + dir := t.TempDir() + CreateServiceDirectories(dir) + // Should not error on second call + CreateServiceDirectories(dir) + + info, err := os.Stat(filepath.Join(dir, "images")) + assert.NoError(t, err) + assert.True(t, info.IsDir()) +} + +// --- Predastore multi-node config --- + +func TestGenerateMultiNodePredastoreConfig_Success(t *testing.T) { + tmpl := `{{range .Nodes}}[[db]] +id = {{.ID}} +host = "{{.Host}}" +{{end}}` + nodes := []PredastoreNodeConfig{ + {ID: 1, Host: "10.0.0.1"}, + {ID: 2, Host: "10.0.0.2"}, + {ID: 3, Host: "10.0.0.3"}, + } + + result, err := GenerateMultiNodePredastoreConfig(tmpl, nodes, "AK", "SK", "us-east-1") + require.NoError(t, err) + assert.Contains(t, result, `host = "10.0.0.1"`) + assert.Contains(t, result, `host = "10.0.0.3"`) +} + +func TestGenerateMultiNodePredastoreConfig_MinimumNodes(t *testing.T) { + tmpl := "{{range .Nodes}}{{.ID}}{{end}}" + + _, err := GenerateMultiNodePredastoreConfig(tmpl, []PredastoreNodeConfig{ + {ID: 1, Host: "10.0.0.1"}, + {ID: 2, Host: "10.0.0.2"}, + }, "AK", "SK", "us-east-1") + assert.Error(t, err) + assert.Contains(t, err.Error(), "at least 3 nodes") +} + +func TestGenerateMultiNodePredastoreConfig_InvalidTemplate(t *testing.T) { + _, err := GenerateMultiNodePredastoreConfig("{{.Unclosed", []PredastoreNodeConfig{ + {ID: 1, Host: "a"}, {ID: 2, Host: "b"}, {ID: 3, Host: "c"}, + }, "AK", "SK", "us-east-1") + assert.Error(t, err) + assert.Contains(t, err.Error(), "failed to parse") +} + +// --- FindNodeIDByIP --- + +func TestFindNodeIDByIP(t *testing.T) { + nodes := []PredastoreNodeConfig{ + {ID: 1, Host: "10.0.0.1"}, + {ID: 2, Host: "10.0.0.2"}, + {ID: 3, Host: "10.0.0.3"}, + } + + assert.Equal(t, 2, FindNodeIDByIP(nodes, "10.0.0.2")) + assert.Equal(t, 0, FindNodeIDByIP(nodes, "10.0.0.99")) + assert.Equal(t, 0, FindNodeIDByIP(nil, "10.0.0.1")) +} + +// --- ParsePredastoreNodeIDFromConfig --- + +func TestParsePredastoreNodeIDFromConfig(t *testing.T) { + tomlContent := ` +[[db]] +id = 1 +host = "10.0.0.1" + +[[db]] +id = 2 +host = "10.0.0.2" + +[[db]] +id = 3 +host = "10.0.0.3" +` + assert.Equal(t, 2, ParsePredastoreNodeIDFromConfig(tomlContent, "10.0.0.2")) + assert.Equal(t, 0, ParsePredastoreNodeIDFromConfig(tomlContent, "10.0.0.99")) + assert.Equal(t, 0, ParsePredastoreNodeIDFromConfig("invalid toml {{{", "10.0.0.1")) + assert.Equal(t, 0, ParsePredastoreNodeIDFromConfig("", "10.0.0.1")) +} + +// --- Integration: Full config generation flow --- + +func TestGenerateConfigFile_HiveTomlTemplate(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "hive.toml") + + tmpl := `version = "1.0" +epoch = 1 +node = "{{.Node}}" + +[nodes.{{.Node}}] +region = "{{.Region}}" +accesskey = "{{.AccessKey}}" +secretkey = "{{.SecretKey}}" +` + + settings := ConfigSettings{ + Node: "node1", + Region: "us-east-1", + AccessKey: "AKIATEST", + SecretKey: "SECRET", + } + + require.NoError(t, GenerateConfigFile(path, tmpl, settings)) + + data, _ := os.ReadFile(path) + content := string(data) + assert.Contains(t, content, `node = "node1"`) + assert.Contains(t, content, `region = "us-east-1"`) + assert.Contains(t, content, fmt.Sprintf(`accesskey = "%s"`, settings.AccessKey)) +} diff --git a/hive/config/config_test.go b/hive/config/config_test.go new file mode 100644 index 000000000..4de701b60 --- /dev/null +++ b/hive/config/config_test.go @@ -0,0 +1,227 @@ +package config + +import ( + "os" + "path/filepath" + "testing" + + "github.com/spf13/viper" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func resetViper(t *testing.T) { + t.Cleanup(func() { viper.Reset() }) +} + +func TestLoadConfig_ValidTOMLFile(t *testing.T) { + resetViper(t) + dir := t.TempDir() + path := filepath.Join(dir, "hive.toml") + + toml := ` +version = "1.0" +epoch = 1 +node = "node1" + +[nodes.node1] +node = "node1" +region = "us-east-1" +az = "us-east-1a" +accesskey = "AKIATEST" +secretkey = "SECRET" + +[nodes.node1.daemon] +host = "127.0.0.1:8080" + +[nodes.node1.nats] +host = "127.0.0.1:4222" + +[nodes.node1.nats.acl] +token = "nats_testtoken" + +[nodes.node1.predastore] +host = "127.0.0.1:8443" +bucket = "predastore" +region = "us-east-1" +` + require.NoError(t, os.WriteFile(path, []byte(toml), 0600)) + + cfg, err := LoadConfig(path) + require.NoError(t, err) + require.NotNil(t, cfg) + + assert.Equal(t, uint64(1), cfg.Epoch) + assert.Equal(t, "node1", cfg.Node) + assert.Equal(t, "1.0", cfg.Version) + + node, ok := cfg.Nodes["node1"] + require.True(t, ok, "node1 should exist in Nodes map") + assert.Equal(t, "us-east-1", node.Region) + assert.Equal(t, "us-east-1a", node.AZ) + assert.Equal(t, "AKIATEST", node.AccessKey) + assert.Equal(t, "127.0.0.1:8080", node.Daemon.Host) + assert.Equal(t, "127.0.0.1:4222", node.NATS.Host) + assert.Equal(t, "nats_testtoken", node.NATS.ACL.Token) + assert.Equal(t, "127.0.0.1:8443", node.Predastore.Host) + assert.Equal(t, "predastore", node.Predastore.Bucket) +} + +func TestLoadConfig_MultipleNodes(t *testing.T) { + resetViper(t) + dir := t.TempDir() + path := filepath.Join(dir, "hive.toml") + + toml := ` +epoch = 2 +node = "leader" +version = "2.0" + +[nodes.leader] +region = "us-east-1" + +[nodes.follower] +region = "us-west-2" +` + require.NoError(t, os.WriteFile(path, []byte(toml), 0600)) + + cfg, err := LoadConfig(path) + require.NoError(t, err) + assert.Len(t, cfg.Nodes, 2) + assert.Equal(t, "us-east-1", cfg.Nodes["leader"].Region) + assert.Equal(t, "us-west-2", cfg.Nodes["follower"].Region) +} + +func TestLoadConfig_EmptyConfigPath(t *testing.T) { + resetViper(t) + cfg, err := LoadConfig("") + require.NoError(t, err) + require.NotNil(t, cfg) + // All zero values + assert.Equal(t, uint64(0), cfg.Epoch) + assert.Empty(t, cfg.Node) +} + +func TestLoadConfig_NonexistentFile(t *testing.T) { + resetViper(t) + cfg, err := LoadConfig("/tmp/nonexistent-hive-config-test-12345.toml") + // Not an error - falls through to defaults + require.NoError(t, err) + require.NotNil(t, cfg) +} + +func TestLoadConfig_MalformedTOML(t *testing.T) { + resetViper(t) + dir := t.TempDir() + path := filepath.Join(dir, "bad.toml") + require.NoError(t, os.WriteFile(path, []byte("this is not valid toml {{{"), 0600)) + + cfg, err := LoadConfig(path) + assert.Error(t, err) + assert.Nil(t, cfg) + assert.Contains(t, err.Error(), "error reading config file") +} + +func TestLoadConfig_PartialConfig(t *testing.T) { + resetViper(t) + dir := t.TempDir() + path := filepath.Join(dir, "partial.toml") + require.NoError(t, os.WriteFile(path, []byte(`node = "partial-node" +epoch = 5 +`), 0600)) + + cfg, err := LoadConfig(path) + require.NoError(t, err) + assert.Equal(t, "partial-node", cfg.Node) + assert.Equal(t, uint64(5), cfg.Epoch) + assert.Empty(t, cfg.Version) + assert.Nil(t, cfg.Nodes) +} + +func TestLoadConfig_EnvVarOverrideWithFile(t *testing.T) { + resetViper(t) + dir := t.TempDir() + path := filepath.Join(dir, "hive.toml") + + // Viper's AutomaticEnv only works for keys Viper already knows about + // (from a config file or explicit BindEnv). Provide a minimal config + // so Viper registers the "epoch" key, then override via env. + require.NoError(t, os.WriteFile(path, []byte(`epoch = 1 +node = "file-node" +`), 0600)) + + t.Setenv("HIVE_NODE", "env-node") + + cfg, err := LoadConfig(path) + require.NoError(t, err) + // Env vars override file values for keys Viper knows about + assert.Equal(t, "env-node", cfg.Node) +} + +func TestLoadConfig_NestedStructParsing(t *testing.T) { + resetViper(t) + dir := t.TempDir() + path := filepath.Join(dir, "nested.toml") + + toml := ` +node = "n1" + +[nodes.n1] +region = "ap-southeast-2" + +[nodes.n1.daemon] +host = "0.0.0.0:8080" +tlskey = "server.key" +tlscert = "server.pem" + +[nodes.n1.nats] +host = "0.0.0.0:4222" + +[nodes.n1.nats.acl] +token = "secret-token" + +[nodes.n1.nats.sub] +subject = "test-subject" + +[nodes.n1.predastore] +host = "0.0.0.0:8443" +bucket = "mybucket" +region = "ap-southeast-2" +accesskey = "AK" +secretkey = "SK" +base_dir = "/data" +node_id = 1 + +[nodes.n1.awsgw] +host = "0.0.0.0:9999" +debug = true +expected_nodes = 3 +` + require.NoError(t, os.WriteFile(path, []byte(toml), 0600)) + + cfg, err := LoadConfig(path) + require.NoError(t, err) + + n := cfg.Nodes["n1"] + assert.Equal(t, "0.0.0.0:8080", n.Daemon.Host) + assert.Equal(t, "server.key", n.Daemon.TLSKey) + assert.Equal(t, "server.pem", n.Daemon.TLSCert) + assert.Equal(t, "0.0.0.0:4222", n.NATS.Host) + assert.Equal(t, "secret-token", n.NATS.ACL.Token) + assert.Equal(t, "test-subject", n.NATS.Sub.Subject) + assert.Equal(t, "0.0.0.0:8443", n.Predastore.Host) + assert.Equal(t, "mybucket", n.Predastore.Bucket) + assert.Equal(t, "AK", n.Predastore.AccessKey) + assert.Equal(t, "/data", n.Predastore.BaseDir) + assert.Equal(t, 1, n.Predastore.NodeID) + assert.Equal(t, "0.0.0.0:9999", n.AWSGW.Host) + assert.True(t, n.AWSGW.Debug) + assert.Equal(t, 3, n.AWSGW.ExpectedNodes) +} + +// Tests for type constants + +func TestNBDTransportConstants(t *testing.T) { + assert.Equal(t, NBDTransport("socket"), NBDTransportSocket) + assert.Equal(t, NBDTransport("tcp"), NBDTransportTCP) +} diff --git a/hive/daemon/daemon_test.go b/hive/daemon/daemon_test.go index a599103dd..54c42eb1d 100644 --- a/hive/daemon/daemon_test.go +++ b/hive/daemon/daemon_test.go @@ -5,6 +5,7 @@ import ( "fmt" "net" "os" + "path/filepath" "runtime" "strings" "sync" @@ -18,6 +19,7 @@ import ( 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/pelletier/go-toml/v2" "github.com/mulgadc/hive/hive/vm" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" @@ -2996,3 +2998,339 @@ func TestAttachVolume_ReplacesStaleEBSRequest(t *testing.T) { instance.EBSRequests.Mu.Unlock() assert.Equal(t, 1, count, "Should have exactly one EBSRequest for the volume, not a duplicate") } + +// --- computeConfigHash --- + +func TestComputeConfigHash_Deterministic(t *testing.T) { + d := &Daemon{clusterConfig: &config.ClusterConfig{ + Epoch: 1, + Version: "1.0", + Nodes: map[string]config.Config{ + "n1": {Region: "us-east-1"}, + }, + }} + + h1, err := d.computeConfigHash() + require.NoError(t, err) + h2, err := d.computeConfigHash() + require.NoError(t, err) + assert.Equal(t, h1, h2) + assert.Len(t, h1, 64) // SHA256 hex +} + +func TestComputeConfigHash_ChangesOnMutation(t *testing.T) { + d := &Daemon{clusterConfig: &config.ClusterConfig{ + Epoch: 1, + Version: "1.0", + Nodes: map[string]config.Config{ + "n1": {Region: "us-east-1"}, + }, + }} + + h1, _ := d.computeConfigHash() + + d.clusterConfig.Epoch = 2 + h2, _ := d.computeConfigHash() + assert.NotEqual(t, h1, h2) + + d.clusterConfig.Epoch = 1 + d.clusterConfig.Nodes["n2"] = config.Config{Region: "eu-west-1"} + h3, _ := d.computeConfigHash() + assert.NotEqual(t, h1, h3) +} + +func TestComputeConfigHash_ExcludesNodeField(t *testing.T) { + d := &Daemon{clusterConfig: &config.ClusterConfig{ + Epoch: 1, + Version: "1.0", + Node: "node-a", + Nodes: map[string]config.Config{ + "n1": {Region: "us-east-1"}, + }, + }} + + h1, _ := d.computeConfigHash() + d.clusterConfig.Node = "node-b" + h2, _ := d.computeConfigHash() + assert.Equal(t, h1, h2, "changing top-level Node should not affect config hash") +} + +// --- saveClusterConfig --- + +func TestSaveClusterConfig_WritesToDisk(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "hive.toml") + + d := &Daemon{ + configPath: path, + clusterConfig: &config.ClusterConfig{ + Epoch: 5, + Version: "2.0", + Node: "test-node", + Nodes: map[string]config.Config{ + "test-node": {Region: "us-west-2"}, + }, + }, + } + + err := d.saveClusterConfig() + require.NoError(t, err) + + data, err := os.ReadFile(path) + require.NoError(t, err) + + var loaded config.ClusterConfig + require.NoError(t, toml.Unmarshal(data, &loaded)) + assert.Equal(t, uint64(5), loaded.Epoch) + assert.Equal(t, "2.0", loaded.Version) + assert.Equal(t, "us-west-2", loaded.Nodes["test-node"].Region) + + // Verify permissions + info, _ := os.Stat(path) + assert.Equal(t, os.FileMode(0600), info.Mode().Perm()) +} + +func TestSaveClusterConfig_ErrorOnEmptyPath(t *testing.T) { + d := &Daemon{ + configPath: "", + clusterConfig: &config.ClusterConfig{}, + } + err := d.saveClusterConfig() + assert.Error(t, err) + assert.Contains(t, err.Error(), "config path not set") +} + +func TestSaveClusterConfig_ErrorOnInvalidPath(t *testing.T) { + d := &Daemon{ + configPath: "/nonexistent/dir/hive.toml", + clusterConfig: &config.ClusterConfig{}, + } + err := d.saveClusterConfig() + assert.Error(t, err) + assert.Contains(t, err.Error(), "failed to write config") +} + +// --- generateInstanceTypes --- + +func TestGenerateInstanceTypes_DualFamily(t *testing.T) { + types := generateInstanceTypes("m7i", "x86_64") + assert.Len(t, types, 14) // 7 m7i + 7 t3 + + m7iCount, t3Count := 0, 0 + for name := range types { + if strings.HasPrefix(name, "m7i.") { + m7iCount++ + } + if strings.HasPrefix(name, "t3.") { + t3Count++ + } + } + assert.Equal(t, 7, m7iCount) + assert.Equal(t, 7, t3Count) +} + +func TestGenerateInstanceTypes_SingleFamily(t *testing.T) { + // When family matches burstable, should not duplicate + types := generateInstanceTypes("t3", "x86_64") + assert.Len(t, types, 7) + for name := range types { + assert.True(t, strings.HasPrefix(name, "t3.")) + } +} + +func TestGenerateInstanceTypes_ARM64(t *testing.T) { + types := generateInstanceTypes("m8g", "arm64") + // m8g is not t4g, so we should get both families + assert.Len(t, types, 14) + + t4gCount := 0 + for name := range types { + if strings.HasPrefix(name, "t4g.") { + t4gCount++ + } + } + assert.Equal(t, 7, t4gCount) +} + +func TestGenerateInstanceTypes_ARM64_SingleFamily(t *testing.T) { + types := generateInstanceTypes("t4g", "arm64") + assert.Len(t, types, 7) + for name := range types { + assert.True(t, strings.HasPrefix(name, "t4g.")) + } +} + +func TestGenerateInstanceTypes_VerifySizes(t *testing.T) { + types := generateInstanceTypes("t3", "x86_64") + + expected := map[string]struct { + vcpus int64 + memMB int64 + }{ + "t3.nano": {2, 512}, + "t3.micro": {2, 1024}, + "t3.small": {2, 2048}, + "t3.medium": {2, 4096}, + "t3.large": {2, 8192}, + "t3.xlarge": {4, 16384}, + "t3.2xlarge": {8, 32768}, + } + + for name, exp := range expected { + it, ok := types[name] + require.True(t, ok, "missing instance type %s", name) + assert.Equal(t, exp.vcpus, *it.VCpuInfo.DefaultVCpus, "%s vCPUs", name) + assert.Equal(t, exp.memMB, *it.MemoryInfo.SizeInMiB, "%s memory", name) + } +} + +// --- getInstanceFamilyFromCPU --- + +func TestGetInstanceFamilyFromCPU_ExactMappings(t *testing.T) { + assert.Equal(t, "m8a", getInstanceFamilyFromCPU("AMD EPYC 7551")) + assert.Equal(t, "m7i", getInstanceFamilyFromCPU("Intel Xeon E5-2686")) + assert.Equal(t, "m8g", getInstanceFamilyFromCPU("Apple M1")) + assert.Equal(t, "m8g", getInstanceFamilyFromCPU("AWS Graviton2")) + assert.Equal(t, "m8g", getInstanceFamilyFromCPU("ARM Cortex-A72")) +} + +func TestGetInstanceFamilyFromCPU_CaseInsensitive(t *testing.T) { + assert.Equal(t, "m8a", getInstanceFamilyFromCPU("amd epyc 9654")) +} + +func TestGetInstanceFamilyFromCPU_DefaultFallback(t *testing.T) { + family := getInstanceFamilyFromCPU("Unknown CPU Model XYZ") + if runtime.GOARCH == "arm64" { + assert.Equal(t, "t4g", family) + } else { + assert.Equal(t, "t3", family) + } +} + +// --- instanceTypeVCPUs / instanceTypeMemoryMiB nil safety --- + +func TestInstanceTypeVCPUs_NilSafety(t *testing.T) { + // Nil VCpuInfo + assert.Equal(t, int64(0), instanceTypeVCPUs(&ec2.InstanceTypeInfo{})) + + // Non-nil VCpuInfo but nil DefaultVCpus + assert.Equal(t, int64(0), instanceTypeVCPUs(&ec2.InstanceTypeInfo{ + VCpuInfo: &ec2.VCpuInfo{}, + })) + + // Valid + assert.Equal(t, int64(4), instanceTypeVCPUs(&ec2.InstanceTypeInfo{ + VCpuInfo: &ec2.VCpuInfo{DefaultVCpus: aws.Int64(4)}, + })) +} + +func TestInstanceTypeMemoryMiB_NilSafety(t *testing.T) { + // Nil MemoryInfo + assert.Equal(t, int64(0), instanceTypeMemoryMiB(&ec2.InstanceTypeInfo{})) + + // Non-nil MemoryInfo but nil SizeInMiB + assert.Equal(t, int64(0), instanceTypeMemoryMiB(&ec2.InstanceTypeInfo{ + MemoryInfo: &ec2.MemoryInfo{}, + })) + + // Valid + assert.Equal(t, int64(8192), instanceTypeMemoryMiB(&ec2.InstanceTypeInfo{ + MemoryInfo: &ec2.MemoryInfo{SizeInMiB: aws.Int64(8192)}, + })) +} + +// --- Daemon.WriteState / Daemon.LoadState nil jsManager --- + +func TestDaemon_WriteState_NilJSManager(t *testing.T) { + d := &Daemon{jsManager: nil} + err := d.WriteState() + assert.Error(t, err) + assert.Contains(t, err.Error(), "JetStream manager not initialized") +} + +func TestDaemon_LoadState_NilJSManager(t *testing.T) { + d := &Daemon{jsManager: nil} + err := d.LoadState() + assert.Error(t, err) + assert.Contains(t, err.Error(), "JetStream manager not initialized") +} + +// --- GetAvailableInstanceTypeInfos edge cases --- + +func TestGetAvailableInstanceTypeInfos_Overcommitted(t *testing.T) { + rm := &ResourceManager{ + availableVCPU: 4, + availableMem: 8.0, + allocatedVCPU: 8, // Over-committed + allocatedMem: 16.0, + instanceTypes: map[string]*ec2.InstanceTypeInfo{ + "t3.micro": { + InstanceType: aws.String("t3.micro"), + VCpuInfo: &ec2.VCpuInfo{DefaultVCpus: aws.Int64(2)}, + MemoryInfo: &ec2.MemoryInfo{SizeInMiB: aws.Int64(1024)}, + }, + }, + } + + infos := rm.GetAvailableInstanceTypeInfos(true) + assert.Empty(t, infos, "overcommitted resources should return 0 available slots") + + infos = rm.GetAvailableInstanceTypeInfos(false) + assert.Empty(t, infos) +} + +func TestGetAvailableInstanceTypeInfos_ShowCapacity(t *testing.T) { + rm := &ResourceManager{ + availableVCPU: 8, + availableMem: 16.0, + allocatedVCPU: 0, + allocatedMem: 0, + instanceTypes: map[string]*ec2.InstanceTypeInfo{ + "t3.micro": { + InstanceType: aws.String("t3.micro"), + VCpuInfo: &ec2.VCpuInfo{DefaultVCpus: aws.Int64(2)}, + MemoryInfo: &ec2.MemoryInfo{SizeInMiB: aws.Int64(1024)}, + }, + }, + } + + // With showCapacity=true, should return multiple entries + infos := rm.GetAvailableInstanceTypeInfos(true) + assert.Greater(t, len(infos), 1) + + // With showCapacity=false, should return exactly 1 + infos = rm.GetAvailableInstanceTypeInfos(false) + assert.Len(t, infos, 1) +} + +// --- NewDaemon --- + +func TestNewDaemon_WalDirDefaultsToBaseDir(t *testing.T) { + cfg := &config.ClusterConfig{ + Node: "n1", + Nodes: map[string]config.Config{ + "n1": { + BaseDir: "/data/hive", + WalDir: "", // Empty - should default to BaseDir + }, + }, + } + + d := NewDaemon(cfg) + assert.Equal(t, "/data/hive", d.config.WalDir) +} + +func TestNewDaemon_WalDirPreservedIfSet(t *testing.T) { + cfg := &config.ClusterConfig{ + Node: "n1", + Nodes: map[string]config.Config{ + "n1": { + BaseDir: "/data/hive", + WalDir: "/fast-ssd/wal", + }, + }, + } + + d := NewDaemon(cfg) + assert.Equal(t, "/fast-ssd/wal", d.config.WalDir) +} diff --git a/hive/services/viperblockd/viperblockd_handlers_test.go b/hive/services/viperblockd/viperblockd_handlers_test.go new file mode 100644 index 000000000..ef6d9f8f2 --- /dev/null +++ b/hive/services/viperblockd/viperblockd_handlers_test.go @@ -0,0 +1,306 @@ +package viperblockd + +// Tests for NATS message handlers: ebs.delete, ebs.sync, ebs.unmount (socket cleanup) +// These extend the existing integration tests to cover untested handler paths. + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" + "time" + + "github.com/mulgadc/hive/hive/config" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// --- ebs.delete handler tests --- + +func TestIntegration_EBSDeleteMountedVolume(t *testing.T) { + t.Parallel() + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + ns, natsURL := setupEmbeddedNATS(t) + defer ns.Shutdown() + + // Create a temp socket file to verify cleanup + tmpDir := t.TempDir() + socketPath := filepath.Join(tmpDir, "vol-del-test.sock") + require.NoError(t, os.WriteFile(socketPath, []byte("fake-socket"), 0600)) + + // Connect a client and create a subscription to use as SnapshotSub + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + snapSub, err := nc.Subscribe("ebs.snapshot.vol-del-test", func(msg *nats.Msg) {}) + require.NoError(t, err) + + cfg := setupTestConfig(t, natsURL) + cfg.MountedVolumes = []MountedVolume{ + { + Name: "vol-del-test", + Port: 10809, + Socket: socketPath, + PID: 99999, // Fake PID + SnapshotSub: snapSub, + }, + } + + go func() { launchService(cfg) }() + time.Sleep(500 * time.Millisecond) + + reqData, _ := json.Marshal(config.EBSDeleteRequest{Volume: "vol-del-test"}) + msg, err := nc.Request("ebs.delete", reqData, 3*time.Second) + require.NoError(t, err) + + var resp config.EBSDeleteResponse + require.NoError(t, json.Unmarshal(msg.Data, &resp)) + + assert.Equal(t, "vol-del-test", resp.Volume) + assert.True(t, resp.Success) + assert.Empty(t, resp.Error) + + // Verify volume removed from config + cfg.mu.Lock() + assert.Len(t, cfg.MountedVolumes, 0) + cfg.mu.Unlock() + + // Verify socket file deleted + assert.False(t, fileExistsCheck(socketPath)) + + // Verify snapshot subscription was unsubscribed + assert.False(t, snapSub.IsValid()) +} + +func TestIntegration_EBSDeleteUnmountedVolume(t *testing.T) { + t.Parallel() + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + ns, natsURL := setupEmbeddedNATS(t) + defer ns.Shutdown() + + cfg := setupTestConfig(t, natsURL) + // No mounted volumes + + go func() { launchService(cfg) }() + time.Sleep(500 * time.Millisecond) + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + reqData, _ := json.Marshal(config.EBSDeleteRequest{Volume: "vol-not-mounted"}) + msg, err := nc.Request("ebs.delete", reqData, 3*time.Second) + require.NoError(t, err) + + var resp config.EBSDeleteResponse + require.NoError(t, json.Unmarshal(msg.Data, &resp)) + assert.True(t, resp.Success) +} + +func TestIntegration_EBSDeleteInvalidJSON(t *testing.T) { + t.Parallel() + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + ns, natsURL := setupEmbeddedNATS(t) + defer ns.Shutdown() + + cfg := setupTestConfig(t, natsURL) + + go func() { launchService(cfg) }() + time.Sleep(500 * time.Millisecond) + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + msg, err := nc.Request("ebs.delete", []byte("not json {{{"), 3*time.Second) + require.NoError(t, err) + + var resp config.EBSDeleteResponse + require.NoError(t, json.Unmarshal(msg.Data, &resp)) + assert.Contains(t, resp.Error, "bad request:") +} + +// --- ebs.sync handler tests --- + +func TestIntegration_EBSSyncVolumeNotMounted(t *testing.T) { + t.Parallel() + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + ns, natsURL := setupEmbeddedNATS(t) + defer ns.Shutdown() + + cfg := setupTestConfig(t, natsURL) + + go func() { launchService(cfg) }() + time.Sleep(500 * time.Millisecond) + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + reqData, _ := json.Marshal(config.EBSSyncRequest{Volume: "vol-not-here"}) + msg, err := nc.Request("ebs.sync", reqData, 3*time.Second) + require.NoError(t, err) + + var resp config.EBSSyncResponse + require.NoError(t, json.Unmarshal(msg.Data, &resp)) + assert.False(t, resp.Synced) + assert.Contains(t, resp.Error, "not mounted") +} + +func TestIntegration_EBSSyncVolumeNoVBInstance(t *testing.T) { + t.Parallel() + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + ns, natsURL := setupEmbeddedNATS(t) + defer ns.Shutdown() + + cfg := setupTestConfig(t, natsURL) + cfg.MountedVolumes = []MountedVolume{ + {Name: "vol-no-vb", VB: nil}, // Volume exists but no VB instance + } + + go func() { launchService(cfg) }() + time.Sleep(500 * time.Millisecond) + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + reqData, _ := json.Marshal(config.EBSSyncRequest{Volume: "vol-no-vb"}) + msg, err := nc.Request("ebs.sync", reqData, 3*time.Second) + require.NoError(t, err) + + var resp config.EBSSyncResponse + require.NoError(t, json.Unmarshal(msg.Data, &resp)) + assert.False(t, resp.Synced) + assert.Contains(t, resp.Error, "not mounted") +} + +func TestIntegration_EBSSyncInvalidJSON(t *testing.T) { + t.Parallel() + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + ns, natsURL := setupEmbeddedNATS(t) + defer ns.Shutdown() + + cfg := setupTestConfig(t, natsURL) + + go func() { launchService(cfg) }() + time.Sleep(500 * time.Millisecond) + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + msg, err := nc.Request("ebs.sync", []byte("garbage"), 3*time.Second) + require.NoError(t, err) + + var resp config.EBSSyncResponse + require.NoError(t, json.Unmarshal(msg.Data, &resp)) + assert.Contains(t, resp.Error, "bad request:") +} + +// --- ebs.unmount socket cleanup --- + +func TestIntegration_EBSUnmountRemovesSocket(t *testing.T) { + t.Parallel() + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + ns, natsURL := setupEmbeddedNATS(t) + defer ns.Shutdown() + + // Create an actual temp file as the socket + tmpDir := t.TempDir() + socketPath := filepath.Join(tmpDir, "vol-unmount-socket.sock") + require.NoError(t, os.WriteFile(socketPath, []byte("fake"), 0600)) + + cfg := setupTestConfig(t, natsURL) + cfg.MountedVolumes = []MountedVolume{ + { + Name: "vol-unmount-socket", + Socket: socketPath, + PID: 99999, + }, + } + + go func() { launchService(cfg) }() + time.Sleep(500 * time.Millisecond) + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + reqData, _ := json.Marshal(config.EBSRequest{Name: "vol-unmount-socket"}) + msg, err := nc.Request("ebs.unmount", reqData, 3*time.Second) + require.NoError(t, err) + + var resp config.EBSUnMountResponse + require.NoError(t, json.Unmarshal(msg.Data, &resp)) + assert.Empty(t, resp.Error) + + // Verify socket file was removed + assert.False(t, fileExistsCheck(socketPath)) +} + +// --- ebs.delete removes socket --- + +func TestIntegration_EBSDeleteRemovesSocket(t *testing.T) { + t.Parallel() + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + ns, natsURL := setupEmbeddedNATS(t) + defer ns.Shutdown() + + tmpDir := t.TempDir() + socketPath := filepath.Join(tmpDir, "vol-del-socket.sock") + require.NoError(t, os.WriteFile(socketPath, []byte("fake"), 0600)) + + cfg := setupTestConfig(t, natsURL) + cfg.MountedVolumes = []MountedVolume{ + {Name: "vol-del-socket", Socket: socketPath, PID: 99999}, + } + + go func() { launchService(cfg) }() + time.Sleep(500 * time.Millisecond) + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + reqData, _ := json.Marshal(config.EBSDeleteRequest{Volume: "vol-del-socket"}) + msg, err := nc.Request("ebs.delete", reqData, 3*time.Second) + require.NoError(t, err) + + var resp config.EBSDeleteResponse + require.NoError(t, json.Unmarshal(msg.Data, &resp)) + assert.True(t, resp.Success) + assert.False(t, fileExistsCheck(socketPath)) +} + +// fileExistsCheck is a helper to check if a file exists on disk. +func fileExistsCheck(path string) bool { + _, err := os.Stat(path) + return err == nil +} diff --git a/hive/services/viperblockd/viperblockd_integration_test.go b/hive/services/viperblockd/viperblockd_integration_test.go index ea6a228e8..ec7c8ebe6 100644 --- a/hive/services/viperblockd/viperblockd_integration_test.go +++ b/hive/services/viperblockd/viperblockd_integration_test.go @@ -84,6 +84,7 @@ func createMockVolumeState(t *testing.T, baseDir, volumeName string) { // TestIntegration_ServiceStartWithEmbeddedNATS tests service startup with embedded NATS func TestIntegration_ServiceStartWithEmbeddedNATS(t *testing.T) { + t.Parallel() if testing.Short() { t.Skip("Skipping integration test in short mode") } @@ -117,6 +118,7 @@ func TestIntegration_ServiceStartWithEmbeddedNATS(t *testing.T) { // TestIntegration_EBSMountRequest tests EBS mount request handling func TestIntegration_EBSMountRequest(t *testing.T) { + t.Parallel() if testing.Short() { t.Skip("Skipping integration test in short mode") } @@ -179,6 +181,7 @@ func TestIntegration_EBSMountRequest(t *testing.T) { // TestIntegration_EBSUnmountRequest tests EBS unmount request handling func TestIntegration_EBSUnmountRequest(t *testing.T) { + t.Parallel() if testing.Short() { t.Skip("Skipping integration test in short mode") } @@ -248,6 +251,7 @@ func TestIntegration_EBSUnmountRequest(t *testing.T) { // TestIntegration_EBSUnmountNonExistentVolume tests unmounting a volume that doesn't exist func TestIntegration_EBSUnmountNonExistentVolume(t *testing.T) { + t.Parallel() // Skip if testing.Short() { @@ -300,6 +304,7 @@ func TestIntegration_EBSUnmountNonExistentVolume(t *testing.T) { // TestIntegration_ConcurrentMountRequests tests multiple concurrent mount requests func TestIntegration_ConcurrentMountRequests(t *testing.T) { + t.Parallel() if testing.Short() { t.Skip("Skipping integration test in short mode") } @@ -366,6 +371,7 @@ func TestIntegration_ConcurrentMountRequests(t *testing.T) { // TestIntegration_MessageSubscriptions tests that all expected subscriptions are active func TestIntegration_MessageSubscriptions(t *testing.T) { + t.Parallel() if testing.Short() { t.Skip("Skipping integration test in short mode") } @@ -419,6 +425,7 @@ func TestIntegration_MessageSubscriptions(t *testing.T) { // TestIntegration_ServiceGracefulShutdown tests graceful shutdown behavior func TestIntegration_ServiceGracefulShutdown(t *testing.T) { + t.Parallel() if testing.Short() { t.Skip("Skipping integration test in short mode") }