pkg/SugarDB
1
0
Fork 0
mirror of https://github.com/EchoVault/SugarDB.git synced 2025-10-05 16:06:57 +08:00
Code Issues Packages Projects Releases Wiki Activity

Updated all test suites to include connection and server shutdown on cleanup.

Browse Source
This commit is contained in:
Kelvin Clement Mwinuka
2024-05-31 01:30:18 +08:00
parent 6f8511632e
commit c7560ce9dd
13 changed files with 13350 additions and 13106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

119
echovault/cluster_test.go
View File

@@ -1,119 +0,0 @@
// Copyright 2024 Kelvin Clement Mwinuka
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package echovault
// func Test_raftApplyDeleteKey(t *testing.T) {
// nodes, err := makeCluster(5)
// if err != nil {
// t.Error(err)
// return
// }
//
// // Prepare the write data for the cluster.
// tests := []struct {
// key string
// value string
// }{
// {
// key: "key1",
// value: "value1",
// },
// {
// key: "key2",
// value: "value2",
// },
// {
// key: "key3",
// value: "value3",
// },
// }
//
// // Write all the data to the cluster leader.
// for i, test := range tests {
// node := nodes[0]
// if err := node.client.WriteArray([]resp.Value{
// resp.StringValue("SET"),
// resp.StringValue(test.key),
// resp.StringValue(test.value),
// }); err != nil {
// t.Errorf("could not write data to leader node (test %d): %v", i, err)
// }
// // Read response and make sure we received "ok" response.
// rd, _, err := node.client.ReadValue()
// if err != nil {
// t.Errorf("could not read response from leader node (test %d): %v", i, err)
// }
// if !strings.EqualFold(rd.String(), "ok") {
// t.Errorf("expected response for test %d to be \"OK\", got %s", i, rd.String())
// }
// }
//
// // Check if the data has been replicated on a quorum (majority of the cluster).
// quorum := int(math.Ceil(float64(len(nodes)/2)) + 1)
// for i, test := range tests {
// count := 0
// for j := 0; j < len(nodes); j++ {
// node := nodes[j]
// if err := node.client.WriteArray([]resp.Value{
// resp.StringValue("GET"),
// resp.StringValue(test.key),
// }); err != nil {
// t.Errorf("could not write data to follower node %d (test %d): %v", j, i, err)
// }
// rd, _, err := node.client.ReadValue()
// if err != nil {
// t.Errorf("could not read data from follower node %d (test %d): %v", j, i, err)
// }
// if rd.String() == test.value {
// count += 1 // If the expected value is found, increment the count.
// }
// }
// // Fail if count is less than quorum.
// if count < quorum {
// t.Errorf("could not find value %s at key %s in cluster quorum", test.value, test.key)
// }
// // Delete key across raft cluster.
// if err = nodes[0].server.raftApplyDeleteKey(nodes[0].server.context, test.key); err != nil {
// t.Error(err)
// }
// }
//
// <-time.After(200 * time.Millisecond) // Yield to give key deletion time to take effect across cluster.
//
// // Check if the data is absent in quorum (majority of the cluster).
// for i, test := range tests {
// count := 0
// for j := 0; j < len(nodes); j++ {
// node := nodes[j]
// if err := node.client.WriteArray([]resp.Value{
// resp.StringValue("GET"),
// resp.StringValue(test.key),
// }); err != nil {
// t.Errorf("could not write command to follower node %d (test %d): %v", j, i, err)
// }
// rd, _, err := node.client.ReadValue()
// if err != nil {
// t.Errorf("could not read data from follower node %d (test %d): %v", j, i, err)
// }
// if rd.IsNull() {
// count += 1 // If the expected value is found, increment the count.
// }
// }
// // Fail if count is less than quorum.
// if count < quorum {
// t.Errorf("found value %s at key %s in cluster quorum", test.value, test.key)
// }
// }
// }

58
echovault/echovault.go
View File

