| 73 | } |
| 74 | |
| 75 | func TestKVWatchAndDelete(t *testing.T) { |
| 76 | testKVs(t, func(t *testing.T, client kv.Client, reg *prometheus.Registry) { |
| 77 | ctx, cancel := context.WithCancel(context.Background()) |
| 78 | defer cancel() |
| 79 | |
| 80 | err := client.CAS(context.Background(), "key-before-watch", func(in interface{}) (out interface{}, retry bool, err error) { |
| 81 | return "value-before-watch", false, nil |
| 82 | }) |
| 83 | require.NoError(t, err) |
| 84 | |
| 85 | w := &watcher{} |
| 86 | wg := &sync.WaitGroup{} |
| 87 | wg.Add(1) |
| 88 | go func() { |
| 89 | defer wg.Done() |
| 90 | w.watch(ctx, client) |
| 91 | }() |
| 92 | |
| 93 | err = client.CAS(context.Background(), "key-to-delete", func(in interface{}) (out interface{}, retry bool, err error) { |
| 94 | return "value-to-delete", false, nil |
| 95 | }) |
| 96 | require.NoError(t, err, "object could not be created") |
| 97 | |
| 98 | // Give watcher time to receive notification. |
| 99 | time.Sleep(500 * time.Millisecond) |
| 100 | |
| 101 | // Now delete it |
| 102 | err = client.Delete(context.Background(), "key-to-delete") |
| 103 | require.NoError(t, err) |
| 104 | |
| 105 | // Give watcher time to receive notification for delete, if any. |
| 106 | time.Sleep(500 * time.Millisecond) |
| 107 | |
| 108 | // Stop the watcher |
| 109 | cancel() |
| 110 | wg.Wait() |
| 111 | |
| 112 | // Consul reports: |
| 113 | // map[key-before-watch:[value-before-watch] key-to-delete:[value-to-delete]] |
| 114 | // |
| 115 | // Etcd reports (before changing etcd client to ignore deletes): |
| 116 | // map[key-to-delete:[value-to-delete <nil>]] |
| 117 | t.Log(w.values) |
| 118 | }) |
| 119 | } |
| 120 | |
| 121 | func setupEtcd(t *testing.T, scenario *e2e.Scenario, reg prometheus.Registerer, logger log.Logger) kv.Client { |
| 122 | t.Helper() |