events_test.go 2.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120
  1. // +build travis
  2. package gocql
  3. import (
  4. "github.com/gocql/gocql/ccm_test"
  5. "log"
  6. "testing"
  7. "time"
  8. )
  9. func TestEventDiscovery(t *testing.T) {
  10. if err := ccm.AllUp(); err != nil {
  11. t.Fatal(err)
  12. }
  13. session := createSession(t)
  14. defer session.Close()
  15. status, err := ccm.Status()
  16. if err != nil {
  17. t.Fatal(err)
  18. }
  19. log.Printf("status=%+v\n", status)
  20. session.pool.mu.RLock()
  21. poolHosts := session.pool.hostConnPools // TODO: replace with session.ring
  22. log.Printf("poolhosts=%+v\n", poolHosts)
  23. // check we discovered all the nodes in the ring
  24. for _, host := range status {
  25. if _, ok := poolHosts[host.Addr]; !ok {
  26. t.Errorf("did not discover %q", host.Addr)
  27. }
  28. }
  29. session.pool.mu.RUnlock()
  30. if t.Failed() {
  31. t.FailNow()
  32. }
  33. }
  34. func TestEventNodeDown(t *testing.T) {
  35. const targetNode = "node1"
  36. if err := ccm.AllUp(); err != nil {
  37. t.Fatal(err)
  38. }
  39. session := createSession(t)
  40. defer session.Close()
  41. if err := ccm.NodeDown(targetNode); err != nil {
  42. t.Fatal(err)
  43. }
  44. log.Println("down")
  45. status, err := ccm.Status()
  46. if err != nil {
  47. t.Fatal(err)
  48. }
  49. log.Printf("status=%+v\n", status)
  50. time.Sleep(5 * time.Second)
  51. session.pool.mu.RLock()
  52. defer session.pool.mu.RUnlock()
  53. poolHosts := session.pool.hostConnPools
  54. node := status[targetNode]
  55. log.Printf("poolhosts=%+v\n", poolHosts)
  56. if _, ok := poolHosts[node.Addr]; ok {
  57. t.Fatal("node not removed after remove event")
  58. }
  59. }
  60. func TestEventNodeUp(t *testing.T) {
  61. if err := ccm.AllUp(); err != nil {
  62. t.Fatal(err)
  63. }
  64. status, err := ccm.Status()
  65. if err != nil {
  66. t.Fatal(err)
  67. }
  68. log.Printf("status=%+v\n", status)
  69. session := createSession(t)
  70. defer session.Close()
  71. if err := ccm.NodeDown("node1"); err != nil {
  72. t.Fatal(err)
  73. }
  74. time.Sleep(5 * time.Second)
  75. session.pool.mu.RLock()
  76. poolHosts := session.pool.hostConnPools
  77. log.Printf("poolhosts=%+v\n", poolHosts)
  78. node1 := status["node1"]
  79. if _, ok := poolHosts[node1.Addr]; ok {
  80. session.pool.mu.RUnlock()
  81. t.Fatal("node1 not removed after remove event")
  82. }
  83. session.pool.mu.RUnlock()
  84. if err := ccm.NodeUp("node1"); err != nil {
  85. t.Fatal(err)
  86. }
  87. time.Sleep(5 * time.Second)
  88. session.pool.mu.RLock()
  89. log.Printf("poolhosts=%+v\n", poolHosts)
  90. if _, ok := poolHosts[node1.Addr]; !ok {
  91. session.pool.mu.RUnlock()
  92. t.Fatal("node1 not added after node added event")
  93. }
  94. session.pool.mu.RUnlock()
  95. }