test(@lib2p/transport-webrtc): add webRTC transport compliance tests

packages/interface-compliance-tests/src/mocks/transport-manager.ts

@@ -0,0 +1,231 @@
+import type { TransportManager } from "@libp2p/interface-internal/transport-manager"
+import type { EventEmitter } from "@libp2p/interface/events"
+import type { Libp2pEvents } from "@libp2p/interface"
+import type { Startable } from "@libp2p/interface/startable"
+import { FaultTolerance, type Listener, type Transport, type Upgrader } from "@libp2p/interface/transport"
+import type { Connection } from "@libp2p/interface/src/connection"
+import type { Multiaddr } from "@multiformats/multiaddr"
+import { CodeError } from "@libp2p/interface/errors"
+
+export interface MockTransportManagerComponents {
+	events: EventEmitter<Libp2pEvents>
+	upgrader: Upgrader
+}
+
+class MockTransportManager implements TransportManager, Startable {
+	private readonly components: MockTransportManagerComponents
+	private readonly transports: Map<string, Transport>
+	private readonly listeners: Map<string, Listener[]>
+	private readonly faultTolerance: FaultTolerance
+	private started: boolean
+
+	constructor(components: MockTransportManagerComponents) {
+		this.components = components
+		this.started = false
+		this.transports = new Map<string, Transport>()
+		this.listeners = new Map<string, Listener[]>()
+		this.faultTolerance = FaultTolerance.FATAL_ALL
+	}
+
+	isStarted(): boolean {
+		return this.started
+	}
+
+	async start(): Promise<void> {
+		this.started = true
+	}
+
+	async stop(): Promise<void> {
+		const tasks = []
+		for (const [_, listeners] of this.listeners) {
+			while (listeners.length > 0) {
+				const listener = listeners.pop()
+
+				if (listener == null) {
+					continue
+				}
+
+				tasks.push(listener.close())
+			}
+		}
+
+		await Promise.all(tasks)
+		for (const key of this.listeners.keys()) {
+			this.listeners.set(key, [])
+		}
+
+		this.started = false
+	}
+
+	getAddrs(): Multiaddr[] {
+		let addrs: Multiaddr[] = []
+		for (const listeners of this.listeners.values()) {
+			for (const listener of listeners) {
+				addrs = [...addrs, ...listener.getAddrs()]
+			}
+		}
+		return addrs
+	}
+
+	/**
+	 * Returns all the transports instances
+	 */
+	getTransports(): Transport[] {
+		return Array.of(...this.transports.values())
+	}
+
+	/**
+	 * Returns all the listener instances
+	 */
+	getListeners(): Listener[] {
+		return Array.of(...this.listeners.values()).flat()
+	}
+
+	add(transport: Transport): void {
+		const tag = transport[Symbol.toStringTag]
+
+		if (tag == null) {
+			throw new CodeError("Transport must have a valid tag", "INVALID_TAG")
+		}
+
+		if (this.transports.has(tag)) {
+			throw new CodeError(`There is already a transport with the tag ${tag}`, "DUPLICATE")
+		}
+
+		this.transports.set(tag, transport)
+
+		if (!this.listeners.has(tag)) {
+			this.listeners.set(tag, [])
+		}
+	}
+
+	transportForMultiaddr(ma: Multiaddr): Transport | undefined {
+		for (const transport of this.transports.values()) {
+			const addrs = transport.filter([ma])
+
+			if (addrs.length > 0) {
+				return transport
+			}
+		}
+	}
+
+	async dial(ma: Multiaddr, options?: any): Promise<Connection> {
+		const transport = this.transportForMultiaddr(ma)
+
+		if (transport == null) {
+			throw new CodeError(`No transport available for address ${String(ma)}`, "TRANSPORT_UNAVAILABLE")
+		}
+
+		try {
+			return await transport.dial(ma, {
+				...options,
+				upgrader: this.components.upgrader,
+			})
+		} catch (err: any) {
+			if (err.code == null) {
+				err.code = "TRANSPORT_DIAL_FAILED"
+			}
+
+			throw err
+		}
+	}
+
+	/**
+	 * Starts listeners for each listen Multiaddr
+	 */
+	async listen(addrs: Multiaddr[]): Promise<void> {
+		if (!this.isStarted()) {
+			throw new CodeError("Not started", "ERR_NODE_NOT_STARTED")
+		}
+
+		if (addrs == null || addrs.length === 0) {
+			return
+		}
+
+		const couldNotListen = []
+
+		for (const [key, transport] of this.transports.entries()) {
+			const supportedAddrs = transport.filter(addrs)
+			const tasks = []
+
+			// For each supported multiaddr, create a listener
+			for (const addr of supportedAddrs) {
+				const listener = transport.createListener({
+					upgrader: this.components.upgrader,
+				})
+
+				let listeners: Listener[] = this.listeners.get(key) ?? []
+
+				if (listeners == null) {
+					listeners = []
+					this.listeners.set(key, listeners)
+				}
+
+				listeners.push(listener)
+
+				// Track listen/close events
+				listener.addEventListener("listening", () => {
+					this.components.events.safeDispatchEvent("transport:listening", {
+						detail: listener,
+					})
+				})
+				listener.addEventListener("close", () => {
+					const index = listeners.findIndex((l) => l === listener)
+
+					// remove the listener
+					listeners.splice(index, 1)
+
+					this.components.events.safeDispatchEvent("transport:close", {
+						detail: listener,
+					})
+				})
+
+				// We need to attempt to listen on everything
+				tasks.push(listener.listen(addr))
+			}
+
+			// Keep track of transports we had no addresses for
+			if (tasks.length === 0) {
+				couldNotListen.push(key)
+				continue
+			}
+
+			const results = await Promise.allSettled(tasks)
+
+			const isListening = results.find((r) => r.status === "fulfilled")
+			if (isListening == null && this.faultTolerance !== FaultTolerance.NO_FATAL) {
+				throw new CodeError(`Transport (${key}) could not listen on any available address`, "ERR_NO_VALID_ADDRESSES")
+			}
+		}
+
+		if (couldNotListen.length === this.transports.size) {
+			const message = `no valid addresses were provided for transports [${couldNotListen.join(", ")}]`
+			if (this.faultTolerance === FaultTolerance.FATAL_ALL) {
+				throw new CodeError(message, "ERR_NO_VALID_ADDRESSES")
+			}
+		}
+	}
+
+	async remove(key: string): Promise<void> {
+		// Close any running listeners
+		for (const listener of this.listeners.get(key) ?? []) {
+			await listener.close()
+		}
+
+		this.transports.delete(key)
+		this.listeners.delete(key)
+	}
+
+	async removeAll(): Promise<void> {
+		const tasks = []
+		for (const key of this.transports.keys()) {
+			tasks.push(this.remove(key))
+		}
+
+		await Promise.all(tasks)
+	}
+}
+
+export function mockTransportManager(components: MockTransportManagerComponents): TransportManager {
+	return new MockTransportManager(components)
+}

