http_test.go 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877
  1. // Copyright 2017 The go-ethereum Authors
  2. // This file is part of the go-ethereum library.
  3. //
  4. // The go-ethereum library is free software: you can redistribute it and/or modify
  5. // it under the terms of the GNU Lesser General Public License as published by
  6. // the Free Software Foundation, either version 3 of the License, or
  7. // (at your option) any later version.
  8. //
  9. // The go-ethereum library is distributed in the hope that it will be useful,
  10. // but WITHOUT ANY WARRANTY; without even the implied warranty of
  11. // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  12. // GNU Lesser General Public License for more details.
  13. //
  14. // You should have received a copy of the GNU Lesser General Public License
  15. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
  16. package simulations
  17. import (
  18. "context"
  19. "flag"
  20. "fmt"
  21. "math/rand"
  22. "net/http/httptest"
  23. "reflect"
  24. "sync"
  25. "sync/atomic"
  26. "testing"
  27. "time"
  28. "github.com/ethereum/go-ethereum/event"
  29. "github.com/ethereum/go-ethereum/log"
  30. "github.com/ethereum/go-ethereum/node"
  31. "github.com/ethereum/go-ethereum/p2p"
  32. "github.com/ethereum/go-ethereum/p2p/enode"
  33. "github.com/ethereum/go-ethereum/p2p/simulations/adapters"
  34. "github.com/ethereum/go-ethereum/rpc"
  35. "github.com/mattn/go-colorable"
  36. )
  37. var (
  38. loglevel = flag.Int("loglevel", 2, "verbosity of logs")
  39. )
  40. func init() {
  41. flag.Parse()
  42. log.PrintOrigins(true)
  43. log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(colorable.NewColorableStderr(), log.TerminalFormat(true))))
  44. }
  45. // testService implements the node.Service interface and provides protocols
  46. // and APIs which are useful for testing nodes in a simulation network
  47. type testService struct {
  48. id enode.ID
  49. // peerCount is incremented once a peer handshake has been performed
  50. peerCount int64
  51. peers map[enode.ID]*testPeer
  52. peersMtx sync.Mutex
  53. // state stores []byte which is used to test creating and loading
  54. // snapshots
  55. state atomic.Value
  56. }
  57. func newTestService(ctx *adapters.ServiceContext) (node.Service, error) {
  58. svc := &testService{
  59. id: ctx.Config.ID,
  60. peers: make(map[enode.ID]*testPeer),
  61. }
  62. svc.state.Store(ctx.Snapshot)
  63. return svc, nil
  64. }
  65. type testPeer struct {
  66. testReady chan struct{}
  67. dumReady chan struct{}
  68. }
  69. func (t *testService) peer(id enode.ID) *testPeer {
  70. t.peersMtx.Lock()
  71. defer t.peersMtx.Unlock()
  72. if peer, ok := t.peers[id]; ok {
  73. return peer
  74. }
  75. peer := &testPeer{
  76. testReady: make(chan struct{}),
  77. dumReady: make(chan struct{}),
  78. }
  79. t.peers[id] = peer
  80. return peer
  81. }
  82. func (t *testService) Protocols() []p2p.Protocol {
  83. return []p2p.Protocol{
  84. {
  85. Name: "test",
  86. Version: 1,
  87. Length: 3,
  88. Run: t.RunTest,
  89. },
  90. {
  91. Name: "dum",
  92. Version: 1,
  93. Length: 1,
  94. Run: t.RunDum,
  95. },
  96. {
  97. Name: "prb",
  98. Version: 1,
  99. Length: 1,
  100. Run: t.RunPrb,
  101. },
  102. }
  103. }
  104. func (t *testService) APIs() []rpc.API {
  105. return []rpc.API{{
  106. Namespace: "test",
  107. Version: "1.0",
  108. Service: &TestAPI{
  109. state: &t.state,
  110. peerCount: &t.peerCount,
  111. },
  112. }}
  113. }
  114. func (t *testService) Start(server *p2p.Server) error {
  115. return nil
  116. }
  117. func (t *testService) Stop() error {
  118. return nil
  119. }
  120. // handshake performs a peer handshake by sending and expecting an empty
  121. // message with the given code
  122. func (t *testService) handshake(rw p2p.MsgReadWriter, code uint64) error {
  123. errc := make(chan error, 2)
  124. go func() { errc <- p2p.Send(rw, code, struct{}{}) }()
  125. go func() { errc <- p2p.ExpectMsg(rw, code, struct{}{}) }()
  126. for i := 0; i < 2; i++ {
  127. if err := <-errc; err != nil {
  128. return err
  129. }
  130. }
  131. return nil
  132. }
  133. func (t *testService) RunTest(p *p2p.Peer, rw p2p.MsgReadWriter) error {
  134. peer := t.peer(p.ID())
  135. // perform three handshakes with three different message codes,
  136. // used to test message sending and filtering
  137. if err := t.handshake(rw, 2); err != nil {
  138. return err
  139. }
  140. if err := t.handshake(rw, 1); err != nil {
  141. return err
  142. }
  143. if err := t.handshake(rw, 0); err != nil {
  144. return err
  145. }
  146. // close the testReady channel so that other protocols can run
  147. close(peer.testReady)
  148. // track the peer
  149. atomic.AddInt64(&t.peerCount, 1)
  150. defer atomic.AddInt64(&t.peerCount, -1)
  151. // block until the peer is dropped
  152. for {
  153. _, err := rw.ReadMsg()
  154. if err != nil {
  155. return err
  156. }
  157. }
  158. }
  159. func (t *testService) RunDum(p *p2p.Peer, rw p2p.MsgReadWriter) error {
  160. peer := t.peer(p.ID())
  161. // wait for the test protocol to perform its handshake
  162. <-peer.testReady
  163. // perform a handshake
  164. if err := t.handshake(rw, 0); err != nil {
  165. return err
  166. }
  167. // close the dumReady channel so that other protocols can run
  168. close(peer.dumReady)
  169. // block until the peer is dropped
  170. for {
  171. _, err := rw.ReadMsg()
  172. if err != nil {
  173. return err
  174. }
  175. }
  176. }
  177. func (t *testService) RunPrb(p *p2p.Peer, rw p2p.MsgReadWriter) error {
  178. peer := t.peer(p.ID())
  179. // wait for the dum protocol to perform its handshake
  180. <-peer.dumReady
  181. // perform a handshake
  182. if err := t.handshake(rw, 0); err != nil {
  183. return err
  184. }
  185. // block until the peer is dropped
  186. for {
  187. _, err := rw.ReadMsg()
  188. if err != nil {
  189. return err
  190. }
  191. }
  192. }
  193. func (t *testService) Snapshot() ([]byte, error) {
  194. return t.state.Load().([]byte), nil
  195. }
  196. // TestAPI provides a test API to:
  197. // * get the peer count
  198. // * get and set an arbitrary state byte slice
  199. // * get and increment a counter
  200. // * subscribe to counter increment events
  201. type TestAPI struct {
  202. state *atomic.Value
  203. peerCount *int64
  204. counter int64
  205. feed event.Feed
  206. }
  207. func (t *TestAPI) PeerCount() int64 {
  208. return atomic.LoadInt64(t.peerCount)
  209. }
  210. func (t *TestAPI) Get() int64 {
  211. return atomic.LoadInt64(&t.counter)
  212. }
  213. func (t *TestAPI) Add(delta int64) {
  214. atomic.AddInt64(&t.counter, delta)
  215. t.feed.Send(delta)
  216. }
  217. func (t *TestAPI) GetState() []byte {
  218. return t.state.Load().([]byte)
  219. }
  220. func (t *TestAPI) SetState(state []byte) {
  221. t.state.Store(state)
  222. }
  223. func (t *TestAPI) Events(ctx context.Context) (*rpc.Subscription, error) {
  224. notifier, supported := rpc.NotifierFromContext(ctx)
  225. if !supported {
  226. return nil, rpc.ErrNotificationsUnsupported
  227. }
  228. rpcSub := notifier.CreateSubscription()
  229. go func() {
  230. events := make(chan int64)
  231. sub := t.feed.Subscribe(events)
  232. defer sub.Unsubscribe()
  233. for {
  234. select {
  235. case event := <-events:
  236. notifier.Notify(rpcSub.ID, event)
  237. case <-sub.Err():
  238. return
  239. case <-rpcSub.Err():
  240. return
  241. case <-notifier.Closed():
  242. return
  243. }
  244. }
  245. }()
  246. return rpcSub, nil
  247. }
  248. var testServices = adapters.Services{
  249. "test": newTestService,
  250. }
  251. func testHTTPServer(t *testing.T) (*Network, *httptest.Server) {
  252. t.Helper()
  253. adapter := adapters.NewSimAdapter(testServices)
  254. network := NewNetwork(adapter, &NetworkConfig{
  255. DefaultService: "test",
  256. })
  257. return network, httptest.NewServer(NewServer(network))
  258. }
  259. // TestHTTPNetwork tests interacting with a simulation network using the HTTP
  260. // API
  261. func TestHTTPNetwork(t *testing.T) {
  262. // start the server
  263. network, s := testHTTPServer(t)
  264. defer s.Close()
  265. // subscribe to events so we can check them later
  266. client := NewClient(s.URL)
  267. events := make(chan *Event, 100)
  268. var opts SubscribeOpts
  269. sub, err := client.SubscribeNetwork(events, opts)
  270. if err != nil {
  271. t.Fatalf("error subscribing to network events: %s", err)
  272. }
  273. defer sub.Unsubscribe()
  274. // check we can retrieve details about the network
  275. gotNetwork, err := client.GetNetwork()
  276. if err != nil {
  277. t.Fatalf("error getting network: %s", err)
  278. }
  279. if gotNetwork.ID != network.ID {
  280. t.Fatalf("expected network to have ID %q, got %q", network.ID, gotNetwork.ID)
  281. }
  282. // start a simulation network
  283. nodeIDs := startTestNetwork(t, client)
  284. // check we got all the events
  285. x := &expectEvents{t, events, sub}
  286. x.expect(
  287. x.nodeEvent(nodeIDs[0], false),
  288. x.nodeEvent(nodeIDs[1], false),
  289. x.nodeEvent(nodeIDs[0], true),
  290. x.nodeEvent(nodeIDs[1], true),
  291. x.connEvent(nodeIDs[0], nodeIDs[1], false),
  292. x.connEvent(nodeIDs[0], nodeIDs[1], true),
  293. )
  294. // reconnect the stream and check we get the current nodes and conns
  295. events = make(chan *Event, 100)
  296. opts.Current = true
  297. sub, err = client.SubscribeNetwork(events, opts)
  298. if err != nil {
  299. t.Fatalf("error subscribing to network events: %s", err)
  300. }
  301. defer sub.Unsubscribe()
  302. x = &expectEvents{t, events, sub}
  303. x.expect(
  304. x.nodeEvent(nodeIDs[0], true),
  305. x.nodeEvent(nodeIDs[1], true),
  306. x.connEvent(nodeIDs[0], nodeIDs[1], true),
  307. )
  308. }
  309. func startTestNetwork(t *testing.T, client *Client) []string {
  310. // create two nodes
  311. nodeCount := 2
  312. nodeIDs := make([]string, nodeCount)
  313. for i := 0; i < nodeCount; i++ {
  314. config := adapters.RandomNodeConfig()
  315. node, err := client.CreateNode(config)
  316. if err != nil {
  317. t.Fatalf("error creating node: %s", err)
  318. }
  319. nodeIDs[i] = node.ID
  320. }
  321. // check both nodes exist
  322. nodes, err := client.GetNodes()
  323. if err != nil {
  324. t.Fatalf("error getting nodes: %s", err)
  325. }
  326. if len(nodes) != nodeCount {
  327. t.Fatalf("expected %d nodes, got %d", nodeCount, len(nodes))
  328. }
  329. for i, nodeID := range nodeIDs {
  330. if nodes[i].ID != nodeID {
  331. t.Fatalf("expected node %d to have ID %q, got %q", i, nodeID, nodes[i].ID)
  332. }
  333. node, err := client.GetNode(nodeID)
  334. if err != nil {
  335. t.Fatalf("error getting node %d: %s", i, err)
  336. }
  337. if node.ID != nodeID {
  338. t.Fatalf("expected node %d to have ID %q, got %q", i, nodeID, node.ID)
  339. }
  340. }
  341. // start both nodes
  342. for _, nodeID := range nodeIDs {
  343. if err := client.StartNode(nodeID); err != nil {
  344. t.Fatalf("error starting node %q: %s", nodeID, err)
  345. }
  346. }
  347. // connect the nodes
  348. for i := 0; i < nodeCount-1; i++ {
  349. peerId := i + 1
  350. if i == nodeCount-1 {
  351. peerId = 0
  352. }
  353. if err := client.ConnectNode(nodeIDs[i], nodeIDs[peerId]); err != nil {
  354. t.Fatalf("error connecting nodes: %s", err)
  355. }
  356. }
  357. return nodeIDs
  358. }
  359. type expectEvents struct {
  360. *testing.T
  361. events chan *Event
  362. sub event.Subscription
  363. }
  364. func (t *expectEvents) nodeEvent(id string, up bool) *Event {
  365. return &Event{
  366. Type: EventTypeNode,
  367. Node: &Node{
  368. Config: &adapters.NodeConfig{
  369. ID: enode.HexID(id),
  370. },
  371. Up: up,
  372. },
  373. }
  374. }
  375. func (t *expectEvents) connEvent(one, other string, up bool) *Event {
  376. return &Event{
  377. Type: EventTypeConn,
  378. Conn: &Conn{
  379. One: enode.HexID(one),
  380. Other: enode.HexID(other),
  381. Up: up,
  382. },
  383. }
  384. }
  385. func (t *expectEvents) expectMsgs(expected map[MsgFilter]int) {
  386. actual := make(map[MsgFilter]int)
  387. timeout := time.After(10 * time.Second)
  388. loop:
  389. for {
  390. select {
  391. case event := <-t.events:
  392. t.Logf("received %s event: %s", event.Type, event)
  393. if event.Type != EventTypeMsg || event.Msg.Received {
  394. continue loop
  395. }
  396. if event.Msg == nil {
  397. t.Fatal("expected event.Msg to be set")
  398. }
  399. filter := MsgFilter{
  400. Proto: event.Msg.Protocol,
  401. Code: int64(event.Msg.Code),
  402. }
  403. actual[filter]++
  404. if actual[filter] > expected[filter] {
  405. t.Fatalf("received too many msgs for filter: %v", filter)
  406. }
  407. if reflect.DeepEqual(actual, expected) {
  408. return
  409. }
  410. case err := <-t.sub.Err():
  411. t.Fatalf("network stream closed unexpectedly: %s", err)
  412. case <-timeout:
  413. t.Fatal("timed out waiting for expected events")
  414. }
  415. }
  416. }
  417. func (t *expectEvents) expect(events ...*Event) {
  418. timeout := time.After(10 * time.Second)
  419. i := 0
  420. for {
  421. select {
  422. case event := <-t.events:
  423. t.Logf("received %s event: %s", event.Type, event)
  424. expected := events[i]
  425. if event.Type != expected.Type {
  426. t.Fatalf("expected event %d to have type %q, got %q", i, expected.Type, event.Type)
  427. }
  428. switch expected.Type {
  429. case EventTypeNode:
  430. if event.Node == nil {
  431. t.Fatal("expected event.Node to be set")
  432. }
  433. if event.Node.ID() != expected.Node.ID() {
  434. t.Fatalf("expected node event %d to have id %q, got %q", i, expected.Node.ID().TerminalString(), event.Node.ID().TerminalString())
  435. }
  436. if event.Node.Up != expected.Node.Up {
  437. t.Fatalf("expected node event %d to have up=%t, got up=%t", i, expected.Node.Up, event.Node.Up)
  438. }
  439. case EventTypeConn:
  440. if event.Conn == nil {
  441. t.Fatal("expected event.Conn to be set")
  442. }
  443. if event.Conn.One != expected.Conn.One {
  444. t.Fatalf("expected conn event %d to have one=%q, got one=%q", i, expected.Conn.One.TerminalString(), event.Conn.One.TerminalString())
  445. }
  446. if event.Conn.Other != expected.Conn.Other {
  447. t.Fatalf("expected conn event %d to have other=%q, got other=%q", i, expected.Conn.Other.TerminalString(), event.Conn.Other.TerminalString())
  448. }
  449. if event.Conn.Up != expected.Conn.Up {
  450. t.Fatalf("expected conn event %d to have up=%t, got up=%t", i, expected.Conn.Up, event.Conn.Up)
  451. }
  452. }
  453. i++
  454. if i == len(events) {
  455. return
  456. }
  457. case err := <-t.sub.Err():
  458. t.Fatalf("network stream closed unexpectedly: %s", err)
  459. case <-timeout:
  460. t.Fatal("timed out waiting for expected events")
  461. }
  462. }
  463. }
  464. // TestHTTPNodeRPC tests calling RPC methods on nodes via the HTTP API
  465. func TestHTTPNodeRPC(t *testing.T) {
  466. // start the server
  467. _, s := testHTTPServer(t)
  468. defer s.Close()
  469. // start a node in the network
  470. client := NewClient(s.URL)
  471. config := adapters.RandomNodeConfig()
  472. node, err := client.CreateNode(config)
  473. if err != nil {
  474. t.Fatalf("error creating node: %s", err)
  475. }
  476. if err := client.StartNode(node.ID); err != nil {
  477. t.Fatalf("error starting node: %s", err)
  478. }
  479. // create two RPC clients
  480. ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
  481. defer cancel()
  482. rpcClient1, err := client.RPCClient(ctx, node.ID)
  483. if err != nil {
  484. t.Fatalf("error getting node RPC client: %s", err)
  485. }
  486. rpcClient2, err := client.RPCClient(ctx, node.ID)
  487. if err != nil {
  488. t.Fatalf("error getting node RPC client: %s", err)
  489. }
  490. // subscribe to events using client 1
  491. events := make(chan int64, 1)
  492. sub, err := rpcClient1.Subscribe(ctx, "test", events, "events")
  493. if err != nil {
  494. t.Fatalf("error subscribing to events: %s", err)
  495. }
  496. defer sub.Unsubscribe()
  497. // call some RPC methods using client 2
  498. if err := rpcClient2.CallContext(ctx, nil, "test_add", 10); err != nil {
  499. t.Fatalf("error calling RPC method: %s", err)
  500. }
  501. var result int64
  502. if err := rpcClient2.CallContext(ctx, &result, "test_get"); err != nil {
  503. t.Fatalf("error calling RPC method: %s", err)
  504. }
  505. if result != 10 {
  506. t.Fatalf("expected result to be 10, got %d", result)
  507. }
  508. // check we got an event from client 1
  509. select {
  510. case event := <-events:
  511. if event != 10 {
  512. t.Fatalf("expected event to be 10, got %d", event)
  513. }
  514. case <-ctx.Done():
  515. t.Fatal(ctx.Err())
  516. }
  517. }
  518. // TestHTTPSnapshot tests creating and loading network snapshots
  519. func TestHTTPSnapshot(t *testing.T) {
  520. // start the server
  521. network, s := testHTTPServer(t)
  522. defer s.Close()
  523. var eventsDone = make(chan struct{})
  524. count := 1
  525. eventsDoneChan := make(chan *Event)
  526. eventSub := network.Events().Subscribe(eventsDoneChan)
  527. go func() {
  528. defer eventSub.Unsubscribe()
  529. for event := range eventsDoneChan {
  530. if event.Type == EventTypeConn && !event.Control {
  531. count--
  532. if count == 0 {
  533. eventsDone <- struct{}{}
  534. return
  535. }
  536. }
  537. }
  538. }()
  539. // create a two-node network
  540. client := NewClient(s.URL)
  541. nodeCount := 2
  542. nodes := make([]*p2p.NodeInfo, nodeCount)
  543. for i := 0; i < nodeCount; i++ {
  544. config := adapters.RandomNodeConfig()
  545. node, err := client.CreateNode(config)
  546. if err != nil {
  547. t.Fatalf("error creating node: %s", err)
  548. }
  549. if err := client.StartNode(node.ID); err != nil {
  550. t.Fatalf("error starting node: %s", err)
  551. }
  552. nodes[i] = node
  553. }
  554. if err := client.ConnectNode(nodes[0].ID, nodes[1].ID); err != nil {
  555. t.Fatalf("error connecting nodes: %s", err)
  556. }
  557. // store some state in the test services
  558. states := make([]string, nodeCount)
  559. for i, node := range nodes {
  560. rpc, err := client.RPCClient(context.Background(), node.ID)
  561. if err != nil {
  562. t.Fatalf("error getting RPC client: %s", err)
  563. }
  564. defer rpc.Close()
  565. state := fmt.Sprintf("%x", rand.Int())
  566. if err := rpc.Call(nil, "test_setState", []byte(state)); err != nil {
  567. t.Fatalf("error setting service state: %s", err)
  568. }
  569. states[i] = state
  570. }
  571. <-eventsDone
  572. // create a snapshot
  573. snap, err := client.CreateSnapshot()
  574. if err != nil {
  575. t.Fatalf("error creating snapshot: %s", err)
  576. }
  577. for i, state := range states {
  578. gotState := snap.Nodes[i].Snapshots["test"]
  579. if string(gotState) != state {
  580. t.Fatalf("expected snapshot state %q, got %q", state, gotState)
  581. }
  582. }
  583. // create another network
  584. network2, s := testHTTPServer(t)
  585. defer s.Close()
  586. client = NewClient(s.URL)
  587. count = 1
  588. eventSub = network2.Events().Subscribe(eventsDoneChan)
  589. go func() {
  590. defer eventSub.Unsubscribe()
  591. for event := range eventsDoneChan {
  592. if event.Type == EventTypeConn && !event.Control {
  593. count--
  594. if count == 0 {
  595. eventsDone <- struct{}{}
  596. return
  597. }
  598. }
  599. }
  600. }()
  601. // subscribe to events so we can check them later
  602. events := make(chan *Event, 100)
  603. var opts SubscribeOpts
  604. sub, err := client.SubscribeNetwork(events, opts)
  605. if err != nil {
  606. t.Fatalf("error subscribing to network events: %s", err)
  607. }
  608. defer sub.Unsubscribe()
  609. // load the snapshot
  610. if err := client.LoadSnapshot(snap); err != nil {
  611. t.Fatalf("error loading snapshot: %s", err)
  612. }
  613. <-eventsDone
  614. // check the nodes and connection exists
  615. net, err := client.GetNetwork()
  616. if err != nil {
  617. t.Fatalf("error getting network: %s", err)
  618. }
  619. if len(net.Nodes) != nodeCount {
  620. t.Fatalf("expected network to have %d nodes, got %d", nodeCount, len(net.Nodes))
  621. }
  622. for i, node := range nodes {
  623. id := net.Nodes[i].ID().String()
  624. if id != node.ID {
  625. t.Fatalf("expected node %d to have ID %s, got %s", i, node.ID, id)
  626. }
  627. }
  628. if len(net.Conns) != 1 {
  629. t.Fatalf("expected network to have 1 connection, got %d", len(net.Conns))
  630. }
  631. conn := net.Conns[0]
  632. if conn.One.String() != nodes[0].ID {
  633. t.Fatalf("expected connection to have one=%q, got one=%q", nodes[0].ID, conn.One)
  634. }
  635. if conn.Other.String() != nodes[1].ID {
  636. t.Fatalf("expected connection to have other=%q, got other=%q", nodes[1].ID, conn.Other)
  637. }
  638. if !conn.Up {
  639. t.Fatal("should be up")
  640. }
  641. // check the node states were restored
  642. for i, node := range nodes {
  643. rpc, err := client.RPCClient(context.Background(), node.ID)
  644. if err != nil {
  645. t.Fatalf("error getting RPC client: %s", err)
  646. }
  647. defer rpc.Close()
  648. var state []byte
  649. if err := rpc.Call(&state, "test_getState"); err != nil {
  650. t.Fatalf("error getting service state: %s", err)
  651. }
  652. if string(state) != states[i] {
  653. t.Fatalf("expected snapshot state %q, got %q", states[i], state)
  654. }
  655. }
  656. // check we got all the events
  657. x := &expectEvents{t, events, sub}
  658. x.expect(
  659. x.nodeEvent(nodes[0].ID, false),
  660. x.nodeEvent(nodes[0].ID, true),
  661. x.nodeEvent(nodes[1].ID, false),
  662. x.nodeEvent(nodes[1].ID, true),
  663. x.connEvent(nodes[0].ID, nodes[1].ID, false),
  664. x.connEvent(nodes[0].ID, nodes[1].ID, true),
  665. )
  666. }
  667. // TestMsgFilterPassMultiple tests streaming message events using a filter
  668. // with multiple protocols
  669. func TestMsgFilterPassMultiple(t *testing.T) {
  670. // start the server
  671. _, s := testHTTPServer(t)
  672. defer s.Close()
  673. // subscribe to events with a message filter
  674. client := NewClient(s.URL)
  675. events := make(chan *Event, 10)
  676. opts := SubscribeOpts{
  677. Filter: "prb:0-test:0",
  678. }
  679. sub, err := client.SubscribeNetwork(events, opts)
  680. if err != nil {
  681. t.Fatalf("error subscribing to network events: %s", err)
  682. }
  683. defer sub.Unsubscribe()
  684. // start a simulation network
  685. startTestNetwork(t, client)
  686. // check we got the expected events
  687. x := &expectEvents{t, events, sub}
  688. x.expectMsgs(map[MsgFilter]int{
  689. {"test", 0}: 2,
  690. {"prb", 0}: 2,
  691. })
  692. }
  693. // TestMsgFilterPassWildcard tests streaming message events using a filter
  694. // with a code wildcard
  695. func TestMsgFilterPassWildcard(t *testing.T) {
  696. // start the server
  697. _, s := testHTTPServer(t)
  698. defer s.Close()
  699. // subscribe to events with a message filter
  700. client := NewClient(s.URL)
  701. events := make(chan *Event, 10)
  702. opts := SubscribeOpts{
  703. Filter: "prb:0,2-test:*",
  704. }
  705. sub, err := client.SubscribeNetwork(events, opts)
  706. if err != nil {
  707. t.Fatalf("error subscribing to network events: %s", err)
  708. }
  709. defer sub.Unsubscribe()
  710. // start a simulation network
  711. startTestNetwork(t, client)
  712. // check we got the expected events
  713. x := &expectEvents{t, events, sub}
  714. x.expectMsgs(map[MsgFilter]int{
  715. {"test", 2}: 2,
  716. {"test", 1}: 2,
  717. {"test", 0}: 2,
  718. {"prb", 0}: 2,
  719. })
  720. }
  721. // TestMsgFilterPassSingle tests streaming message events using a filter
  722. // with a single protocol and code
  723. func TestMsgFilterPassSingle(t *testing.T) {
  724. // start the server
  725. _, s := testHTTPServer(t)
  726. defer s.Close()
  727. // subscribe to events with a message filter
  728. client := NewClient(s.URL)
  729. events := make(chan *Event, 10)
  730. opts := SubscribeOpts{
  731. Filter: "dum:0",
  732. }
  733. sub, err := client.SubscribeNetwork(events, opts)
  734. if err != nil {
  735. t.Fatalf("error subscribing to network events: %s", err)
  736. }
  737. defer sub.Unsubscribe()
  738. // start a simulation network
  739. startTestNetwork(t, client)
  740. // check we got the expected events
  741. x := &expectEvents{t, events, sub}
  742. x.expectMsgs(map[MsgFilter]int{
  743. {"dum", 0}: 2,
  744. })
  745. }
  746. // TestMsgFilterPassSingle tests streaming message events using an invalid
  747. // filter
  748. func TestMsgFilterFailBadParams(t *testing.T) {
  749. // start the server
  750. _, s := testHTTPServer(t)
  751. defer s.Close()
  752. client := NewClient(s.URL)
  753. events := make(chan *Event, 10)
  754. opts := SubscribeOpts{
  755. Filter: "foo:",
  756. }
  757. _, err := client.SubscribeNetwork(events, opts)
  758. if err == nil {
  759. t.Fatalf("expected event subscription to fail but succeeded!")
  760. }
  761. opts.Filter = "bzz:aa"
  762. _, err = client.SubscribeNetwork(events, opts)
  763. if err == nil {
  764. t.Fatalf("expected event subscription to fail but succeeded!")
  765. }
  766. opts.Filter = "invalid"
  767. _, err = client.SubscribeNetwork(events, opts)
  768. if err == nil {
  769. t.Fatalf("expected event subscription to fail but succeeded!")
  770. }
  771. }