server_test.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903
  1. package etcdserver
  2. import (
  3. "fmt"
  4. "math/rand"
  5. "reflect"
  6. "sync"
  7. "testing"
  8. "time"
  9. pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
  10. "github.com/coreos/etcd/raft"
  11. "github.com/coreos/etcd/raft/raftpb"
  12. "github.com/coreos/etcd/store"
  13. "github.com/coreos/etcd/testutil"
  14. "github.com/coreos/etcd/third_party/code.google.com/p/go.net/context"
  15. )
  16. func TestGetExpirationTime(t *testing.T) {
  17. tests := []struct {
  18. r pb.Request
  19. want time.Time
  20. }{
  21. {
  22. pb.Request{Expiration: 0},
  23. time.Time{},
  24. },
  25. {
  26. pb.Request{Expiration: 60000},
  27. time.Unix(0, 60000),
  28. },
  29. {
  30. pb.Request{Expiration: -60000},
  31. time.Unix(0, -60000),
  32. },
  33. }
  34. for i, tt := range tests {
  35. got := getExpirationTime(&tt.r)
  36. if !reflect.DeepEqual(tt.want, got) {
  37. t.Errorf("#%d: incorrect expiration time: want=%v got=%v", i, tt.want, got)
  38. }
  39. }
  40. }
  41. // TestDoLocalAction tests requests which do not need to go through raft to be applied,
  42. // and are served through local data.
  43. func TestDoLocalAction(t *testing.T) {
  44. tests := []struct {
  45. req pb.Request
  46. wresp Response
  47. werr error
  48. waction []string
  49. }{
  50. {
  51. pb.Request{Method: "GET", Id: 1, Wait: true},
  52. Response{Watcher: &stubWatcher{}}, nil, []string{"Watch"},
  53. },
  54. {
  55. pb.Request{Method: "GET", Id: 1},
  56. Response{Event: &store.Event{}}, nil, []string{"Get"},
  57. },
  58. {
  59. pb.Request{Method: "BADMETHOD", Id: 1},
  60. Response{}, ErrUnknownMethod, []string{},
  61. },
  62. }
  63. for i, tt := range tests {
  64. st := &storeRecorder{}
  65. srv := &EtcdServer{Store: st}
  66. resp, err := srv.Do(context.TODO(), tt.req)
  67. if err != tt.werr {
  68. t.Fatalf("#%d: err = %+v, want %+v", i, err, tt.werr)
  69. }
  70. if !reflect.DeepEqual(resp, tt.wresp) {
  71. t.Errorf("#%d: resp = %+v, want %+v", i, resp, tt.wresp)
  72. }
  73. action := st.Action()
  74. if !reflect.DeepEqual(action, tt.waction) {
  75. t.Errorf("#%d: action = %+v, want %+v", i, action, tt.waction)
  76. }
  77. }
  78. }
  79. // TestDoBadLocalAction tests server requests which do not need to go through consensus,
  80. // and return errors when they fetch from local data.
  81. func TestDoBadLocalAction(t *testing.T) {
  82. storeErr := fmt.Errorf("bah")
  83. tests := []struct {
  84. req pb.Request
  85. waction []string
  86. }{
  87. {
  88. pb.Request{Method: "GET", Id: 1, Wait: true},
  89. []string{"Watch"},
  90. },
  91. {
  92. pb.Request{Method: "GET", Id: 1},
  93. []string{"Get"},
  94. },
  95. }
  96. for i, tt := range tests {
  97. st := &errStoreRecorder{err: storeErr}
  98. srv := &EtcdServer{Store: st}
  99. resp, err := srv.Do(context.Background(), tt.req)
  100. if err != storeErr {
  101. t.Fatalf("#%d: err = %+v, want %+v", i, err, storeErr)
  102. }
  103. if !reflect.DeepEqual(resp, Response{}) {
  104. t.Errorf("#%d: resp = %+v, want %+v", i, resp, Response{})
  105. }
  106. action := st.Action()
  107. if !reflect.DeepEqual(action, tt.waction) {
  108. t.Errorf("#%d: action = %+v, want %+v", i, action, tt.waction)
  109. }
  110. }
  111. }
  112. func TestApplyRequest(t *testing.T) {
  113. tests := []struct {
  114. req pb.Request
  115. wresp Response
  116. waction []string
  117. }{
  118. {
  119. pb.Request{Method: "POST", Id: 1},
  120. Response{Event: &store.Event{}}, []string{"Create"},
  121. },
  122. {
  123. pb.Request{Method: "PUT", Id: 1, PrevExist: boolp(true), PrevIndex: 1},
  124. Response{Event: &store.Event{}}, []string{"Update"},
  125. },
  126. {
  127. pb.Request{Method: "PUT", Id: 1, PrevExist: boolp(false), PrevIndex: 1},
  128. Response{Event: &store.Event{}}, []string{"Create"},
  129. },
  130. {
  131. pb.Request{Method: "PUT", Id: 1, PrevExist: boolp(true)},
  132. Response{Event: &store.Event{}}, []string{"Update"},
  133. },
  134. {
  135. pb.Request{Method: "PUT", Id: 1, PrevExist: boolp(false)},
  136. Response{Event: &store.Event{}}, []string{"Create"},
  137. },
  138. {
  139. pb.Request{Method: "PUT", Id: 1, PrevIndex: 1},
  140. Response{Event: &store.Event{}}, []string{"CompareAndSwap"},
  141. },
  142. {
  143. pb.Request{Method: "PUT", Id: 1, PrevValue: "bar"},
  144. Response{Event: &store.Event{}}, []string{"CompareAndSwap"},
  145. },
  146. {
  147. pb.Request{Method: "PUT", Id: 1},
  148. Response{Event: &store.Event{}}, []string{"Set"},
  149. },
  150. {
  151. pb.Request{Method: "DELETE", Id: 1, PrevIndex: 1},
  152. Response{Event: &store.Event{}}, []string{"CompareAndDelete"},
  153. },
  154. {
  155. pb.Request{Method: "DELETE", Id: 1, PrevValue: "bar"},
  156. Response{Event: &store.Event{}}, []string{"CompareAndDelete"},
  157. },
  158. {
  159. pb.Request{Method: "DELETE", Id: 1},
  160. Response{Event: &store.Event{}}, []string{"Delete"},
  161. },
  162. {
  163. pb.Request{Method: "QGET", Id: 1},
  164. Response{Event: &store.Event{}}, []string{"Get"},
  165. },
  166. {
  167. pb.Request{Method: "SYNC", Id: 1},
  168. Response{}, []string{"DeleteExpiredKeys"},
  169. },
  170. {
  171. pb.Request{Method: "BADMETHOD", Id: 1},
  172. Response{err: ErrUnknownMethod}, []string{},
  173. },
  174. }
  175. for i, tt := range tests {
  176. st := &storeRecorder{}
  177. srv := &EtcdServer{Store: st}
  178. resp := srv.applyRequest(tt.req)
  179. if !reflect.DeepEqual(resp, tt.wresp) {
  180. t.Errorf("#%d: resp = %+v, want %+v", i, resp, tt.wresp)
  181. }
  182. action := st.Action()
  183. if !reflect.DeepEqual(action, tt.waction) {
  184. t.Errorf("#%d: action = %+v, want %+v", i, action, tt.waction)
  185. }
  186. }
  187. }
  188. func TestClusterOf1(t *testing.T) { testServer(t, 1) }
  189. func TestClusterOf3(t *testing.T) { testServer(t, 3) }
  190. func testServer(t *testing.T, ns int64) {
  191. ctx, cancel := context.WithCancel(context.Background())
  192. defer cancel()
  193. ss := make([]*EtcdServer, ns)
  194. send := func(msgs []raftpb.Message) {
  195. for _, m := range msgs {
  196. t.Logf("m = %+v\n", m)
  197. ss[m.To-1].Node.Step(ctx, m)
  198. }
  199. }
  200. peers := make([]int64, ns)
  201. for i := int64(0); i < ns; i++ {
  202. peers[i] = i + 1
  203. }
  204. for i := int64(0); i < ns; i++ {
  205. id := i + 1
  206. n := raft.StartNode(id, peers, 10, 1)
  207. tk := time.NewTicker(10 * time.Millisecond)
  208. defer tk.Stop()
  209. srv := &EtcdServer{
  210. Node: n,
  211. Store: store.New(),
  212. Send: send,
  213. Storage: &storageRecorder{},
  214. Ticker: tk.C,
  215. }
  216. srv.Start()
  217. // TODO(xiangli): randomize election timeout
  218. // then remove this sleep.
  219. time.Sleep(1 * time.Millisecond)
  220. ss[i] = srv
  221. }
  222. for i := 1; i <= 10; i++ {
  223. r := pb.Request{
  224. Method: "PUT",
  225. Id: int64(i),
  226. Path: "/foo",
  227. Val: "bar",
  228. }
  229. j := rand.Intn(len(ss))
  230. t.Logf("ss = %d", j)
  231. resp, err := ss[j].Do(ctx, r)
  232. if err != nil {
  233. t.Fatal(err)
  234. }
  235. g, w := resp.Event.Node, &store.NodeExtern{
  236. Key: "/foo",
  237. ModifiedIndex: uint64(i),
  238. CreatedIndex: uint64(i),
  239. Value: stringp("bar"),
  240. }
  241. if !reflect.DeepEqual(g, w) {
  242. t.Error("value:", *g.Value)
  243. t.Errorf("g = %+v, w %+v", g, w)
  244. }
  245. }
  246. time.Sleep(10 * time.Millisecond)
  247. var last interface{}
  248. for i, sv := range ss {
  249. sv.Stop()
  250. g, _ := sv.Store.Get("/", true, true)
  251. if last != nil && !reflect.DeepEqual(last, g) {
  252. t.Errorf("server %d: Root = %#v, want %#v", i, g, last)
  253. }
  254. last = g
  255. }
  256. }
  257. func TestDoProposal(t *testing.T) {
  258. tests := []pb.Request{
  259. pb.Request{Method: "POST", Id: 1},
  260. pb.Request{Method: "PUT", Id: 1},
  261. pb.Request{Method: "DELETE", Id: 1},
  262. pb.Request{Method: "GET", Id: 1, Quorum: true},
  263. }
  264. for i, tt := range tests {
  265. ctx, _ := context.WithCancel(context.Background())
  266. n := raft.StartNode(0xBAD0, []int64{0xBAD0}, 10, 1)
  267. st := &storeRecorder{}
  268. tk := make(chan time.Time)
  269. // this makes <-tk always successful, which accelerates internal clock
  270. close(tk)
  271. srv := &EtcdServer{
  272. Node: n,
  273. Store: st,
  274. Send: func(_ []raftpb.Message) {},
  275. Storage: &storageRecorder{},
  276. Ticker: tk,
  277. }
  278. srv.Start()
  279. resp, err := srv.Do(ctx, tt)
  280. srv.Stop()
  281. action := st.Action()
  282. if len(action) != 1 {
  283. t.Errorf("#%d: len(action) = %d, want 1", i, len(action))
  284. }
  285. if err != nil {
  286. t.Fatalf("#%d: err = %v, want nil", i, err)
  287. }
  288. wresp := Response{Event: &store.Event{}}
  289. if !reflect.DeepEqual(resp, wresp) {
  290. t.Errorf("#%d: resp = %v, want %v", i, resp, wresp)
  291. }
  292. }
  293. }
  294. func TestDoProposalCancelled(t *testing.T) {
  295. ctx, cancel := context.WithCancel(context.Background())
  296. // node cannot make any progress because there are two nodes
  297. n := raft.StartNode(0xBAD0, []int64{0xBAD0, 0xBAD1}, 10, 1)
  298. st := &storeRecorder{}
  299. wait := &waitRecorder{}
  300. srv := &EtcdServer{
  301. // TODO: use fake node for better testability
  302. Node: n,
  303. Store: st,
  304. w: wait,
  305. }
  306. done := make(chan struct{})
  307. var err error
  308. go func() {
  309. _, err = srv.Do(ctx, pb.Request{Method: "PUT", Id: 1})
  310. close(done)
  311. }()
  312. cancel()
  313. <-done
  314. action := st.Action()
  315. if len(action) != 0 {
  316. t.Errorf("len(action) = %v, want 0", len(action))
  317. }
  318. if err != context.Canceled {
  319. t.Fatalf("err = %v, want %v", err, context.Canceled)
  320. }
  321. w := []string{"Register1", "Trigger1"}
  322. if !reflect.DeepEqual(wait.action, w) {
  323. t.Errorf("wait.action = %+v, want %+v", wait.action, w)
  324. }
  325. }
  326. func TestDoProposalStopped(t *testing.T) {
  327. ctx, cancel := context.WithCancel(context.Background())
  328. defer cancel()
  329. // node cannot make any progress because there are two nodes
  330. n := raft.StartNode(0xBAD0, []int64{0xBAD0, 0xBAD1}, 10, 1)
  331. st := &storeRecorder{}
  332. tk := make(chan time.Time)
  333. // this makes <-tk always successful, which accelarates internal clock
  334. close(tk)
  335. srv := &EtcdServer{
  336. // TODO: use fake node for better testability
  337. Node: n,
  338. Store: st,
  339. Send: func(_ []raftpb.Message) {},
  340. Storage: &storageRecorder{},
  341. Ticker: tk,
  342. }
  343. srv.Start()
  344. done := make(chan struct{})
  345. var err error
  346. go func() {
  347. _, err = srv.Do(ctx, pb.Request{Method: "PUT", Id: 1})
  348. close(done)
  349. }()
  350. srv.Stop()
  351. <-done
  352. action := st.Action()
  353. if len(action) != 0 {
  354. t.Errorf("len(action) = %v, want 0", len(action))
  355. }
  356. if err != ErrStopped {
  357. t.Errorf("err = %v, want %v", err, ErrStopped)
  358. }
  359. }
  360. // TestSync tests sync 1. is nonblocking 2. sends out SYNC request.
  361. func TestSync(t *testing.T) {
  362. n := &nodeProposeDataRecorder{}
  363. srv := &EtcdServer{
  364. Node: n,
  365. }
  366. start := time.Now()
  367. srv.sync(defaultSyncTimeout)
  368. // check that sync is non-blocking
  369. if d := time.Since(start); d > time.Millisecond {
  370. t.Errorf("CallSyncTime = %v, want < %v", d, time.Millisecond)
  371. }
  372. testutil.ForceGosched()
  373. data := n.data()
  374. if len(data) != 1 {
  375. t.Fatalf("len(proposeData) = %d, want 1", len(data))
  376. }
  377. var r pb.Request
  378. if err := r.Unmarshal(data[0]); err != nil {
  379. t.Fatalf("unmarshal request error: %v", err)
  380. }
  381. if r.Method != "SYNC" {
  382. t.Errorf("method = %s, want SYNC", r.Method)
  383. }
  384. }
  385. // TestSyncTimeout tests the case that sync 1. is non-blocking 2. cancel request
  386. // after timeout
  387. func TestSyncTimeout(t *testing.T) {
  388. n := &nodeProposalBlockerRecorder{}
  389. srv := &EtcdServer{
  390. Node: n,
  391. }
  392. start := time.Now()
  393. srv.sync(0)
  394. // check that sync is non-blocking
  395. if d := time.Since(start); d > time.Millisecond {
  396. t.Errorf("CallSyncTime = %v, want < %v", d, time.Millisecond)
  397. }
  398. // give time for goroutine in sync to cancel
  399. // TODO: use fake clock
  400. testutil.ForceGosched()
  401. w := []string{"Propose blocked"}
  402. if g := n.Action(); !reflect.DeepEqual(g, w) {
  403. t.Errorf("action = %v, want %v", g, w)
  404. }
  405. }
  406. // TODO: TestNoSyncWhenNoLeader
  407. func TestSyncTriggerDeleteExpriedKeys(t *testing.T) {
  408. n := raft.StartNode(0xBAD0, []int64{0xBAD0}, 10, 1)
  409. n.Campaign(context.TODO())
  410. st := &storeRecorder{}
  411. srv := &EtcdServer{
  412. // TODO: use fake node for better testability
  413. Node: n,
  414. Store: st,
  415. Send: func(_ []raftpb.Message) {},
  416. Storage: &storageRecorder{},
  417. SyncTicker: time.After(0),
  418. }
  419. srv.Start()
  420. // give time for sync request to be proposed and performed
  421. // TODO: use fake clock
  422. testutil.ForceGosched()
  423. srv.Stop()
  424. action := st.Action()
  425. if len(action) != 1 {
  426. t.Fatalf("len(action) = %d, want 1", len(action))
  427. }
  428. if action[0] != "DeleteExpiredKeys" {
  429. t.Errorf("action = %s, want DeleteExpiredKeys", action[0])
  430. }
  431. }
  432. // snapshot should snapshot the store and cut the persistent
  433. // TODO: node.Compact is called... we need to make the node an interface
  434. func TestSnapshot(t *testing.T) {
  435. n := raft.StartNode(0xBAD0, []int64{0xBAD0}, 10, 1)
  436. defer n.Stop()
  437. st := &storeRecorder{}
  438. p := &storageRecorder{}
  439. s := &EtcdServer{
  440. Store: st,
  441. Storage: p,
  442. Node: n,
  443. }
  444. s.snapshot()
  445. action := st.Action()
  446. if len(action) != 1 {
  447. t.Fatalf("len(action) = %d, want 1", len(action))
  448. }
  449. if action[0] != "Save" {
  450. t.Errorf("action = %s, want Save", action[0])
  451. }
  452. action = p.Action()
  453. if len(action) != 1 {
  454. t.Fatalf("len(action) = %d, want 1", len(action))
  455. }
  456. if action[0] != "Cut" {
  457. t.Errorf("action = %s, want Cut", action[0])
  458. }
  459. }
  460. // Applied > SnapCount should trigger a SaveSnap event
  461. func TestTriggerSnap(t *testing.T) {
  462. ctx := context.Background()
  463. n := raft.StartNode(0xBAD0, []int64{0xBAD0}, 10, 1)
  464. n.Campaign(ctx)
  465. st := &storeRecorder{}
  466. p := &storageRecorder{}
  467. s := &EtcdServer{
  468. Store: st,
  469. Send: func(_ []raftpb.Message) {},
  470. Storage: p,
  471. Node: n,
  472. SnapCount: 10,
  473. }
  474. s.Start()
  475. for i := 0; int64(i) < s.SnapCount; i++ {
  476. s.Do(ctx, pb.Request{Method: "PUT", Id: 1})
  477. }
  478. time.Sleep(time.Millisecond)
  479. s.Stop()
  480. action := p.Action()
  481. // each operation is recorded as a Save
  482. // Nop + SnapCount * Puts + Cut + SaveSnap = Save + SnapCount * Save + Cut + SaveSnap
  483. if len(action) != 3+int(s.SnapCount) {
  484. t.Fatalf("len(action) = %d, want %d", len(action), 3+int(s.SnapCount))
  485. }
  486. if action[12] != "SaveSnap" {
  487. t.Errorf("action = %s, want SaveSnap", action[12])
  488. }
  489. }
  490. // TestRecvSnapshot tests when it receives a snapshot from raft leader,
  491. // it should trigger storage.SaveSnap and also store.Recover.
  492. func TestRecvSnapshot(t *testing.T) {
  493. n := newReadyNode()
  494. st := &storeRecorder{}
  495. p := &storageRecorder{}
  496. s := &EtcdServer{
  497. Store: st,
  498. Send: func(_ []raftpb.Message) {},
  499. Storage: p,
  500. Node: n,
  501. }
  502. s.Start()
  503. n.readyc <- raft.Ready{Snapshot: raftpb.Snapshot{Index: 1}}
  504. // make goroutines move forward to receive snapshot
  505. testutil.ForceGosched()
  506. s.Stop()
  507. waction := []string{"Recovery"}
  508. if g := st.Action(); !reflect.DeepEqual(g, waction) {
  509. t.Errorf("store action = %v, want %v", g, waction)
  510. }
  511. waction = []string{"Save", "SaveSnap"}
  512. if g := p.Action(); !reflect.DeepEqual(g, waction) {
  513. t.Errorf("storage action = %v, want %v", g, waction)
  514. }
  515. }
  516. // TestRecvSlowSnapshot tests that slow snapshot will not be applied
  517. // to store.
  518. func TestRecvSlowSnapshot(t *testing.T) {
  519. n := newReadyNode()
  520. st := &storeRecorder{}
  521. s := &EtcdServer{
  522. Store: st,
  523. Send: func(_ []raftpb.Message) {},
  524. Storage: &storageRecorder{},
  525. Node: n,
  526. }
  527. s.Start()
  528. n.readyc <- raft.Ready{Snapshot: raftpb.Snapshot{Index: 1}}
  529. // make goroutines move forward to receive snapshot
  530. testutil.ForceGosched()
  531. action := st.Action()
  532. n.readyc <- raft.Ready{Snapshot: raftpb.Snapshot{Index: 1}}
  533. // make goroutines move forward to receive snapshot
  534. testutil.ForceGosched()
  535. s.Stop()
  536. if g := st.Action(); !reflect.DeepEqual(g, action) {
  537. t.Errorf("store action = %v, want %v", g, action)
  538. }
  539. }
  540. // TestAddNode tests AddNode could propose configuration and add node to raft.
  541. func TestAddNode(t *testing.T) {
  542. n := newNodeCommitterRecorder()
  543. s := &EtcdServer{
  544. Node: n,
  545. Store: &storeRecorder{},
  546. Send: func(_ []raftpb.Message) {},
  547. Storage: &storageRecorder{},
  548. }
  549. s.Start()
  550. s.AddNode(context.TODO(), 1, []byte("foo"))
  551. action := n.Action()
  552. s.Stop()
  553. waction := []string{"Configure", "AddNode"}
  554. if !reflect.DeepEqual(action, waction) {
  555. t.Errorf("action = %v, want %v", action, waction)
  556. }
  557. }
  558. // TestRemoveNode tests RemoveNode could propose configuration and remove node from raft.
  559. func TestRemoveNode(t *testing.T) {
  560. n := newNodeCommitterRecorder()
  561. s := &EtcdServer{
  562. Node: n,
  563. Store: &storeRecorder{},
  564. Send: func(_ []raftpb.Message) {},
  565. Storage: &storageRecorder{},
  566. }
  567. s.Start()
  568. s.RemoveNode(context.TODO(), 1)
  569. action := n.Action()
  570. s.Stop()
  571. waction := []string{"Configure", "RemoveNode"}
  572. if !reflect.DeepEqual(action, waction) {
  573. t.Errorf("action = %v, want %v", action, waction)
  574. }
  575. }
  576. // TODO: test wait trigger correctness in multi-server case
  577. func TestGetBool(t *testing.T) {
  578. tests := []struct {
  579. b *bool
  580. wb bool
  581. wset bool
  582. }{
  583. {nil, false, false},
  584. {boolp(true), true, true},
  585. {boolp(false), false, true},
  586. }
  587. for i, tt := range tests {
  588. b, set := getBool(tt.b)
  589. if b != tt.wb {
  590. t.Errorf("#%d: value = %v, want %v", i, b, tt.wb)
  591. }
  592. if set != tt.wset {
  593. t.Errorf("#%d: set = %v, want %v", i, set, tt.wset)
  594. }
  595. }
  596. }
  597. func TestGenID(t *testing.T) {
  598. // Sanity check that the GenID function has been seeded appropriately
  599. // (math/rand is seeded with 1 by default)
  600. r := rand.NewSource(int64(1))
  601. var n int64
  602. for n == 0 {
  603. n = r.Int63()
  604. }
  605. if n == GenID() {
  606. t.Fatalf("GenID's rand seeded with 1!")
  607. }
  608. }
  609. type recorder struct {
  610. sync.Mutex
  611. action []string
  612. }
  613. func (r *recorder) record(action string) {
  614. r.Lock()
  615. r.action = append(r.action, action)
  616. r.Unlock()
  617. }
  618. func (r *recorder) Action() []string {
  619. r.Lock()
  620. cpy := make([]string, len(r.action))
  621. copy(cpy, r.action)
  622. r.Unlock()
  623. return cpy
  624. }
  625. type storeRecorder struct {
  626. recorder
  627. }
  628. func (s *storeRecorder) Version() int { return 0 }
  629. func (s *storeRecorder) Index() uint64 { return 0 }
  630. func (s *storeRecorder) Get(_ string, _, _ bool) (*store.Event, error) {
  631. s.record("Get")
  632. return &store.Event{}, nil
  633. }
  634. func (s *storeRecorder) Set(_ string, _ bool, _ string, _ time.Time) (*store.Event, error) {
  635. s.record("Set")
  636. return &store.Event{}, nil
  637. }
  638. func (s *storeRecorder) Update(_, _ string, _ time.Time) (*store.Event, error) {
  639. s.record("Update")
  640. return &store.Event{}, nil
  641. }
  642. func (s *storeRecorder) Create(_ string, _ bool, _ string, _ bool, _ time.Time) (*store.Event, error) {
  643. s.record("Create")
  644. return &store.Event{}, nil
  645. }
  646. func (s *storeRecorder) CompareAndSwap(_, _ string, _ uint64, _ string, _ time.Time) (*store.Event, error) {
  647. s.record("CompareAndSwap")
  648. return &store.Event{}, nil
  649. }
  650. func (s *storeRecorder) Delete(_ string, _, _ bool) (*store.Event, error) {
  651. s.record("Delete")
  652. return &store.Event{}, nil
  653. }
  654. func (s *storeRecorder) CompareAndDelete(_, _ string, _ uint64) (*store.Event, error) {
  655. s.record("CompareAndDelete")
  656. return &store.Event{}, nil
  657. }
  658. func (s *storeRecorder) Watch(_ string, _, _ bool, _ uint64) (store.Watcher, error) {
  659. s.record("Watch")
  660. return &stubWatcher{}, nil
  661. }
  662. func (s *storeRecorder) Save() ([]byte, error) {
  663. s.record("Save")
  664. return nil, nil
  665. }
  666. func (s *storeRecorder) Recovery(b []byte) error {
  667. s.record("Recovery")
  668. return nil
  669. }
  670. func (s *storeRecorder) TotalTransactions() uint64 { return 0 }
  671. func (s *storeRecorder) JsonStats() []byte { return nil }
  672. func (s *storeRecorder) DeleteExpiredKeys(cutoff time.Time) {
  673. s.record("DeleteExpiredKeys")
  674. }
  675. type stubWatcher struct{}
  676. func (w *stubWatcher) EventChan() chan *store.Event { return nil }
  677. func (w *stubWatcher) Remove() {}
  678. // errStoreRecorder returns an store error on Get, Watch request
  679. type errStoreRecorder struct {
  680. storeRecorder
  681. err error
  682. }
  683. func (s *errStoreRecorder) Get(_ string, _, _ bool) (*store.Event, error) {
  684. s.record("Get")
  685. return nil, s.err
  686. }
  687. func (s *errStoreRecorder) Watch(_ string, _, _ bool, _ uint64) (store.Watcher, error) {
  688. s.record("Watch")
  689. return nil, s.err
  690. }
  691. type waitRecorder struct {
  692. action []string
  693. }
  694. func (w *waitRecorder) Register(id int64) <-chan interface{} {
  695. w.action = append(w.action, fmt.Sprint("Register", id))
  696. return nil
  697. }
  698. func (w *waitRecorder) Trigger(id int64, x interface{}) {
  699. w.action = append(w.action, fmt.Sprint("Trigger", id))
  700. }
  701. func boolp(b bool) *bool { return &b }
  702. func stringp(s string) *string { return &s }
  703. type storageRecorder struct {
  704. recorder
  705. }
  706. func (p *storageRecorder) Save(st raftpb.HardState, ents []raftpb.Entry) {
  707. p.record("Save")
  708. }
  709. func (p *storageRecorder) Cut() error {
  710. p.record("Cut")
  711. return nil
  712. }
  713. func (p *storageRecorder) SaveSnap(st raftpb.Snapshot) {
  714. if raft.IsEmptySnap(st) {
  715. return
  716. }
  717. p.record("SaveSnap")
  718. }
  719. type readyNode struct {
  720. readyc chan raft.Ready
  721. }
  722. func newReadyNode() *readyNode {
  723. readyc := make(chan raft.Ready, 1)
  724. return &readyNode{readyc: readyc}
  725. }
  726. func (n *readyNode) Tick() {}
  727. func (n *readyNode) Campaign(ctx context.Context) error { return nil }
  728. func (n *readyNode) Propose(ctx context.Context, data []byte) error { return nil }
  729. func (n *readyNode) Configure(ctx context.Context, data []byte) error { return nil }
  730. func (n *readyNode) Step(ctx context.Context, msg raftpb.Message) error { return nil }
  731. func (n *readyNode) Ready() <-chan raft.Ready { return n.readyc }
  732. func (n *readyNode) Stop() {}
  733. func (n *readyNode) Compact(d []byte) {}
  734. func (n *readyNode) AddNode(id int64) {}
  735. func (n *readyNode) RemoveNode(id int64) {}
  736. type nodeRecorder struct {
  737. recorder
  738. }
  739. func (n *nodeRecorder) Tick() {
  740. n.record("Tick")
  741. }
  742. func (n *nodeRecorder) Campaign(ctx context.Context) error {
  743. n.record("Campaign")
  744. return nil
  745. }
  746. func (n *nodeRecorder) Propose(ctx context.Context, data []byte) error {
  747. n.record("Propose")
  748. return nil
  749. }
  750. func (n *nodeRecorder) Configure(ctx context.Context, data []byte) error {
  751. n.record("Configure")
  752. return nil
  753. }
  754. func (n *nodeRecorder) Step(ctx context.Context, msg raftpb.Message) error {
  755. n.record("Step")
  756. return nil
  757. }
  758. func (n *nodeRecorder) Ready() <-chan raft.Ready { return nil }
  759. func (n *nodeRecorder) Stop() {
  760. n.record("Stop")
  761. }
  762. func (n *nodeRecorder) Compact(d []byte) {
  763. n.record("Compact")
  764. }
  765. func (n *nodeRecorder) AddNode(id int64) {
  766. n.record("AddNode")
  767. }
  768. func (n *nodeRecorder) RemoveNode(id int64) {
  769. n.record("RemoveNode")
  770. }
  771. type nodeProposeDataRecorder struct {
  772. nodeRecorder
  773. sync.Mutex
  774. d [][]byte
  775. }
  776. func (n *nodeProposeDataRecorder) data() [][]byte {
  777. n.Lock()
  778. d := n.d
  779. n.Unlock()
  780. return d
  781. }
  782. func (n *nodeProposeDataRecorder) Propose(ctx context.Context, data []byte) error {
  783. n.nodeRecorder.Propose(ctx, data)
  784. n.Lock()
  785. n.d = append(n.d, data)
  786. n.Unlock()
  787. return nil
  788. }
  789. type nodeProposalBlockerRecorder struct {
  790. nodeRecorder
  791. }
  792. func (n *nodeProposalBlockerRecorder) Propose(ctx context.Context, data []byte) error {
  793. <-ctx.Done()
  794. n.record("Propose blocked")
  795. return nil
  796. }
  797. type nodeCommitterRecorder struct {
  798. nodeRecorder
  799. readyc chan raft.Ready
  800. }
  801. func newNodeCommitterRecorder() *nodeCommitterRecorder {
  802. readyc := make(chan raft.Ready, 1)
  803. readyc <- raft.Ready{SoftState: &raft.SoftState{RaftState: raft.StateLeader}}
  804. return &nodeCommitterRecorder{readyc: readyc}
  805. }
  806. func (n *nodeCommitterRecorder) Propose(ctx context.Context, data []byte) error {
  807. n.readyc <- raft.Ready{CommittedEntries: []raftpb.Entry{{Data: data}}}
  808. return n.nodeRecorder.Propose(ctx, data)
  809. }
  810. func (n *nodeCommitterRecorder) Configure(ctx context.Context, data []byte) error {
  811. n.readyc <- raft.Ready{CommittedEntries: []raftpb.Entry{{Type: raftpb.EntryConfig, Data: data}}}
  812. return n.nodeRecorder.Configure(ctx, data)
  813. }
  814. func (n *nodeCommitterRecorder) Ready() <-chan raft.Ready {
  815. return n.readyc
  816. }