1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128 | import Foundation
import Testing
@testable import IrdP2PSignaling
import IrdP2PCrypto
/// Against the real Go signal server on loopback (needs `AP2P_REPO` and Go).
@Suite(.enabled(if: LocalServers.isAvailable), .serialized) struct SignalServerTests {
func session(_ servers: LocalServers, peer: PeerId, origin: String? = LocalServers.origin, room: RoomId = testRoom) -> RelaySession {
let cfg = RelaySession.Configuration(url: servers.wsURL, origin: origin, room: room, peer: peer, limits: .defaults)
return RelaySession(configuration: cfg, transport: URLSessionWebSocketTransport(), clock: SystemClock())
}
@Test func originGatesTheUpgrade() async throws {
let servers = try LocalServers()
defer { servers.stop() }
try await servers.waitUntilReady()
let refused = session(servers, peer: testPeer, origin: nil)
await #expect(throws: P2PError.upgradeRejected(status: 403)) { try await refused.start() }
let wrong = session(servers, peer: testPeer, origin: "https://evil.example")
await #expect(throws: P2PError.upgradeRejected(status: 403)) { try await wrong.start() }
let ok = session(servers, peer: testPeer)
try await ok.start()
#expect(await ok.isJoined)
await ok.leave()
}
@Test func twoPeersSeeEachOtherAndRelaySignals() async throws {
let servers = try LocalServers()
defer { servers.stop() }
try await servers.waitUntilReady()
let a = session(servers, peer: testPeer), b = session(servers, peer: otherPeer)
var ea = a.events.makeAsyncIterator(), eb = b.events.makeAsyncIterator()
try await a.start()
#expect(await ea.next() == .joined)
#expect(await ea.next() == .roster([]))
try await b.start()
#expect(await eb.next() == .joined)
#expect(await eb.next() == .roster([testPeer]))
#expect(await ea.next() == .peerJoined(otherPeer))
try await a.send(signal: .object(["hello": .string("b")]), to: otherPeer)
#expect(await eb.next() == .signal(from: testPeer, payload: .object(["hello": .string("b")])))
// signaling to a stranger is silently dropped: no error frame, no event
try await a.send(signal: .null, to: try PeerId(String(repeating: "f", count: 32)))
await b.leave()
#expect(await ea.next() == .peerLeft(otherPeer))
await a.leave()
}
@Test func rejoiningWithTheSamePeerIdReplacesTheOldConnection() async throws {
let servers = try LocalServers()
defer { servers.stop() }
try await servers.waitUntilReady()
let watcher = session(servers, peer: otherPeer)
var ew = watcher.events.makeAsyncIterator()
try await watcher.start()
_ = await ew.next(); _ = await ew.next()
let first = session(servers, peer: testPeer)
try await first.start()
#expect(await ew.next() == .peerJoined(testPeer))
let second = session(servers, peer: testPeer)
var e2 = second.events.makeAsyncIterator()
try await second.start()
#expect(await e2.next() == .joined)
#expect(await e2.next() == .roster([otherPeer]))
try await second.send(signal: .int(1), to: otherPeer)
// the hub re-announces the replacing connection as peer-joined; skip roster noise until the signal
var got: RelayEvent? = nil
for _ in 0..<4 {
let e = await ew.next()
if case .signal = e { got = e; break }
}
#expect(got == .signal(from: testPeer, payload: .int(1)))
await second.leave()
await first.leave()
await watcher.leave()
}
@Test func rateLimitedFramesArriveAsErrorsWhenUnpaced() async throws {
let servers = try LocalServers()
defer { servers.stop() }
try await servers.waitUntilReady()
var cfg = RelaySession.Configuration(url: servers.wsURL, origin: LocalServers.origin, room: testRoom, peer: testPeer, limits: .defaults)
cfg.pacing = false
let s = RelaySession(configuration: cfg, transport: URLSessionWebSocketTransport())
var events = s.events.makeAsyncIterator()
try await s.start()
_ = await events.next(); _ = await events.next()
for _ in 0..<120 { try await s.send(signal: .int(1), to: otherPeer) }
var sawRateLimited = false
for _ in 0..<5 {
if case .relayError(.rateLimited)? = await events.next() { sawRateLimited = true; break }
}
#expect(sawRateLimited)
await s.leave()
}
@Test func roomFullIsReportedOnJoin() async throws {
let servers = try LocalServers()
defer { servers.stop() }
try await servers.waitUntilReady()
var sessions: [RelaySession] = []
for i in 0..<4 {
let s = session(servers, peer: try PeerId(String(format: "%032x", i + 1)))
try await s.start()
sessions.append(s)
}
let fifth = session(servers, peer: try PeerId(String(format: "%032x", 99)))
await #expect(throws: P2PError.relay(.roomFull)) { try await fifth.start() }
for s in sessions { await s.leave() }
}
@Test(.timeLimit(.minutes(3))) func idleConnectionSurvivesTheServerPingCycle() async throws {
// The server pings every 25 s and closes after 60 s without a pong; URLSession must answer.
let servers = try LocalServers()
defer { servers.stop() }
try await servers.waitUntilReady()
var cfg = RelaySession.Configuration(url: servers.wsURL, origin: LocalServers.origin, room: testRoom, peer: testPeer, limits: .defaults)
cfg.keepaliveInterval = .seconds(3600) // no app-level pings: only protocol-level PONGs keep it alive
let s = RelaySession(configuration: cfg, transport: URLSessionWebSocketTransport())
var events = s.events.makeAsyncIterator()
try await s.start()
_ = await events.next(); _ = await events.next()
try await Task.sleep(for: .seconds(70))
#expect(await s.isJoined)
try await s.send(signal: .int(1), to: otherPeer)
await s.leave()
}
}
|