diff --git a/go.mod b/go.mod
index f632a8d1..d8e0c298 100644
--- a/go.mod
+++ b/go.mod
@@ -88,6 +88,7 @@ require (
 	github.com/opencontainers/runc v1.1.7 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/stretchr/objx v0.5.0 // indirect
 	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
 	github.com/valyala/bytebufferpool v1.0.0 // indirect
 	github.com/valyala/fasttemplate v1.2.2 // indirect
diff --git a/testing/eventtools/mock_connection.go b/testing/eventtools/mock_connection.go
new file mode 100644
index 00000000..82610665
--- /dev/null
+++ b/testing/eventtools/mock_connection.go
@@ -0,0 +1,86 @@
+package eventtools
+
+import (
+	"context"
+
+	"github.com/stretchr/testify/mock"
+
+	"go.infratographer.com/x/events"
+)
+
+var _ events.Connection = (*MockConnection)(nil)
+
+// MockConnection implements events.Connection
+type MockConnection struct {
+	mock.Mock
+}
+
+// Close implements events.Connection
+func (c *MockConnection) Close(_ context.Context) error {
+	args := c.Called()
+
+	return args.Error(0)
+}
+
+// Publish implements events.Connection
+func (c *MockConnection) Publish(_ context.Context, topic string, message any) (events.Message[any], error) {
+	args := c.Called(topic, message)
+
+	return args.Get(0).(events.Message[any]), args.Error(1)
+}
+
+// PublishAuthRelationshipRequest implements events.Connection
+func (c *MockConnection) PublishAuthRelationshipRequest(_ context.Context, topic string, message events.AuthRelationshipRequest) (events.Message[events.AuthRelationshipResponse], error) {
+	args := c.Called(topic, message)
+
+	return args.Get(0).(events.Message[events.AuthRelationshipResponse]), args.Error(1)
+}
+
+// PublishChange implements events.Connection
+func (c *MockConnection) PublishChange(_ context.Context, topic string, message events.ChangeMessage) (events.Message[events.ChangeMessage], error) {
+	args := c.Called(topic, message)
+
+	return args.Get(0).(events.Message[events.ChangeMessage]), args.Error(1)
+}
+
+// PublishEvent implements events.Connection
+func (c *MockConnection) PublishEvent(_ context.Context, topic string, message events.EventMessage) (events.Message[events.EventMessage], error) {
+	args := c.Called(topic, message)
+
+	return args.Get(0).(events.Message[events.EventMessage]), args.Error(1)
+}
+
+// Source implements events.Connection
+func (c *MockConnection) Source() any {
+	args := c.Called()
+
+	return args.Error(0)
+}
+
+// Subscribe implements events.Connection
+func (c *MockConnection) Subscribe(_ context.Context, topic string) (<-chan events.Message[any], error) {
+	args := c.Called(topic)
+
+	return args.Get(0).(<-chan events.Message[any]), args.Error(1)
+}
+
+// SubscribeAuthRelationshipRequests implements events.Connection
+func (c *MockConnection) SubscribeAuthRelationshipRequests(_ context.Context, topic string) (<-chan events.Message[events.AuthRelationshipRequest], error) {
+	args := c.Called(topic)
+
+	return args.Get(0).(<-chan events.Message[events.AuthRelationshipRequest]), args.Error(1)
+}
+
+// SubscribeChanges implements events.Connection
+func (c *MockConnection) SubscribeChanges(_ context.Context, topic string) (<-chan events.Message[events.ChangeMessage], error) {
+	args := c.Called(topic)
+
+	return args.Get(0).(<-chan events.Message[events.ChangeMessage]), args.Error(1)
+}
+
+// SubscribeEvents implements events.Connection
+func (c *MockConnection) SubscribeEvents(_ context.Context, topic string) (<-chan events.Message[events.EventMessage], error) {
+	args := c.Called(topic)
+
+	return args.Get(0).(<-chan events.Message[events.EventMessage]), args.Error(1)
+}
diff --git a/testing/eventtools/mock_message.go b/testing/eventtools/mock_message.go
new file mode 100644
index 00000000..7e49d90a
--- /dev/null
+++ b/testing/eventtools/mock_message.go
@@ -0,0 +1,101 @@
+package eventtools
+
+import (
+	"context"
+	"time"
+
+	"github.com/stretchr/testify/mock"
+
+	"go.infratographer.com/x/events"
+)
+
+var _ events.Message[any] = (*MockMessage[interface{}])(nil)
+
+// MockMessage implements events.Message.
+type MockMessage[T any] struct {
+	mock.Mock
+}
+
+// Connection implements events.Message.
+func (m *MockMessage[T]) Connection() events.Connection {
+	args := m.Called()
+
+	return args.Get(0).(events.Connection)
+}
+
+// ID implements events.Message.
+func (m *MockMessage[T]) ID() string {
+	args := m.Called()
+
+	return args.String(0)
+}
+
+// Topic implements events.Message.
+func (m *MockMessage[T]) Topic() string {
+	args := m.Called()
+
+	return args.String(0)
+}
+
+// Message implements events.Message.
+func (m *MockMessage[T]) Message() T {
+	args := m.Called()
+
+	return args.Get(0).(T)
+}
+
+// Ack implements events.Message.
+func (m *MockMessage[T]) Ack() error {
+	args := m.Called()
+
+	return args.Error(0)
+}
+
+// Nak implements events.Message.
+func (m *MockMessage[T]) Nak(delay time.Duration) error {
+	args := m.Called(delay)
+
+	return args.Error(0)
+}
+
+// Term implements events.Message.
+func (m *MockMessage[T]) Term() error {
+	args := m.Called()
+
+	return args.Error(0)
+}
+
+// Timestamp implements events.Message.
+func (m *MockMessage[T]) Timestamp() time.Time {
+	args := m.Called()
+
+	return args.Get(0).(time.Time)
+}
+
+// Deliveries implements events.Message.
+func (m *MockMessage[T]) Deliveries() uint64 {
+	args := m.Called()
+
+	return args.Get(0).(uint64)
+}
+
+// Error implements events.Message.
+func (m *MockMessage[T]) Error() error {
+	args := m.Called()
+
+	return args.Error(0)
+}
+
+// ReplyAuthRelationshipRequest implements events.Message.
+func (m *MockMessage[T]) ReplyAuthRelationshipRequest(_ context.Context, message events.AuthRelationshipResponse) (events.Message[events.AuthRelationshipResponse], error) {
+	args := m.Called(message)
+
+	return args.Get(0).(events.Message[events.AuthRelationshipResponse]), args.Error(1)
+}
+
+// Source implements events.Message.
+func (m *MockMessage[T]) Source() any {
+	args := m.Called()
+
+	return args.Error(0)
+}