Add integration tests for MultiThreaded RESP server

integration_tests/commands/resp/abort/server_abort_test.go

@@ -0,0 +1,130 @@
+package abort
+
+import (
+	"context"
+	"fmt"
+	"github.com/dicedb/dice/integration_tests/commands/resp"
+	"net"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/dicedb/dice/config"
+)
+
+var testServerOptions = resp.TestServerOptions{
+	Port: 8740,
+}
+
+func TestAbortCommand(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer ctx.Done()
+	t.Cleanup(cancel)
+
+	var wg sync.WaitGroup
+	resp.RunTestServer(&wg, testServerOptions)
+
+	time.Sleep(2 * time.Second)
+
+	// Test 1: Ensure the server is running
+	t.Run("ServerIsRunning", func(t *testing.T) {
+		conn, err := net.Dial("tcp", fmt.Sprintf("127.0.0.1:%d", config.DiceConfig.Server.Port))
+		if err != nil {
+			t.Fatalf("Failed to connect to server: %v", err)
+		}
+		conn.Close()
+	})
+
+	//Test 2: Send ABORT command and check if the server shuts down
+	t.Run("AbortCommandShutdown", func(t *testing.T) {
+		conn, err := net.Dial("tcp", fmt.Sprintf("127.0.0.1:%d", config.DiceConfig.Server.Port))
+		if err != nil {
+			t.Fatalf("Failed to connect to server: %v", err)
+		}
+		defer conn.Close()
+
+		// Send ABORT command
+		result := resp.FireCommand(conn, "ABORT")
+		if result != "OK" {
+			t.Fatalf("Unexpected response to ABORT command: %v", result)
+		}
+
+		// Wait for the server to shut down
+		time.Sleep(1 * time.Second)
+
+		// Try to connect again, it should fail
+		_, err = net.Dial("tcp", fmt.Sprintf("127.0.0.1:%d", config.DiceConfig.Server.Port))
+		if err == nil {
+			t.Fatal("Server did not shut down as expected")
+		}
+	})
+
+	// Test 3: Ensure the server port is released
+	t.Run("PortIsReleased", func(t *testing.T) {
+		// Try to bind to the same port
+		listener, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", config.DiceConfig.Server.Port))
+		if err != nil {
+			t.Fatalf("Port should be available after server shutdown: %v", err)
+		}
+		listener.Close()
+	})
+
+	wg.Wait()
+}
+
+func TestServerRestartAfterAbort(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer ctx.Done()
+	t.Cleanup(cancel)
+
+	// start test server.
+	var wg sync.WaitGroup
+	resp.RunTestServer(&wg, testServerOptions)
+
+	time.Sleep(1 * time.Second)
+
+	conn, err := net.Dial("tcp", fmt.Sprintf("127.0.0.1:%d", config.DiceConfig.Server.Port))
+	if err != nil {
+		t.Fatalf("Server should be running after restart: %v", err)
+	}
+
+	// Send ABORT command to shut down server
+	result := resp.FireCommand(conn, "ABORT")
+	if result != "OK" {
+		t.Fatalf("Unexpected response to ABORT command: %v", result)
+	}
+	conn.Close()
+
+	// wait for the server to shutdown
+	time.Sleep(2 * time.Second)
+
+	wg.Wait()
+
+	// restart server
+	ctx2, cancel2 := context.WithCancel(context.Background())
+	defer ctx2.Done()
+	t.Cleanup(cancel2)
+
+	// start test server.
+	// use different waitgroups and contexts to avoid race conditions.;
+	var wg2 sync.WaitGroup
+	resp.RunTestServer(&wg2, testServerOptions)
+
+	// wait for the server to start up
+	time.Sleep(2 * time.Second)
+
+	// Check if the server is running
+	conn2, err := net.Dial("tcp", fmt.Sprintf("127.0.0.1:%d", config.DiceConfig.Server.Port))
+	if err != nil {
+		t.Fatalf("Server should be running after restart: %v", err)
+	}
+
+	// Clean up
+	result = resp.FireCommand(conn2, "ABORT")
+	if result != "OK" {
+		t.Fatalf("Unexpected response to ABORT command: %v", result)
+	}
+	conn2.Close()
+
+	wg2.Wait()
+}

