-
Notifications
You must be signed in to change notification settings - Fork 3
/
mvds_interactive_test.go
103 lines (83 loc) · 2.91 KB
/
mvds_interactive_test.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
package main
import (
"testing"
"time"
"github.com/stretchr/testify/suite"
"github.com/vacp2p/mvds/dependency"
"github.com/vacp2p/mvds/node"
"github.com/vacp2p/mvds/peers"
"github.com/vacp2p/mvds/state"
"github.com/vacp2p/mvds/store"
"github.com/vacp2p/mvds/transport"
"go.uber.org/zap"
)
func TestMVDSInteractiveSuite(t *testing.T) {
suite.Run(t, new(MVDSInteractiveSuite))
}
type MVDSInteractiveSuite struct {
suite.Suite
client1 *node.Node
client2 *node.Node
ds1 store.MessageStore
ds2 store.MessageStore
state1 state.SyncState
state2 state.SyncState
peers1 peers.Persistence
peers2 peers.Persistence
groupID state.GroupID
}
func (s *MVDSInteractiveSuite) SetupTest() {
logger := zap.NewNop()
in1 := make(chan transport.Packet)
t1 := transport.NewChannelTransport(0, in1)
s.ds1 = store.NewMemoryMessageStore()
s.state1 = state.NewMemorySyncState()
s.peers1 = peers.NewMemoryPersistence()
p1 := [65]byte{0x01}
s.client1 = node.NewNode(s.ds1, t1, s.state1, Calc, 0, p1, node.InteractiveMode, s.peers1, dependency.NewInMemoryTracker(), node.EventualMode, logger)
in2 := make(chan transport.Packet)
t2 := transport.NewChannelTransport(0, in2)
s.ds2 = store.NewMemoryMessageStore()
s.state2 = state.NewMemorySyncState()
p2 := [65]byte{0x02}
s.peers2 = peers.NewMemoryPersistence()
s.client2 = node.NewNode(s.ds2, t2, s.state2, Calc, 0, p2, node.InteractiveMode, s.peers2, dependency.NewInMemoryTracker(), node.EventualMode, logger)
t2.AddOutput(p1, in1)
t1.AddOutput(p2, in2)
s.groupID = [32]byte{0x01, 0x2, 0x3, 0x4}
s.Require().NoError(s.client1.AddPeer(s.groupID, p2))
s.Require().NoError(s.client2.AddPeer(s.groupID, p1))
s.client1.Start(10 * time.Millisecond)
s.client2.Start(10 * time.Millisecond)
}
func (s *MVDSInteractiveSuite) TearDownTest() {
s.client1.Stop()
s.client2.Stop()
}
func (s *MVDSInteractiveSuite) TestInteractiveMode() {
subscription := s.client2.Subscribe()
messageID, err := s.client1.AppendMessage(s.groupID, []byte("message 1"))
s.Require().NoError(err)
// Check message is in store
message1Sender, err := s.ds1.Get(messageID)
s.Require().NoError(err)
s.Require().NotNil(message1Sender)
// Check state is updated correctly
states, err := s.state1.All(s.client1.CurrentEpoch())
s.Require().NoError(err)
s.Require().Equal(1, len(states))
// Check we store the request
s.Require().Eventually(func() bool {
states, err := s.state2.All(s.client2.CurrentEpoch())
return err == nil && len(states) == 1 && states[0].Type == state.REQUEST
}, 1*time.Second, 10*time.Millisecond, "An request is stored in the state")
<-subscription
message1Receiver, err := s.ds2.Get(messageID)
s.Require().NoError(err)
s.Require().NotNil(message1Receiver)
// Check state is removed
s.Require().Eventually(func() bool {
states, err := s.state1.All(s.client1.CurrentEpoch())
return err == nil && len(states) == 0
}, 1*time.Second, 10*time.Millisecond, "We clear all the state")
}