packages/interface-compliance-tests/src/transport/dial-test.ts

@@ -19,9 +19,11 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
     let upgrader: Upgrader
     let registrar: Registrar
     let addrs: Multiaddr[]
+    let listeningAddrs: Multiaddr[]
     let transport: Transport
     let connector: Connector
     let listener: Listener
+    let hasListener: boolean
 
     before(async () => {
       registrar = mockRegistrar()
@@ -30,24 +32,27 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
         events: new EventEmitter()
       });
 
-      ({ addrs, transport, connector } = await common.setup())
+      ({ addrs, transport, connector, listeningAddrs =[], hasListener = true } = await common.setup())
     })
 
     after(async () => {
       await common.teardown()
     })
 
     beforeEach(async () => {
-      listener = transport.createListener({
-        upgrader
-      })
-      await listener.listen(addrs[0])
+      if (hasListener) {
+        listener = transport.createListener({
+          upgrader
+        })
+        listeningAddrs.length > 0 ? await listener.listen(listeningAddrs[0]) : await listener.listen(addrs[0])
+      }
     })
 
     afterEach(async () => {
       sinon.restore()
       connector.restore()
-      await listener.close()
+      if (hasListener)
+        await listener.close()
     })
 
     it('simple', async () => {
@@ -56,13 +61,13 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
         void pipe([
           uint8ArrayFromString('hey')
         ],
-        data.stream,
-        drain
+          data.stream,
+          drain
         )
       })
 
       const upgradeSpy = sinon.spy(upgrader, 'upgradeOutbound')