integration_tests/commands/resp/command_getkeys_test.go

@@ -0,0 +1,50 @@
+package resp
+
+import (
+	"testing"
+
+	"gotest.tools/v3/assert"
+)
+
+var getKeysTestCases = []struct {
+	name     string
+	inCmd    string
+	expected interface{}
+}{
+	{"Set command", "set 1 2 3 4", []interface{}{"1"}},
+	{"Get command", "get key", []interface{}{"key"}},
+	{"TTL command", "ttl key", []interface{}{"key"}},
+	{"Del command", "del 1 2 3 4 5 6", []interface{}{"1", "2", "3", "4", "5", "6"}},
+	{"MSET command", "MSET key1 val1 key2 val2", []interface{}{"key1", "key2"}},
+	{"Expire command", "expire key time extra", []interface{}{"key"}},
+	{"BFINIT command", "BFINIT bloom some parameters", []interface{}{"bloom"}},
+	{"Ping command", "ping", "ERR the command has no key arguments"},
+	{"Invalid Get command", "get", "ERR invalid number of arguments specified for command"},
+	{"Abort command", "abort", "ERR the command has no key arguments"},
+	{"Invalid command", "NotValidCommand", "ERR invalid command specified"},
+	{"Wrong number of arguments", "", "ERR wrong number of arguments for 'command|getkeys' command"},
+}
+
+func TestCommandGetKeys(t *testing.T) {
+	conn := getLocalConnection()
+	defer conn.Close()
+
+	for _, tc := range getKeysTestCases {
+		t.Run(tc.name, func(t *testing.T) {
+			result := FireCommand(conn, "COMMAND GETKEYS "+tc.inCmd)
+			assert.DeepEqual(t, tc.expected, result)
+		})
+	}
+}
+
+func BenchmarkGetKeysMatch(b *testing.B) {
+	conn := getLocalConnection()
+	defer conn.Close()
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		for _, tc := range getKeysTestCases {
+			FireCommand(conn, "COMMAND GETKEYS "+tc.inCmd)
+		}
+	}
+}

integration_tests/commands/resp/command_info_test.go

@@ -0,0 +1,50 @@
+package resp
+
+import (
+	"testing"
+
+	"gotest.tools/v3/assert"
+)
+
+var getInfoTestCases = []struct {
+	name     string
+	inCmd    string
+	expected interface{}
+}{
+	{"Set command", "SET", []interface{}{[]interface{}{"SET", int64(-3), int64(1), int64(0), int64(0)}}},
+	{"Get command", "GET", []interface{}{[]interface{}{"GET", int64(2), int64(1), int64(0), int64(0)}}},
+	{"Ping command", "PING", []interface{}{[]interface{}{"PING", int64(-1), int64(0), int64(0), int64(0)}}},
+	{"Invalid command", "INVALID_CMD", []interface{}{string("(nil)")}},
+	{"Combination of valid and Invalid command", "SET INVALID_CMD", []interface{}{
+		[]interface{}{"SET", int64(-3), int64(1), int64(0), int64(0)},
+		string("(nil)"),
+	}},
+	{"Combination of multiple valid commands", "SET GET", []interface{}{
+		[]interface{}{"SET", int64(-3), int64(1), int64(0), int64(0)},
+		[]interface{}{"GET", int64(2), int64(1), int64(0), int64(0)},
+	}},
+}
+
+func TestCommandInfo(t *testing.T) {
+	conn := getLocalConnection()
+	defer conn.Close()
+
+	for _, tc := range getInfoTestCases {
+		t.Run(tc.name, func(t *testing.T) {
+			result := FireCommand(conn, "COMMAND INFO "+tc.inCmd)
+			assert.DeepEqual(t, tc.expected, result)
+		})
+	}
+}
+
+func BenchmarkCommandInfo(b *testing.B) {
+	conn := getLocalConnection()
+	defer conn.Close()
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		for _, tc := range getKeysTestCases {
+			FireCommand(conn, "COMMAND INFO "+tc.inCmd)
+		}
+	}
+}

