fix: dispatch connection event from mock upgrader (#345)

* Dispatch a connection event as per the upgrader interface
* Simplify args for MockConnectionManager
* Use base64pad for MockMuxer data transfer

Author: achingbrain

packages/interface-mocks/src/connection-manager.ts

@@ -40,12 +40,17 @@ class MockNetwork {
 
 export const mockNetwork = new MockNetwork()
 
+export interface MockConnectionManagerComponents {
+  peerId: PeerId
+  registrar: Registrar
+}
+
 class MockConnectionManager extends EventEmitter<ConnectionManagerEvents> implements ConnectionManager, Startable {
   private connections: Connection[] = []
-  private readonly components: MockNetworkComponents
+  private readonly components: MockConnectionManagerComponents
   private started = false
 
-  constructor (components: MockNetworkComponents) {
+  constructor (components: MockConnectionManagerComponents) {
     super()
 
     this.components = components
@@ -99,7 +104,7 @@ class MockConnectionManager extends EventEmitter<ConnectionManagerEvents> implem
     this.connections.push(aToB)
     ;(componentsB.connectionManager as MockConnectionManager).connections.push(bToA)
 
-    this.components.connectionManager.safeDispatchEvent<Connection>('peer:connect', {
+    this.safeDispatchEvent<Connection>('peer:connect', {
       detail: aToB
     })
 
@@ -159,6 +164,6 @@ class MockConnectionManager extends EventEmitter<ConnectionManagerEvents> implem
   }
 }
 
-export function mockConnectionManager (components: MockNetworkComponents): ConnectionManager {
+export function mockConnectionManager (components: MockConnectionManagerComponents): ConnectionManager {
   return new MockConnectionManager(components)
 }

packages/interface-mocks/src/multiaddr-connection.ts

@@ -37,7 +37,7 @@ export function mockMultiaddrConnPair (opts: MockMultiaddrConnPairOptions): { in
 
   const outbound: MultiaddrConnection = {
     ...outboundStream,
-    remoteAddr: remoteAddr.encapsulate(`/p2p/${remotePeer.toString()}`),
+    remoteAddr: remoteAddr.toString().includes(`/p2p/${remotePeer.toString()}`) ? remoteAddr : remoteAddr.encapsulate(`/p2p/${remotePeer.toString()}`),
     timeline: {
       open: Date.now()
     },

packages/interface-mocks/src/muxer.ts

@@ -152,11 +152,10 @@ class MuxedStream {
 
             while (list.length > 0) {
               const available = Math.min(list.length, MAX_MESSAGE_SIZE)
-              const subList = list.subarray(0, available)
               const dataMsg: DataMessage = {
                 id,
                 type: 'data',
-                chunk: uint8ArrayToString(subList.slice(), 'base64'),
+                chunk: uint8ArrayToString(list.subarray(0, available), 'base64pad'),
                 direction: this.type
               }
 
@@ -344,7 +343,7 @@ class MockMuxer implements StreamMuxer {
     }
 
     if (message.type === 'data') {
-      muxedStream.input.push(new Uint8ArrayList(uint8ArrayFromString(message.chunk, 'base64')))
+      muxedStream.input.push(new Uint8ArrayList(uint8ArrayFromString(message.chunk, 'base64pad')))
     } else if (message.type === 'reset') {
       this.log('-> reset stream %s %s', muxedStream.type, muxedStream.stream.id)
       muxedStream.stream.reset()

packages/interface-mocks/src/upgrader.ts

@@ -2,7 +2,7 @@ import { mockConnection } from './connection.js'
 import type { Upgrader, UpgraderEvents, UpgraderOptions } from '@libp2p/interface-transport'
 import type { Connection, MultiaddrConnection } from '@libp2p/interface-connection'
 import type { Registrar } from '@libp2p/interface-registrar'
-import { EventEmitter } from '@libp2p/interfaces/events'
+import { CustomEvent, EventEmitter } from '@libp2p/interfaces/events'
 
 export interface MockUpgraderInit {
   registrar?: Registrar
@@ -18,19 +18,27 @@ class MockUpgrader extends EventEmitter<UpgraderEvents> implements Upgrader {
   }
 
   async upgradeOutbound (multiaddrConnection: MultiaddrConnection, opts: UpgraderOptions = {}): Promise<Connection> {
-    return mockConnection(multiaddrConnection, {
+    const connection = mockConnection(multiaddrConnection, {
       direction: 'outbound',
       registrar: this.registrar,
       ...opts
     })
+
+    this.dispatchEvent(new CustomEvent<Connection>('connection', { detail: connection }))
+
+    return connection
   }
 
   async upgradeInbound (multiaddrConnection: MultiaddrConnection, opts: UpgraderOptions = {}): Promise<Connection> {
-    return mockConnection(multiaddrConnection, {
+    const connection = mockConnection(multiaddrConnection, {
       direction: 'inbound',
       registrar: this.registrar,
       ...opts
     })
+
+    this.dispatchEvent(new CustomEvent<Connection>('connection', { detail: connection }))
+
+    return connection
   }
 }
 

packages/interface-stream-muxer-compliance-tests/src/close-test.ts

@@ -47,7 +47,7 @@ export default (common: TestSetup<StreamMuxerFactory>): void => {
       void pipe(p[0], dialer, p[0])
       void pipe(p[1], listener, p[1])
 
-      const streams = await Promise.all(Array(expectedStreams).fill(0).map(() => dialer.newStream()))
+      const streams = await Promise.all(Array(expectedStreams).fill(0).map(async () => await dialer.newStream()))
 
       void Promise.all(
         streams.map(async stream => {
@@ -89,7 +89,7 @@ export default (common: TestSetup<StreamMuxerFactory>): void => {
       void pipe(p[0], dialer, p[0])
       void pipe(p[1], listener, p[1])
 
-      const streams = await Promise.all(Array(expectedStreams).fill(0).map(() => dialer.newStream()))
+      const streams = await Promise.all(Array(expectedStreams).fill(0).map(async () => await dialer.newStream()))
 
       void Promise.all(
         streams.map(async stream => {
@@ -132,7 +132,7 @@ export default (common: TestSetup<StreamMuxerFactory>): void => {
       void pipe(p[0], dialer, p[0])
       void pipe(p[1], listener, p[1])
 
-      const streams = await Promise.all(Array(expectedStreams).fill(0).map(() => dialer.newStream()))
+      const streams = await Promise.all(Array(expectedStreams).fill(0).map(async () => await dialer.newStream()))
 
       const streamPipes = streams.map(async stream => {
         await pipe(
@@ -201,7 +201,7 @@ export default (common: TestSetup<StreamMuxerFactory>): void => {
       void pipe(p[1], listener, p[1])
 
       const stream = await dialer.newStream()
-      const streams = await Promise.all(Array.from(Array(5), () => dialer.newStream()))
+      const streams = await Promise.all(Array.from(Array(5), async () => await dialer.newStream()))
       let closed = false
       const controllers: AbortController[] = []