fix: move connection manager mock to connection manager module (#205)

Also add mock network to enable dialing other peers by mocks.

Components now control the start/stop lifecycle of their components.

Author: achingbrain

packages/libp2p-interface-compliance-tests/src/mocks/connection-manager.ts

@@ -1,23 +1,137 @@
-import { EventEmitter } from '@libp2p/interfaces'
+import { CustomEvent, EventEmitter, Startable } from '@libp2p/interfaces'
 import type { Connection } from '@libp2p/interfaces/connection'
 import type { PeerId } from '@libp2p/interfaces/peer-id'
-import type { ConnectionManager, ConnectionManagerEvents } from '@libp2p/interfaces/registrar'
+import type { ConnectionManager, ConnectionManagerEvents } from '@libp2p/interfaces/connection-manager'
+import type { Components, Initializable } from '@libp2p/interfaces/src/components'
+import { connectionPair } from './connection.js'
+import errCode from 'err-code'
 
-class MockConnectionManager extends EventEmitter<ConnectionManagerEvents> implements ConnectionManager {
-  getConnectionMap (): Map<string, Connection[]> {
-    return new Map<string, Connection[]>()
+class MockNetwork {
+  private components: Components[] = []
+
+  addNode (components: Components): void {
+    this.components.push(components)
+  }
+
+  getNode (peerId: PeerId): Components {
+    for (const components of this.components) {
+      if (peerId.equals(components.getPeerId())) {
+        return components
+      }
+    }
+
+    throw errCode(new Error('Peer not found'), 'ERR_PEER_NOT_FOUND')
+  }
+
+  reset () {
+    this.components = []
+  }
+}
+
+export const mockNetwork = new MockNetwork()
+
+class MockConnectionManager extends EventEmitter<ConnectionManagerEvents> implements ConnectionManager, Initializable, Startable {
+  private connections: Connection[] = []
+  private components?: Components
+  private started = false
+
+  init (components: Components) {
+    this.components = components
+  }
+
+  isStarted () {
+    return this.started
+  }
+
+  async start () {
+    this.started = true
+  }
+
+  async stop () {
+    for (const connection of this.connections) {
+      await this.closeConnections(connection.remotePeer)
+    }
+
+    this.started = false
   }
 
-  getConnectionList (): Connection[] {
-    return []
+  getConnections (peerId?: PeerId): Connection[] {
+    if (peerId != null) {
+      return this.connections
+        .filter(c => c.remotePeer.toString() === peerId.toString())
+    }
+
+    return this.connections
   }
 
-  getConnections (): Connection[] {
-    return []
+  async openConnection (peerId: PeerId) {
+    if (this.components == null) {
+      throw errCode(new Error('Not initialized'), 'ERR_NOT_INITIALIZED')
+    }
+
+    const existingConnections = this.getConnections(peerId)
+
+    if (existingConnections.length > 0) {
+      return existingConnections[0]
+    }
+
+    const componentsB = mockNetwork.getNode(peerId)
+
+    const [aToB, bToA] = connectionPair(this.components, componentsB)
+
+    // track connections
+    this.connections.push(aToB)
+    ;(componentsB.getConnectionManager() as MockConnectionManager).connections.push(bToA)
+
+    this.components.getConnectionManager().dispatchEvent(new CustomEvent<Connection>('peer:connect', {
+      detail: aToB
+    }))
+
+    for (const protocol of this.components.getRegistrar().getProtocols()) {
+      for (const topology of this.components.getRegistrar().getTopologies(protocol)) {
+        topology.onConnect(componentsB.getPeerId(), aToB)
+      }
+    }
+
+    componentsB.getConnectionManager().dispatchEvent(new CustomEvent<Connection>('peer:connect', {
+      detail: bToA
+    }))
+
+    for (const protocol of componentsB.getRegistrar().getProtocols()) {
+      for (const topology of componentsB.getRegistrar().getTopologies(protocol)) {
+        topology.onConnect(this.components.getPeerId(), bToA)
+      }
+    }
+
+    return aToB
   }
 
-  getConnection (peerId: PeerId): Connection | undefined {
-    return undefined
+  async closeConnections (peerId: PeerId) {
+    if (this.components == null) {
+      throw errCode(new Error('Not initialized'), 'ERR_NOT_INITIALIZED')
+    }
+
+    const connections = this.getConnections(peerId)
+
+    if (connections.length === 0) {
+      return
+    }
+
+    const componentsB = mockNetwork.getNode(peerId)
+
+    for (const protocol of this.components.getRegistrar().getProtocols()) {
+      this.components.getRegistrar().getTopologies(protocol).forEach(topology => {
+        topology.onDisconnect(componentsB.getPeerId())
+      })
+    }
+
+    for (const conn of connections) {
+      await conn.close()
+    }
+
+    this.connections = this.connections.filter(c => !c.remotePeer.equals(peerId))
+
+    await componentsB.getConnectionManager().closeConnections(peerId)
   }
 }
 

packages/libp2p-interface-compliance-tests/src/mocks/index.ts

@@ -1,6 +1,6 @@
 
 export { mockConnectionGater } from './connection-gater.js'
-export { mockConnectionManager } from './connection-manager.js'
+export { mockConnectionManager, mockNetwork } from './connection-manager.js'
 export { mockConnection, mockStream, connectionPair } from './connection.js'
 export { mockMultiaddrConnection, mockMultiaddrConnPair } from './multiaddr-connection.js'
 export { mockMuxer } from './muxer.js'

packages/libp2p-interface-compliance-tests/src/mocks/registrar.ts

@@ -2,8 +2,6 @@ import type { IncomingStreamData, Registrar, StreamHandler } from '@libp2p/inter
 import type { Connection } from '@libp2p/interfaces/connection'
 import type { PeerId } from '@libp2p/interfaces/peer-id'
 import type { Topology } from '@libp2p/interfaces/topology'
-import { connectionPair } from './connection.js'
-import type { Components } from '@libp2p/interfaces/src/components'
 
 export class MockRegistrar implements Registrar {
   private readonly topologies: Map<string, { topology: Topology, protocols: string[] }> = new Map()
@@ -89,7 +87,7 @@ export class MockRegistrar implements Registrar {
       return output
     }
 
-    throw new Error(`No topologies registered for protocol ${protocol}`)
+    return []
   }
 }
 
@@ -106,16 +104,3 @@ export async function mockIncomingStreamEvent (protocol: string, conn: Connectio
     }
   }
 }
-
-export async function connectPeers (protocol: string, a: Components, b: Components) {
-  // Notify peers of connection
-  const [aToB, bToA] = connectionPair(a, b)
-
-  for (const topology of a.getRegistrar().getTopologies(protocol)) {
-    await topology.onConnect(b.getPeerId(), aToB)
-  }
-
-  for (const topology of b.getRegistrar().getTopologies(protocol)) {
-    await topology.onConnect(a.getPeerId(), bToA)
-  }
-}

packages/libp2p-interface-compliance-tests/src/pubsub/api.ts

@@ -11,6 +11,7 @@ import type { Components } from '@libp2p/interfaces/components'
 import { createComponents } from './utils.js'
 import { start, stop } from '../index.js'
 import { isStartable } from '@libp2p/interfaces'
+import { mockNetwork } from '../mocks/connection-manager.js'
 
 const topic = 'foo'
 const data = uint8ArrayFromString('bar')
@@ -22,20 +23,22 @@ export default (common: TestSetup<PubSub, PubSubArgs>) => {
 
     // Create pubsub router
     beforeEach(async () => {
+      mockNetwork.reset()
       components = await createComponents()
 
-      pubsub = await common.setup({
+      pubsub = components.setPubSub(await common.setup({
         components,
         init: {
           emitSelf: true
         }
-      })
+      }))
     })
 
     afterEach(async () => {
       sinon.restore()
-      await stop(pubsub)
+      await stop(common)
       await common.teardown()
+      mockNetwork.reset()
     })
 
     it('can start correctly', async () => {
@@ -45,7 +48,7 @@ export default (common: TestSetup<PubSub, PubSubArgs>) => {
 
       sinon.spy(components.getRegistrar(), 'register')
 
-      await pubsub.start()
+      await start(components)
 
       expect(pubsub.isStarted()).to.equal(true)
       expect(components.getRegistrar().register).to.have.property('callCount', 1)
@@ -58,8 +61,8 @@ export default (common: TestSetup<PubSub, PubSubArgs>) => {
 
       sinon.spy(components.getRegistrar(), 'unregister')
 
-      await pubsub.start()
-      await pubsub.stop()
+      await start(components)
+      await stop(components)
 
       expect(pubsub.isStarted()).to.equal(false)
       expect(components.getRegistrar().unregister).to.have.property('callCount', 1)
@@ -70,7 +73,7 @@ export default (common: TestSetup<PubSub, PubSubArgs>) => {
         throw new Error('a message should not be received')
       }
 
-      await start(pubsub)
+      await start(components)
       pubsub.subscribe(topic)
       pubsub.addEventListener('message', handler)
 
@@ -90,13 +93,13 @@ export default (common: TestSetup<PubSub, PubSubArgs>) => {
       // handlers are called async
       await delay(100)
 
-      await stop(pubsub)
+      await stop(components)
     })
 
     it('can subscribe and publish correctly', async () => {
       const defer = pDefer()
 
-      await start(pubsub)
+      await start(components)
 
       pubsub.subscribe(topic)
       pubsub.addEventListener('message', (evt) => {
@@ -107,7 +110,7 @@ export default (common: TestSetup<PubSub, PubSubArgs>) => {
       await pubsub.publish(topic, data)
       await defer.promise
 
-      await stop(pubsub)
+      await stop(components)
     })
   })
 }