p2p-cljs / test / ird / p2p / libp2p_test.cljs
530 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
97
98
(ns ird.p2p.libp2p-test
  "The libp2p adapter. Its full frame-routing and presence logic runs
  offline over an injected in-process pubsub bus (always on): two rooms
  meet on the room topic, run the hello/verify handshake and exchange
  messages through the same public API as the /ws relay. The real two-node
  gossipsub path is gated on IRD_P2P_LIBP2P_TEST=1 because the pinned
  gossipsub (v2 interface) does not deliver over libp2p v3 — see the README."
  (:require [cljs.test :refer [deftest is testing]]
            [cljs.core.async :as async :refer [<!]]
            [ird.p2p :as p2p]
            [ird.p2p.crypto :as crypto]
            [ird.p2p.descriptor :as desc]
            [ird.p2p.impl.bytes :as bytes]
            [ird.p2p.impl.libp2p :as impl]
            [ird.p2p.test-util :as tu :refer [deftest-async]]
            [ird.p2p.transport :as t]
            [ird.p2p.transport.libp2p :as lp]))

(def low "00000000000000000000000000000001")
(def high "0f1e2d3c4b5a69788796a5b4c3d2e1f0")
(def app-salt "ird-cljs-tests/libp2p")

;; ---- an in-process pubsub bus that actually delivers -------------------------------

(defn make-bus []
  (atom {:subs {}}))

(defn bus-node
  "A node handle over BUS: publish fans a message out to every other
  subscriber of the topic, exactly the gossipsub contract the transport
  relies on."
  [bus id]
  (let [{:keys [out put! close!]} (ird.p2p.async/unbounded-chan)
        topics (atom #{})]
    (swap! bus assoc-in [:subs id] {:put! put! :topics topics})
    {:peer-id id
     :hex (impl/peer-id->hex id)
     :multiaddrs (fn [] (async/go []))
     :dial (fn [_] (async/go true))
     :subscribe (fn [topic] (swap! topics conj topic) nil)
     :unsubscribe (fn [topic] (swap! topics disj topic) nil)
     :publish (fn [topic b64]
                (async/go
                  (doseq [[oid {:keys [put! topics]}] (:subs @bus)]
                    (when (and (not= oid id) (contains? @topics topic))
                      (put! {:topic topic :data b64})))
                  true))
     :messages out
     :stop (fn [] (async/go (swap! bus update :subs dissoc id) (close!) nil))}))

(defn node-fn-for [bus counter]
  (fn [_opts] (async/go (bus-node bus (str "node-" (swap! counter inc))))))

(defn app [bus counter peer]
  (p2p/make-app {:host "app.example.test" :app-salt app-salt :origin nil
                 :descriptor (desc/parse (tu/fixture "descriptor-full.json"))
                 :transport (lp/transport {:node-fn (node-fn-for bus counter)})
                 :http-fn (tu/responding 404 {:error "not_found"})}))

(deftest topic-derivation
  (is (= "ap2p/room/Xgt16oloGeosHfjLPLb9pPNh0BYNdIknBbtc2ZJm88Q"
         (lp/room-topic "Xgt16oloGeosHfjLPLb9pPNh0BYNdIknBbtc2ZJm88Q"))))

(deftest peer-id-hex-derivation
  (is (= 32 (count (impl/peer-id->hex "12D3KooWExample"))))
  (is (crypto/valid-peer-id? (impl/peer-id->hex "12D3KooWExample")))
  (is (= (impl/peer-id->hex "x") (impl/peer-id->hex "x")) "stable")
  (is (not= (impl/peer-id->hex "a") (impl/peer-id->hex "b"))))

(deftest-async two-rooms-handshake-and-message-over-the-bus
  (let [bus (make-bus)
        counter (atom 0)
        secret (crypto/random-secret)
        ra (<! (p2p/join (app bus counter low) secret {:peer-id low :options {:pacing? false}}))
        ca (tu/collector (p2p/events ra))
        rb (<! (p2p/join (app bus counter high) secret {:peer-id high :options {:pacing? false}}))
        cb (tu/collector (p2p/events rb))]
    (is (not (tu/error? ra)))
    (is (not (tu/error? rb)))
    (is (some? (<! (tu/wait-for ca #(= {:type :peer-joined :peer high} %)))))
    (is (some? (<! (tu/wait-for cb #(= {:type :peer-joined :peer low} %)))))
    (is (some? (<! (tu/wait-for ca #(= {:type :peer-ready :peer high :via :relayed} %)))))
    (is (some? (<! (tu/wait-for cb #(= {:type :peer-ready :peer low :via :relayed} %)))))
    (is (= [high] (p2p/members ra)))
    (is (true? (<! (p2p/send ra high (p2p/json {:text "over gossip"})))))
    (is (some? (<! (tu/wait-for cb #(and (= :message (:type %)) (= {:text "over gossip"} (get-in % [:message :value])))))))
    ;; a chunked message rides the same topic
    (let [big (apply str (repeat 50000 "z"))]
      (is (true? (<! (p2p/send rb low (p2p/json {:big big})))))
      (is (some? (<! (tu/wait-for ca #(= big (get-in % [:message :value :big])) 10000)))))
    (<! (p2p/leave rb))
    (is (some? (<! (tu/wait-for ca #(= {:type :peer-left :peer high} %)))))
    (<! (p2p/leave ra))))

(deftest-async transport-rejects-a-bad-peer-id
  (let [tr (lp/transport {:node-fn (node-fn-for (make-bus) (atom 0))})
        e (<! (t/connect tr {:room "Xgt16oloGeosHfjLPLb9pPNh0BYNdIknBbtc2ZJm88Q" :peer "nope"}))]
    (is (tu/error? e))))

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