1 files changed, 23 insertions, 0 deletions

diff --git a/core/operator/harness_test.go b/core/operator/harness_test.go
new file mode 100644
index 0000000..383ab8c
--- /dev/null
+++ b/core/operator/harness_test.go
@@ -0,0 +1,23 @@
+package operator_test
+
+import (
+        "context"
+
+        "flux/core"
+)
+
+type captureEmitter[T any] struct {
+        records []core.StreamRecord[T]
+}
+
+func (e *captureEmitter[T]) Emit(_ context.Context, record core.StreamRecord[T]) error {
+        e.records = append(e.records, record)
+        return nil
+}
+
+func (e *captureEmitter[T]) Records() []core.StreamRecord[T] {
+        result := make([]core.StreamRecord[T], len(e.records))
+        copy(result, e.records)
+
+        return result
+}