-      const conn = await transport.dial(addrs[0], {
+      const conn = await transport.dial(listeningAddrs[0], {
         upgrader
       })
 

packages/interface-compliance-tests/src/transport/index.ts

@@ -12,8 +12,10 @@ export interface Connector {
 
 export interface TransportTestFixtures {
   addrs: Multiaddr[]
+  listeningAddrs?: Multiaddr[]
   transport: Transport
   connector: Connector
+  hasListener?: boolean
 }
 
 export default (common: TestSetup<TransportTestFixtures>): void => {

packages/interface-compliance-tests/src/transport/listen-test.ts

@@ -20,6 +20,7 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
   describe('listen', () => {
     let upgrader: Upgrader
     let addrs: Multiaddr[]
+    let listeningAddrs: Multiaddr[]
     let transport: Transport
     let registrar: Registrar
 
@@ -30,7 +31,7 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
         events: new EventEmitter()
       });
 
-      ({ transport, addrs } = await common.setup())
+      ({ transport, addrs, listeningAddrs = [] } = await common.setup())
     })
 
     after(async () => {
@@ -45,7 +46,7 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
       const listener = transport.createListener({
         upgrader
       })
-      await listener.listen(addrs[0])
+      listeningAddrs.length > 0 ? await listener.listen(listeningAddrs[0]) : await listener.listen(addrs[0])
       await listener.close()
     })
 
@@ -66,7 +67,7 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
       })
 
       // Listen
-      await listener.listen(addrs[0])
+      listeningAddrs.length > 0 ? await listener.listen(listeningAddrs[0]) : await listener.listen(addrs[0])
 
       // Create two connections to the listener
       const [conn1] = await Promise.all([
@@ -113,7 +114,7 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
       })
 
       // Listen
-      await listener.listen(addrs[0])
+      listeningAddrs.length > 0 ? await listener.listen(listeningAddrs[0]) : await listener.listen(addrs[0])
 
       // Create a connection to the listener
       const conn = await transport.dial(addrs[0], {
@@ -139,7 +140,7 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
         })
 
         void (async () => {
-          await listener.listen(addrs[0])
+          listeningAddrs.length > 0 ? await listener.listen(listeningAddrs[0]) : await listener.listen(addrs[0])
           await transport.dial(addrs[0], {
             upgrader
           })
@@ -159,7 +160,8 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
         listener.addEventListener('listening', () => {
           listener.close().then(done, done)
         })
-        void listener.listen(addrs[0])
+        const addrToListenOn = listeningAddrs.length > 0 ? listeningAddrs[0] : addrs[0]
+        void listener.listen(addrToListenOn)
       })
 
       it('error', (done) => {
@@ -182,7 +184,7 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
         listener.addEventListener('close', () => { done() })
 
         void (async () => {
-          await listener.listen(addrs[0])
+          listeningAddrs.length > 0 ? await listener.listen(listeningAddrs[0]) : await listener.listen(addrs[0])
           await listener.close()
         })()
       })

packages/interface/package.json

@@ -163,11 +163,15 @@
     "it-stream-types": "^2.0.1",
     "multiformats": "^12.0.1",
     "p-defer": "^4.0.0",
+    "race-signal": "^1.0.0",
     "uint8arraylist": "^2.4.3"
   },
   "devDependencies": {
     "@types/sinon": "^10.0.15",
     "aegir": "^40.0.8",
+    "delay": "^6.0.0",
+    "it-all": "^3.0.3",
+    "it-drain": "^3.0.3",
     "sinon": "^16.0.0",
     "sinon-ts": "^1.0.0"
   }