12350021459

Committed 16 Dec 2024 09:06AM UTC coverage: 73.845%. First build

Build # 12350021459

Build Type

push

github

Committed by

oleg-jukovec

Commit Message

Release v2.2.0

Run Details

6042 of 8182 relevant lines covered (73.85%)

4998.76 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/test_helpers/pool_helper.go

package test_helpers

import (
        "context"
        "fmt"
        "reflect"
        "time"

        "github.com/tarantool/go-tarantool/v2"
        "github.com/tarantool/go-tarantool/v2/pool"
)

type ListenOnInstanceArgs struct {
        ConnPool      *pool.ConnectionPool
        Mode          pool.Mode
        ServersNumber int
        ExpectedPorts map[string]bool
}

type CheckStatusesArgs struct {
        ConnPool           *pool.ConnectionPool
        Servers            []string
        Mode               pool.Mode
        ExpectedPoolStatus bool
        ExpectedStatuses   map[string]bool
}

func compareTuples(expectedTpl []interface{}, actualTpl []interface{}) error {
        if len(actualTpl) != len(expectedTpl) {
                return fmt.Errorf("unexpected body of Insert (tuple len)")
        }

        for i, field := range actualTpl {
                if field != expectedTpl[i] {
                        return fmt.Errorf("unexpected field, expected: %v actual: %v", expectedTpl[i], field)
                }
        }

        return nil
}

func CheckPoolStatuses(args interface{}) error {
        checkArgs, ok := args.(CheckStatusesArgs)
        if !ok {
                return fmt.Errorf("incorrect args")
        }

        connected, _ := checkArgs.ConnPool.ConnectedNow(checkArgs.Mode)
        if connected != checkArgs.ExpectedPoolStatus {
                return fmt.Errorf(
                        "incorrect connection pool status: expected status %t actual status %t",
                        checkArgs.ExpectedPoolStatus, connected)
        }

        poolInfo := checkArgs.ConnPool.GetInfo()
        for _, server := range checkArgs.Servers {
                status := poolInfo[server].ConnectedNow
                if checkArgs.ExpectedStatuses[server] != status {
                        return fmt.Errorf(
                                "incorrect conn status: addr %s expected status %t actual status %t",
                                server, checkArgs.ExpectedStatuses[server], status)
                }
        }

        return nil
}

// ProcessListenOnInstance helper calls "return box.cfg.listen"
// as many times as there are servers in the connection pool
// with specified mode.
// For RO mode expected received ports equals to replica ports.
// For RW mode expected received ports equals to master ports.
// For PreferRO mode expected received ports equals to replica
// ports or to all ports.
// For PreferRW mode expected received ports equals to master ports
// or to all ports.
func ProcessListenOnInstance(args interface{}) error {
        actualPorts := map[string]bool{}

        listenArgs, ok := args.(ListenOnInstanceArgs)
        if !ok {
                return fmt.Errorf("incorrect args")
        }

        for i := 0; i < listenArgs.ServersNumber; i++ {
                req := tarantool.NewEvalRequest("return box.cfg.listen")
                data, err := listenArgs.ConnPool.Do(req, listenArgs.Mode).Get()
                if err != nil {
                        return fmt.Errorf("fail to Eval: %s", err.Error())
                }
                if len(data) < 1 {
                        return fmt.Errorf("response.Data is empty after Eval")
                }

                port, ok := data[0].(string)
                if !ok {
                        return fmt.Errorf("response.Data is incorrect after Eval")
                }

                actualPorts[port] = true
        }

        equal := reflect.DeepEqual(actualPorts, listenArgs.ExpectedPorts)
        if !equal {
                return fmt.Errorf("expected ports: %v, actual ports: %v",
                        listenArgs.ExpectedPorts, actualPorts)
        }

        return nil
}

func Retry(f func(interface{}) error, args interface{}, count int, timeout time.Duration) error {
        var err error

        for i := 0; ; i++ {
                err = f(args)
                if err == nil {
                        return err
                }

                if i >= (count - 1) {
                        break
                }

                time.Sleep(timeout)
        }

        return err
}