@@ -73,7 +73,7 @@ type EchoVault struct {
mutex sync.Mutex // Mutex as only one goroutine can edit the LFU cache at a time. mutex sync.Mutex // Mutex as only one goroutine can edit the LFU cache at a time.
cache eviction.CacheLFU // LFU cache represented by a min head. cache eviction.CacheLFU // LFU cache represented by a min head.
} }
// LRU cache used when eviction policy is allkeys-lru or volatile-lru // LRU cache used when eviction policy is allkeys-lru or volatile-lru.
lruCache struct { lruCache struct {
mutex sync.Mutex // Mutex as only one goroutine can edit the LRU at a time. mutex sync.Mutex // Mutex as only one goroutine can edit the LRU at a time.
cache eviction.CacheLRU // LRU cache represented by a max head. cache eviction.CacheLRU // LRU cache represented by a max head.
@@ -95,9 +95,12 @@ type EchoVault struct {
rewriteAOFInProgress atomic.Bool // Atomic boolean that's true when actively rewriting AOF file is in progress. rewriteAOFInProgress atomic.Bool // Atomic boolean that's true when actively rewriting AOF file is in progress.
stateCopyInProgress atomic.Bool // Atomic boolean that's true when actively copying state for snapshotting or preamble generation. stateCopyInProgress atomic.Bool // Atomic boolean that's true when actively copying state for snapshotting or preamble generation.
stateMutationInProgress atomic.Bool // Atomic boolean that is set to true when state mutation is in progress. stateMutationInProgress atomic.Bool // Atomic boolean that is set to true when state mutation is in progress.
latestSnapshotMilliseconds atomic.Int64 // Unix epoch in milliseconds latestSnapshotMilliseconds atomic.Int64 // Unix epoch in milliseconds.
snapshotEngine *snapshot.Engine // Snapshot engine for standalone mode snapshotEngine *snapshot.Engine // Snapshot engine for standalone mode.
aofEngine *aof.Engine // AOF engine for standalone mode aofEngine *aof.Engine // AOF engine for standalone mode.
listener net.Listener // TCP listener.
quit chan struct{} // Channel that signals the closing of all client connections.
} }
// WithContext is an options that for the NewEchoVault function that allows you to // WithContext is an options that for the NewEchoVault function that allows you to
@@ -142,6 +145,7 @@ func NewEchoVault(options ...func(echovault *EchoVault)) (*EchoVault, error) {
commands = append(commands, str.Commands()...) commands = append(commands, str.Commands()...)
return commands return commands
}(), }(),
quit: make(chan struct{}),
} }
for _, option := range options { for _, option := range options {
@@ -320,30 +324,35 @@ func (server *EchoVault) startTCP() {
KeepAlive: 200 * time.Millisecond, KeepAlive: 200 * time.Millisecond,
} }
listener, err := listenConfig.Listen(server.context, "tcp", fmt.Sprintf("%s:%d", conf.BindAddr, conf.Port)) listener, err := listenConfig.Listen(
server.context,
"tcp",
fmt.Sprintf("%s:%d", conf.BindAddr, conf.Port),
)
if err != nil { if err != nil {
log.Fatal(err) log.Printf("listener error: %v", err)
return
} }
if !conf.TLS { if !conf.TLS {
// TCP // TCP
log.Printf("Starting TCP echovault at Address %s, Port %d...\n", conf.BindAddr, conf.Port) log.Printf("Starting TCP server at Address %s, Port %d...\n", conf.BindAddr, conf.Port)
} }
if conf.TLS || conf.MTLS { if conf.TLS || conf.MTLS {
// TLS // TLS
if conf.TLS { if conf.TLS {
log.Printf("Starting mTLS echovault at Address %s, Port %d...\n", conf.BindAddr, conf.Port) log.Printf("Starting mTLS server at Address %s, Port %d...\n", conf.BindAddr, conf.Port)
} else { } else {
log.Printf("Starting TLS echovault at Address %s, Port %d...\n", conf.BindAddr, conf.Port) log.Printf("Starting TLS server at Address %s, Port %d...\n", conf.BindAddr, conf.Port)
} }
var certificates []tls.Certificate var certificates []tls.Certificate
for _, certKeyPair := range conf.CertKeyPairs { for _, certKeyPair := range conf.CertKeyPairs {
c, err := tls.LoadX509KeyPair(certKeyPair[0], certKeyPair[1]) c, err := tls.LoadX509KeyPair(certKeyPair[0], certKeyPair[1])
if err != nil { if err != nil {
log.Fatal(err) log.Printf("load cert key pair: %v\n", err)
return
} }
certificates = append(certificates, c) certificates = append(certificates, c)
} }
@@ -356,14 +365,15 @@ func (server *EchoVault) startTCP() {
for _, c := range conf.ClientCAs { for _, c := range conf.ClientCAs {
ca, err := os.Open(c) ca, err := os.Open(c)
if err != nil { if err != nil {
log.Fatal(err) log.Printf("client cert open: %v\n", err)
return
} }
certBytes, err := io.ReadAll(ca) certBytes, err := io.ReadAll(ca)
if err != nil { if err != nil {
log.Fatal(err) log.Printf("client cert read: %v\n", err)
} }
if ok := clientCerts.AppendCertsFromPEM(certBytes); !ok { if ok := clientCerts.AppendCertsFromPEM(certBytes); !ok {
log.Fatal(err) log.Printf("client cert append: %v\n", err)
} }
} }
} }
@@ -375,17 +385,24 @@ func (server *EchoVault) startTCP() {
}) })
} }
// Listen to connection server.listener = listener
// Listen to connection.
for { for {
conn, err := listener.Accept() select {
case <-server.quit:
return
default:
conn, err := server.listener.Accept()
if err != nil { if err != nil {
log.Println("Could not establish connection") log.Printf("listener error: %v\n", err)
continue continue
} }
// Read loop for connection // Read loop for connection
go server.handleConnection(conn) go server.handleConnection(conn)
} }
} }
}
func (server *EchoVault) handleConnection(conn net.Conn) { func (server *EchoVault) handleConnection(conn net.Conn) {
// If ACL module is loaded, register the connection with the ACL // If ACL module is loaded, register the connection with the ACL
@@ -536,6 +553,13 @@ func (server *EchoVault) rewriteAOF() error {
// ShutDown gracefully shuts down the EchoVault instance. // ShutDown gracefully shuts down the EchoVault instance.
// This function shuts down the memberlist and raft layers. // This function shuts down the memberlist and raft layers.
func (server *EchoVault) ShutDown() { func (server *EchoVault) ShutDown() {
if server.listener != nil {
go func() { server.quit <- struct{}{} }()
log.Println("closing tcp listener...")
if err := server.listener.Close(); err != nil {
log.Printf("listener close: %v\n", err)
}
}
if server.isInCluster() { if server.isInCluster() {
server.raft.RaftShutdown() server.raft.RaftShutdown()
server.memberList.MemberListShutdown() server.memberList.MemberListShutdown()

513
echovault/echovault_test.go
View File

@@ -23,12 +23,14 @@ import (
"github.com/echovault/echovault/internal" "github.com/echovault/echovault/internal"
"github.com/tidwall/resp" "github.com/tidwall/resp"
"io" "io"
"math"
"net" "net"
"os" "os"
"path" "path"
"strings" "strings"
"sync" "sync"
"testing" "testing"
"time"
) )
type ClientServerPair struct { type ClientServerPair struct {
@@ -38,6 +40,7 @@ type ClientServerPair struct {
raftPort int raftPort int
mlPort int mlPort int
bootstrapCluster bool bootstrapCluster bool
raw net.Conn
client *resp.Conn client *resp.Conn
server *EchoVault server *EchoVault
} }
@@ -62,8 +65,6 @@ func getBindAddr() net.IP {
return getBindAddrNet(0) return getBindAddrNet(0)
} }
var setupLock sync.Mutex
func setupServer( func setupServer(
serverId string, serverId string,
bootstrapCluster bool, bootstrapCluster bool,
@@ -73,25 +74,20 @@ func setupServer(
raftPort, raftPort,
mlPort int, mlPort int,
) (*EchoVault, error) { ) (*EchoVault, error) {
setupLock.Lock()
defer setupLock.Unlock()
ctx := context.Background()
config := DefaultConfig() config := DefaultConfig()
config.DataDir = "./testdata" config.DataDir = "./testdata"
config.ForwardCommand = true config.ForwardCommand = true
config.BindAddr = bindAddr config.BindAddr = bindAddr
config.JoinAddr = joinAddr config.JoinAddr = joinAddr
config.Port = uint16(port) config.Port = uint16(port)
// config.InMemory = true config.InMemory = true
config.ServerID = serverId config.ServerID = serverId
config.RaftBindPort = uint16(raftPort) config.RaftBindPort = uint16(raftPort)
config.MemberListBindPort = uint16(mlPort) config.MemberListBindPort = uint16(mlPort)
config.BootstrapCluster = bootstrapCluster config.BootstrapCluster = bootstrapCluster
return NewEchoVault( return NewEchoVault(
WithContext(ctx), WithContext(context.Background()),
WithConfig(config), WithConfig(config),
) )
} }
@@ -163,6 +159,7 @@ func makeCluster(size int) ([]ClientServerPair, error) {
raftPort: raftPort, raftPort: raftPort,
mlPort: memberlistPort, mlPort: memberlistPort,
bootstrapCluster: bootstrapCluster, bootstrapCluster: bootstrapCluster,
raw: conn,
client: client, client: client,
server: server, server: server,
} }
@@ -171,224 +168,268 @@ func makeCluster(size int) ([]ClientServerPair, error) {
return pairs, nil return pairs, nil
} }
// func Test_ClusterReplication(t *testing.T) { func Test_Cluster(t *testing.T) {
// nodes, err := makeCluster(5) nodes, err := makeCluster(5)
// if err != nil { if err != nil {
// t.Error(err) t.Error(err)
// return return
// } }
//
// // Prepare the write data for the cluster.
// tests := []struct {
// key string
// value string
// }{
// {
// key: "key1",
// value: "value1",
// },
// {
// key: "key2",
// value: "value2",
// },
// {
// key: "key3",
// value: "value3",
// },
// }
//
// // Write all the data to the cluster leader
// for i, test := range tests {
// node := nodes[0]
// if err := node.client.WriteArray([]resp.Value{
// resp.StringValue("SET"),
// resp.StringValue(test.key),
// resp.StringValue(test.value),
// }); err != nil {
// t.Errorf("could not write data to leader node (test %d): %v", i, err)
// }
// // Read response and make sure we received "ok" response.
// rd, _, err := node.client.ReadValue()
// if err != nil {
// t.Errorf("could not read response from leader node (test %d): %v", i, err)
// }
// if !strings.EqualFold(rd.String(), "ok") {
// t.Errorf("expected response for test %d to be \"OK\", got %s", i, rd.String())
// }
// }
//
// // Check if the data has been replicated on a quorum (majority of the cluster).
// quorum := int(math.Ceil(float64(len(nodes)/2)) + 1)
// for i, test := range tests {
// count := 0
// for j := 0; j < len(nodes); j++ {
// node := nodes[j]
// if err := node.client.WriteArray([]resp.Value{
// resp.StringValue("GET"),
// resp.StringValue(test.key),
// }); err != nil {
// t.Errorf("could not write data to follower node %d (test %d): %v", j, i, err)
// }
// rd, _, err := node.client.ReadValue()
// if err != nil {
// t.Errorf("could not read data from follower node %d (test %d): %v", j, i, err)
// }
// if rd.String() == test.value {
// count += 1 // If the expected value is found, increment the count.
// }
// }
// // Fail if count is less than quorum.
// if count < quorum {
// t.Errorf("could not find value %s at key %s in cluster quorum", test.value, test.key)
// }
// }
// }
// func Test_ClusterDeleteKey(t *testing.T) { defer func() {
// nodes, err := makeCluster(5) for _, node := range nodes {
// if err != nil { _ = node.raw.Close()
// t.Error(err) node.server.ShutDown()
// return }
// } }()
//
// // Prepare the write data for the cluster
// tests := []struct {
// key string
// value string
// }{
// {
// key: "key1",
// value: "value1",
// },
// {
// key: "key2",
// value: "value2",
// },
// {
// key: "key3",
// value: "value3",
// },
// }
//
// // Write all the data to the cluster leader
// for i, test := range tests {
// node := nodes[0]
// if err := node.client.WriteArray([]resp.Value{
// resp.StringValue("SET"),
// resp.StringValue(test.key),
// resp.StringValue(test.value),
// }); err != nil {
// t.Errorf("could not write command to leader node (test %d): %v", i, err)
// }
// // Read response and make sure we received "ok" response.
// rd, _, err := node.client.ReadValue()
// if err != nil {
// t.Errorf("could not read response from leader node (test %d): %v", i, err)
// }
// if !strings.EqualFold(rd.String(), "ok") {
// t.Errorf("expected response for test %d to be \"OK\", got %s", i, rd.String())
// }
// }
//
// quorum := int(math.Ceil(float64(len(nodes)/2)) + 1)
//
// // Check if the data has been replicated on a quorum (majority of the cluster).
// for i, test := range tests {
// count := 0
// for j := 0; j < len(nodes); j++ {
// node := nodes[j]
// if err := node.client.WriteArray([]resp.Value{
// resp.StringValue("GET"),
// resp.StringValue(test.key),
// }); err != nil {
// t.Errorf("could not write command to follower node %d (test %d): %v", j, i, err)
// }
// rd, _, err := node.client.ReadValue()
// if err != nil {
// t.Errorf("could not read data from follower node %d (test %d): %v", j, i, err)
// }
// if rd.String() == test.value {
// count += 1 // If the expected value is found, increment the count.
// }
// }
// // Fail if count is less than quorum.
// if count < quorum {
// t.Errorf("could not find value %s at key %s in cluster quorum", test.value, test.key)
// return
// }
// }
//
// // Delete the key on the leader node
// for i, test := range tests {
// node := nodes[0]
// if err := node.client.WriteArray([]resp.Value{
// resp.StringValue("DEL"),
// resp.StringValue(test.key),
// }); err != nil {
// t.Errorf("could not write command to leader node (test %d): %v", i, err)
// }
// // Read response and make sure we received "ok" response.
// rd, _, err := node.client.ReadValue()
// if err != nil {
// t.Errorf("could not read response from leader node (test %d): %v", i, err)
// }
// if rd.Integer() != 1 {
// t.Errorf("expected response for test %d to be 1, got %d", i, rd.Integer())
// }
// }
//
// // Check if the data is absent in quorum (majority of the cluster).
// for i, test := range tests {
// count := 0
// for j := 0; j < len(nodes); j++ {
// node := nodes[j]
// if err := node.client.WriteArray([]resp.Value{
// resp.StringValue("GET"),
// resp.StringValue(test.key),
// }); err != nil {
// t.Errorf("could not write command to follower node %d (test %d): %v", j, i, err)
// }
// rd, _, err := node.client.ReadValue()
// if err != nil {
// t.Errorf("could not read data from follower node %d (test %d): %v", j, i, err)
// }
// if rd.IsNull() {
// count += 1 // If the expected value is found, increment the count.
// }
// }
// // Fail if count is less than quorum.
// if count < quorum {
// t.Errorf("could not find value %s at key %s in cluster quorum", test.value, test.key)
// }
// }
// }
// func Test_CommandForwarded(t *testing.T) { // Prepare the write data for the cluster.
// nodes, err := makeCluster(5) tests := map[string][]struct {
// if err != nil { key string
// t.Error(err) value string
// return }{
// } "replication": {
// {key: "key1", value: "value1"},
// // Prepare the write data for the cluster {key: "key2", value: "value2"},
// tests := []struct { {key: "key3", value: "value3"},
// key string },
// value string "deletion": {
// }{ {key: "key4", value: "value4"},
// { {key: "key5", value: "value4"},
// key: "key1", {key: "key6", value: "value5"},
// value: "value1", },
// }, "raft-apply-delete": {
// { {key: "key7", value: "value7"},
// key: "key2", {key: "key8", value: "value8"},
// value: "value2", {key: "key9", value: "value9"},
// }, },
// { "forward": {
// key: "key3", {key: "key10", value: "value10"},
// value: "value3", {key: "key11", value: "value11"},
// }, {key: "key12", value: "value12"},
// } },
// }
t.Run("Test_Replication", func(t *testing.T) {
tests := tests["replication"]
// Write all the data to the cluster leader.
for i, test := range tests {
node := nodes[0]
if err := node.client.WriteArray([]resp.Value{
resp.StringValue("SET"), resp.StringValue(test.key), resp.StringValue(test.value),
}); err != nil {
t.Errorf("could not write data to leader node (test %d): %v", i, err)
}
// Read response and make sure we received "ok" response.
rd, _, err := node.client.ReadValue()
if err != nil {
t.Errorf("could not read response from leader node (test %d): %v", i, err)
}
if !strings.EqualFold(rd.String(), "ok") {
t.Errorf("expected response for test %d to be \"OK\", got %s", i, rd.String())
}
}
<-time.After(200 * time.Millisecond) // Yield
// Check if the data has been replicated on a quorum (majority of the cluster).
quorum := int(math.Ceil(float64(len(nodes)/2)) + 1)
for i, test := range tests {
count := 0
for j := 0; j < len(nodes); j++ {
node := nodes[j]
if err := node.client.WriteArray([]resp.Value{
resp.StringValue("GET"),
resp.StringValue(test.key),
}); err != nil {
t.Errorf("could not write data to follower node %d (test %d): %v", j, i, err)
}
rd, _, err := node.client.ReadValue()
if err != nil {
t.Errorf("could not read data from follower node %d (test %d): %v", j, i, err)
}
if rd.String() == test.value {
count += 1 // If the expected value is found, increment the count.
}
}
// Fail if count is less than quorum.
if count < quorum {
t.Errorf("could not find value %s at key %s in cluster quorum", test.value, test.key)
}
}
})
t.Run("Test_DeleteKey", func(t *testing.T) {
tests := tests["deletion"]
// Write all the data to the cluster leader.
for i, test := range tests {
node := nodes[0]
_, ok, err := node.server.Set(test.key, test.value, SetOptions{})
if err != nil {
t.Errorf("could not write command to leader node (test %d): %v", i, err)
}
if !ok {
t.Errorf("expected set for test %d ok = true, got ok = false", i)
}
}
<-time.After(200 * time.Millisecond) // Yield
// Check if the data has been replicated on a quorum (majority of the cluster).
quorum := int(math.Ceil(float64(len(nodes)/2)) + 1)
for i, test := range tests {
count := 0
for j := 0; j < len(nodes); j++ {
node := nodes[j]
if err := node.client.WriteArray([]resp.Value{
resp.StringValue("GET"),
resp.StringValue(test.key),
}); err != nil {
t.Errorf("could not write command to follower node %d (test %d): %v", j, i, err)
}
rd, _, err := node.client.ReadValue()
if err != nil {
t.Errorf("could not read data from follower node %d (test %d): %v", j, i, err)
}
if rd.String() == test.value {
count += 1 // If the expected value is found, increment the count.
}
}
// Fail if count is less than quorum.
if count < quorum {
t.Errorf("could not find value %s at key %s in cluster quorum", test.value, test.key)
return
}
}
// Delete the key on the leader node
// 1. Prepare delete command.
command := []resp.Value{resp.StringValue("DEL")}
for _, test := range tests {
command = append(command, resp.StringValue(test.key))
}
// 2. Send delete command.
if err := nodes[0].client.WriteArray(command); err != nil {
t.Error(err)
return
}
res, _, err := nodes[0].client.ReadValue()
if err != nil {
t.Error(err)
return
}
// 3. Check the delete count is equal to length of tests.
if res.Integer() != len(tests) {
t.Errorf("expected delete response to be %d, got %d", len(tests), res.Integer())
}
<-time.After(200 * time.Millisecond) // Yield
// Check if the data is absent in quorum (majority of the cluster).
for i, test := range tests {
count := 0
for j := 0; j < len(nodes); j++ {
node := nodes[j]
if err := node.client.WriteArray([]resp.Value{
resp.StringValue("GET"),
resp.StringValue(test.key),
}); err != nil {
t.Errorf("could not write command to follower node %d (test %d): %v", j, i, err)
}
rd, _, err := node.client.ReadValue()
if err != nil {
t.Errorf("could not read data from follower node %d (test %d): %v", j, i, err)
}
if rd.IsNull() {
count += 1 // If the expected value is found, increment the count.
}
}
// Fail if count is less than quorum.
if count < quorum {
t.Errorf("could not find value %s at key %s in cluster quorum", test.value, test.key)
}
}
})
t.Run("Test_raftApplyDeleteKey", func(t *testing.T) {
tests := tests["raft-apply-delete"]
// Write all the data to the cluster leader.
for i, test := range tests {
node := nodes[0]
_, ok, err := node.server.Set(test.key, test.value, SetOptions{})
if err != nil {
t.Errorf("could not write command to leader node (test %d): %v", i, err)
}
if !ok {
t.Errorf("expected set for test %d ok = true, got ok = false", i)
}
}
<-time.After(200 * time.Millisecond) // Yield
// Check if the data has been replicated on a quorum (majority of the cluster).
quorum := int(math.Ceil(float64(len(nodes)/2)) + 1)
for i, test := range tests {
count := 0
for j := 0; j < len(nodes); j++ {
node := nodes[j]
if err := node.client.WriteArray([]resp.Value{
resp.StringValue("GET"),
resp.StringValue(test.key),
}); err != nil {
t.Errorf("could not write command to follower node %d (test %d): %v", j, i, err)
}
rd, _, err := node.client.ReadValue()
if err != nil {
t.Errorf("could not read data from follower node %d (test %d): %v", j, i, err)
}
if rd.String() == test.value {
count += 1 // If the expected value is found, increment the count.
}
}
// Fail if count is less than quorum.
if count < quorum {
t.Errorf("could not find value %s at key %s in cluster quorum", test.value, test.key)
return
}
}
// Delete the keys using raftApplyDelete method.
for _, test := range tests {
if err := nodes[0].server.raftApplyDeleteKey(nodes[0].server.context, test.key); err != nil {
t.Error(err)
}
}
<-time.After(200 * time.Millisecond) // Yield to give key deletion time to take effect across cluster.
// Check if the data is absent in quorum (majority of the cluster).
for i, test := range tests {
count := 0
for j := 0; j < len(nodes); j++ {
node := nodes[j]
if err := node.client.WriteArray([]resp.Value{
resp.StringValue("GET"),
resp.StringValue(test.key),
}); err != nil {
t.Errorf("could not write command to follower node %d (test %d): %v", j, i, err)
}
rd, _, err := node.client.ReadValue()
if err != nil {
t.Errorf("could not read data from follower node %d (test %d): %v", j, i, err)
}
if rd.IsNull() {
count += 1 // If the expected value is found, increment the count.
}
}
// Fail if count is less than quorum.
if count < quorum {
t.Errorf("found value %s at key %s in cluster quorum", test.value, test.key)
}
}
})
// t.Run("Test_ForwardCommand", func(t *testing.T) {
// tests := tests["forward"]
// // Write all the data a random cluster follower. // // Write all the data a random cluster follower.
// for i, test := range tests { // for i, test := range tests {
// // Send write command to follower node. // // Send write command to follower node.
@@ -398,19 +439,19 @@ func makeCluster(size int) ([]ClientServerPair, error) {
// resp.StringValue(test.key), // resp.StringValue(test.key),
// resp.StringValue(test.value), // resp.StringValue(test.value),
// }); err != nil { // }); err != nil {
// t.Errorf("could not write data to leader node (test %d): %v", i, err) // t.Errorf("could not write data to follower node (test %d): %v", i, err)
// } // }
// // Read response and make sure we received "ok" response. // // Read response and make sure we received "ok" response.
// rd, _, err := node.client.ReadValue() // rd, _, err := node.client.ReadValue()
// if err != nil { // if err != nil {
// t.Errorf("could not read response from leader node (test %d): %v", i, err) // t.Errorf("could not read response from follower node (test %d): %v", i, err)
// } // }
// if !strings.EqualFold(rd.String(), "ok") { // if !strings.EqualFold(rd.String(), "ok") {
// t.Errorf("expected response for test %d to be \"OK\", got %s", i, rd.String()) // t.Errorf("expected response for test %d to be \"OK\", got %s", i, rd.String())
// } // }
// } // }
// //
// <-time.After(250 * time.Millisecond) // Short yield to allow change to take effect. // <-time.After(200 * time.Millisecond) // Short yield to allow change to take effect.
// //
// // Check if the data has been replicated on a quorum (majority of the cluster). // // Check if the data has been replicated on a quorum (majority of the cluster).
// quorum := int(math.Ceil(float64(len(nodes)/2)) + 1) // quorum := int(math.Ceil(float64(len(nodes)/2)) + 1)
@@ -437,7 +478,8 @@ func makeCluster(size int) ([]ClientServerPair, error) {
// t.Errorf("could not find value %s at key %s in cluster quorum", test.value, test.key) // t.Errorf("could not find value %s at key %s in cluster quorum", test.value, test.key)
// } // }
// } // }
// } // })
}
func Test_TLS(t *testing.T) { func Test_TLS(t *testing.T) {
port, err := internal.GetFreePort() port, err := internal.GetFreePort()
@@ -464,6 +506,7 @@ func Test_TLS(t *testing.T) {
server, err := NewEchoVault(WithConfig(conf)) server, err := NewEchoVault(WithConfig(conf))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return
} }
wg := sync.WaitGroup{} wg := sync.WaitGroup{}
@@ -494,8 +537,12 @@ func Test_TLS(t *testing.T) {
}) })
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return
} }
defer func() {
_ = conn.Close()
server.ShutDown()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
// Test that we can set and get a value from the server. // Test that we can set and get a value from the server.
@@ -561,6 +608,7 @@ func Test_MTLS(t *testing.T) {
server, err := NewEchoVault(WithConfig(conf)) server, err := NewEchoVault(WithConfig(conf))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return
} }
wg := sync.WaitGroup{} wg := sync.WaitGroup{}
@@ -613,7 +661,10 @@ func Test_MTLS(t *testing.T) {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
server.ShutDown()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
// Test that we can set and get a value from the server. // Test that we can set and get a value from the server.

8
internal/memberlist/memberlist.go
View File

@@ -160,13 +160,15 @@ func (m *MemberList) MemberListShutdown() {
// Gracefully leave memberlist cluster // Gracefully leave memberlist cluster
err := m.memberList.Leave(500 * time.Millisecond) err := m.memberList.Leave(500 * time.Millisecond)
if err != nil { if err != nil {
log.Fatal("Could not gracefully leave memberlist cluster") log.Printf("memberlist leave: %v\n", err)
return
} }
err = m.memberList.Shutdown() err = m.memberList.Shutdown()
if err != nil { if err != nil {
log.Fatal("Could not gracefully shutdown memberlist background maintenance") log.Printf("memberlist shutdown: %v\n", err)
return
} }
log.Println("Successfully shutdown memberlist") log.Println("successfully shutdown memberlist")
} }

133
internal/modules/admin/commands_test.go
View File

@@ -32,7 +32,6 @@ import (
str "github.com/echovault/echovault/internal/modules/string" str "github.com/echovault/echovault/internal/modules/string"
"github.com/tidwall/resp" "github.com/tidwall/resp"
"net" "net"
"os"
"path" "path"
"slices" "slices"
"strings" "strings"
@@ -50,13 +49,6 @@ func setupServer(port uint16) (*echovault.EchoVault, error) {
} }
func Test_AdminCommands(t *testing.T) { func Test_AdminCommands(t *testing.T) {
t.Cleanup(func() {
_ = os.RemoveAll("./testdata")
})
t.Run("Test COMMANDS command", func(t *testing.T) {
t.Parallel()
port, err := internal.GetFreePort() port, err := internal.GetFreePort()
if err != nil { if err != nil {
t.Error(err) t.Error(err)
@@ -77,11 +69,21 @@ func Test_AdminCommands(t *testing.T) {
}() }()
wg.Wait() wg.Wait()
t.Cleanup(func() {
mockServer.ShutDown()
})
t.Run("Test COMMANDS command", func(t *testing.T) {
t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port)) conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
if err = client.WriteArray([]resp.Value{resp.StringValue("COMMANDS")}); err != nil { if err = client.WriteArray([]resp.Value{resp.StringValue("COMMANDS")}); err != nil {
@@ -128,31 +130,14 @@ func Test_AdminCommands(t *testing.T) {
t.Run("Test COMMAND COUNT command", func(t *testing.T) { t.Run("Test COMMAND COUNT command", func(t *testing.T) {
t.Parallel() t.Parallel()
port, err := internal.GetFreePort()
if err != nil {
t.Error(err)
return
}
mockServer, err := setupServer(uint16(port))
if err != nil {
t.Error(err)
return
}
wg := sync.WaitGroup{}
wg.Add(1)
go func() {
wg.Done()
mockServer.Start()
}()
wg.Wait()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port)) conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
if err = client.WriteArray([]resp.Value{resp.StringValue("COMMAND"), resp.StringValue("COUNT")}); err != nil { if err = client.WriteArray([]resp.Value{resp.StringValue("COMMAND"), resp.StringValue("COUNT")}); err != nil {
@@ -199,31 +184,14 @@ func Test_AdminCommands(t *testing.T) {
t.Run("Test COMMAND LIST command", func(t *testing.T) { t.Run("Test COMMAND LIST command", func(t *testing.T) {
t.Parallel() t.Parallel()
port, err := internal.GetFreePort()
if err != nil {
t.Error(err)
return
}
mockServer, err := setupServer(uint16(port))
if err != nil {
t.Error(err)
return
}
wg := sync.WaitGroup{}
wg.Add(1)
go func() {
wg.Done()
mockServer.Start()
}()
wg.Wait()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port)) conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
// Get all the commands from the existing modules. // Get all the commands from the existing modules.
@@ -336,17 +304,6 @@ func Test_AdminCommands(t *testing.T) {
}) })
t.Run("Test MODULE LOAD command", func(t *testing.T) { t.Run("Test MODULE LOAD command", func(t *testing.T) {
port, err := internal.GetFreePort()
if err != nil {
t.Error(err)
return
}
mockServer, err := setupServer(uint16(port))
if err != nil {
t.Error(err)
return
}
tests := []struct { tests := []struct {
name string name string
execCommand []resp.Value execCommand []resp.Value
@@ -433,20 +390,14 @@ func Test_AdminCommands(t *testing.T) {
}, },
} }
wg := sync.WaitGroup{}
wg.Add(1)
go func() {
wg.Done()
mockServer.Start()
}()
wg.Wait()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port)) conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
respConn := resp.NewConn(conn) respConn := resp.NewConn(conn)
for i := 0; i < len(tests); i++ { for i := 0; i < len(tests); i++ {
@@ -505,31 +456,14 @@ func Test_AdminCommands(t *testing.T) {
}) })
t.Run("Test MODULE UNLOAD command", func(t *testing.T) { t.Run("Test MODULE UNLOAD command", func(t *testing.T) {
port, err := internal.GetFreePort()
if err != nil {
t.Error(err)
return
}
mockServer, err := setupServer(uint16(port))
if err != nil {
t.Error(err)
return
}
wg := sync.WaitGroup{}
wg.Add(1)
go func() {
wg.Done()
mockServer.Start()
}()
wg.Wait()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port)) conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
respConn := resp.NewConn(conn) respConn := resp.NewConn(conn)
// Load module.set module // Load module.set module
@@ -693,31 +627,14 @@ func Test_AdminCommands(t *testing.T) {
}) })
t.Run("Test MODULE LIST command", func(t *testing.T) { t.Run("Test MODULE LIST command", func(t *testing.T) {
port, err := internal.GetFreePort()
if err != nil {
t.Error(err)
return
}
mockServer, err := setupServer(uint16(port))
if err != nil {
t.Error(err)
return
}
wg := sync.WaitGroup{}
wg.Add(1)
go func() {
wg.Done()
mockServer.Start()
}()
wg.Wait()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port)) conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
respConn := resp.NewConn(conn) respConn := resp.NewConn(conn)
// Load module.get module with arg // Load module.get module with arg

34
internal/modules/connection/commands_test.go
View File

@@ -28,20 +28,26 @@ import (
"testing" "testing"
) )
var mockServer *echovault.EchoVault func Test_Connection(t *testing.T) {
var port int port, err := internal.GetFreePort()
var addr = "localhost" if err != nil {
t.Error(err)
return
}
func init() { mockServer, err := echovault.NewEchoVault(
port, _ = internal.GetFreePort()
mockServer, _ = echovault.NewEchoVault(
echovault.WithConfig(config.Config{ echovault.WithConfig(config.Config{
DataDir: "", DataDir: "",
EvictionPolicy: constants.NoEviction, EvictionPolicy: constants.NoEviction,
BindAddr: addr, BindAddr: "localhost",
Port: uint16(port), Port: uint16(port),
}), }),
) )
if err != nil {
t.Error(err)
return
}
wg := sync.WaitGroup{} wg := sync.WaitGroup{}
wg.Add(1) wg.Add(1)
go func() { go func() {
@@ -49,14 +55,20 @@ func init() {
mockServer.Start() mockServer.Start()
}() }()
wg.Wait() wg.Wait()
}
func Test_HandlePing(t *testing.T) { t.Cleanup(func() {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) mockServer.ShutDown()
})
t.Run("Test_HandlePing", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -107,4 +119,6 @@ func Test_HandlePing(t *testing.T) {
t.Errorf("expected response \"%s\", got \"%s\"", test.expected, res.String()) t.Errorf("expected response \"%s\", got \"%s\"", test.expected, res.String())
} }
} }
})
} }

130
internal/modules/generic/commands_test.go
View File

@@ -30,27 +30,32 @@ import (
"time" "time"
) )
var addr string
var port int
var mockServer *echovault.EchoVault
var mockClock clock.Clock
type KeyData struct { type KeyData struct {
Value interface{} Value interface{}
ExpireAt time.Time ExpireAt time.Time
} }
func init() { func Test_Generic(t *testing.T) {
mockClock = clock.NewClock() mockClock := clock.NewClock()
port, _ = internal.GetFreePort() port, err := internal.GetFreePort()
mockServer, _ = echovault.NewEchoVault( if err != nil {
t.Error(err)
return
}
mockServer, err := echovault.NewEchoVault(
echovault.WithConfig(config.Config{ echovault.WithConfig(config.Config{
BindAddr: addr, BindAddr: "localhost",
Port: uint16(port), Port: uint16(port),
DataDir: "", DataDir: "",
EvictionPolicy: constants.NoEviction, EvictionPolicy: constants.NoEviction,
}), }),
) )
if err != nil {
t.Error(err)
return
}
wg := sync.WaitGroup{} wg := sync.WaitGroup{}
wg.Add(1) wg.Add(1)
go func() { go func() {
@@ -58,14 +63,21 @@ func init() {
mockServer.Start() mockServer.Start()
}() }()
wg.Wait() wg.Wait()
}
func Test_HandleSET(t *testing.T) { t.Cleanup(func() {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) mockServer.ShutDown()
})
t.Run("Test_HandleSET", func(t *testing.T) {
t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -459,14 +471,18 @@ func Test_HandleSET(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleMSET(t *testing.T) { t.Run("Test_HandleMSET", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -547,14 +563,18 @@ func Test_HandleMSET(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleGET(t *testing.T) { t.Run("Test_HandleGET", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -662,14 +682,18 @@ func Test_HandleGET(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleMGET(t *testing.T) { t.Run("Test_HandleMGET", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -765,14 +789,18 @@ func Test_HandleMGET(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleDEL(t *testing.T) { t.Run("Test_HandleDEL", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -873,14 +901,18 @@ func Test_HandleDEL(t *testing.T) {
} }
}) })
} }
} })
func Test_HandlePERSIST(t *testing.T) { t.Run("Test_HandlePERSIST", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1026,14 +1058,18 @@ func Test_HandlePERSIST(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleEXPIRETIME(t *testing.T) { t.Run("Test_HandleEXPIRETIME", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1143,14 +1179,18 @@ func Test_HandleEXPIRETIME(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleTTL(t *testing.T) { t.Run("Test_HandleTTL", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1260,14 +1300,18 @@ func Test_HandleTTL(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleEXPIRE(t *testing.T) { t.Run("Test_HandleEXPIRE", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1543,14 +1587,18 @@ func Test_HandleEXPIRE(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleEXPIREAT(t *testing.T) { t.Run("Test_HandleEXPIREAT", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1850,4 +1898,6 @@ func Test_HandleEXPIREAT(t *testing.T) {
} }
}) })
} }
})
} }

134
internal/modules/hash/commands_test.go
View File

@@ -30,20 +30,26 @@ import (
"testing" "testing"
) )
var mockServer *echovault.EchoVault func Test_Hash(t *testing.T) {
var addr = "localhost" port, err := internal.GetFreePort()
var port int if err != nil {
t.Error(err)
return
}
func init() { mockServer, err := echovault.NewEchoVault(
port, _ = internal.GetFreePort()
mockServer, _ = echovault.NewEchoVault(
echovault.WithConfig(config.Config{ echovault.WithConfig(config.Config{
BindAddr: addr, BindAddr: "localhost",
Port: uint16(port), Port: uint16(port),
DataDir: "", DataDir: "",
EvictionPolicy: constants.NoEviction, EvictionPolicy: constants.NoEviction,
}), }),
) )
if err != nil {
t.Error(err)
return
}
wg := sync.WaitGroup{} wg := sync.WaitGroup{}
wg.Add(1) wg.Add(1)
go func() { go func() {
@@ -51,14 +57,21 @@ func init() {
mockServer.Start() mockServer.Start()
}() }()
wg.Wait() wg.Wait()
}
func Test_HandleHSET(t *testing.T) { t.Cleanup(func() {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) mockServer.ShutDown()
})
t.Run("Test_HandleHSET", func(t *testing.T) {
t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
// Tests for both HSet and HSetNX // Tests for both HSet and HSetNX
@@ -231,14 +244,18 @@ func Test_HandleHSET(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleHINCRBY(t *testing.T) { t.Run("Test_HandleHINCRBY", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
// Tests for both HIncrBy and HIncrByFloat // Tests for both HIncrBy and HIncrByFloat
@@ -429,14 +446,18 @@ func Test_HandleHINCRBY(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleHGET(t *testing.T) { t.Run("Test_HandleHGET", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -581,14 +602,18 @@ func Test_HandleHGET(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleHSTRLEN(t *testing.T) { t.Run("Test_HandleHSTRLEN", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -735,14 +760,18 @@ func Test_HandleHSTRLEN(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleHVALS(t *testing.T) { t.Run("Test_HandleHVALS", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -873,14 +902,18 @@ func Test_HandleHVALS(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleHRANDFIELD(t *testing.T) { t.Run("Test_HandleHRANDFIELD", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1078,14 +1111,18 @@ func Test_HandleHRANDFIELD(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleHLEN(t *testing.T) { t.Run("Test_HandleHLEN", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1200,14 +1237,18 @@ func Test_HandleHLEN(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleHKeys(t *testing.T) { t.Run("Test_HandleHKeys", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1324,14 +1365,18 @@ func Test_HandleHKeys(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleHGETALL(t *testing.T) { t.Run("Test_HandleHGETALL", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1461,14 +1506,18 @@ func Test_HandleHGETALL(t *testing.T) {
}) })
} }
} })
func Test_HandleHEXISTS(t *testing.T) { t.Run("Test_HandleHEXISTS", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1584,14 +1633,18 @@ func Test_HandleHEXISTS(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleHDEL(t *testing.T) { t.Run("Test_HandleHDEL", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1724,4 +1777,5 @@ func Test_HandleHDEL(t *testing.T) {
} }
}) })
} }
})
} }

124
internal/modules/list/commands_test.go
View File

@@ -30,20 +30,26 @@ import (
"testing" "testing"
) )
var mockServer *echovault.EchoVault func Test_List(t *testing.T) {
var addr = "localhost" port, err := internal.GetFreePort()
var port int if err != nil {
t.Error(err)
return
}
func init() { mockServer, err := echovault.NewEchoVault(
port, _ = internal.GetFreePort()
mockServer, _ = echovault.NewEchoVault(
echovault.WithConfig(config.Config{ echovault.WithConfig(config.Config{
BindAddr: addr, BindAddr: "localhost",
Port: uint16(port), Port: uint16(port),
DataDir: "", DataDir: "",
EvictionPolicy: constants.NoEviction, EvictionPolicy: constants.NoEviction,
}), }),
) )
if err != nil {
t.Error(err)
return
}
wg := sync.WaitGroup{} wg := sync.WaitGroup{}
wg.Add(1) wg.Add(1)
go func() { go func() {
@@ -51,14 +57,21 @@ func init() {
mockServer.Start() mockServer.Start()
}() }()
wg.Wait() wg.Wait()
}
func Test_HandleLLEN(t *testing.T) { t.Cleanup(func() {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) mockServer.ShutDown()
})
t.Run("Test_HandleLLEN", func(t *testing.T) {
t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -171,14 +184,18 @@ func Test_HandleLLEN(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleLINDEX(t *testing.T) { t.Run("Test_HandleLINDEX", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -331,14 +348,18 @@ func Test_HandleLINDEX(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleLRANGE(t *testing.T) { t.Run("Test_HandleLRANGE", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -516,14 +537,18 @@ func Test_HandleLRANGE(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleLSET(t *testing.T) { t.Run("Test_HandleLSET", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -701,14 +726,18 @@ func Test_HandleLSET(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleLTRIM(t *testing.T) { t.Run("Test_HandleLTRIM", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -897,14 +926,18 @@ func Test_HandleLTRIM(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleLREM(t *testing.T) { t.Run("Test_HandleLREM", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1058,14 +1091,18 @@ func Test_HandleLREM(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleLMOVE(t *testing.T) { t.Run("Test_HandleLMOVE", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1285,14 +1322,18 @@ func Test_HandleLMOVE(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleLPUSH(t *testing.T) { t.Run("Test_HandleLPUSH", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1436,14 +1477,18 @@ func Test_HandleLPUSH(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleRPUSH(t *testing.T) { t.Run("Test_HandleRPUSH", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1587,14 +1632,18 @@ func Test_HandleRPUSH(t *testing.T) {
} }
}) })
} }
} })
func Test_HandlePOP(t *testing.T) { t.Run("Test_HandlePOP", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1747,4 +1796,5 @@ func Test_HandlePOP(t *testing.T) {
} }
}) })
} }
})
} }

184
internal/modules/set/commands_test.go
View File

@@ -31,20 +31,26 @@ import (
"testing" "testing"
) )
var mockServer *echovault.EchoVault func Test_Set(t *testing.T) {
var addr = "localhost" port, err := internal.GetFreePort()
var port int if err != nil {
t.Error(err)
return
}
func init() { mockServer, err := echovault.NewEchoVault(
port, _ = internal.GetFreePort()
mockServer, _ = echovault.NewEchoVault(
echovault.WithConfig(config.Config{ echovault.WithConfig(config.Config{
BindAddr: addr, BindAddr: "localhost",
Port: uint16(port), Port: uint16(port),
DataDir: "", DataDir: "",
EvictionPolicy: constants.NoEviction, EvictionPolicy: constants.NoEviction,
}), }),
) )
if err != nil {
t.Error(err)
return
}
wg := sync.WaitGroup{} wg := sync.WaitGroup{}
wg.Add(1) wg.Add(1)
go func() { go func() {
@@ -52,14 +58,21 @@ func init() {
mockServer.Start() mockServer.Start()
}() }()
wg.Wait() wg.Wait()
}
func Test_HandleSADD(t *testing.T) { t.Cleanup(func() {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) mockServer.ShutDown()
})
t.Run("Test_HandleSADD", func(t *testing.T) {
t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -196,14 +209,18 @@ func Test_HandleSADD(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleSCARD(t *testing.T) { t.Run("Test_HandleSCARD", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -319,14 +336,18 @@ func Test_HandleSCARD(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleSDIFF(t *testing.T) { t.Run("Test_HandleSDIFF", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -465,14 +486,18 @@ func Test_HandleSDIFF(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleSDIFFSTORE(t *testing.T) { t.Run("Test_HandleSDIFFSTORE", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -643,14 +668,18 @@ func Test_HandleSDIFFSTORE(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleSINTER(t *testing.T) { t.Run("Test_HandleSINTER", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -789,14 +818,18 @@ func Test_HandleSINTER(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleSINTERCARD(t *testing.T) { t.Run("Test_HandleSINTERCARD", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -940,14 +973,18 @@ func Test_HandleSINTERCARD(t *testing.T) {
}) })
} }
} })
func Test_HandleSINTERSTORE(t *testing.T) { t.Run("Test_HandleSINTERSTORE", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1118,14 +1155,18 @@ func Test_HandleSINTERSTORE(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleSISMEMBER(t *testing.T) { t.Run("Test_HandleSISMEMBER", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1236,14 +1277,18 @@ func Test_HandleSISMEMBER(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleSMEMBERS(t *testing.T) { t.Run("Test_HandleSMEMBERS", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1359,14 +1404,18 @@ func Test_HandleSMEMBERS(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleSMISMEMBER(t *testing.T) { t.Run("Test_HandleSMISMEMBER", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1482,14 +1531,18 @@ func Test_HandleSMISMEMBER(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleSMOVE(t *testing.T) { t.Run("Test_HandleSMOVE", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1674,14 +1727,18 @@ func Test_HandleSMOVE(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleSPOP(t *testing.T) { t.Run("Test_HandleSPOP", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1812,14 +1869,18 @@ func Test_HandleSPOP(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleSRANDMEMBER(t *testing.T) { t.Run("Test_HandleSRANDMEMBER", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1973,14 +2034,18 @@ func Test_HandleSRANDMEMBER(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleSREM(t *testing.T) { t.Run("Test_HandleSREM", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -2110,14 +2175,18 @@ func Test_HandleSREM(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleSUNION(t *testing.T) { t.Run("Test_HandleSUNION", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -2249,14 +2318,18 @@ func Test_HandleSUNION(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleSUNIONSTORE(t *testing.T) { t.Run("Test_HandleSUNIONSTORE", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -2405,4 +2478,5 @@ func Test_HandleSUNIONSTORE(t *testing.T) {
} }
}) })
} }
})
} }

254
internal/modules/sorted_set/commands_test.go
View File

@@ -32,20 +32,26 @@ import (
"testing" "testing"
) )
var mockServer *echovault.EchoVault func Test_SortedSet(t *testing.T) {
var addr = "localhost" port, err := internal.GetFreePort()
var port int if err != nil {
t.Error(err)
return
}
func init() { mockServer, err := echovault.NewEchoVault(
port, _ = internal.GetFreePort()
mockServer, _ = echovault.NewEchoVault(
echovault.WithConfig(config.Config{ echovault.WithConfig(config.Config{
BindAddr: addr, BindAddr: "localhost",
Port: uint16(port), Port: uint16(port),
DataDir: "", DataDir: "",
EvictionPolicy: constants.NoEviction, EvictionPolicy: constants.NoEviction,
}), }),
) )
if err != nil {
t.Error(err)
return
}
wg := sync.WaitGroup{} wg := sync.WaitGroup{}
wg.Add(1) wg.Add(1)
go func() { go func() {
@@ -53,14 +59,21 @@ func init() {
mockServer.Start() mockServer.Start()
}() }()
wg.Wait() wg.Wait()
}
func Test_HandleZADD(t *testing.T) { t.Cleanup(func() {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) mockServer.ShutDown()
})
t.Run("Test_HandleZADD", func(t *testing.T) {
t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -265,14 +278,18 @@ func Test_HandleZADD(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleZCARD(t *testing.T) { t.Run("Test_HandleZCARD", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -392,14 +409,18 @@ func Test_HandleZCARD(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleZCOUNT(t *testing.T) { t.Run("Test_HandleZCOUNT", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -563,14 +584,18 @@ func Test_HandleZCOUNT(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleZLEXCOUNT(t *testing.T) { t.Run("Test_HandleZLEXCOUNT", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -710,14 +735,18 @@ func Test_HandleZLEXCOUNT(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleZDIFF(t *testing.T) { t.Run("Test_HandleZDIFF", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -926,14 +955,18 @@ func Test_HandleZDIFF(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleZDIFFSTORE(t *testing.T) { t.Run("Test_HandleZDIFFSTORE", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1165,14 +1198,18 @@ func Test_HandleZDIFFSTORE(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleZINCRBY(t *testing.T) { t.Run("Test_HandleZINCRBY", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1444,14 +1481,18 @@ func Test_HandleZINCRBY(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleZMPOP(t *testing.T) { t.Run("Test_HandleZMPOP", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -1755,14 +1796,18 @@ func Test_HandleZMPOP(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleZPOP(t *testing.T) { t.Run("Test_HandleZPOP", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -2012,14 +2057,18 @@ func Test_HandleZPOP(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleZMSCORE(t *testing.T) { t.Run("Test_HandleZMSCORE", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -2136,14 +2185,18 @@ func Test_HandleZMSCORE(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleZSCORE(t *testing.T) { t.Run("Test_HandleZSCORE", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -2269,14 +2322,18 @@ func Test_HandleZSCORE(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleZRANDMEMBER(t *testing.T) { t.Run("Test_HandleZRANDMEMBER", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -2462,14 +2519,18 @@ func Test_HandleZRANDMEMBER(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleZRANK(t *testing.T) { t.Run("Test_HandleZRANK", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -2615,14 +2676,18 @@ func Test_HandleZRANK(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleZREM(t *testing.T) { t.Run("Test_HandleZREM", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -2784,14 +2849,18 @@ func Test_HandleZREM(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleZREMRANGEBYSCORE(t *testing.T) { t.Run("Test_HandleZREMRANGEBYSCORE", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -2956,14 +3025,18 @@ func Test_HandleZREMRANGEBYSCORE(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleZREMRANGEBYRANK(t *testing.T) { t.Run("Test_HandleZREMRANGEBYRANK", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -3182,14 +3255,18 @@ func Test_HandleZREMRANGEBYRANK(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleZREMRANGEBYLEX(t *testing.T) { t.Run("Test_HandleZREMRANGEBYLEX", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -3379,14 +3456,18 @@ func Test_HandleZREMRANGEBYLEX(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleZRANGE(t *testing.T) { t.Run("Test_HandleZRANGE", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -3676,14 +3757,18 @@ func Test_HandleZRANGE(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleZRANGESTORE(t *testing.T) { t.Run("Test_HandleZRANGESTORE", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -4023,14 +4108,18 @@ func Test_HandleZRANGESTORE(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleZINTER(t *testing.T) { t.Run("Test_HandleZINTER", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -4387,14 +4476,18 @@ func Test_HandleZINTER(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleZINTERSTORE(t *testing.T) { t.Run("Test_HandleZINTERSTORE", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -4798,14 +4891,18 @@ func Test_HandleZINTERSTORE(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleZUNION(t *testing.T) { t.Run("Test_HandleZUNION", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -5187,14 +5284,18 @@ func Test_HandleZUNION(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleZUNIONSTORE(t *testing.T) { t.Run("Test_HandleZUNIONSTORE", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error() t.Error()
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -5649,4 +5750,5 @@ func Test_HandleZUNIONSTORE(t *testing.T) {
} }
}) })
} }
})
} }

56
internal/modules/string/commands_test.go
View File

@@ -29,20 +29,26 @@ import (
"testing" "testing"
) )
var mockServer *echovault.EchoVault func Test_String(t *testing.T) {
var addr = "localhost" port, err := internal.GetFreePort()
var port int if err != nil {
t.Error()
return
}
func init() { mockServer, err := echovault.NewEchoVault(
port, _ = internal.GetFreePort()
mockServer, _ = echovault.NewEchoVault(
echovault.WithConfig(config.Config{ echovault.WithConfig(config.Config{
BindAddr: addr, BindAddr: "localhost",
Port: uint16(port), Port: uint16(port),
DataDir: "", DataDir: "",
EvictionPolicy: constants.NoEviction, EvictionPolicy: constants.NoEviction,
}), }),
) )
if err != nil {
t.Error(err)
return
}
wg := sync.WaitGroup{} wg := sync.WaitGroup{}
wg.Add(1) wg.Add(1)
go func() { go func() {
@@ -50,14 +56,21 @@ func init() {
mockServer.Start() mockServer.Start()
}() }()
wg.Wait() wg.Wait()
}
func Test_HandleSetRange(t *testing.T) { t.Cleanup(func() {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) mockServer.ShutDown()
})
t.Run("Test_HandleSetRange", func(t *testing.T) {
t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -196,13 +209,18 @@ func Test_HandleSetRange(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleStrLen(t *testing.T) { t.Run("Test_HandleStrLen", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -292,13 +310,18 @@ func Test_HandleStrLen(t *testing.T) {
} }
}) })
} }
} })
func Test_HandleSubStr(t *testing.T) { t.Run("Test_HandleSubStr", func(t *testing.T) {
conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", addr, port)) t.Parallel()
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil { if err != nil {
t.Error(err) t.Error(err)
return
} }
defer func() {
_ = conn.Close()
}()
client := resp.NewConn(conn) client := resp.NewConn(conn)
tests := []struct { tests := []struct {
@@ -433,4 +456,5 @@ func Test_HandleSubStr(t *testing.T) {
} }
}) })
} }
})
} }

7
internal/raft/raft.go
View File

@@ -216,12 +216,13 @@ func (r *Raft) TakeSnapshot() error {
} }
func (r *Raft) RaftShutdown() { func (r *Raft) RaftShutdown() {
// Leadership transfer if current node is the leader // Leadership transfer if current node is the leader.
if r.IsRaftLeader() { if r.IsRaftLeader() {
err := r.raft.LeadershipTransfer().Error() err := r.raft.LeadershipTransfer().Error()
if err != nil { if err != nil {
log.Fatal(err) log.Printf("raft shutdown: %v\n", err)
return
} }
log.Println("Leadership transfer successful.") log.Println("leadership transfer successful.")
} }
} }