feat(testing): FakeBroadcastManager records listeners and can dispatch to them

.claude/rules/broadcasting.md

@@ -81,9 +81,14 @@ Register via `Echo.addInterceptor()` or `driver.addInterceptor()` in a ServicePr
 
 - `Echo.fake()` — binds `FakeBroadcastManager` in container; returns the fake for assertions
 - `Echo.unfake()` — removes fake binding (or use `MagicApp.reset()` + `Magic.flush()` in `setUp()`)
-- Assertions: `assertConnected()`, `assertDisconnected()`, `assertSubscribed(channel)`, `assertNotSubscribed(channel)`, `assertInterceptorAdded()` — all throw `AssertionError` with descriptive messages
+- Assertions: `assertConnected()`, `assertDisconnected()`, `assertSubscribed(channel)`, `assertNotSubscribed(channel)`, `assertListening(channel, event)`, `assertNotListening(channel, event)`, `assertInterceptorAdded()` — all throw `AssertionError` with descriptive messages
+- `fake.dispatch(channel, event, data)` — run the registered handler with a decoded payload, as the driver would on an incoming frame. Throws if nothing is listening, because dispatching into silence is the failure it exists to reveal
 - `fake.reset()` — clear all recorded state
-- `fake.driver` — access underlying `FakeBroadcastDriver` for low-level inspection (`.subscribedChannels`, `.addedInterceptors`, `.isConnected`)
+- `fake.driver` — access underlying `FakeBroadcastDriver` for low-level inspection (`.subscribedChannels`, `.addedInterceptors`, `.isConnected`, `.listeners`)
+
+**Subscribing and listening are separate, and only `assertListening` covers the second.** `assertSubscribed` passes for an app that holds a live channel and registers no handler at all, so a deleted `listen()` line is invisible to it. Reach for `assertListening` on the line your realtime feature depends on, and `dispatch()` when you need to prove a handler actually runs rather than merely exists.
+
+A second `listen()` for one event REPLACES the first, matching `ReverbBroadcastDriver`, which cancels the previous subscription before storing the new one.
 
 ## Config
 

lib/src/testing/fake_broadcast_manager.dart

@@ -78,6 +78,61 @@ class FakeBroadcastManager extends BroadcastManager {
     }
   }
 
+  /// Assert that a handler for [event] is registered on [channel].
+  ///
+  /// Subscribing to a channel and listening for an event are separate steps, and
+  /// [assertSubscribed] only covers the first: an application can hold a live
+  /// channel and register no handler at all. This is the assertion that fails
+  /// when a `listen()` line is deleted.
+  ///
+  /// Throws [AssertionError] if no handler is registered.
+  void assertListening(String channel, String event) {
+    final Map<String, void Function(BroadcastEvent)>? handlers =
+        _driver._listeners[channel];
+
+    if (handlers == null || !handlers.containsKey(event)) {
+      throw AssertionError(
+        'Expected a handler for "$event" on channel "$channel" but none was '
+        'registered. Registered: ${_driver.listeners}',
+      );
+    }
+  }
+
+  /// Assert that NO handler for [event] is registered on [channel].
+  ///
+  /// Throws [AssertionError] if a handler is registered.
+  void assertNotListening(String channel, String event) {
+    if (_driver._listeners[channel]?.containsKey(event) ?? false) {
+      throw AssertionError(
+        'Expected no handler for "$event" on channel "$channel" but one was '
+        'registered.',
+      );
+    }
+  }
+
+  /// Deliver [data] to the handler registered for [event] on [channel], exactly
+  /// as the driver would on an incoming frame.
+  ///
+  /// This is what turns "a handler exists" into "a handler runs on a frame".
+  /// [data] is handed over already decoded, matching [BroadcastEvent.data] on the
+  /// real driver, which performs the Pusher double-JSON-decode before a handler
+  /// ever sees it.
+  ///
+  /// Throws [AssertionError] if no handler is registered, because dispatching
+  /// into silence is the failure this method exists to make visible.
+  void dispatch(String channel, String event, Map<String, dynamic> data) {
+    assertListening(channel, event);
+
+    _driver._listeners[channel]![event]!(
+      BroadcastEvent(
+        event: event,
+        channel: channel,
+        data: data,
+        receivedAt: DateTime.now(),
+      ),
+    );
+  }
+
   /// Assert that at least one interceptor has been added to the driver.
   ///
   /// Throws [AssertionError] if no interceptors have been added.