integration_tests/commands/resp/get_test.go

@@ -0,0 +1,39 @@
+package resp
+
+import (
+	"testing"
+	"time"
+
+	"gotest.tools/v3/assert"
+)
+
+func TestGet(t *testing.T) {
+	conn := getLocalConnection()
+	defer conn.Close()
+
+	testCases := []struct {
+		name   string
+		cmds   []string
+		expect []interface{}
+		delays []time.Duration
+	}{
+		{
+			name:   "Get with expiration",
+			cmds:   []string{"SET k v EX 4", "GET k", "GET k"},
+			expect: []interface{}{"OK", "v", "(nil)"},
+			delays: []time.Duration{0, 0, 5 * time.Second},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			for i, cmd := range tc.cmds {
+				if tc.delays[i] > 0 {
+					time.Sleep(tc.delays[i])
+				}
+				result := FireCommand(conn, cmd)
+				assert.Equal(t, tc.expect[i], result, "Value mismatch for cmd %s", cmd)
+			}
+		})
+	}
+}

integration_tests/commands/resp/getset_test.go

@@ -0,0 +1,57 @@
+package resp
+
+import (
+	"testing"
+	"time"
+
+	"gotest.tools/v3/assert"
+)
+
+func TestGetSet(t *testing.T) {
+	conn := getLocalConnection()
+	defer conn.Close()
+
+	testCases := []struct {
+		name   string
+		cmds   []string
+		expect []interface{}
+		delays []time.Duration
+	}{
+		{
+			name:   "GETSET with INCR",
+			cmds:   []string{"INCR mycounter", "GETSET mycounter \"0\"", "GET mycounter"},
+			expect: []interface{}{int64(1), int64(1), int64(0)},
+			delays: []time.Duration{0, 0, 0},
+		},
+		{
+			name:   "GETSET with SET",
+			cmds:   []string{"SET mykey \"Hello\"", "GETSET mykey \"world\"", "GET mykey"},
+			expect: []interface{}{"OK", "Hello", "world"},
+			delays: []time.Duration{0, 0, 0},
+		},
+		{
+			name:   "GETSET with TTL",
+			cmds:   []string{"SET k v EX 60", "GETSET k v1", "TTL k"},
+			expect: []interface{}{"OK", "v", int64(-1)},
+			delays: []time.Duration{0, 0, 0},
+		},
+		{
+			name:   "GETSET error when key exists but does not hold a string value",
+			cmds:   []string{"LPUSH k1 \"somevalue\"", "GETSET k1 \"v1\""},
+			expect: []interface{}{"OK", "WRONGTYPE Operation against a key holding the wrong kind of value"},
+			delays: []time.Duration{0, 0, 0},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			for i, cmd := range tc.cmds {
+				if tc.delays[i] > 0 {
+					time.Sleep(tc.delays[i])
+				}
+				result := FireCommand(conn, cmd)
+				assert.Equal(t, tc.expect[i], result, "Value mismatch for cmd %s", cmd)
+			}
+		})
+	}
+}

integration_tests/commands/resp/main_test.go

@@ -0,0 +1,47 @@
+package resp
+
+import (
+	"log/slog"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/dicedb/dice/internal/logger"
+)
+
+func TestMain(m *testing.M) {
+	logger := logger.New(logger.Opts{WithTimestamp: false})
+	slog.SetDefault(logger)
+
+	var wg sync.WaitGroup
+	// Run the test server
+	// This is a synchronous method, because internally it
+	// checks for available port and then forks a goroutine
+	// to start the server
+	opts := TestServerOptions{
+		Port:   8739,
+		Logger: logger,
+	}
+	RunTestServer(&wg, opts)
+
+	// Wait for the server to start
+	time.Sleep(2 * time.Second)
+
+	conn := getLocalConnection()
+	if conn == nil {
+		panic("Failed to connect to the test server")
+	}
+	defer conn.Close()
+
+	// Run the test suite
+	exitCode := m.Run()
+
+	result := FireCommand(conn, "ABORT")
+	if result != "OK" {
+		panic("Failed to abort the server")
+	}
+
+	wg.Wait()
+	os.Exit(exitCode)
+}