func InsertOnInstance(ctx context.Context, dialer tarantool.Dialer, connOpts tarantool.Opts,
        space interface{}, tuple interface{}) error {
        conn, err := tarantool.Connect(ctx, dialer, connOpts)
        if err != nil {
                return fmt.Errorf("fail to connect: %s", err.Error())
        }
        if conn == nil {
                return fmt.Errorf("conn is nil after Connect")
        }
        defer conn.Close()

        data, err := conn.Do(tarantool.NewInsertRequest(space).Tuple(tuple)).Get()
        if err != nil {
                return fmt.Errorf("failed to Insert: %s", err.Error())
        }
        if len(data) != 1 {
                return fmt.Errorf("response Body len != 1")
        }
        if tpl, ok := data[0].([]interface{}); !ok {
                return fmt.Errorf("unexpected body of Insert")
        } else {
                expectedTpl, ok := tuple.([]interface{})
                if !ok {
                        return fmt.Errorf("failed to cast")
                }

                err = compareTuples(expectedTpl, tpl)
                if err != nil {
                        return err
                }
        }

        return nil
}

func InsertOnInstances(
        dialers []tarantool.Dialer,
        connOpts tarantool.Opts,
        space interface{},
        tuple interface{}) error {
        serversNumber := len(dialers)
        roles := make([]bool, serversNumber)
        for i := 0; i < serversNumber; i++ {
                roles[i] = false
        }

        err := SetClusterRO(dialers, connOpts, roles)
        if err != nil {
                return fmt.Errorf("fail to set roles for cluster: %s", err.Error())
        }

        for _, dialer := range dialers {
                ctx, cancel := GetConnectContext()
                err := InsertOnInstance(ctx, dialer, connOpts, space, tuple)
                cancel()
                if err != nil {
                        return err
                }
        }

        return nil
}

func SetInstanceRO(ctx context.Context, dialer tarantool.Dialer, connOpts tarantool.Opts,
        isReplica bool) error {
        conn, err := tarantool.Connect(ctx, dialer, connOpts)
        if err != nil {
                return err
        }

        defer conn.Close()

        req := tarantool.NewCallRequest("box.cfg").
                Args([]interface{}{map[string]bool{"read_only": isReplica}})
        if _, err := conn.Do(req).Get(); err != nil {
                return err
        }

        return nil
}

func SetClusterRO(dialers []tarantool.Dialer, connOpts tarantool.Opts,
        roles []bool) error {
        if len(dialers) != len(roles) {
                return fmt.Errorf("number of servers should be equal to number of roles")
        }

        for i, dialer := range dialers {
                ctx, cancel := GetConnectContext()
                err := SetInstanceRO(ctx, dialer, connOpts, roles[i])
                cancel()
                if err != nil {
                        return err
                }
        }

        return nil
}

func StartTarantoolInstances(instsOpts []StartOpts) ([]TarantoolInstance, error) {
        instances := make([]TarantoolInstance, 0, len(instsOpts))

        for _, opts := range instsOpts {
                instance, err := StartTarantool(opts)
                if err != nil {
                        StopTarantoolInstances(instances)
                        return nil, err
                }

                instances = append(instances, instance)
        }

        return instances, nil
}

func StopTarantoolInstances(instances []TarantoolInstance) {
        for _, instance := range instances {
                StopTarantoolWithCleanup(instance)
        }
}

func GetPoolConnectContext() (context.Context, context.CancelFunc) {
        return context.WithTimeout(context.Background(), 500*time.Millisecond)
}