@@ -138,21 +193,51 @@ class FakeBroadcastDriver implements BroadcastDriver {
   @override
   BroadcastChannel channel(String name) {
     _subscribedChannels.add(name);
-    return _FakeBroadcastChannel(name);
+    return _FakeBroadcastChannel(name, this);
   }
 
   @override
   BroadcastChannel private(String name) {
     _subscribedChannels.add('private-$name');
-    return _FakeBroadcastChannel('private-$name');
+    return _FakeBroadcastChannel('private-$name', this);
   }
 
   @override
   BroadcastPresenceChannel join(String name) {
     _subscribedChannels.add('presence-$name');
-    return _FakeBroadcastPresenceChannel('presence-$name');
+    return _FakeBroadcastPresenceChannel('presence-$name', this);
+  }
+
+  /// Event handlers registered through [BroadcastChannel.listen], keyed by
+  /// channel name and then by event name.
+  ///
+  /// The driver owns this rather than the channel because the channel factories
+  /// above mint a NEW channel object on every call and keep no reference, so a
+  /// registration recorded on the channel would vanish the moment the caller let
+  /// go of it. That is what made a listener registration untestable.
+  final Map<String, Map<String, void Function(BroadcastEvent)>> _listeners = {};
+
+  /// Registered event names per channel, for low-level inspection.
+  Map<String, List<String>> get listeners => Map.unmodifiable({
+    for (final MapEntry<String, Map<String, void Function(BroadcastEvent)>> e
+        in _listeners.entries)
+      e.key: List<String>.unmodifiable(e.value.keys),
+  });
+
+  void _register(
+    String channel,
+    String event,
+    void Function(BroadcastEvent) callback,
+  ) {
+    // Last registration wins, matching ReverbBroadcastDriver, which cancels the
+    // previous subscription before storing the new one. A fake that appended
+    // instead would hide a double-registration bug rather than reproduce it.
+    (_listeners[channel] ??= {})[event] = callback;
   }
 
+  void _unregister(String channel, String event) =>
+      _listeners[channel]?.remove(event);
+
   @override
   void leave(String name) => _subscribedChannels.remove(name);
 
@@ -164,6 +249,7 @@ class FakeBroadcastDriver implements BroadcastDriver {
     _connected = false;
     _subscribedChannels.clear();
     _addedInterceptors.clear();
+    _listeners.clear();
   }
 }
 
@@ -172,9 +258,10 @@ class FakeBroadcastDriver implements BroadcastDriver {
 // ---------------------------------------------------------------------------
 
 class _FakeBroadcastChannel implements BroadcastChannel {
-  _FakeBroadcastChannel(this._name);
+  _FakeBroadcastChannel(this._name, this._driver);
 
   final String _name;
+  final FakeBroadcastDriver _driver;
 
   @override
   String get name => _name;
@@ -186,10 +273,19 @@ class _FakeBroadcastChannel implements BroadcastChannel {
   BroadcastChannel listen(
     String event,
     void Function(BroadcastEvent) callback,
-  ) => this;
+  ) {
+    // Recorded rather than discarded, and the difference is what a consumer's
+    // test can prove. This used to return `this` and drop both arguments, so a
+    // subscription line could be deleted from an application and its whole suite
+    // stayed green: the one line every broadcast depends on was the one line
+    // nothing covered.
+    _driver._register(_name, event, callback);
+
+    return this;
+  }
 
   @override
-  void stopListening(String event) {}
+  void stopListening(String event) => _driver._unregister(_name, event);
 }
 
 // ---------------------------------------------------------------------------
@@ -198,7 +294,7 @@ class _FakeBroadcastChannel implements BroadcastChannel {
 
 class _FakeBroadcastPresenceChannel extends _FakeBroadcastChannel
     implements BroadcastPresenceChannel {
-  _FakeBroadcastPresenceChannel(super.name);
+  _FakeBroadcastPresenceChannel(super.name, super.driver);
 
   @override
   List<Map<String, dynamic>> get members => const [];

test/testing/fake_broadcast_manager_test.dart

@@ -375,4 +375,108 @@ void main() {
       expect(presence.onLeave, isA<Stream>());
     });
   });
