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
129
130
131
132
133
134
135
136
137
138
139 | (ns ird.p2p.test-util
"Shared test tooling: fixture loading, a manual test clock, scripted
:http-fn transports, and `eventually` polling. Tests may use JS interop
freely; the purity rule applies to src/ only."
(:require-macros [ird.p2p.test-util])
(:require [clojure.string :as str]
[cljs.core.async :as async :refer [go go-loop <! >!]]
[ird.p2p.async :as a]
[ird.p2p.impl.json :as json]))
(def ^:private fs (js/require "node:fs"))
(defn read-text
"The text of a file under the repo root."
[path]
(.readFileSync fs path "utf8"))
(defn vector-file
"A golden-vector file (keyword keys)."
[name]
(json/parse (read-text (str "test/ird/p2p/vectors/" name ".json"))))
(defn fixture
"A fixture file's text."
[name]
(read-text (str "test/ird/p2p/fixtures/" name)))
(defn env [name]
(let [v (aget (.-env js/process) name)]
(when (and (string? v) (not (str/blank? v))) v)))
;; ---- clock ----------------------------------------------------------------------
(defn test-clock
"A manual clock: {:now-fn :sleep-fn :advance! :sleepers}. `sleep-fn`
returns a channel that closes once `advance!` has moved time past its
deadline (immediately for ms <= 0)."
[]
(let [time (atom 0)
waiters (atom [])]
{:now-fn (fn [] @time)
:sleep-fn (fn [ms]
(let [c (async/chan)]
(if (<= ms 0)
(async/close! c)
(swap! waiters conj {:deadline (+ @time ms) :ch c}))
c))
:advance! (fn [ms]
(swap! time + ms)
(let [now @time
[due later] ((juxt filter remove) #(<= (:deadline %) now) @waiters)]
(reset! waiters (vec later))
(doseq [w due] (async/close! (:ch w)))
nil))
:sleepers (fn [] (count @waiters))}))
(defn no-sleep
"A :sleep-fn that never waits."
[_ms]
(doto (async/chan) async/close!))
;; ---- http ---------------------------------------------------------------------------
(defn ->json [body] (if (string? body) body (json/stringify body)))
(defn responding
"An :http-fn answering STATUS and BODY to every request, recording into CAPTURED."
([status body] (responding (atom []) status body))
([captured status body]
(fn [req]
(swap! captured conj req)
(go {:status status :headers {} :body (->json body)}))))
(defn scripted
"An :http-fn answering successive requests with the next [status body] of
RESPONSES (or a function of the request returning [status body]),
recording into CAPTURED; once exhausted it delivers an Error."
[captured responses]
(let [queue (atom (vec responses))]
(fn [req]
(swap! captured conj req)
(let [next (first @queue)
[status body] (if (fn? next) (next req) next)]
(swap! queue #(if (seq %) (subvec % 1) %))
(go (if status
{:status status :headers {} :body (->json body)}
(ex-info "Scripted responses exhausted" {:url (:url req)})))))))
(defn routing
"An :http-fn dispatching on the URL: ROUTES is a vector of
[substring (fn [req] -> [status body])], first match wins, 404 otherwise."
([routes] (routing (atom []) routes))
([captured routes]
(fn [req]
(swap! captured conj req)
(let [[status body] (or (some (fn [[needle f]] (when (str/includes? (:url req) needle) (f req))) routes)
[404 {:error "not_found"}])]
(go {:status status :headers {} :body (->json body)})))))
(defn failing
"An :http-fn whose channel delivers ERR."
[err]
(fn [_] (go err)))
;; ---- polling -------------------------------------------------------------------------
(defn eventually
"Channel of true once (pred) is truthy, polling every 5 ms of real time
up to TIMEOUT-MS (default 5000); false otherwise."
([pred] (eventually pred 5000))
([pred timeout-ms]
(let [deadline (+ (js/Date.now) timeout-ms)]
(go-loop []
(cond
(pred) true
(>= (js/Date.now) deadline) (boolean (pred))
:else (do (<! (async/timeout 5)) (recur)))))))
(defn collector
"Drain CH into an atom vector; returns the atom."
[ch]
(let [seen (atom [])]
(go-loop []
(when-let [v (<! ch)]
(swap! seen conj v)
(recur)))
seen))
(defn wait-for
"Channel of the first collected event satisfying PRED (real-time polling), or nil."
([seen pred] (wait-for seen pred 5000))
([seen pred timeout-ms]
(go
(<! (eventually #(some pred @seen) timeout-ms))
(some #(when (pred %) %) @seen))))
(defn count-of [seen pred] (count (filter pred @seen)))
(def error? a/error?)
|