http_test.go 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876
  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. colorable "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. adapter := adapters.NewSimAdapter(testServices)
  253. network := NewNetwork(adapter, &NetworkConfig{
  254. DefaultService: "test",
  255. })
  256. return network, httptest.NewServer(NewServer(network))
  257. }
  258. // TestHTTPNetwork tests interacting with a simulation network using the HTTP
  259. // API
  260. func TestHTTPNetwork(t *testing.T) {
  261. // start the server
  262. network, s := testHTTPServer(t)
  263. defer s.Close()
  264. // subscribe to events so we can check them later
  265. client := NewClient(s.URL)
  266. events := make(chan *Event, 100)
  267. var opts SubscribeOpts
  268. sub, err := client.SubscribeNetwork(events, opts)
  269. if err != nil {
  270. t.Fatalf("error subscribing to network events: %s", err)
  271. }
  272. defer sub.Unsubscribe()
  273. // check we can retrieve details about the network
  274. gotNetwork, err := client.GetNetwork()
  275. if err != nil {
  276. t.Fatalf("error getting network: %s", err)
  277. }
  278. if gotNetwork.ID != network.ID {
  279. t.Fatalf("expected network to have ID %q, got %q", network.ID, gotNetwork.ID)
  280. }
  281. // start a simulation network
  282. nodeIDs := startTestNetwork(t, client)
  283. // check we got all the events
  284. x := &expectEvents{t, events, sub}
  285. x.expect(
  286. x.nodeEvent(nodeIDs[0], false),
  287. x.nodeEvent(nodeIDs[1], false),
  288. x.nodeEvent(nodeIDs[0], true),
  289. x.nodeEvent(nodeIDs[1], true),
  290. x.connEvent(nodeIDs[0], nodeIDs[1], false),
  291. x.connEvent(nodeIDs[0], nodeIDs[1], true),
  292. )
  293. // reconnect the stream and check we get the current nodes and conns
  294. events = make(chan *Event, 100)
  295. opts.Current = true
  296. sub, err = client.SubscribeNetwork(events, opts)
  297. if err != nil {
  298. t.Fatalf("error subscribing to network events: %s", err)
  299. }
  300. defer sub.Unsubscribe()
  301. x = &expectEvents{t, events, sub}
  302. x.expect(
  303. x.nodeEvent(nodeIDs[0], true),
  304. x.nodeEvent(nodeIDs[1], true),
  305. x.connEvent(nodeIDs[0], nodeIDs[1], true),
  306. )
  307. }
  308. func startTestNetwork(t *testing.T, client *Client) []string {
  309. // create two nodes
  310. nodeCount := 2
  311. nodeIDs := make([]string, nodeCount)
  312. for i := 0; i < nodeCount; i++ {
  313. config := adapters.RandomNodeConfig()
  314. node, err := client.CreateNode(config)
  315. if err != nil {
  316. t.Fatalf("error creating node: %s", err)
  317. }
  318. nodeIDs[i] = node.ID
  319. }
  320. // check both nodes exist
  321. nodes, err := client.GetNodes()
  322. if err != nil {
  323. t.Fatalf("error getting nodes: %s", err)
  324. }
  325. if len(nodes) != nodeCount {
  326. t.Fatalf("expected %d nodes, got %d", nodeCount, len(nodes))
  327. }
  328. for i, nodeID := range nodeIDs {
  329. if nodes[i].ID != nodeID {
  330. t.Fatalf("expected node %d to have ID %q, got %q", i, nodeID, nodes[i].ID)
  331. }
  332. node, err := client.GetNode(nodeID)
  333. if err != nil {
  334. t.Fatalf("error getting node %d: %s", i, err)
  335. }
  336. if node.ID != nodeID {
  337. t.Fatalf("expected node %d to have ID %q, got %q", i, nodeID, node.ID)
  338. }
  339. }
  340. // start both nodes
  341. for _, nodeID := range nodeIDs {
  342. if err := client.StartNode(nodeID); err != nil {
  343. t.Fatalf("error starting node %q: %s", nodeID, err)
  344. }
  345. }
  346. // connect the nodes
  347. for i := 0; i < nodeCount-1; i++ {
  348. peerId := i + 1
  349. if i == nodeCount-1 {
  350. peerId = 0
  351. }
  352. if err := client.ConnectNode(nodeIDs[i], nodeIDs[peerId]); err != nil {
  353. t.Fatalf("error connecting nodes: %s", err)
  354. }
  355. }
  356. return nodeIDs
  357. }
  358. type expectEvents struct {
  359. *testing.T
  360. events chan *Event
  361. sub event.Subscription
  362. }
  363. func (t *expectEvents) nodeEvent(id string, up bool) *Event {
  364. return &Event{
  365. Type: EventTypeNode,
  366. Node: &Node{
  367. Config: &adapters.NodeConfig{
  368. ID: enode.HexID(id),
  369. },
  370. Up: up,
  371. },
  372. }
  373. }
  374. func (t *expectEvents) connEvent(one, other string, up bool) *Event {
  375. return &Event{
  376. Type: EventTypeConn,
  377. Conn: &Conn{
  378. One: enode.HexID(one),
  379. Other: enode.HexID(other),
  380. Up: up,
  381. },
  382. }
  383. }
  384. func (t *expectEvents) expectMsgs(expected map[MsgFilter]int) {
  385. actual := make(map[MsgFilter]int)
  386. timeout := time.After(10 * time.Second)
  387. loop:
  388. for {
  389. select {
  390. case event := <-t.events:
  391. t.Logf("received %s event: %s", event.Type, event)
  392. if event.Type != EventTypeMsg || event.Msg.Received {
  393. continue loop
  394. }
  395. if event.Msg == nil {
  396. t.Fatal("expected event.Msg to be set")
  397. }
  398. filter := MsgFilter{
  399. Proto: event.Msg.Protocol,
  400. Code: int64(event.Msg.Code),
  401. }
  402. actual[filter]++
  403. if actual[filter] > expected[filter] {
  404. t.Fatalf("received too many msgs for filter: %v", filter)
  405. }
  406. if reflect.DeepEqual(actual, expected) {
  407. return
  408. }
  409. case err := <-t.sub.Err():
  410. t.Fatalf("network stream closed unexpectedly: %s", err)
  411. case <-timeout:
  412. t.Fatal("timed out waiting for expected events")
  413. }
  414. }
  415. }
  416. func (t *expectEvents) expect(events ...*Event) {
  417. timeout := time.After(10 * time.Second)
  418. i := 0
  419. for {
  420. select {
  421. case event := <-t.events:
  422. t.Logf("received %s event: %s", event.Type, event)
  423. expected := events[i]
  424. if event.Type != expected.Type {
  425. t.Fatalf("expected event %d to have type %q, got %q", i, expected.Type, event.Type)
  426. }
  427. switch expected.Type {
  428. case EventTypeNode:
  429. if event.Node == nil {
  430. t.Fatal("expected event.Node to be set")
  431. }
  432. if event.Node.ID() != expected.Node.ID() {
  433. t.Fatalf("expected node event %d to have id %q, got %q", i, expected.Node.ID().TerminalString(), event.Node.ID().TerminalString())
  434. }
  435. if event.Node.Up != expected.Node.Up {
  436. t.Fatalf("expected node event %d to have up=%t, got up=%t", i, expected.Node.Up, event.Node.Up)
  437. }
  438. case EventTypeConn:
  439. if event.Conn == nil {
  440. t.Fatal("expected event.Conn to be set")
  441. }
  442. if event.Conn.One != expected.Conn.One {
  443. t.Fatalf("expected conn event %d to have one=%q, got one=%q", i, expected.Conn.One.TerminalString(), event.Conn.One.TerminalString())
  444. }
  445. if event.Conn.Other != expected.Conn.Other {
  446. t.Fatalf("expected conn event %d to have other=%q, got other=%q", i, expected.Conn.Other.TerminalString(), event.Conn.Other.TerminalString())
  447. }
  448. if event.Conn.Up != expected.Conn.Up {
  449. t.Fatalf("expected conn event %d to have up=%t, got up=%t", i, expected.Conn.Up, event.Conn.Up)
  450. }
  451. }
  452. i++
  453. if i == len(events) {
  454. return
  455. }
  456. case err := <-t.sub.Err():
  457. t.Fatalf("network stream closed unexpectedly: %s", err)
  458. case <-timeout:
  459. t.Fatal("timed out waiting for expected events")
  460. }
  461. }
  462. }
  463. // TestHTTPNodeRPC tests calling RPC methods on nodes via the HTTP API
  464. func TestHTTPNodeRPC(t *testing.T) {
  465. // start the server
  466. _, s := testHTTPServer(t)
  467. defer s.Close()
  468. // start a node in the network
  469. client := NewClient(s.URL)
  470. config := adapters.RandomNodeConfig()
  471. node, err := client.CreateNode(config)
  472. if err != nil {
  473. t.Fatalf("error creating node: %s", err)
  474. }
  475. if err := client.StartNode(node.ID); err != nil {
  476. t.Fatalf("error starting node: %s", err)
  477. }
  478. // create two RPC clients
  479. ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
  480. defer cancel()
  481. rpcClient1, err := client.RPCClient(ctx, node.ID)
  482. if err != nil {
  483. t.Fatalf("error getting node RPC client: %s", err)
  484. }
  485. rpcClient2, err := client.RPCClient(ctx, node.ID)
  486. if err != nil {
  487. t.Fatalf("error getting node RPC client: %s", err)
  488. }
  489. // subscribe to events using client 1
  490. events := make(chan int64, 1)
  491. sub, err := rpcClient1.Subscribe(ctx, "test", events, "events")
  492. if err != nil {
  493. t.Fatalf("error subscribing to events: %s", err)
  494. }
  495. defer sub.Unsubscribe()
  496. // call some RPC methods using client 2
  497. if err := rpcClient2.CallContext(ctx, nil, "test_add", 10); err != nil {
  498. t.Fatalf("error calling RPC method: %s", err)
  499. }
  500. var result int64
  501. if err := rpcClient2.CallContext(ctx, &result, "test_get"); err != nil {
  502. t.Fatalf("error calling RPC method: %s", err)
  503. }
  504. if result != 10 {
  505. t.Fatalf("expected result to be 10, got %d", result)
  506. }
  507. // check we got an event from client 1
  508. select {
  509. case event := <-events:
  510. if event != 10 {
  511. t.Fatalf("expected event to be 10, got %d", event)
  512. }
  513. case <-ctx.Done():
  514. t.Fatal(ctx.Err())
  515. }
  516. }
  517. // TestHTTPSnapshot tests creating and loading network snapshots
  518. func TestHTTPSnapshot(t *testing.T) {
  519. // start the server
  520. network, s := testHTTPServer(t)
  521. defer s.Close()
  522. var eventsDone = make(chan struct{})
  523. count := 1
  524. eventsDoneChan := make(chan *Event)
  525. eventSub := network.Events().Subscribe(eventsDoneChan)
  526. go func() {
  527. defer eventSub.Unsubscribe()
  528. for event := range eventsDoneChan {
  529. if event.Type == EventTypeConn && !event.Control {
  530. count--
  531. if count == 0 {
  532. eventsDone <- struct{}{}
  533. return
  534. }
  535. }
  536. }
  537. }()
  538. // create a two-node network
  539. client := NewClient(s.URL)
  540. nodeCount := 2
  541. nodes := make([]*p2p.NodeInfo, nodeCount)
  542. for i := 0; i < nodeCount; i++ {
  543. config := adapters.RandomNodeConfig()
  544. node, err := client.CreateNode(config)
  545. if err != nil {
  546. t.Fatalf("error creating node: %s", err)
  547. }
  548. if err := client.StartNode(node.ID); err != nil {
  549. t.Fatalf("error starting node: %s", err)
  550. }
  551. nodes[i] = node
  552. }
  553. if err := client.ConnectNode(nodes[0].ID, nodes[1].ID); err != nil {
  554. t.Fatalf("error connecting nodes: %s", err)
  555. }
  556. // store some state in the test services
  557. states := make([]string, nodeCount)
  558. for i, node := range nodes {
  559. rpc, err := client.RPCClient(context.Background(), node.ID)
  560. if err != nil {
  561. t.Fatalf("error getting RPC client: %s", err)
  562. }
  563. defer rpc.Close()
  564. state := fmt.Sprintf("%x", rand.Int())
  565. if err := rpc.Call(nil, "test_setState", []byte(state)); err != nil {
  566. t.Fatalf("error setting service state: %s", err)
  567. }
  568. states[i] = state
  569. }
  570. <-eventsDone
  571. // create a snapshot
  572. snap, err := client.CreateSnapshot()
  573. if err != nil {
  574. t.Fatalf("error creating snapshot: %s", err)
  575. }
  576. for i, state := range states {
  577. gotState := snap.Nodes[i].Snapshots["test"]
  578. if string(gotState) != state {
  579. t.Fatalf("expected snapshot state %q, got %q", state, gotState)
  580. }
  581. }
  582. // create another network
  583. network2, s := testHTTPServer(t)
  584. defer s.Close()
  585. client = NewClient(s.URL)
  586. count = 1
  587. eventSub = network2.Events().Subscribe(eventsDoneChan)
  588. go func() {
  589. defer eventSub.Unsubscribe()
  590. for event := range eventsDoneChan {
  591. if event.Type == EventTypeConn && !event.Control {
  592. count--
  593. if count == 0 {
  594. eventsDone <- struct{}{}
  595. return
  596. }
  597. }
  598. }
  599. }()
  600. // subscribe to events so we can check them later
  601. events := make(chan *Event, 100)
  602. var opts SubscribeOpts
  603. sub, err := client.SubscribeNetwork(events, opts)
  604. if err != nil {
  605. t.Fatalf("error subscribing to network events: %s", err)
  606. }
  607. defer sub.Unsubscribe()
  608. // load the snapshot
  609. if err := client.LoadSnapshot(snap); err != nil {
  610. t.Fatalf("error loading snapshot: %s", err)
  611. }
  612. <-eventsDone
  613. // check the nodes and connection exists
  614. net, err := client.GetNetwork()
  615. if err != nil {
  616. t.Fatalf("error getting network: %s", err)
  617. }
  618. if len(net.Nodes) != nodeCount {
  619. t.Fatalf("expected network to have %d nodes, got %d", nodeCount, len(net.Nodes))
  620. }
  621. for i, node := range nodes {
  622. id := net.Nodes[i].ID().String()
  623. if id != node.ID {
  624. t.Fatalf("expected node %d to have ID %s, got %s", i, node.ID, id)
  625. }
  626. }
  627. if len(net.Conns) != 1 {
  628. t.Fatalf("expected network to have 1 connection, got %d", len(net.Conns))
  629. }
  630. conn := net.Conns[0]
  631. if conn.One.String() != nodes[0].ID {
  632. t.Fatalf("expected connection to have one=%q, got one=%q", nodes[0].ID, conn.One)
  633. }
  634. if conn.Other.String() != nodes[1].ID {
  635. t.Fatalf("expected connection to have other=%q, got other=%q", nodes[1].ID, conn.Other)
  636. }
  637. if !conn.Up {
  638. t.Fatal("should be up")
  639. }
  640. // check the node states were restored
  641. for i, node := range nodes {
  642. rpc, err := client.RPCClient(context.Background(), node.ID)
  643. if err != nil {
  644. t.Fatalf("error getting RPC client: %s", err)
  645. }
  646. defer rpc.Close()
  647. var state []byte
  648. if err := rpc.Call(&state, "test_getState"); err != nil {
  649. t.Fatalf("error getting service state: %s", err)
  650. }
  651. if string(state) != states[i] {
  652. t.Fatalf("expected snapshot state %q, got %q", states[i], state)
  653. }
  654. }
  655. // check we got all the events
  656. x := &expectEvents{t, events, sub}
  657. x.expect(
  658. x.nodeEvent(nodes[0].ID, false),
  659. x.nodeEvent(nodes[0].ID, true),
  660. x.nodeEvent(nodes[1].ID, false),
  661. x.nodeEvent(nodes[1].ID, true),
  662. x.connEvent(nodes[0].ID, nodes[1].ID, false),
  663. x.connEvent(nodes[0].ID, nodes[1].ID, true),
  664. )
  665. }
  666. // TestMsgFilterPassMultiple tests streaming message events using a filter
  667. // with multiple protocols
  668. func TestMsgFilterPassMultiple(t *testing.T) {
  669. // start the server
  670. _, s := testHTTPServer(t)
  671. defer s.Close()
  672. // subscribe to events with a message filter
  673. client := NewClient(s.URL)
  674. events := make(chan *Event, 10)
  675. opts := SubscribeOpts{
  676. Filter: "prb:0-test:0",
  677. }
  678. sub, err := client.SubscribeNetwork(events, opts)
  679. if err != nil {
  680. t.Fatalf("error subscribing to network events: %s", err)
  681. }
  682. defer sub.Unsubscribe()
  683. // start a simulation network
  684. startTestNetwork(t, client)
  685. // check we got the expected events
  686. x := &expectEvents{t, events, sub}
  687. x.expectMsgs(map[MsgFilter]int{
  688. {"test", 0}: 2,
  689. {"prb", 0}: 2,
  690. })
  691. }
  692. // TestMsgFilterPassWildcard tests streaming message events using a filter
  693. // with a code wildcard
  694. func TestMsgFilterPassWildcard(t *testing.T) {
  695. // start the server
  696. _, s := testHTTPServer(t)
  697. defer s.Close()
  698. // subscribe to events with a message filter
  699. client := NewClient(s.URL)
  700. events := make(chan *Event, 10)
  701. opts := SubscribeOpts{
  702. Filter: "prb:0,2-test:*",
  703. }
  704. sub, err := client.SubscribeNetwork(events, opts)
  705. if err != nil {
  706. t.Fatalf("error subscribing to network events: %s", err)
  707. }
  708. defer sub.Unsubscribe()
  709. // start a simulation network
  710. startTestNetwork(t, client)
  711. // check we got the expected events
  712. x := &expectEvents{t, events, sub}
  713. x.expectMsgs(map[MsgFilter]int{
  714. {"test", 2}: 2,
  715. {"test", 1}: 2,
  716. {"test", 0}: 2,
  717. {"prb", 0}: 2,
  718. })
  719. }
  720. // TestMsgFilterPassSingle tests streaming message events using a filter
  721. // with a single protocol and code
  722. func TestMsgFilterPassSingle(t *testing.T) {
  723. // start the server
  724. _, s := testHTTPServer(t)
  725. defer s.Close()
  726. // subscribe to events with a message filter
  727. client := NewClient(s.URL)
  728. events := make(chan *Event, 10)
  729. opts := SubscribeOpts{
  730. Filter: "dum:0",
  731. }
  732. sub, err := client.SubscribeNetwork(events, opts)
  733. if err != nil {
  734. t.Fatalf("error subscribing to network events: %s", err)
  735. }
  736. defer sub.Unsubscribe()
  737. // start a simulation network
  738. startTestNetwork(t, client)
  739. // check we got the expected events
  740. x := &expectEvents{t, events, sub}
  741. x.expectMsgs(map[MsgFilter]int{
  742. {"dum", 0}: 2,
  743. })
  744. }
  745. // TestMsgFilterPassSingle tests streaming message events using an invalid
  746. // filter
  747. func TestMsgFilterFailBadParams(t *testing.T) {
  748. // start the server
  749. _, s := testHTTPServer(t)
  750. defer s.Close()
  751. client := NewClient(s.URL)
  752. events := make(chan *Event, 10)
  753. opts := SubscribeOpts{
  754. Filter: "foo:",
  755. }
  756. _, err := client.SubscribeNetwork(events, opts)
  757. if err == nil {
  758. t.Fatalf("expected event subscription to fail but succeeded!")
  759. }
  760. opts.Filter = "bzz:aa"
  761. _, err = client.SubscribeNetwork(events, opts)
  762. if err == nil {
  763. t.Fatalf("expected event subscription to fail but succeeded!")
  764. }
  765. opts.Filter = "invalid"
  766. _, err = client.SubscribeNetwork(events, opts)
  767. if err == nil {
  768. t.Fatalf("expected event subscription to fail but succeeded!")
  769. }
  770. }