package store import ( "context" "fmt" "os" "sync" "testing" "time" "github.com/google/uuid" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) // TestEtcdStore tests the basic operations of the EtcdStore implementation // This is an integration test that requires starting an embedded etcd server func TestEtcdStore(t *testing.T) { // Create a temporary directory for etcd data tempDir, err := os.MkdirTemp("", "etcd-test-*") require.NoError(t, err) defer os.RemoveAll(tempDir) // Configure and start embedded etcd etcdConfig := EtcdEmbedConfig{ Name: "test-node", DataDir: tempDir, ClientURLs: []string{"http://localhost:0"}, // Use port 0 to get a random available port PeerURLs: []string{"http://localhost:0"}, } etcdServer, err := StartEmbeddedEtcd(etcdConfig) require.NoError(t, err) // Use a cleanup function instead of defer to avoid double-close var once sync.Once t.Cleanup(func() { once.Do(func() { if etcdServer != nil { // Wrap in a recover to handle potential "close of closed channel" panic func() { defer func() { if r := recover(); r != nil { // Log the panic but continue - the server was likely already closed t.Logf("Recovered from panic while closing etcd server: %v", r) } }() etcdServer.Close() }() } }) }) // Get the actual client URL that was assigned clientURL := etcdServer.Clients[0].Addr().String() // Create the store store, err := NewEtcdStore([]string{clientURL}, etcdServer) require.NoError(t, err) defer store.Close() // Test context with timeout ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() // Test Put and Get t.Run("PutAndGet", func(t *testing.T) { key := "/test/key1" value := []byte("test-value-1") err := store.Put(ctx, key, value) require.NoError(t, err) kv, err := store.Get(ctx, key) require.NoError(t, err) assert.Equal(t, key, kv.Key) assert.Equal(t, value, kv.Value) assert.Greater(t, kv.Version, int64(0)) }) // Test List t.Run("List", func(t *testing.T) { // Put multiple keys with same prefix prefix := "/test/list/" for i := 0; i < 3; i++ { key := fmt.Sprintf("%s%d", prefix, i) value := []byte(fmt.Sprintf("value-%d", i)) err := store.Put(ctx, key, value) require.NoError(t, err) } // List keys with prefix kvs, err := store.List(ctx, prefix) require.NoError(t, err) assert.Len(t, kvs, 3) // Verify each key starts with prefix for _, kv := range kvs { assert.True(t, len(kv.Key) > len(prefix)) assert.Equal(t, prefix, kv.Key[:len(prefix)]) } }) // Test Delete t.Run("Delete", func(t *testing.T) { key := "/test/delete-key" value := []byte("delete-me") // Put a key err := store.Put(ctx, key, value) require.NoError(t, err) // Verify it exists _, err = store.Get(ctx, key) require.NoError(t, err) // Delete it err = store.Delete(ctx, key) require.NoError(t, err) // Verify it's gone _, err = store.Get(ctx, key) require.Error(t, err) }) // Test Watch t.Run("Watch", func(t *testing.T) { prefix := "/test/watch/" key := prefix + "key1" // Start watching before any changes watchCh, err := store.Watch(ctx, prefix, 0) require.NoError(t, err) // Make changes in a goroutine go func() { time.Sleep(100 * time.Millisecond) store.Put(ctx, key, []byte("watch-value-1")) time.Sleep(100 * time.Millisecond) store.Put(ctx, key, []byte("watch-value-2")) time.Sleep(100 * time.Millisecond) store.Delete(ctx, key) }() // Collect events var events []WatchEvent timeout := time.After(2 * time.Second) eventLoop: for { select { case event, ok := <-watchCh: if !ok { break eventLoop } events = append(events, event) if len(events) >= 3 { break eventLoop } case <-timeout: t.Fatal("Timed out waiting for watch events") break eventLoop } } // Verify events require.Len(t, events, 3) // First event: Put watch-value-1 assert.Equal(t, EventTypePut, events[0].Type) assert.Equal(t, key, events[0].KV.Key) assert.Equal(t, []byte("watch-value-1"), events[0].KV.Value) // Second event: Put watch-value-2 assert.Equal(t, EventTypePut, events[1].Type) assert.Equal(t, key, events[1].KV.Key) assert.Equal(t, []byte("watch-value-2"), events[1].KV.Value) // Third event: Delete assert.Equal(t, EventTypeDelete, events[2].Type) assert.Equal(t, key, events[2].KV.Key) }) // Test DoTransaction t.Run("DoTransaction", func(t *testing.T) { key1 := "/test/txn/key1" key2 := "/test/txn/key2" // Put key1 first err := store.Put(ctx, key1, []byte("txn-value-1")) require.NoError(t, err) // Get key1 to get its version kv, err := store.Get(ctx, key1) require.NoError(t, err) version := kv.Version // Transaction: If key1 has expected version, put key2 checks := []Compare{ {Key: key1, ExpectedVersion: version}, } onSuccess := []Op{ {Type: OpPut, Key: key2, Value: []byte("txn-value-2")}, } onFailure := []Op{} // Empty for this test committed, err := store.DoTransaction(ctx, checks, onSuccess, onFailure) require.NoError(t, err) assert.True(t, committed) // Verify key2 was created kv2, err := store.Get(ctx, key2) require.NoError(t, err) assert.Equal(t, []byte("txn-value-2"), kv2.Value) // Now try a transaction that should fail checks = []Compare{ {Key: key1, ExpectedVersion: version + 100}, // Wrong version } committed, err = store.DoTransaction(ctx, checks, onSuccess, onFailure) require.NoError(t, err) assert.False(t, committed) }) } // TestLeaderElection tests the Campaign, Resign, and GetLeader methods func TestLeaderElection(t *testing.T) { // Create a temporary directory for etcd data tempDir, err := os.MkdirTemp("", "etcd-election-test-*") require.NoError(t, err) defer os.RemoveAll(tempDir) // Configure and start embedded etcd etcdConfig := EtcdEmbedConfig{ Name: "election-test-node", DataDir: tempDir, ClientURLs: []string{"http://localhost:0"}, PeerURLs: []string{"http://localhost:0"}, } etcdServer, err := StartEmbeddedEtcd(etcdConfig) require.NoError(t, err) // Use a cleanup function instead of defer to avoid double-close var once sync.Once t.Cleanup(func() { once.Do(func() { if etcdServer != nil { // Wrap in a recover to handle potential "close of closed channel" panic func() { defer func() { if r := recover(); r != nil { // Log the panic but continue - the server was likely already closed t.Logf("Recovered from panic while closing etcd server: %v", r) } }() etcdServer.Close() }() } }) }) // Get the actual client URL that was assigned clientURL := etcdServer.Clients[0].Addr().String() // Create the store store, err := NewEtcdStore([]string{clientURL}, etcdServer) require.NoError(t, err) defer store.Close() // Test context with timeout ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() // Test Campaign and GetLeader t.Run("CampaignAndGetLeader", func(t *testing.T) { leaderID := "test-leader-" + uuid.New().String()[:8] // Campaign for leadership leadershipCtx, err := store.Campaign(ctx, leaderID, 5) require.NoError(t, err) require.NotNil(t, leadershipCtx) // Wait a moment for leadership to be established time.Sleep(100 * time.Millisecond) // Verify we are the leader currentLeader, err := store.GetLeader(ctx) require.NoError(t, err) assert.Equal(t, leaderID, currentLeader) // Resign leadership err = store.Resign(ctx) require.NoError(t, err) // Wait a moment for resignation to take effect time.Sleep(500 * time.Millisecond) // Verify leadership context is cancelled select { case <-leadershipCtx.Done(): // Expected default: t.Fatal("Leadership context should be cancelled after resign") } // Verify no leader or different leader currentLeader, err = store.GetLeader(ctx) require.NoError(t, err) assert.NotEqual(t, leaderID, currentLeader, "Should not still be leader after resigning") }) // Test multiple candidates t.Run("MultipleLeaderCandidates", func(t *testing.T) { // Create a second store client store2, err := NewEtcdStore([]string{clientURL}, nil) // No embedded server for this one require.NoError(t, err) defer store2.Close() leaderID1 := "leader1-" + uuid.New().String()[:8] leaderID2 := "leader2-" + uuid.New().String()[:8] // First store campaigns leadershipCtx1, err := store.Campaign(ctx, leaderID1, 5) require.NoError(t, err) // Wait a moment for leadership to be established time.Sleep(100 * time.Millisecond) // Verify first store is leader currentLeader, err := store.GetLeader(ctx) require.NoError(t, err) assert.Equal(t, leaderID1, currentLeader) // Second store campaigns but shouldn't become leader yet leadershipCtx2, err := store2.Campaign(ctx, leaderID2, 5) require.NoError(t, err) // Wait a moment to ensure leadership state is stable time.Sleep(100 * time.Millisecond) // Verify first store is still leader currentLeader, err = store.GetLeader(ctx) require.NoError(t, err) assert.Equal(t, leaderID1, currentLeader) // First store resigns err = store.Resign(ctx) require.NoError(t, err) // Wait for second store to become leader deadline := time.Now().Add(3 * time.Second) var leaderFound bool for time.Now().Before(deadline) { currentLeader, err = store2.GetLeader(ctx) if err == nil && currentLeader == leaderID2 { leaderFound = true break } time.Sleep(100 * time.Millisecond) } // Verify second store is now leader assert.True(t, leaderFound, "Second candidate should have become leader") assert.Equal(t, leaderID2, currentLeader) // Verify first leadership context is cancelled select { case <-leadershipCtx1.Done(): // Expected default: t.Fatal("First leadership context should be cancelled after resign") } // Second store resigns err = store2.Resign(ctx) require.NoError(t, err) // Verify second leadership context is cancelled select { case <-leadershipCtx2.Done(): // Expected default: t.Fatal("Second leadership context should be cancelled after resign") } }) } // TestEtcdStoreWithMockEmbeddedEtcd tests the EtcdStore with a mock embedded etcd // This is a unit test that doesn't require starting a real etcd server func TestEtcdStoreWithMockEmbeddedEtcd(t *testing.T) { // This test would use mocks to test the EtcdStore without starting a real etcd server // For brevity, we'll skip the implementation of this test t.Skip("Mock-based unit test not implemented") }