p2p-swift / Tests / IrdP2PTests / RoomMessagingTests.swift
595 words
 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
import Foundation
import Testing
@testable import IrdP2P
@testable import IrdP2PSignaling
import IrdP2PCrypto

@Suite struct RoomMessagingTests {
    struct Line: Codable, Equatable { var text: String; var n: Int }

    func pair(_ h: Harness, direct: Bool) async throws -> (Room, Collector, Room, Collector) {
        let (a, ca) = try await h.join(peer: lowPeer, preferDataChannel: direct)
        let (b, cb) = try await h.join(peer: highPeer, preferDataChannel: direct)
        let want: PeerTransport = direct ? .direct : .relayed
        let okA = await eventuallyTrue(timeout: .seconds(5)) { await a.transport(of: highPeer) == want }
        let okB = await eventuallyTrue(timeout: .seconds(5)) { await b.transport(of: lowPeer) == want }
        if !(okA && okB) { Collector.dump("pair(direct: \(direct)) did not settle", h, [("a", ca), ("b", cb)]) }
        #expect(okA && okB)
        return (a, ca, b, cb)
    }

    @Test(arguments: [false, true]) func jsonAndBinaryBothWays(direct: Bool) async throws {
        let h = Harness()
        let (a, ca, b, cb) = try await pair(h, direct: direct)
        try await a.send(to: highPeer, json: Line(text: "salut", n: 1))
        try await b.send(to: lowPeer, .data(Data([0, 1, 2, 255])))
        try await b.send(to: lowPeer, .json(.object(["t": .string("ack")])))
        #expect(await cb.waitFor { if case .message(lowPeer, .json(let v)) = $0 { return v["text"]?.stringValue == "salut" }; return false } != nil)
        #expect(await ca.waitFor(.message(from: highPeer, .data(Data([0, 1, 2, 255])))))
        #expect(await ca.waitFor(.message(from: highPeer, .json(.object(["t": .string("ack")])))))
        let line = try cb.messages.first!.1.decode(Line.self)
        #expect(line == Line(text: "salut", n: 1))
        await a.leave(); await b.leave()
    }

    @Test(arguments: [false, true]) func largeMessagesAreChunked(direct: Bool) async throws {
        let h = Harness()
        let (a, ca, b, cb) = try await pair(h, direct: direct)
        let bigText = String(repeating: "ă", count: 100_000)                       // 200 KB of UTF-8
        let bigData = Data((0..<100_000).map { UInt8($0 % 251) })
        try await a.send(to: highPeer, .json(.object(["text": .string(bigText)])))
        try await b.send(to: lowPeer, .data(bigData))
        #expect(await cb.waitFor(timeout: .seconds(10)) { if case .message(lowPeer, .json(let v)) = $0 { return v["text"]?.stringValue == bigText }; return false } != nil)
        #expect(await ca.waitFor(.message(from: highPeer, .data(bigData)), timeout: .seconds(10)))
        if !direct {
            // every relayed frame stayed under the node's cap
            #expect(h.relay.frames.allSatisfy { $0.text.utf8.count <= 65_536 })
        }
        await a.leave(); await b.leave()
    }

    @Test func broadcastReachesEveryVerifiedPeer() async throws {
        let h = Harness()
        let (a, _) = try await h.join(peer: lowPeer, preferDataChannel: false)
        let (b, cb) = try await h.join(peer: highPeer, preferDataChannel: false)
        let (c, cc) = try await h.join(peer: thirdPeer, preferDataChannel: false)
        #expect(await eventuallyTrue { await a.members == [highPeer, thirdPeer] })
        await a.broadcast(.json(.string("hi all")))
        #expect(await cb.waitFor(.message(from: lowPeer, .json(.string("hi all")))))
        #expect(await cc.waitFor(.message(from: lowPeer, .json(.string("hi all")))))
        await a.leave(); await b.leave(); await c.leave()
    }

    @Test func replayedFramesAreDroppedAndUnknownPeersRefused() async throws {
        let h = Harness()
        h.relay.duplicateSignals = true
        let (a, ca, b, cb) = try await pair(h, direct: false)
        try await a.send(to: highPeer, .json(.string("once")))
        #expect(await cb.waitFor(.message(from: lowPeer, .json(.string("once")))))
        try? await Task.sleep(for: .milliseconds(50))
        #expect(cb.count(.message(from: lowPeer, .json(.string("once")))) == 1)
        await #expect(throws: P2PError.peerUnknown(thirdPeer)) { try await a.send(to: thirdPeer, .json(.null)) }
        _ = ca
        await a.leave(); await b.leave()
        await #expect(throws: P2PError.closed) { try await a.send(to: highPeer, .json(.null)) }
    }

    @Test func peerLossWhileDirectKeepsTheDataChannel() async throws {
        let h = Harness()
        let (a, ca, b, cb) = try await pair(h, direct: true)
        // b's relay socket dies but the data channel is up: no peerLeft, messages still flow
        h.relay.dropSocket(of: highPeer)
        #expect(await cb.waitFor { if case .status(.reconnecting) = $0 { return true }; return false } != nil)
        try await a.send(to: highPeer, .json(.string("still here")))
        #expect(await cb.waitFor(.message(from: lowPeer, .json(.string("still here")))))
        #expect(!ca.contains(.peerLeft(highPeer)))
        // b reconnects (same peer id) after the backoff; advance the clock until it rejoins
        for _ in 0..<5 {
            if await eventuallyTrue(timeout: .seconds(1), { h.clock.sleeperCount >= 1 }) { h.clock.advance(by: .seconds(15)) }
            if cb.count(.status(.joined)) >= 2 { break }
            try? await Task.sleep(for: .milliseconds(50))
        }
        #expect(cb.count(.status(.joined)) >= 2)
        #expect(await eventuallyTrue { await a.members == [highPeer] })
        await a.leave(); await b.leave()
    }
}

static mirror of HEAD · about · clone: git clone https://ird.cod-sursa.ro/p2p-swift.git