aboutsummaryrefslogtreecommitdiffstats
path: root/swarm/network/stream/streamer_test.go
diff options
context:
space:
mode:
Diffstat (limited to 'swarm/network/stream/streamer_test.go')
-rw-r--r--swarm/network/stream/streamer_test.go183
1 files changed, 173 insertions, 10 deletions
diff --git a/swarm/network/stream/streamer_test.go b/swarm/network/stream/streamer_test.go
index 0390a7b9b..0bdebefa7 100644
--- a/swarm/network/stream/streamer_test.go
+++ b/swarm/network/stream/streamer_test.go
@@ -20,6 +20,7 @@ import (
"bytes"
"context"
"errors"
+ "strconv"
"testing"
"time"
@@ -28,7 +29,7 @@ import (
)
func TestStreamerSubscribe(t *testing.T) {
- tester, streamer, _, teardown, err := newStreamerTester(t)
+ tester, streamer, _, teardown, err := newStreamerTester(t, nil)
defer teardown()
if err != nil {
t.Fatal(err)
@@ -42,7 +43,7 @@ func TestStreamerSubscribe(t *testing.T) {
}
func TestStreamerRequestSubscription(t *testing.T) {
- tester, streamer, _, teardown, err := newStreamerTester(t)
+ tester, streamer, _, teardown, err := newStreamerTester(t, nil)
defer teardown()
if err != nil {
t.Fatal(err)
@@ -127,7 +128,7 @@ func (self *testServer) Close() {
}
func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
- tester, streamer, _, teardown, err := newStreamerTester(t)
+ tester, streamer, _, teardown, err := newStreamerTester(t, nil)
defer teardown()
if err != nil {
t.Fatal(err)
@@ -220,7 +221,7 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
}
func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
- tester, streamer, _, teardown, err := newStreamerTester(t)
+ tester, streamer, _, teardown, err := newStreamerTester(t, nil)
defer teardown()
if err != nil {
t.Fatal(err)
@@ -287,7 +288,7 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
}
func TestStreamerUpstreamSubscribeUnsubscribeMsgExchangeLive(t *testing.T) {
- tester, streamer, _, teardown, err := newStreamerTester(t)
+ tester, streamer, _, teardown, err := newStreamerTester(t, nil)
defer teardown()
if err != nil {
t.Fatal(err)
@@ -353,7 +354,7 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchangeLive(t *testing.T) {
}
func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) {
- tester, streamer, _, teardown, err := newStreamerTester(t)
+ tester, streamer, _, teardown, err := newStreamerTester(t, nil)
defer teardown()
if err != nil {
t.Fatal(err)
@@ -397,7 +398,7 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) {
}
func TestStreamerUpstreamSubscribeLiveAndHistory(t *testing.T) {
- tester, streamer, _, teardown, err := newStreamerTester(t)
+ tester, streamer, _, teardown, err := newStreamerTester(t, nil)
defer teardown()
if err != nil {
t.Fatal(err)
@@ -462,7 +463,7 @@ func TestStreamerUpstreamSubscribeLiveAndHistory(t *testing.T) {
}
func TestStreamerDownstreamCorruptHashesMsgExchange(t *testing.T) {
- tester, streamer, _, teardown, err := newStreamerTester(t)
+ tester, streamer, _, teardown, err := newStreamerTester(t, nil)
defer teardown()
if err != nil {
t.Fatal(err)
@@ -527,7 +528,7 @@ func TestStreamerDownstreamCorruptHashesMsgExchange(t *testing.T) {
}
func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
- tester, streamer, _, teardown, err := newStreamerTester(t)
+ tester, streamer, _, teardown, err := newStreamerTester(t, nil)
defer teardown()
if err != nil {
t.Fatal(err)
@@ -626,7 +627,7 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
}
func TestStreamerRequestSubscriptionQuitMsgExchange(t *testing.T) {
- tester, streamer, _, teardown, err := newStreamerTester(t)
+ tester, streamer, _, teardown, err := newStreamerTester(t, nil)
defer teardown()
if err != nil {
t.Fatal(err)
@@ -752,3 +753,165 @@ func TestStreamerRequestSubscriptionQuitMsgExchange(t *testing.T) {
t.Fatal(err)
}
}
+
+// TestMaxPeerServersWithUnsubscribe creates a registry with a limited
+// number of stream servers, and performs a test with subscriptions and
+// unsubscriptions, checking if unsubscriptions will remove streams,
+// leaving place for new streams.
+func TestMaxPeerServersWithUnsubscribe(t *testing.T) {
+ var maxPeerServers = 6
+ tester, streamer, _, teardown, err := newStreamerTester(t, &RegistryOptions{
+ MaxPeerServers: maxPeerServers,
+ })
+ defer teardown()
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ streamer.RegisterServerFunc("foo", func(p *Peer, t string, live bool) (Server, error) {
+ return newTestServer(t), nil
+ })
+
+ node := tester.Nodes[0]
+
+ for i := 0; i < maxPeerServers+10; i++ {
+ stream := NewStream("foo", strconv.Itoa(i), true)
+
+ err = tester.TestExchanges(p2ptest.Exchange{
+ Label: "Subscribe message",
+ Triggers: []p2ptest.Trigger{
+ {
+ Code: 4,
+ Msg: &SubscribeMsg{
+ Stream: stream,
+ Priority: Top,
+ },
+ Peer: node.ID(),
+ },
+ },
+ Expects: []p2ptest.Expect{
+ {
+ Code: 1,
+ Msg: &OfferedHashesMsg{
+ Stream: stream,
+ HandoverProof: &HandoverProof{
+ Handover: &Handover{},
+ },
+ Hashes: make([]byte, HashSize),
+ From: 1,
+ To: 1,
+ },
+ Peer: node.ID(),
+ },
+ },
+ })
+
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ err = tester.TestExchanges(p2ptest.Exchange{
+ Label: "unsubscribe message",
+ Triggers: []p2ptest.Trigger{
+ {
+ Code: 0,
+ Msg: &UnsubscribeMsg{
+ Stream: stream,
+ },
+ Peer: node.ID(),
+ },
+ },
+ })
+
+ if err != nil {
+ t.Fatal(err)
+ }
+ }
+}
+
+// TestMaxPeerServersWithoutUnsubscribe creates a registry with a limited
+// number of stream servers, and performs subscriptions to detect subscriptions
+// error message exchange.
+func TestMaxPeerServersWithoutUnsubscribe(t *testing.T) {
+ var maxPeerServers = 6
+ tester, streamer, _, teardown, err := newStreamerTester(t, &RegistryOptions{
+ MaxPeerServers: maxPeerServers,
+ })
+ defer teardown()
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ streamer.RegisterServerFunc("foo", func(p *Peer, t string, live bool) (Server, error) {
+ return newTestServer(t), nil
+ })
+
+ node := tester.Nodes[0]
+
+ for i := 0; i < maxPeerServers+10; i++ {
+ stream := NewStream("foo", strconv.Itoa(i), true)
+
+ if i >= maxPeerServers {
+ err = tester.TestExchanges(p2ptest.Exchange{
+ Label: "Subscribe message",
+ Triggers: []p2ptest.Trigger{
+ {
+ Code: 4,
+ Msg: &SubscribeMsg{
+ Stream: stream,
+ Priority: Top,
+ },
+ Peer: node.ID(),
+ },
+ },
+ Expects: []p2ptest.Expect{
+ {
+ Code: 7,
+ Msg: &SubscribeErrorMsg{
+ Error: ErrMaxPeerServers.Error(),
+ },
+ Peer: node.ID(),
+ },
+ },
+ })
+
+ if err != nil {
+ t.Fatal(err)
+ }
+ continue
+ }
+
+ err = tester.TestExchanges(p2ptest.Exchange{
+ Label: "Subscribe message",
+ Triggers: []p2ptest.Trigger{
+ {
+ Code: 4,
+ Msg: &SubscribeMsg{
+ Stream: stream,
+ Priority: Top,
+ },
+ Peer: node.ID(),
+ },
+ },
+ Expects: []p2ptest.Expect{
+ {
+ Code: 1,
+ Msg: &OfferedHashesMsg{
+ Stream: stream,
+ HandoverProof: &HandoverProof{
+ Handover: &Handover{},
+ },
+ Hashes: make([]byte, HashSize),
+ From: 1,
+ To: 1,
+ },
+ Peer: node.ID(),
+ },
+ },
+ })
+
+ if err != nil {
+ t.Fatal(err)
+ }
+ }
+}
generated by cgit v1.2.3 (git 2.25.1) at 2024-08-26 20:20:44 +0800