func (t *testResourceWatcher) expectEvents(kind string, anyOrder bool, expectedEvents []watch.Event) {
	By("Waiting for the correct number of events")
	log.Infof("Start waiting at %s", time.Now())
	t.lock.Lock()
	cur := len(t.events)
	t.lock.Unlock()
	for ii := 0; ii < 10 && cur != len(expectedEvents); ii++ {
		time.Sleep(100 * time.Millisecond)
		t.lock.Lock()
		newcur := len(t.events)
		t.lock.Unlock()
		if newcur != cur {
			ii = 0
			cur = newcur
		}
	}
	log.Infof("Finish waiting at %s", time.Now())
	t.lock.Lock()
	defer t.lock.Unlock()
	var actualEvents []watch.Event
	log.Infof("Received %s events, expected %d", len(t.events), len(expectedEvents))
	if len(t.events) != len(expectedEvents) {
		log.Errorf("Number of received events does not match expected.")
		for _, e := range t.events {
			var o runtime.Object
			if e.Type == watch.Deleted {
				o = e.Previous
			} else {
				o = e.Object
			}
			log.Infof(
				"Received event: EventType:%s; Kind:%s; Name:%s; Namespace:%s",
				e.Type,
				o.GetObjectKind().GroupVersionKind(),
				o.(v1.ObjectMetaAccessor).GetObjectMeta().GetName(),
				o.(v1.ObjectMetaAccessor).GetObjectMeta().GetNamespace(),
			)
		}
		Expect(len(t.events)).To(Equal(len(expectedEvents)))
	} else {
		actualEvents = t.events
	}
	if anyOrder {
		log.Info("Ordering events")
		expectedEvents = t.sortEvents(expectedEvents)
		actualEvents = t.sortEvents(actualEvents)
	}
	log.Info("Comparing actual events against expected events.  Summary:")
	var expectedObject runtime.Object
	var actualObject runtime.Object
	for i, expectedEvent := range expectedEvents {
		if expectedEvent.Type == watch.Deleted {
			expectedObject = expectedEvent.Previous
		} else {
			expectedObject = expectedEvent.Object
		}
		log.Infof(
			"Expected: EventType:%s; Kind:%s; Name:%s; Namespace:%s",
			expectedEvent.Type,
			expectedObject.GetObjectKind().GroupVersionKind(),
			expectedObject.(v1.ObjectMetaAccessor).GetObjectMeta().GetName(),
			expectedObject.(v1.ObjectMetaAccessor).GetObjectMeta().GetNamespace(),
		)
		if i < len(actualEvents) {
			actualEvent := actualEvents[i]
			if actualEvent.Type == watch.Deleted {
				actualObject = actualEvent.Previous
			} else {
				actualObject = actualEvent.Object
			}
			if actualObject != nil {
				log.Infof(
					"Actual:   EventType:%s; Kind:%s; Name:%s; Namespace:%s",
					actualEvent.Type,
					actualObject.GetObjectKind().GroupVersionKind(),
					actualObject.(v1.ObjectMetaAccessor).GetObjectMeta().GetName(),
					actualObject.(v1.ObjectMetaAccessor).GetObjectMeta().GetNamespace(),
				)
			} else {
				log.Warnf("Actual:    EventType:%s, Object: <nil>; Error: %s", actualEvent.Type, actualEvent.Error)
			}
		} else {
			log.Error("Actual:   Event missing")
		}
	}
	Expect(actualEvents).To(HaveLen(len(expectedEvents)))
	for i, expectedEvent := range expectedEvents {
		actualEvent := actualEvents[i]
		actualYaml, _ := yaml.Marshal(actualEvent)
		expectedYaml, _ := yaml.Marshal(expectedEvent)
		traceString := fmt.Sprintf("\nTracing out event details\nActual event: %s\nExpected event: %s\n", actualYaml, expectedYaml)
		Expect(actualEvent.Type).To(Equal(expectedEvent.Type), traceString)
		if expectedEvent.Object != nil {
			Expect(actualEvent.Object).NotTo(BeNil(), traceString)
			Expect(actualEvent.Object).To(MatchResourceWithStatus(
				kind,
				expectedEvent.Object.(v1.ObjectMetaAccessor).GetObjectMeta().GetNamespace(),
				expectedEvent.Object.(v1.ObjectMetaAccessor).GetObjectMeta().GetName(),
				getSpec(expectedEvent.Object),
				getStatus(expectedEvent.Object),
				traceString,
			))
		} else {
			Expect(actualEvent.Object).To(BeNil(), traceString)
		}
		if expectedEvent.Previous != nil && (expectedEvent.Type == watch.Deleted || t.datastoreType != apiconfig.Kubernetes) {
			Expect(actualEvent.Previous).NotTo(BeNil(), traceString)
			Expect(actualEvent.Previous).To(MatchResourceWithStatus(
				kind,
				expectedEvent.Previous.(v1.ObjectMetaAccessor).GetObjectMeta().GetNamespace(),
				expectedEvent.Previous.(v1.ObjectMetaAccessor).GetObjectMeta().GetName(),
				getSpec(expectedEvent.Previous),
				getStatus(expectedEvent.Previous),
				traceString,
			))
		} else {
			Expect(actualEvent.Previous).To(BeNil(), traceString)
		}
	}
	t.events = t.events[len(expectedEvents):]
}
ExpectEvents validates the received events match those expected.  This should be called
within a Ginkgo test.
package testutils
import (
	"reflect"
	"sort"
	"sync"
	"time"
	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	log "github.com/sirupsen/logrus"
	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"fmt"
	"k8s.io/apimachinery/pkg/conversion"
	apiv3 "github.com/projectcalico/api/pkg/apis/projectcalico/v3"
	"github.com/projectcalico/go-yaml-wrapper"
	"github.com/projectcalico/libcalico-go/lib/apiconfig"
	"github.com/projectcalico/libcalico-go/lib/watch"
)
const ExpectNoNamespace = ""
type resourceMatcher struct {
	kind, namespace, name string
	spec                  interface{}
	status                interface{}
}
func Resource(kind, namespace, name string, spec interface{}, optionalDescription ...interface{}) *resourceMatcher {
	return &resourceMatcher{kind, namespace, name, spec, nil}
}
func ResourceWithStatus(kind, namespace, name string, spec, status interface{}, optionalDescription ...interface{}) *resourceMatcher {
	return &resourceMatcher{kind, namespace, name, spec, status}
}
var MatchResource = Resource
var MatchResourceWithStatus = ResourceWithStatus
func (m *resourceMatcher) Match(actual interface{}) (success bool, err error) {
	res, ok := actual.(runtime.Object)
	if !ok {
		ptr := reflect.New(reflect.TypeOf(actual))
		ptr.Elem().Set(reflect.ValueOf(actual))
		res = ptr.Interface().(runtime.Object)
	}
	ma := res.(v1.ObjectMetaAccessor)
	success = (ma.GetObjectMeta().GetNamespace() == m.namespace) &&
		(ma.GetObjectMeta().GetName() == m.name) &&
		(ma.GetObjectMeta().GetResourceVersion() != "") &&
		(res.GetObjectKind().GroupVersionKind().Kind == m.kind) &&
		(res.GetObjectKind().GroupVersionKind().Group == apiv3.Group) &&
		(res.GetObjectKind().GroupVersionKind().Version == apiv3.VersionCurrent) &&
		(m.spec == nil || reflect.DeepEqual(getSpec(res), m.spec)) &&
		(m.status == nil || reflect.DeepEqual(getStatus(res), m.status))
	return
}
func (m *resourceMatcher) FailureMessage(actual interface{}) (message string) {
	message = fmt.Sprintf("Expected\n\t%#v\nto match\n\t%#v", actual, m)
	return
}
func (m *resourceMatcher) NegatedFailureMessage(actual interface{}) (message string) {
	message = fmt.Sprintf("Expected\n\t%#v\nnot to match\n\t%#v", actual, m)
	return
}
func NewTestResourceWatch(datastoreType apiconfig.DatastoreType, w watch.Interface) TestResourceWatchInterface {
	tw := &testResourceWatcher{
		datastoreType: datastoreType,
		watch:         w,
		events:        []watch.Event{},
		watchClosedCh: make(chan struct{}),
	}
	go tw.run()
	return tw
}
type TestResourceWatchInterface interface {
	Stop()
	ExpectEvents(kind string, events []watch.Event)
	ExpectEventsAnyOrder(kind string, events []watch.Event)
}
type testResourceWatcher struct {
	datastoreType apiconfig.DatastoreType
	watch         watch.Interface
	events        []watch.Event
	watchClosedCh chan struct{}
	closing       bool
	lock          sync.Mutex
}
func (t *testResourceWatcher) run() {
	for {
		select {
		case event := <-t.watch.ResultChan():
			t.lock.Lock()
			t.events = append(t.events, event)
			t.lock.Unlock()
		case <-t.watchClosedCh:
			log.Info("Exiting test watch loop")
			return
		}
	}
}
func (t *testResourceWatcher) Stop() {
	t.lock.Lock()
	defer t.lock.Unlock()
	if !t.closing {
		t.watch.Stop()
		close(t.watchClosedCh)
		t.closing = true
	}
}
func (t *testResourceWatcher) ExpectEvents(kind string, expectedEvents []watch.Event) {
	t.expectEvents(kind, false, expectedEvents)
}
func (t *testResourceWatcher) ExpectEventsAnyOrder(kind string, expectedEvents []watch.Event) {
	for _, e := range expectedEvents {
		Expect(e.Type).To(Equal(watch.Added))
	}
	t.expectEvents(kind, true, expectedEvents)
}