open-consul/api/watch/funcs_test.go

894 lines
17 KiB
Go
Raw Normal View History

package watch_test
2014-08-20 22:22:22 +00:00
import (
"bytes"
"crypto/ecdsa"
"crypto/elliptic"
"crypto/rand"
"crypto/sha256"
"crypto/x509"
"crypto/x509/pkix"
"encoding/json"
"encoding/pem"
"errors"
"fmt"
"math/big"
"net/url"
"strings"
"sync"
2014-08-20 22:22:22 +00:00
"testing"
"time"
"github.com/hashicorp/consul/api"
"github.com/hashicorp/consul/api/watch"
"github.com/hashicorp/consul/sdk/testutil"
"github.com/stretchr/testify/require"
2014-08-20 22:22:22 +00:00
)
func makeClient(t *testing.T) (*api.Client, *testutil.TestServer) {
// Make client config
conf := api.DefaultConfig()
// Create server
server, err := testutil.NewTestServerConfigT(t, nil)
require.NoError(t, err)
conf.Address = server.HTTPAddr
// Create client
client, err := api.NewClient(conf)
if err != nil {
server.Stop()
// guaranteed to fail but will be a nice error message
require.NoError(t, err)
}
return client, server
}
var errBadContent = errors.New("bad content")
var errTimeout = errors.New("timeout")
var timeout = 5 * time.Second
func makeInvokeCh() chan error {
ch := make(chan error)
time.AfterFunc(timeout, func() { ch <- errTimeout })
return ch
}
func updateConnectCA(t *testing.T, client *api.Client) {
t.Helper()
connect := client.Connect()
cfg, _, err := connect.CAGetConfig(nil)
require.NoError(t, err)
require.NotNil(t, cfg.Config)
// update the cert
// Create the private key we'll use for this CA cert.
pk, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
require.NoError(t, err)
bs, err := x509.MarshalECPrivateKey(pk)
require.NoError(t, err)
var buf bytes.Buffer
require.NoError(t, pem.Encode(&buf, &pem.Block{Type: "EC PRIVATE KEY", Bytes: bs}))
cfg.Config["PrivateKey"] = buf.String()
bs, err = x509.MarshalPKIXPublicKey(pk.Public())
require.NoError(t, err)
kID := []byte(strings.Replace(fmt.Sprintf("% x", sha256.Sum256(bs)), " ", ":", -1))
// Create the CA cert
template := x509.Certificate{
SerialNumber: big.NewInt(42),
Subject: pkix.Name{CommonName: "CA Modified"},
URIs: []*url.URL{&url.URL{Scheme: "spiffe", Host: fmt.Sprintf("11111111-2222-3333-4444-555555555555.%s", "consul")}},
BasicConstraintsValid: true,
KeyUsage: x509.KeyUsageCertSign |
x509.KeyUsageCRLSign |
x509.KeyUsageDigitalSignature,
IsCA: true,
NotAfter: time.Now().AddDate(10, 0, 0),
NotBefore: time.Now(),
AuthorityKeyId: kID,
SubjectKeyId: kID,
}
bs, err = x509.CreateCertificate(rand.Reader, &template, &template, pk.Public(), pk)
require.NoError(t, err)
buf.Reset()
require.NoError(t, pem.Encode(&buf, &pem.Block{Type: "CERTIFICATE", Bytes: bs}))
cfg.Config["RootCert"] = buf.String()
_, err = connect.CASetConfig(cfg, nil)
require.NoError(t, err)
}
2014-08-20 22:22:22 +00:00
func TestKeyWatch(t *testing.T) {
2017-10-24 13:24:57 +00:00
t.Parallel()
c, s := makeClient(t)
defer s.Stop()
invoke := makeInvokeCh()
2014-08-21 18:38:44 +00:00
plan := mustParse(t, `{"type":"key", "key":"foo/bar/baz"}`)
plan.Handler = func(idx uint64, raw interface{}) {
if raw == nil {
return // ignore
}
v, ok := raw.(*api.KVPair)
if !ok || v == nil {
return // ignore
}
if string(v.Value) != "test" {
invoke <- errBadContent
return
2014-08-20 22:22:22 +00:00
}
invoke <- nil
2014-08-20 22:22:22 +00:00
}
var wg sync.WaitGroup
wg.Add(1)
2014-08-20 22:22:22 +00:00
go func() {
defer wg.Done()
kv := c.KV()
time.Sleep(20 * time.Millisecond)
pair := &api.KVPair{
2014-08-20 22:22:22 +00:00
Key: "foo/bar/baz",
Value: []byte("test"),
}
if _, err := kv.Put(pair, nil); err != nil {
2014-08-20 22:22:22 +00:00
t.Fatalf("err: %v", err)
}
}()
2014-08-20 22:22:22 +00:00
wg.Add(1)
go func() {
defer wg.Done()
if err := plan.Run(s.HTTPAddr); err != nil {
2014-08-20 22:22:22 +00:00
t.Fatalf("err: %v", err)
}
}()
if err := <-invoke; err != nil {
2014-08-20 22:22:22 +00:00
t.Fatalf("err: %v", err)
}
plan.Stop()
wg.Wait()
2014-08-20 22:22:22 +00:00
}
func TestKeyWatch_With_PrefixDelete(t *testing.T) {
2017-10-24 13:24:57 +00:00
t.Parallel()
c, s := makeClient(t)
defer s.Stop()
invoke := makeInvokeCh()
plan := mustParse(t, `{"type":"key", "key":"foo/bar/baz"}`)
plan.Handler = func(idx uint64, raw interface{}) {
if raw == nil {
return // ignore
}
v, ok := raw.(*api.KVPair)
if !ok || v == nil {
return // ignore
}
if string(v.Value) != "test" {
invoke <- errBadContent
return
}
invoke <- nil
}
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
kv := c.KV()
time.Sleep(20 * time.Millisecond)
pair := &api.KVPair{
Key: "foo/bar/baz",
Value: []byte("test"),
}
if _, err := kv.Put(pair, nil); err != nil {
t.Fatalf("err: %v", err)
}
}()
wg.Add(1)
go func() {
defer wg.Done()
if err := plan.Run(s.HTTPAddr); err != nil {
t.Fatalf("err: %v", err)
}
}()
if err := <-invoke; err != nil {
t.Fatalf("err: %v", err)
}
plan.Stop()
wg.Wait()
}
2014-08-20 22:22:22 +00:00
func TestKeyPrefixWatch(t *testing.T) {
2017-10-24 13:24:57 +00:00
t.Parallel()
c, s := makeClient(t)
defer s.Stop()
invoke := makeInvokeCh()
2014-08-21 18:38:44 +00:00
plan := mustParse(t, `{"type":"keyprefix", "prefix":"foo/"}`)
plan.Handler = func(idx uint64, raw interface{}) {
if raw == nil {
return // ignore
}
v, ok := raw.(api.KVPairs)
if !ok || len(v) == 0 {
return
2014-08-20 22:22:22 +00:00
}
if string(v[0].Key) != "foo/bar" {
invoke <- errBadContent
return
}
invoke <- nil
2014-08-20 22:22:22 +00:00
}
var wg sync.WaitGroup
wg.Add(1)
2014-08-20 22:22:22 +00:00
go func() {
defer wg.Done()
kv := c.KV()
time.Sleep(20 * time.Millisecond)
pair := &api.KVPair{
2014-08-20 22:22:22 +00:00
Key: "foo/bar",
}
if _, err := kv.Put(pair, nil); err != nil {
2014-08-20 22:22:22 +00:00
t.Fatalf("err: %v", err)
}
}()
2014-08-20 22:22:22 +00:00
wg.Add(1)
go func() {
defer wg.Done()
if err := plan.Run(s.HTTPAddr); err != nil {
2014-08-20 22:22:22 +00:00
t.Fatalf("err: %v", err)
}
}()
if err := <-invoke; err != nil {
2014-08-20 22:22:22 +00:00
t.Fatalf("err: %v", err)
}
plan.Stop()
wg.Wait()
2014-08-20 22:22:22 +00:00
}
2014-08-20 22:29:31 +00:00
func TestServicesWatch(t *testing.T) {
2017-10-24 13:24:57 +00:00
t.Parallel()
c, s := makeClient(t)
defer s.Stop()
invoke := makeInvokeCh()
2014-08-21 18:38:44 +00:00
plan := mustParse(t, `{"type":"services"}`)
plan.Handler = func(idx uint64, raw interface{}) {
if raw == nil {
return // ignore
}
v, ok := raw.(map[string][]string)
if !ok || len(v) == 0 {
return // ignore
}
if v["consul"] == nil {
invoke <- errBadContent
return
2014-08-20 22:29:31 +00:00
}
invoke <- nil
2014-08-20 22:29:31 +00:00
}
var wg sync.WaitGroup
wg.Add(1)
2014-08-20 22:29:31 +00:00
go func() {
defer wg.Done()
agent := c.Agent()
time.Sleep(20 * time.Millisecond)
reg := &api.AgentServiceRegistration{
2014-08-20 22:29:31 +00:00
ID: "foo",
Name: "foo",
}
if err := agent.ServiceRegister(reg); err != nil {
t.Fatalf("err: %v", err)
}
2014-08-20 22:29:31 +00:00
}()
wg.Add(1)
go func() {
defer wg.Done()
if err := plan.Run(s.HTTPAddr); err != nil {
t.Fatalf("err: %v", err)
}
}()
if err := <-invoke; err != nil {
2014-08-20 22:29:31 +00:00
t.Fatalf("err: %v", err)
}
plan.Stop()
wg.Wait()
2014-08-20 22:29:31 +00:00
}
2014-08-20 22:33:13 +00:00
func TestNodesWatch(t *testing.T) {
2017-10-24 13:24:57 +00:00
t.Parallel()
c, s := makeClient(t)
defer s.Stop()
invoke := makeInvokeCh()
2014-08-21 18:38:44 +00:00
plan := mustParse(t, `{"type":"nodes"}`)
plan.Handler = func(idx uint64, raw interface{}) {
if raw == nil {
return // ignore
}
v, ok := raw.([]*api.Node)
if !ok || len(v) == 0 {
return // ignore
2014-08-20 22:33:13 +00:00
}
invoke <- nil
2014-08-20 22:33:13 +00:00
}
var wg sync.WaitGroup
wg.Add(1)
2014-08-20 22:33:13 +00:00
go func() {
defer wg.Done()
catalog := c.Catalog()
time.Sleep(20 * time.Millisecond)
reg := &api.CatalogRegistration{
2014-08-20 22:33:13 +00:00
Node: "foobar",
Address: "1.1.1.1",
Datacenter: "dc1",
}
if _, err := catalog.Register(reg, nil); err != nil {
t.Fatalf("err: %v", err)
2014-08-20 22:33:13 +00:00
}
}()
wg.Add(1)
go func() {
defer wg.Done()
if err := plan.Run(s.HTTPAddr); err != nil {
t.Fatalf("err: %v", err)
}
}()
if err := <-invoke; err != nil {
2014-08-20 22:33:13 +00:00
t.Fatalf("err: %v", err)
}
plan.Stop()
wg.Wait()
2014-08-20 22:33:13 +00:00
}
2014-08-20 22:50:32 +00:00
func TestServiceWatch(t *testing.T) {
2017-10-24 13:24:57 +00:00
t.Parallel()
c, s := makeClient(t)
defer s.Stop()
invoke := makeInvokeCh()
2014-08-21 18:38:44 +00:00
plan := mustParse(t, `{"type":"service", "service":"foo", "tag":"bar", "passingonly":true}`)
plan.Handler = func(idx uint64, raw interface{}) {
if raw == nil {
return // ignore
2014-08-20 22:50:32 +00:00
}
v, ok := raw.([]*api.ServiceEntry)
if !ok || len(v) == 0 {
return // ignore
}
if v[0].Service.ID != "foo" {
invoke <- errBadContent
return
}
invoke <- nil
2014-08-20 22:50:32 +00:00
}
var wg sync.WaitGroup
2014-08-20 22:50:32 +00:00
wg.Add(1)
go func() {
defer wg.Done()
agent := c.Agent()
time.Sleep(20 * time.Millisecond)
reg := &api.AgentServiceRegistration{
2014-08-20 22:50:32 +00:00
ID: "foo",
Name: "foo",
Tags: []string{"bar"},
}
if err := agent.ServiceRegister(reg); err != nil {
t.Fatalf("err: %v", err)
}
}()
2014-08-20 22:50:32 +00:00
wg.Add(1)
go func() {
defer wg.Done()
if err := plan.Run(s.HTTPAddr); err != nil {
t.Fatalf("err: %v", err)
}
2014-08-20 22:50:32 +00:00
}()
if err := <-invoke; err != nil {
2014-08-20 22:50:32 +00:00
t.Fatalf("err: %v", err)
}
plan.Stop()
wg.Wait()
2014-08-20 22:50:32 +00:00
}
2014-08-20 23:32:12 +00:00
func TestServiceMultipleTagsWatch(t *testing.T) {
t.Parallel()
c, s := makeClient(t)
defer s.Stop()
invoke := makeInvokeCh()
plan := mustParse(t, `{"type":"service", "service":"foo", "tag":["bar","buzz"], "passingonly":true}`)
plan.Handler = func(idx uint64, raw interface{}) {
if raw == nil {
return // ignore
}
v, ok := raw.([]*api.ServiceEntry)
if !ok || len(v) == 0 {
return // ignore
}
if v[0].Service.ID != "foobarbuzzbiff" {
invoke <- errBadContent
return
}
if len(v[0].Service.Tags) == 0 {
invoke <- errBadContent
return
}
// test for our tags
barFound := false
buzzFound := false
for _, t := range v[0].Service.Tags {
if t == "bar" {
barFound = true
} else if t == "buzz" {
buzzFound = true
}
}
if !barFound || !buzzFound {
invoke <- errBadContent
return
}
invoke <- nil
}
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
agent := c.Agent()
// we do not want to find this one.
time.Sleep(20 * time.Millisecond)
reg := &api.AgentServiceRegistration{
ID: "foobarbiff",
Name: "foo",
Tags: []string{"bar", "biff"},
}
if err := agent.ServiceRegister(reg); err != nil {
t.Fatalf("err: %v", err)
}
// we do not want to find this one.
reg = &api.AgentServiceRegistration{
ID: "foobuzzbiff",
Name: "foo",
Tags: []string{"buzz", "biff"},
}
if err := agent.ServiceRegister(reg); err != nil {
t.Fatalf("err: %v", err)
}
// we want to find this one
reg = &api.AgentServiceRegistration{
ID: "foobarbuzzbiff",
Name: "foo",
Tags: []string{"bar", "buzz", "biff"},
}
if err := agent.ServiceRegister(reg); err != nil {
t.Fatalf("err: %v", err)
}
}()
wg.Add(1)
go func() {
defer wg.Done()
if err := plan.Run(s.HTTPAddr); err != nil {
t.Fatalf("err: %v", err)
}
}()
if err := <-invoke; err != nil {
t.Fatalf("err: %v", err)
}
plan.Stop()
wg.Wait()
}
2014-08-20 23:32:12 +00:00
func TestChecksWatch_State(t *testing.T) {
2017-10-24 13:24:57 +00:00
t.Parallel()
c, s := makeClient(t)
defer s.Stop()
invoke := makeInvokeCh()
2014-08-21 18:38:44 +00:00
plan := mustParse(t, `{"type":"checks", "state":"warning"}`)
plan.Handler = func(idx uint64, raw interface{}) {
if raw == nil {
return // ignore
2014-08-20 23:32:12 +00:00
}
v, ok := raw.([]*api.HealthCheck)
if !ok || len(v) == 0 {
return // ignore
}
if v[0].CheckID != "foobar" || v[0].Status != "warning" {
invoke <- errBadContent
return
}
invoke <- nil
2014-08-20 23:32:12 +00:00
}
var wg sync.WaitGroup
wg.Add(1)
2014-08-20 23:32:12 +00:00
go func() {
defer wg.Done()
catalog := c.Catalog()
time.Sleep(20 * time.Millisecond)
reg := &api.CatalogRegistration{
2014-08-20 23:32:12 +00:00
Node: "foobar",
Address: "1.1.1.1",
Datacenter: "dc1",
Check: &api.AgentCheck{
2014-08-20 23:32:12 +00:00
Node: "foobar",
CheckID: "foobar",
Name: "foobar",
Status: api.HealthWarning,
2014-08-20 23:32:12 +00:00
},
}
if _, err := catalog.Register(reg, nil); err != nil {
t.Fatalf("err: %v", err)
}
}()
2014-08-20 23:32:12 +00:00
wg.Add(1)
go func() {
defer wg.Done()
if err := plan.Run(s.HTTPAddr); err != nil {
t.Fatalf("err: %v", err)
2014-08-20 23:32:12 +00:00
}
}()
if err := <-invoke; err != nil {
2014-08-20 23:32:12 +00:00
t.Fatalf("err: %v", err)
}
plan.Stop()
wg.Wait()
2014-08-20 23:32:12 +00:00
}
func TestChecksWatch_Service(t *testing.T) {
2017-10-24 13:24:57 +00:00
t.Parallel()
c, s := makeClient(t)
defer s.Stop()
invoke := makeInvokeCh()
2014-08-21 18:38:44 +00:00
plan := mustParse(t, `{"type":"checks", "service":"foobar"}`)
plan.Handler = func(idx uint64, raw interface{}) {
if raw == nil {
return // ignore
2014-08-20 23:32:12 +00:00
}
v, ok := raw.([]*api.HealthCheck)
if !ok || len(v) == 0 {
return // ignore
}
if v[0].CheckID != "foobar" {
invoke <- errBadContent
return
}
invoke <- nil
2014-08-20 23:32:12 +00:00
}
var wg sync.WaitGroup
wg.Add(1)
2014-08-20 23:32:12 +00:00
go func() {
defer wg.Done()
catalog := c.Catalog()
time.Sleep(20 * time.Millisecond)
reg := &api.CatalogRegistration{
2014-08-20 23:32:12 +00:00
Node: "foobar",
Address: "1.1.1.1",
Datacenter: "dc1",
Service: &api.AgentService{
2014-08-20 23:32:12 +00:00
ID: "foobar",
Service: "foobar",
},
Check: &api.AgentCheck{
2014-08-20 23:32:12 +00:00
Node: "foobar",
CheckID: "foobar",
Name: "foobar",
Status: api.HealthPassing,
2014-08-20 23:32:12 +00:00
ServiceID: "foobar",
},
}
if _, err := catalog.Register(reg, nil); err != nil {
2014-08-20 23:32:12 +00:00
t.Fatalf("err: %v", err)
}
}()
2014-08-20 23:32:12 +00:00
wg.Add(1)
go func() {
defer wg.Done()
if err := plan.Run(s.HTTPAddr); err != nil {
t.Fatalf("err: %v", err)
2014-08-20 23:32:12 +00:00
}
}()
if err := <-invoke; err != nil {
2014-08-20 23:32:12 +00:00
t.Fatalf("err: %v", err)
}
plan.Stop()
wg.Wait()
2014-08-20 23:32:12 +00:00
}
2014-08-28 22:41:06 +00:00
func TestEventWatch(t *testing.T) {
2017-10-24 13:24:57 +00:00
t.Parallel()
c, s := makeClient(t)
defer s.Stop()
invoke := makeInvokeCh()
2014-08-28 22:41:06 +00:00
plan := mustParse(t, `{"type":"event", "name": "foo"}`)
plan.Handler = func(idx uint64, raw interface{}) {
if raw == nil {
return
}
v, ok := raw.([]*api.UserEvent)
if !ok || len(v) == 0 {
return // ignore
2014-08-28 22:41:06 +00:00
}
if string(v[len(v)-1].Name) != "foo" {
invoke <- errBadContent
return
}
invoke <- nil
2014-08-28 22:41:06 +00:00
}
var wg sync.WaitGroup
wg.Add(1)
2014-08-28 22:41:06 +00:00
go func() {
defer wg.Done()
event := c.Event()
time.Sleep(20 * time.Millisecond)
params := &api.UserEvent{Name: "foo"}
2014-08-28 22:41:06 +00:00
if _, _, err := event.Fire(params, nil); err != nil {
t.Fatalf("err: %v", err)
}
}()
wg.Add(1)
go func() {
defer wg.Done()
if err := plan.Run(s.HTTPAddr); err != nil {
t.Fatalf("err: %v", err)
}
}()
if err := <-invoke; err != nil {
2014-08-28 22:41:06 +00:00
t.Fatalf("err: %v", err)
}
plan.Stop()
wg.Wait()
2014-08-28 22:41:06 +00:00
}
func TestConnectRootsWatch(t *testing.T) {
t.Parallel()
// makeClient will bootstrap a CA
c, s := makeClient(t)
defer s.Stop()
var originalCAID string
invoke := makeInvokeCh()
plan := mustParse(t, `{"type":"connect_roots"}`)
plan.Handler = func(idx uint64, raw interface{}) {
if raw == nil {
return // ignore
}
v, ok := raw.(*api.CARootList)
if !ok || v == nil || len(v.Roots) == 0 {
return // ignore
}
// Only 1 CA is the bootstrapped state (i.e. first response). Ignore this
// state and wait for the new CA to show up too.
if len(v.Roots) == 1 {
originalCAID = v.ActiveRootID
return
}
require.NotEmpty(t, originalCAID)
require.NotEqual(t, originalCAID, v.ActiveRootID)
invoke <- nil
}
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
time.Sleep(100 * time.Millisecond)
updateConnectCA(t, c)
}()
wg.Add(1)
go func() {
defer wg.Done()
if err := plan.Run(s.HTTPAddr); err != nil {
t.Fatalf("err: %v", err)
}
}()
if err := <-invoke; err != nil {
t.Fatalf("err: %v", err)
}
plan.Stop()
wg.Wait()
}
func TestConnectLeafWatch(t *testing.T) {
t.Parallel()
// makeClient will bootstrap a CA
c, s := makeClient(t)
defer s.Stop()
// Register a web service to get certs for
{
agent := c.Agent()
reg := api.AgentServiceRegistration{
ID: "web",
Name: "web",
Port: 9090,
}
err := agent.ServiceRegister(&reg)
require.Nil(t, err)
}
var lastCert *api.LeafCert
//invoke := makeInvokeCh()
invoke := make(chan error)
2018-05-19 07:11:51 +00:00
plan := mustParse(t, `{"type":"connect_leaf", "service":"web"}`)
plan.Handler = func(idx uint64, raw interface{}) {
if raw == nil {
return // ignore
}
v, ok := raw.(*api.LeafCert)
if !ok || v == nil {
return // ignore
}
if lastCert == nil {
// Initial fetch, just store the cert and return
lastCert = v
return
}
// TODO(banks): right now the root rotation actually causes Serial numbers
// to reset so these end up all being the same. That needs fixing but it's
// a bigger task than I want to bite off for this PR.
//require.NotEqual(t, lastCert.SerialNumber, v.SerialNumber)
require.NotEqual(t, lastCert.CertPEM, v.CertPEM)
require.NotEqual(t, lastCert.PrivateKeyPEM, v.PrivateKeyPEM)
require.NotEqual(t, lastCert.ModifyIndex, v.ModifyIndex)
invoke <- nil
}
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
time.Sleep(100 * time.Millisecond)
updateConnectCA(t, c)
}()
wg.Add(1)
go func() {
defer wg.Done()
if err := plan.Run(s.HTTPAddr); err != nil {
t.Fatalf("err: %v", err)
}
}()
if err := <-invoke; err != nil {
t.Fatalf("err: %v", err)
}
plan.Stop()
wg.Wait()
}
func TestAgentServiceWatch(t *testing.T) {
t.Parallel()
c, s := makeClient(t)
defer s.Stop()
// Register a local agent service
reg := &api.AgentServiceRegistration{
Name: "web",
Port: 8080,
}
client := c
agent := client.Agent()
err := agent.ServiceRegister(reg)
require.NoError(t, err)
invoke := makeInvokeCh()
plan := mustParse(t, `{"type":"agent_service", "service_id":"web"}`)
plan.HybridHandler = func(blockParamVal watch.BlockingParamVal, raw interface{}) {
if raw == nil {
return // ignore
}
v, ok := raw.(*api.AgentService)
if !ok || v == nil {
return // ignore
}
invoke <- nil
}
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
time.Sleep(20 * time.Millisecond)
// Change the service definition
reg.Port = 9090
err := agent.ServiceRegister(reg)
require.NoError(t, err)
}()
wg.Add(1)
go func() {
defer wg.Done()
if err := plan.Run(s.HTTPAddr); err != nil {
t.Fatalf("err: %v", err)
}
}()
if err := <-invoke; err != nil {
t.Fatalf("err: %v", err)
}
plan.Stop()
wg.Wait()
}
func mustParse(t *testing.T, q string) *watch.Plan {
t.Helper()
var params map[string]interface{}
if err := json.Unmarshal([]byte(q), &params); err != nil {
t.Fatal(err)
}
plan, err := watch.Parse(params)
if err != nil {
t.Fatalf("err: %v", err)
}
return plan
}