util/eventbus: add test helpers to simplify testing events (#16294)

Instead of every module having to come up with a set of test methods for
the event bus, this handful of test helpers hides a lot of the needed
setup for the testing of the event bus.

The tests in portmapper is also ported over to the new helpers.

Updates #15160

Signed-off-by: Claus Lensbøl <claus@tailscale.com>

net/portmapper/portmapper.go

@@ -515,7 +515,7 @@ func (c *Client) createMapping() {
 			GoodUntil: mapping.GoodUntil(),
 		})
 	}
-	if c.onChange != nil {
+	if c.onChange != nil && c.pubClient == nil {
 		go c.onChange()
 	}
 }

net/portmapper/portmapper_test.go

@@ -12,7 +12,7 @@ import (
 	"time"
 
 	"tailscale.com/control/controlknobs"
-	"tailscale.com/util/eventbus"
+	"tailscale.com/util/eventbus/eventbustest"
 )
 
 func TestCreateOrGetMapping(t *testing.T) {
@@ -142,22 +142,15 @@ func TestUpdateEvent(t *testing.T) {
 		t.Fatalf("Create test gateway: %v", err)
 	}
 
-	bus := eventbus.New()
-	defer bus.Close()
+	bus := eventbustest.NewBus(t)
+	tw := eventbustest.NewWatcher(t, bus)
 
-	sub := eventbus.Subscribe[Mapping](bus.Client("TestUpdateEvent"))
 	c := newTestClient(t, igd, bus)
 	if _, err := c.Probe(t.Context()); err != nil {
 		t.Fatalf("Probe failed: %v", err)
 	}
 	c.GetCachedMappingOrStartCreatingOne()
-
-	select {
-	case evt := <-sub.Events():
-		t.Logf("Received portmap update: %+v", evt)
-	case <-sub.Done():
-		t.Error("Subscriber closed prematurely")
-	case <-time.After(5 * time.Second):
-		t.Error("Timed out waiting for an update event")
+	if err := eventbustest.Expect(tw, eventbustest.Type[Mapping]()); err != nil {
+		t.Error(err.Error())
 	}
 }

util/eventbus/doc.go

@@ -89,4 +89,14 @@
 // The [Debugger], obtained through [Bus.Debugger], provides
 // introspection facilities to monitor events flowing through the bus,
 // and inspect publisher and subscriber state.
+//
+// Additionally, a debug command exists for monitoring the eventbus:
+//
+//	tailscale debug daemon-bus-events
+//
+// # Testing facilities
+//
+// Helpers for testing code with the eventbus can be found in:
+//
+//	eventbus/eventbustest
 package eventbus

util/eventbus/eventbustest/doc.go

@@ -0,0 +1,45 @@
+// Copyright (c) Tailscale Inc & AUTHORS
+// SPDX-License-Identifier: BSD-3-Clause
+
+// Package eventbustest provides helper methods for testing an [eventbus.Bus].
+//
+// # Usage
+//
+// A [Watcher] presents a set of generic helpers for testing events.
+//
+// To test code that generates events, create a [Watcher] from the [eventbus.Bus]
+// used by the code under test, run the code to generate events, then use the watcher
+// to verify that the expected events were produced. In outline:
+//
+//	bus := eventbustest.NewBus(t)
+//	tw := eventbustest.NewWatcher(t, bus)
+//	somethingThatEmitsSomeEvent()
+//	if err := eventbustest.Expect(tw, eventbustest.Type[EventFoo]()); err != nil {
+//	  t.Error(err.Error())
+//	}
+//
+// As shown, [Expect] checks that at least one event of the given type occurs
+// in the stream generated by the code under test.
+//
+// The following functions all take an any parameter representing a function.
+// This function will take an argument of the expected type and is used to test
+// for the events on the eventbus being of the given type. The function can
+// take the shape described in [Expect].
+//
+// [Type] is a helper for only testing event type.
+//
+// To check for specific properties of an event, use [Expect], and pass a function
+// as the second argument that tests for those properties.
+//
+// To test for multiple events, use [Expect], which checks that the stream
+// contains the given events in the given order, possibly with other events
+// interspersed.
+//
+// To test the complete contents of the stream, use [ExpectExactly], which
+// checks that the stream contains exactly the given events in the given order,
+// and no others.
+//
+// See the [usage examples].
+//
+// [usage examples]: https://github.com/tailscale/tailscale/blob/main/util/eventbus/eventbustest/examples_test.go
+package eventbustest

util/eventbus/eventbustest/eventbustest.go

@@ -0,0 +1,203 @@
+// Copyright (c) Tailscale Inc & AUTHORS
+// SPDX-License-Identifier: BSD-3-Clause
+
+package eventbustest
+
+import (
+	"errors"
+	"fmt"
+	"reflect"
+	"testing"
+	"time"
+
+	"tailscale.com/util/eventbus"
+)
+
+// NewBus constructs an [eventbus.Bus] that will be shut automatically when
+// its controlling test ends.
+func NewBus(t *testing.T) *eventbus.Bus {
+	bus := eventbus.New()
+	t.Cleanup(bus.Close)
+	return bus
+}
+
+// NewTestWatcher constructs a [Watcher] that can be used to check the stream of
+// events generated by code under test. After construction the caller may use
+// [Expect] and [ExpectExactly], to verify that the desired events were captured.
+func NewWatcher(t *testing.T, bus *eventbus.Bus) *Watcher {
+	tw := &Watcher{
+		mon:     bus.Debugger().WatchBus(),
+		TimeOut: 5 * time.Second,
+		chDone:  make(chan bool, 1),
+		events:  make(chan any, 100),
+	}
+	if deadline, ok := t.Deadline(); ok {
+		tw.TimeOut = deadline.Sub(time.Now())
+	}
+	t.Cleanup(tw.done)
+	go tw.watch()
+	return tw
+}
+
+// Watcher monitors and holds events for test expectations.
+type Watcher struct {
+	mon    *eventbus.Subscriber[eventbus.RoutedEvent]
+	events chan any
+	chDone chan bool
+	// TimeOut defines when the Expect* functions should stop looking for events
+	// coming from the Watcher. The value is set by [NewWatcher] and defaults to
+	// the deadline passed in by [testing.T]. If looking to verify the absence
+	// of an event, the TimeOut can be set to a lower value after creating the
+	// Watcher.
+	TimeOut time.Duration
+}
+
+// Type is a helper representing the expectation to see an event of type T, without
+// caring about the content of the event.
+// It makes it possible to use helpers like:
+//
+//	eventbustest.ExpectFilter(tw, eventbustest.Type[EventFoo]())
+func Type[T any]() func(T) { return func(T) {} }
+
+// Expect verifies that the given events are a subsequence of the events
+// observed by tw. That is, tw must contain at least one event matching the type
+// of each argument in the given order, other event types are allowed to occur in
+// between without error. The given events are represented by a function
+// that must have one of the following forms:
+//
+//	// Tests for the event type only
+//	func(e ExpectedType)
+//
+//	// Tests for event type and whatever is defined in the body.
+//	// If return is false, the test will look for other events of that type
+//	// If return is true, the test will look for the next given event
+//	// if a list is given
+//	func(e ExpectedType) bool
+//
+//	// Tests for event type and whatever is defined in the body.
+//	// The boolean return works as above.
+//	// The if error != nil, the test helper will return that error immediately.
+//	func(e ExpectedType) (bool, error)
+//
+// If the list of events must match exactly with no extra events,
+// use [ExpectExactly].
+func Expect(tw *Watcher, filters ...any) error {
+	if len(filters) == 0 {
+		return errors.New("no event filters were provided")
+	}
+	eventCount := 0
+	head := 0
+	for head < len(filters) {
+		eventFunc := eventFilter(filters[head])
+		select {
+		case event := <-tw.events:
+			eventCount++
+			if ok, err := eventFunc(event); err != nil {
+				return err
+			} else if ok {
+				head++
+			}
+		case <-time.After(tw.TimeOut):
+			return fmt.Errorf(
+				"timed out waiting for event, saw %d events, %d was expected",
+				eventCount, head)
+		case <-tw.chDone:
+			return errors.New("watcher closed while waiting for events")
+		}
+	}
+	return nil
+}
+
+// ExpectExactly checks for some number of events showing up on the event bus
+// in a given order, returning an error if the events does not match the given list
+// exactly. The given events are represented by a function as described in
+// [Expect]. Use [Expect] if other events are allowed.
+func ExpectExactly(tw *Watcher, filters ...any) error {
+	if len(filters) == 0 {
+		return errors.New("no event filters were provided")
+	}
+	eventCount := 0
+	for pos, next := range filters {
+		eventFunc := eventFilter(next)
+		fnType := reflect.TypeOf(next)
+		argType := fnType.In(0)
+		select {
+		case event := <-tw.events:
+			eventCount++
+			typeEvent := reflect.TypeOf(event)
+			if typeEvent != argType {
+				return fmt.Errorf(
+					"expected event type %s, saw %s, at index %d",
+					argType, typeEvent, pos)
+			} else if ok, err := eventFunc(event); err != nil {
+				return err
+			} else if !ok {
+				return fmt.Errorf(
+					"expected test ok for type %s, at index %d", argType, pos)
+			}
+		case <-time.After(tw.TimeOut):
+			return fmt.Errorf(
+				"timed out waiting for event, saw %d events, %d was expected",
+				eventCount, pos)
+		case <-tw.chDone:
+			return errors.New("watcher closed while waiting for events")
+		}
+	}
+	return nil
+}
+
+func (tw *Watcher) watch() {
+	for {
+		select {
+		case event := <-tw.mon.Events():
+			tw.events <- event.Event
+		case <-tw.chDone:
+			tw.mon.Close()
+			return
+		}
+	}
+}
+
+// done tells the watcher to stop monitoring for new events.
+func (tw *Watcher) done() {
+	close(tw.chDone)
+}
+
+type filter = func(any) (bool, error)
+
+func eventFilter(f any) filter {
+	ft := reflect.TypeOf(f)
+	if ft.Kind() != reflect.Func {
+		panic("filter is not a function")
+	} else if ft.NumIn() != 1 {
+		panic(fmt.Sprintf("function takes %d arguments, want 1", ft.NumIn()))
+	}
+	var fixup func([]reflect.Value) []reflect.Value
+	switch ft.NumOut() {
+	case 0:
+		fixup = func([]reflect.Value) []reflect.Value {
+			return []reflect.Value{reflect.ValueOf(true), reflect.Zero(reflect.TypeFor[error]())}
+		}
+	case 1:
+		if ft.Out(0) != reflect.TypeFor[bool]() {
+			panic(fmt.Sprintf("result is %T, want bool", ft.Out(0)))
+		}
+		fixup = func(vals []reflect.Value) []reflect.Value {
+			return append(vals, reflect.Zero(reflect.TypeFor[error]()))
+		}
+	case 2:
+		if ft.Out(0) != reflect.TypeFor[bool]() || ft.Out(1) != reflect.TypeFor[error]() {
+			panic(fmt.Sprintf("results are %T, %T; want bool, error", ft.Out(0), ft.Out(1)))
+		}
+		fixup = func(vals []reflect.Value) []reflect.Value { return vals }
+	default:
+		panic(fmt.Sprintf("function returns %d values", ft.NumOut()))
+	}
+	fv := reflect.ValueOf(f)
+	return reflect.MakeFunc(reflect.TypeFor[filter](), func(args []reflect.Value) []reflect.Value {
+		if !args[0].IsValid() || args[0].Elem().Type() != ft.In(0) {
+			return []reflect.Value{reflect.ValueOf(false), reflect.Zero(reflect.TypeFor[error]())}
+		}
+		return fixup(fv.Call([]reflect.Value{args[0].Elem()}))
+	}).Interface().(filter)
+}

util/eventbus/eventbustest/eventbustest_test.go

@@ -0,0 +1,366 @@
+// Copyright (c) Tailscale Inc & AUTHORS
+// SPDX-License-Identifier: BSD-3-Clause
+
+package eventbustest_test
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"tailscale.com/util/eventbus"
+	"tailscale.com/util/eventbus/eventbustest"
+)
+
+type EventFoo struct {
+	Value int
+}
+
+type EventBar struct {
+	Value string
+}
+
+type EventBaz struct {
+	Value []float64
+}
+
+func TestExpectFilter(t *testing.T) {
+	tests := []struct {
+		name       string
+		events     []int
+		expectFunc any
+		wantErr    bool
+	}{
+		{
+			name:       "single event",
+			events:     []int{42},
+			expectFunc: eventbustest.Type[EventFoo](),
+			wantErr:    false,
+		},
+		{
+			name:       "multiple events, single expectation",
+			events:     []int{42, 1, 2, 3, 4, 5},
+			expectFunc: eventbustest.Type[EventFoo](),
+			wantErr:    false,
+		},
+		{
+			name:   "filter on event with function",
+			events: []int{24, 42},
+			expectFunc: func(event EventFoo) (bool, error) {
+				if event.Value == 42 {
+					return true, nil
+				}
+				return false, nil
+			},
+			wantErr: false,
+		},
+		{
+			name:   "first event has to be func",
+			events: []int{24, 42},
+			expectFunc: func(event EventFoo) (bool, error) {
+				if event.Value != 42 {
+					return false, fmt.Errorf("expected 42, got %d", event.Value)
+				}
+				return false, nil
+			},
+			wantErr: true,
+		},
+		{
+			name:   "no events",
+			events: []int{},
+			expectFunc: func(event EventFoo) (bool, error) {
+				return true, nil
+			},
+			wantErr: true,
+		},
+	}
+
+	bus := eventbustest.NewBus(t)
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tw := eventbustest.NewWatcher(t, bus)
+			// TODO(cmol): When synctest is out of experimental, use that instead:
+			// https://go.dev/blog/synctest
+			tw.TimeOut = 10 * time.Millisecond
+
+			client := bus.Client("testClient")
+			defer client.Close()
+			updater := eventbus.Publish[EventFoo](client)
+
+			for _, i := range tt.events {
+				updater.Publish(EventFoo{i})
+			}
+
+			if err := eventbustest.Expect(tw, tt.expectFunc); (err != nil) != tt.wantErr {
+				t.Errorf("ExpectFilter[EventFoo]: error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}
+
+func TestExpectEvents(t *testing.T) {
+	tests := []struct {
+		name         string
+		events       []any
+		expectEvents []any
+		wantErr      bool
+	}{
+		{
+			name:         "No expectations",
+			events:       []any{EventFoo{}},
+			expectEvents: []any{},
+			wantErr:      true,
+		},
+		{
+			name:         "One event",
+			events:       []any{EventFoo{}},
+			expectEvents: []any{eventbustest.Type[EventFoo]()},
+			wantErr:      false,
+		},
+		{
+			name:         "Two events",
+			events:       []any{EventFoo{}, EventBar{}},
+			expectEvents: []any{eventbustest.Type[EventFoo](), eventbustest.Type[EventBar]()},
+			wantErr:      false,
+		},
+		{
+			name:         "Two expected events with another in the middle",
+			events:       []any{EventFoo{}, EventBaz{}, EventBar{}},
+			expectEvents: []any{eventbustest.Type[EventFoo](), eventbustest.Type[EventBar]()},
+			wantErr:      false,
+		},
+		{
+			name:         "Missing event",
+			events:       []any{EventFoo{}, EventBaz{}},
+			expectEvents: []any{eventbustest.Type[EventFoo](), eventbustest.Type[EventBar]()},
+			wantErr:      true,
+		},
+		{
+			name:   "One event with specific value",
+			events: []any{EventFoo{42}},
+			expectEvents: []any{
+				func(ev EventFoo) (bool, error) {
+					if ev.Value == 42 {
+						return true, nil
+					}
+					return false, nil
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name:   "Two event with one specific value",
+			events: []any{EventFoo{43}, EventFoo{42}},
+			expectEvents: []any{
+				func(ev EventFoo) (bool, error) {
+					if ev.Value == 42 {
+						return true, nil
+					}
+					return false, nil
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name:   "One event with wrong value",
+			events: []any{EventFoo{43}},
+			expectEvents: []any{
+				func(ev EventFoo) (bool, error) {
+					if ev.Value == 42 {
+						return true, nil
+					}
+					return false, nil
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name:   "Two events with specific values",
+			events: []any{EventFoo{42}, EventFoo{42}, EventBar{"42"}},
+			expectEvents: []any{
+				func(ev EventFoo) (bool, error) {
+					if ev.Value == 42 {
+						return true, nil
+					}
+					return false, nil
+				},
+				func(ev EventBar) (bool, error) {
+					if ev.Value == "42" {
+						return true, nil
+					}
+					return false, nil
+				},
+			},
+			wantErr: false,
+		},
+	}
+
+	bus := eventbustest.NewBus(t)
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tw := eventbustest.NewWatcher(t, bus)
+			// TODO(cmol): When synctest is out of experimental, use that instead:
+			// https://go.dev/blog/synctest
+			tw.TimeOut = 10 * time.Millisecond
+
+			client := bus.Client("testClient")
+			defer client.Close()
+			updaterFoo := eventbus.Publish[EventFoo](client)
+			updaterBar := eventbus.Publish[EventBar](client)
+			updaterBaz := eventbus.Publish[EventBaz](client)
+
+			for _, ev := range tt.events {
+				switch ev.(type) {
+				case EventFoo:
+					evCast := ev.(EventFoo)
+					updaterFoo.Publish(evCast)
+				case EventBar:
+					evCast := ev.(EventBar)
+					updaterBar.Publish(evCast)
+				case EventBaz:
+					evCast := ev.(EventBaz)
+					updaterBaz.Publish(evCast)
+				}
+			}
+
+			if err := eventbustest.Expect(tw, tt.expectEvents...); (err != nil) != tt.wantErr {
+				t.Errorf("ExpectEvents: error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}
+
+func TestExpectExactlyEventsFilter(t *testing.T) {
+	tests := []struct {
+		name         string
+		events       []any
+		expectEvents []any
+		wantErr      bool
+	}{
+		{
+			name:         "No expectations",
+			events:       []any{EventFoo{}},
+			expectEvents: []any{},
+			wantErr:      true,
+		},
+		{
+			name:         "One event",
+			events:       []any{EventFoo{}},
+			expectEvents: []any{eventbustest.Type[EventFoo]()},
+			wantErr:      false,
+		},
+		{
+			name:         "Two events",
+			events:       []any{EventFoo{}, EventBar{}},
+			expectEvents: []any{eventbustest.Type[EventFoo](), eventbustest.Type[EventBar]()},
+			wantErr:      false,
+		},
+		{
+			name:         "Two expected events with another in the middle",
+			events:       []any{EventFoo{}, EventBaz{}, EventBar{}},
+			expectEvents: []any{eventbustest.Type[EventFoo](), eventbustest.Type[EventBar]()},
+			wantErr:      true,
+		},
+		{
+			name:         "Missing event",
+			events:       []any{EventFoo{}, EventBaz{}},
+			expectEvents: []any{eventbustest.Type[EventFoo](), eventbustest.Type[EventBar]()},
+			wantErr:      true,
+		},
+		{
+			name:   "One event with value",
+			events: []any{EventFoo{42}},
+			expectEvents: []any{
+				func(ev EventFoo) (bool, error) {
+					if ev.Value == 42 {
+						return true, nil
+					}
+					return false, nil
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name:   "Two event with one specific value",
+			events: []any{EventFoo{43}, EventFoo{42}},
+			expectEvents: []any{
+				func(ev EventFoo) (bool, error) {
+					if ev.Value == 42 {
+						return true, nil
+					}
+					return false, nil
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name:   "One event with wrong value",
+			events: []any{EventFoo{43}},
+			expectEvents: []any{
+				func(ev EventFoo) (bool, error) {
+					if ev.Value == 42 {
+						return true, nil
+					}
+					return false, nil
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name:   "Two events with specific values",
+			events: []any{EventFoo{42}, EventFoo{42}, EventBar{"42"}},
+			expectEvents: []any{
+				func(ev EventFoo) (bool, error) {
+					if ev.Value == 42 {
+						return true, nil
+					}
+					return false, nil
+				},
+				func(ev EventBar) (bool, error) {
+					if ev.Value == "42" {
+						return true, nil
+					}
+					return false, nil
+				},
+			},
+			wantErr: true,
+		},
+	}
+
+	bus := eventbustest.NewBus(t)
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tw := eventbustest.NewWatcher(t, bus)
+			// TODO(cmol): When synctest is out of experimental, use that instead:
+			// https://go.dev/blog/synctest
+			tw.TimeOut = 10 * time.Millisecond
+
+			client := bus.Client("testClient")
+			defer client.Close()
+			updaterFoo := eventbus.Publish[EventFoo](client)
+			updaterBar := eventbus.Publish[EventBar](client)
+			updaterBaz := eventbus.Publish[EventBaz](client)
+
+			for _, ev := range tt.events {
+				switch ev.(type) {
+				case EventFoo:
+					evCast := ev.(EventFoo)
+					updaterFoo.Publish(evCast)
+				case EventBar:
+					evCast := ev.(EventBar)
+					updaterBar.Publish(evCast)
+				case EventBaz:
+					evCast := ev.(EventBaz)
+					updaterBaz.Publish(evCast)
+				}
+			}
+
+			if err := eventbustest.ExpectExactly(tw, tt.expectEvents...); (err != nil) != tt.wantErr {
+				t.Errorf("ExpectEvents: error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}

util/eventbus/eventbustest/examples_test.go

@@ -0,0 +1,201 @@
+// Copyright (c) Tailscale Inc & AUTHORS
+// SPDX-License-Identifier: BSD-3-Clause
+
+package eventbustest_test
+
+import (
+	"testing"
+
+	"tailscale.com/util/eventbus"
+	"tailscale.com/util/eventbus/eventbustest"
+)
+
+func TestExample_Expect(t *testing.T) {
+	type eventOfInterest struct{}
+
+	bus := eventbustest.NewBus(t)
+	tw := eventbustest.NewWatcher(t, bus)
+
+	client := bus.Client("testClient")
+	updater := eventbus.Publish[eventOfInterest](client)
+	updater.Publish(eventOfInterest{})
+
+	if err := eventbustest.Expect(tw, eventbustest.Type[eventOfInterest]()); err != nil {
+		t.Log(err.Error())
+	} else {
+		t.Log("OK")
+	}
+	// Output:
+	// OK
+}
+
+func TestExample_Expect_WithFunction(t *testing.T) {
+	type eventOfInterest struct {
+		value int
+	}
+
+	bus := eventbustest.NewBus(t)
+	tw := eventbustest.NewWatcher(t, bus)
+
+	client := bus.Client("testClient")
+	updater := eventbus.Publish[eventOfInterest](client)
+	updater.Publish(eventOfInterest{43})
+	updater.Publish(eventOfInterest{42})
+
+	// Look for an event of eventOfInterest with a specific value
+	if err := eventbustest.Expect(tw, func(event eventOfInterest) (bool, error) {
+		if event.value != 42 {
+			return false, nil // Look for another event with the expected value.
+			// You could alternatively return an error here to ensure that the
+			// first seen eventOfInterest matches the value:
+			// return false, fmt.Errorf("expected 42, got %d", event.value)
+		}
+		return true, nil
+	}); err != nil {
+		t.Log(err.Error())
+	} else {
+		t.Log("OK")
+	}
+	// Output:
+	// OK
+}
+
+func TestExample_Expect_MultipleEvents(t *testing.T) {
+	type eventOfInterest struct{}
+	type eventOfNoConcern struct{}
+	type eventOfCuriosity struct{}
+
+	bus := eventbustest.NewBus(t)
+	tw := eventbustest.NewWatcher(t, bus)
+
+	client := bus.Client("testClient")
+	updaterInterest := eventbus.Publish[eventOfInterest](client)
+	updaterConcern := eventbus.Publish[eventOfNoConcern](client)
+	updaterCuriosity := eventbus.Publish[eventOfCuriosity](client)
+	updaterInterest.Publish(eventOfInterest{})
+	updaterConcern.Publish(eventOfNoConcern{})
+	updaterCuriosity.Publish(eventOfCuriosity{})
+
+	// Even though three events was published, we just care about the two
+	if err := eventbustest.Expect(tw,
+		eventbustest.Type[eventOfInterest](),
+		eventbustest.Type[eventOfCuriosity]()); err != nil {
+		t.Log(err.Error())
+	} else {
+		t.Log("OK")
+	}
+	// Output:
+	// OK
+}
+
+func TestExample_ExpectExactly_MultipleEvents(t *testing.T) {
+	type eventOfInterest struct{}
+	type eventOfNoConcern struct{}
+	type eventOfCuriosity struct{}
+
+	bus := eventbustest.NewBus(t)
+	tw := eventbustest.NewWatcher(t, bus)
+
+	client := bus.Client("testClient")
+	updaterInterest := eventbus.Publish[eventOfInterest](client)
+	updaterConcern := eventbus.Publish[eventOfNoConcern](client)
+	updaterCuriosity := eventbus.Publish[eventOfCuriosity](client)
+	updaterInterest.Publish(eventOfInterest{})
+	updaterConcern.Publish(eventOfNoConcern{})
+	updaterCuriosity.Publish(eventOfCuriosity{})
+
+	// Will fail as more events than the two expected comes in
+	if err := eventbustest.ExpectExactly(tw,
+		eventbustest.Type[eventOfInterest](),
+		eventbustest.Type[eventOfCuriosity]()); err != nil {
+		t.Log(err.Error())
+	} else {
+		t.Log("OK")
+	}
+}
+
+func TestExample_Expect_WithMultipleFunctions(t *testing.T) {
+	type eventOfInterest struct {
+		value int
+	}
+	type eventOfNoConcern struct{}
+	type eventOfCuriosity struct {
+		value string
+	}
+
+	bus := eventbustest.NewBus(t)
+	tw := eventbustest.NewWatcher(t, bus)
+
+	client := bus.Client("testClient")
+	updaterInterest := eventbus.Publish[eventOfInterest](client)
+	updaterConcern := eventbus.Publish[eventOfNoConcern](client)
+	updaterCuriosity := eventbus.Publish[eventOfCuriosity](client)
+	updaterInterest.Publish(eventOfInterest{42})
+	updaterConcern.Publish(eventOfNoConcern{})
+	updaterCuriosity.Publish(eventOfCuriosity{"42"})
+
+	interest := func(event eventOfInterest) (bool, error) {
+		if event.value == 42 {
+			return true, nil
+		}
+		return false, nil
+	}
+	curiosity := func(event eventOfCuriosity) (bool, error) {
+		if event.value == "42" {
+			return true, nil
+		}
+		return false, nil
+	}
+
+	// Will fail as more events than the two expected comes in
+	if err := eventbustest.Expect(tw, interest, curiosity); err != nil {
+		t.Log(err.Error())
+	} else {
+		t.Log("OK")
+	}
+	// Output:
+	// OK
+}
+
+func TestExample_ExpectExactly_WithMultipleFuncions(t *testing.T) {
+	type eventOfInterest struct {
+		value int
+	}
+	type eventOfNoConcern struct{}
+	type eventOfCuriosity struct {
+		value string
+	}
+
+	bus := eventbustest.NewBus(t)
+	tw := eventbustest.NewWatcher(t, bus)
+
+	client := bus.Client("testClient")
+	updaterInterest := eventbus.Publish[eventOfInterest](client)
+	updaterConcern := eventbus.Publish[eventOfNoConcern](client)
+	updaterCuriosity := eventbus.Publish[eventOfCuriosity](client)
+	updaterInterest.Publish(eventOfInterest{42})
+	updaterConcern.Publish(eventOfNoConcern{})
+	updaterCuriosity.Publish(eventOfCuriosity{"42"})
+
+	interest := func(event eventOfInterest) (bool, error) {
+		if event.value == 42 {
+			return true, nil
+		}
+		return false, nil
+	}
+	curiosity := func(event eventOfCuriosity) (bool, error) {
+		if event.value == "42" {
+			return true, nil
+		}
+		return false, nil
+	}
+
+	// Will fail as more events than the two expected comes in
+	if err := eventbustest.ExpectExactly(tw, interest, curiosity); err != nil {
+		t.Log(err.Error())
+	} else {
+		t.Log("OK")
+	}
+	// Output:
+	// expected event type eventbustest.eventOfCuriosity, saw eventbustest.eventOfNoConcern, at index 1
+}