+
+  group('listener registration', () {
+    test('listen() records the event so a deleted subscription is visible', () {
+      // The reason this exists: `listen()` used to return `this` and drop both
+      // arguments, so an application could delete its subscription line and its
+      // whole suite stayed green. Subscribing and listening are separate steps
+      // and `assertSubscribed` only covers the first.
+      final fake = FakeBroadcastManager();
+
+      fake.connection().private('teams.1').listen('order.shipped', (_) {});
+
+      fake.assertListening('private-teams.1', 'order.shipped');
+      expect(fake.driver.listeners, {
+        'private-teams.1': ['order.shipped'],
+      });
+    });
+
+    test('assertListening throws when nothing registered that event', () {
+      final fake = FakeBroadcastManager();
+      fake.connection().private('teams.1');
+
+      expect(
+        () => fake.assertListening('private-teams.1', 'order.shipped'),
+        throwsA(isA<AssertionError>()),
+      );
+    });
+
+    test('assertNotListening throws once a handler is registered', () {
+      final fake = FakeBroadcastManager();
+      fake.connection().channel('public').listen('ping', (_) {});
+
+      fake.assertNotListening('public', 'other');
+      expect(
+        () => fake.assertNotListening('public', 'ping'),
+        throwsA(isA<AssertionError>()),
+      );
+    });
+
+    test('dispatch() runs the registered handler with the decoded payload', () {
+      // What turns "a handler exists" into "a handler runs on a frame", which is
+      // the only thing that proves a consumer's wiring end to end.
+      final fake = FakeBroadcastManager();
+      BroadcastEvent? received;
+
+      fake
+          .connection()
+          .private('teams.1')
+          .listen('order.shipped', (e) => received = e);
+
+      fake.dispatch('private-teams.1', 'order.shipped', {'id': 7});
+
+      expect(received, isNotNull);
+      expect(received!.event, 'order.shipped');
+      expect(received!.channel, 'private-teams.1');
+      expect(received!.data, {'id': 7});
+    });
+
+    test('dispatch() into silence throws rather than passing quietly', () {
+      final fake = FakeBroadcastManager();
+      fake.connection().private('teams.1');
+
+      expect(
+        () => fake.dispatch('private-teams.1', 'order.shipped', const {}),
+        throwsA(isA<AssertionError>()),
+      );
+    });
+
+    test('a second listen() for one event replaces the first', () {
+      // Matches ReverbBroadcastDriver, which cancels the previous subscription
+      // before storing the new one. A fake that appended would hide a
+      // double-registration bug instead of reproducing it.
+      final fake = FakeBroadcastManager();
+      final List<String> calls = [];
+
+      fake.connection().private('teams.1')
+        ..listen('ping', (_) => calls.add('first'))
+        ..listen('ping', (_) => calls.add('second'));
+
+      fake.dispatch('private-teams.1', 'ping', const {});
+
+      expect(calls, ['second']);
+      expect(fake.driver.listeners['private-teams.1'], ['ping']);
+    });
+
+    test('stopListening() removes the handler', () {
+      final fake = FakeBroadcastManager();
+      final channel = fake.connection().private('teams.1')
+        ..listen('ping', (_) {});
+
+      channel.stopListening('ping');
+
+      fake.assertNotListening('private-teams.1', 'ping');
+    });
+
+    test('reset() clears registrations along with the rest', () {
+      final fake = FakeBroadcastManager();
+      fake.connection().private('teams.1').listen('ping', (_) {});
+
+      fake.reset();
+
+      expect(fake.driver.listeners, isEmpty);
+      fake.assertNotListening('private-teams.1', 'ping');
+    });
+  });
 }