1	package test_helpers
2
3	import (
4	"context"
5	"fmt"
6	"reflect"
7	"time"
8
9	"github.com/tarantool/go-tarantool/v2"
10	"github.com/tarantool/go-tarantool/v2/pool"
11	)
12
13	type ListenOnInstanceArgs struct {
14	ConnPool *pool.ConnectionPool
15	Mode pool.Mode
16	ServersNumber int
17	ExpectedPorts map[string]bool
18	}
19
20	type CheckStatusesArgs struct {
21	ConnPool *pool.ConnectionPool
22	Servers []string
23	Mode pool.Mode
24	ExpectedPoolStatus bool
25	ExpectedStatuses map[string]bool
26	}
27
28	func compareTuples(expectedTpl []interface{}, actualTpl []interface{}) error {	×
29	if len(actualTpl) != len(expectedTpl) {	×
30	return fmt.Errorf("unexpected body of Insert (tuple len)")	×
31	}	×
32
33	for i, field := range actualTpl {	×
34	if field != expectedTpl[i] {	×
35	return fmt.Errorf("unexpected field, expected: %v actual: %v", expectedTpl[i], field)	×
36	}	×
37	}
38
39	return nil	×
40	}
41
42	func CheckPoolStatuses(args interface{}) error {	×
43	checkArgs, ok := args.(CheckStatusesArgs)	×
44	if !ok {	×
45	return fmt.Errorf("incorrect args")	×
46	}	×
47
48	connected, _ := checkArgs.ConnPool.ConnectedNow(checkArgs.Mode)	×
49	if connected != checkArgs.ExpectedPoolStatus {	×
50	return fmt.Errorf(	×
51	"incorrect connection pool status: expected status %t actual status %t",	×
52	checkArgs.ExpectedPoolStatus, connected)	×
53	}	×
54
55	poolInfo := checkArgs.ConnPool.GetInfo()	×
56	for _, server := range checkArgs.Servers {	×
57	status := poolInfo[server].ConnectedNow	×
58	if checkArgs.ExpectedStatuses[server] != status {	×
59	return fmt.Errorf(	×
60	"incorrect conn status: addr %s expected status %t actual status %t",	×
61	server, checkArgs.ExpectedStatuses[server], status)	×
62	}	×
63	}
64
65	return nil	×
66	}
67
68	// ProcessListenOnInstance helper calls "return box.cfg.listen"
69	// as many times as there are servers in the connection pool
70	// with specified mode.
71	// For RO mode expected received ports equals to replica ports.
72	// For RW mode expected received ports equals to master ports.
73	// For PreferRO mode expected received ports equals to replica
74	// ports or to all ports.
75	// For PreferRW mode expected received ports equals to master ports
76	// or to all ports.
77	func ProcessListenOnInstance(args interface{}) error {	×
78	actualPorts := map[string]bool{}	×
79		×
80	listenArgs, ok := args.(ListenOnInstanceArgs)	×
81	if !ok {	×
82	return fmt.Errorf("incorrect args")	×
83	}	×
84
85	for i := 0; i < listenArgs.ServersNumber; i++ {	×
86	req := tarantool.NewEvalRequest("return box.cfg.listen")	×
87	data, err := listenArgs.ConnPool.Do(req, listenArgs.Mode).Get()	×
88	if err != nil {	×
89	return fmt.Errorf("fail to Eval: %s", err.Error())	×
90	}	×
91	if len(data) < 1 {	×
92	return fmt.Errorf("response.Data is empty after Eval")	×
93	}	×
94
95	port, ok := data[0].(string)	×
96	if !ok {	×
97	return fmt.Errorf("response.Data is incorrect after Eval")	×
98	}	×
99
100	actualPorts[port] = true	×
101	}
102
103	equal := reflect.DeepEqual(actualPorts, listenArgs.ExpectedPorts)	×
104	if !equal {	×
105	return fmt.Errorf("expected ports: %v, actual ports: %v",	×
106	listenArgs.ExpectedPorts, actualPorts)	×
107	}	×
108
109	return nil	×
110	}
111
112	func Retry(f func(interface{}) error, args interface{}, count int, timeout time.Duration) error {	×
113	var err error	×
114		×
115	for i := 0; ; i++ {	×
116	err = f(args)	×
117	if err == nil {	×
118	return err	×
119	}	×
120
121	if i >= (count - 1) {	×
122	break	×
123	}
124
125	time.Sleep(timeout)	×
126	}
127
128	return err	×
129	}
130
131	func InsertOnInstance(ctx context.Context, dialer tarantool.Dialer, connOpts tarantool.Opts,
132	space interface{}, tuple interface{}) error {	×
133	conn, err := tarantool.Connect(ctx, dialer, connOpts)	×
134	if err != nil {	×
135	return fmt.Errorf("fail to connect: %s", err.Error())	×
136	}	×
137	if conn == nil {	×
138	return fmt.Errorf("conn is nil after Connect")	×
139	}	×
140	defer conn.Close()	×
141		×
142	data, err := conn.Do(tarantool.NewInsertRequest(space).Tuple(tuple)).Get()	×
143	if err != nil {	×
144	return fmt.Errorf("failed to Insert: %s", err.Error())	×
145	}	×
146	if len(data) != 1 {	×
147	return fmt.Errorf("response Body len != 1")	×
148	}	×
149	if tpl, ok := data[0].([]interface{}); !ok {	×
150	return fmt.Errorf("unexpected body of Insert")	×
151	} else {	×
152	expectedTpl, ok := tuple.([]interface{})	×
153	if !ok {	×
154	return fmt.Errorf("failed to cast")	×
155	}	×
156
157	err = compareTuples(expectedTpl, tpl)	×
158	if err != nil {	×
159	return err	×
160	}	×
161	}
162
163	return nil	×
164	}
165
166	func InsertOnInstances(
167	dialers []tarantool.Dialer,
168	connOpts tarantool.Opts,
169	space interface{},
170	tuple interface{}) error {	×
171	serversNumber := len(dialers)	×
172	roles := make([]bool, serversNumber)	×
173	for i := 0; i < serversNumber; i++ {	×
174	roles[i] = false	×
175	}	×
176
177	err := SetClusterRO(dialers, connOpts, roles)	×
178	if err != nil {	×
179	return fmt.Errorf("fail to set roles for cluster: %s", err.Error())	×
180	}	×
181
182	for _, dialer := range dialers {	×
183	ctx, cancel := GetConnectContext()	×
184	err := InsertOnInstance(ctx, dialer, connOpts, space, tuple)	×
185	cancel()	×
186	if err != nil {	×
187	return err	×
188	}	×
189	}
190
191	return nil	×
192	}
193
194	func SetInstanceRO(ctx context.Context, dialer tarantool.Dialer, connOpts tarantool.Opts,
195	isReplica bool) error {	×
196	conn, err := tarantool.Connect(ctx, dialer, connOpts)	×
197	if err != nil {	×
198	return err	×
199	}	×
200
201	defer conn.Close()	×
202		×
203	req := tarantool.NewCallRequest("box.cfg").	×
204	Args([]interface{}{map[string]bool{"read_only": isReplica}})	×
205	if _, err := conn.Do(req).Get(); err != nil {	×
206	return err	×
207	}	×
208
209	return nil	×
210	}
211
212	func SetClusterRO(dialers []tarantool.Dialer, connOpts tarantool.Opts,
213	roles []bool) error {	×
214	if len(dialers) != len(roles) {	×
215	return fmt.Errorf("number of servers should be equal to number of roles")	×
216	}	×
217
218	for i, dialer := range dialers {	×
219	ctx, cancel := GetConnectContext()	×
220	err := SetInstanceRO(ctx, dialer, connOpts, roles[i])	×
221	cancel()	×
222	if err != nil {	×
223	return err	×
224	}	×
225	}
226
227	return nil	×
228	}
229
230	func StartTarantoolInstances(instsOpts []StartOpts) ([]TarantoolInstance, error) {	×
231	instances := make([]TarantoolInstance, 0, len(instsOpts))	×
232		×
233	for _, opts := range instsOpts {	×
234	instance, err := StartTarantool(opts)	×
235	if err != nil {	×
236	StopTarantoolInstances(instances)	×
237	return nil, err	×
238	}	×
239
240	instances = append(instances, instance)	×
241	}
242
243	return instances, nil	×
244	}
245
246	func StopTarantoolInstances(instances []TarantoolInstance) {	×
247	for _, instance := range instances {	×
248	StopTarantoolWithCleanup(instance)	×
249	}	×
250	}
251
252	func GetPoolConnectContext() (context.Context, context.CancelFunc) {	×
253	return context.WithTimeout(context.Background(), 500*time.Millisecond)	×
254	}	×

tarantool / go-tarantool / 12350021459

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous