github/penpot
1
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2026-10-29 05:50:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

✨ Add test hooks for websocket listener options

Browse Source 
The generic websocket listener now returns the handler map
with the resolved options attached in its metadata (under the
::options key), exposing the codecs and internal state for
testing and introspection purposes without affecting the
public api.

The http-handler of the notifications endpoint is now public,
and the websocket tests are extended to cover the wiring of
the negotiated payload format into the listener: json codecs
selected via Accept header or _fmt parameter, and the transit
default for unsignaled requests.

Closes #11409

AI-assisted-by: glm-5.3-flash
This commit is contained in:
Andrey Antukh 2026-08-28 10:16:02 +00:00
parent 5e8f6ef186
commit a8ae691875
3 changed files with 180 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
backend/src/app/http/websocket.clj
View File

@ -340,7 +340,10 @@
:inc 1)
message)
(defn- http-handler
(defn http-handler
"The websocket upgrade request handler. Returns the upgrade
response containing the websocket listener built with the
codecs of the negotiated payload format."
[cfg {:keys [params ::session/profile-id] :as request}]
(let [session-id (some-> params :session-id uuid/parse*)
format (cnegot/negotiate-format request)]

62
backend/src/app/util/websocket.clj
View File

@ -63,7 +63,11 @@
It also accepts some options that allows you parametrize the
protocol behavior. The options map will be used as-as for the
initial data of the `ws` data structure"
initial data of the `ws` data structure.
The returned map has the resolved options attached in its
metadata under the `::options` key; it is intended for testing
and introspection purposes only."
[request & {:keys [::on-rcv-message
::on-snd-message
::on-connect
@ -110,38 +114,40 @@
(assoc ::encode-fn encode-fn)
(assoc ::decode-fn decode-fn))]
{:on-open
(fn on-open [channel]
(l/dbg :fn "on-open" :conn-id (str id))
(let [options (-> options
(assoc ::channel channel)
(on-connect))
timeout (ct/duration idle-timeout)]
(with-meta
{:on-open
(fn on-open [channel]
(l/dbg :fn "on-open" :conn-id (str id))
(let [options (-> options
(assoc ::channel channel)
(on-connect))
timeout (ct/duration idle-timeout)]
(yws/set-idle-timeout! channel timeout)
(px/submit! :vthread (partial start-io-loop! options))))
(yws/set-idle-timeout! channel timeout)
(px/submit! :vthread (partial start-io-loop! options))))
:on-close
(fn on-close [_channel code reason]
(l/dbg :fn "on-close"
:conn-id (str id)
:code code
:reason reason)
(sp/close! close-ch))
:on-close
(fn on-close [_channel code reason]
(l/dbg :fn "on-close"
:conn-id (str id)
:code code
:reason reason)
(sp/close! close-ch))
:on-error
(fn on-error [_channel cause]
(sp/close! close-ch cause))
:on-error
(fn on-error [_channel cause]
(sp/close! close-ch cause))
:on-message
(fn on-message [_channel message]
(when (string? message)
(sp/offer! input-ch message)
(swap! state assoc ::last-activity-at (ct/now))))
:on-message
(fn on-message [_channel message]
(when (string? message)
(sp/offer! input-ch message)
(swap! state assoc ::last-activity-at (ct/now))))
:on-pong
(fn on-pong [_channel data]
(sp/put! hbeat-ch data))}))
:on-pong
(fn on-pong [_channel data]
(sp/put! hbeat-ch data))}
{::options options})))
(defn- handle-ping!
[{:keys [::id ::beats ::channel] :as wsp} beat-id]

160
backend/test/backend_tests/http_websocket_test.clj
View File

@ -10,27 +10,29 @@
[app.common.transit :as tr]
[app.common.uuid :as uuid]
[app.db :as db]
[app.http.websocket :as ws]
[app.http.session :as-alias session]
[app.http.websocket :as http.ws]
[app.msgbus :as mbus]
[app.nitrate :as nitrate]
[app.rpc :as-alias rpc]
[app.rpc.commands.files :as files]
[app.rpc.commands.teams :as teams]
[app.util.websocket :as util-ws]
[app.util.websocket :as ws]
[backend-tests.helpers :as th]
[clojure.test :as t]
[promesa.exec.csp :as sp]))
[promesa.exec.csp :as sp]
[yetti.websocket :as yws]))
(t/use-fixtures :once th/state-init)
(t/use-fixtures :each th/database-reset)
(defn make-wsp
[profile-id state output-ch]
{::util-ws/id (uuid/next)
::util-ws/state state
::util-ws/output-ch output-ch
::ws/profile-id profile-id
::ws/session-id (uuid/next)})
{::ws/id (uuid/next)
::ws/state state
::ws/output-ch output-ch
::http.ws/profile-id profile-id
::http.ws/session-id (uuid/next)})
(t/deftest subscribe-file-permission-check
(let [profile1 (th/create-profile* 1 {:is-active true})
@ -46,7 +48,7 @@
(t/is (thrown-with-msg?
clojure.lang.ExceptionInfo
#"not found"
((get-method ws/handle-message :subscribe-file)
((get-method http.ws/handle-message :subscribe-file)
cfg wsp {:file-id (:id file)})))))
(t/testing "permission check passes for authorized user"
@ -65,7 +67,7 @@
(t/is (thrown-with-msg?
clojure.lang.ExceptionInfo
#"not found"
((get-method ws/handle-message :subscribe-team)
((get-method http.ws/handle-message :subscribe-team)
cfg wsp {:team-id (:id team)})))))
(t/testing "permission check passes for authorized user"
@ -82,11 +84,11 @@
(with-redefs [nitrate/call nitrate-call
mbus/sub! (fn [_ & {:keys [topics]}]
(swap! calls conj topics))]
((get-method ws/handle-message :subscribe-team)
((get-method http.ws/handle-message :subscribe-team)
th/*system* wsp {:team-id team-id}))
(some-> @state ::ws/team-subscription :channel sp/close!)
(some-> @state ::http.ws/team-subscription :channel sp/close!)
{:topics @calls
:subscription (::ws/team-subscription @state)}))
:subscription (::http.ws/team-subscription @state)}))
(t/deftest subscribe-team-subscribes-to-team-organization
(let [profile (th/create-profile* 1 {:is-active true})
@ -122,9 +124,9 @@
cfg th/*system*
file-id (:id file)
sub-ch (sp/chan :buf (sp/dropping-buffer 64))
state (atom {::ws/file-subscription {:file-id file-id
:channel sub-ch
:topic file-id}})
state (atom {::http.ws/file-subscription {:file-id file-id
:channel sub-ch
:topic file-id}})
output-ch (sp/chan :buf (sp/dropping-buffer 64))
wsp (make-wsp (:id profile) state output-ch)]
@ -134,7 +136,7 @@
:position {:x 10 :y 20}
:zoom 1.0}]
(t/is (nil?
((get-method ws/handle-message :pointer-update)
((get-method http.ws/handle-message :pointer-update)
cfg wsp wrong-msg)))))
(t/testing "does nothing when no file subscription exists"
@ -145,7 +147,7 @@
:position {:x 10 :y 20}
:zoom 1.0}]
(t/is (nil?
((get-method ws/handle-message :pointer-update)
((get-method http.ws/handle-message :pointer-update)
cfg empty-wsp msg)))))))
(t/deftest resolve-encoder-defaults-to-transit
@ -212,3 +214,125 @@
{:type "subscribe-team" :team-id (str team-id)})]
(t/is (= :subscribe-team (:type message)))
(t/is (= team-id (:team-id message)))))
(t/deftest resolve-encoder-defaults-to-transit
(let [encode (http.ws/resolve-encoder :transit)
msg {:type :join-file
:file-id (uuid/next)
:profile-id (uuid/next)}]
(t/is (= msg (tr/decode-str (encode msg))))))
(t/deftest resolve-encoder-unexpected-format-falls-back-to-transit
(let [encode (http.ws/resolve-encoder :foo)
msg {:type :join-file :file-id (uuid/next)}]
(t/is (= msg (tr/decode-str (encode msg))))))
(t/deftest resolve-encoder-json
(let [encode (http.ws/resolve-encoder :json)
file-id (uuid/next)
encoded (encode {:type :join-file :file-id file-id})]
(t/is (string? encoded))
(t/is (= "join-file" (get (json/decode encoded) "type")))
(t/is (= (str file-id) (get (json/decode encoded) "fileId")))))
(t/deftest resolve-decoder-defaults-to-transit
(let [decode (http.ws/resolve-decoder :transit)
msg {:type :subscribe-file :file-id (uuid/next)}]
(t/is (= msg (decode (tr/encode-str msg {:type :json-verbose}))))))
(t/deftest resolve-decoder-unexpected-format-falls-back-to-transit
(let [decode (http.ws/resolve-decoder :foo)
msg {:type :subscribe-file :file-id (uuid/next)}]
(t/is (= msg (decode (tr/encode-str msg {:type :json-verbose}))))))
(t/deftest resolve-decoder-json
(let [decode (http.ws/resolve-decoder :json)
file-id (uuid/next)
decoded (decode (str "{\"type\":\"subscribe-file\","
"\"fileId\":\"" file-id "\"}"))]
(t/is (= "subscribe-file" (:type decoded)))
(t/is (= (str file-id) (:file-id decoded)))))
(t/deftest normalize-message-from-json-decoding
(let [file-id (uuid/next)
decoded {:type "subscribe-file"
:file-id (str file-id)}
message (http.ws/normalize-message decoded)]
(t/is (= :subscribe-file (:type message)))
(t/is (uuid? (:file-id message)))
(t/is (= file-id (:file-id message)))))
(t/deftest normalize-message-noop-for-transit-decoding
(let [file-id (uuid/next)
decoded {:type :subscribe-file :file-id file-id}]
(t/is (= decoded (http.ws/normalize-message decoded)))))
(t/deftest normalize-message-with-invalid-uuid
(let [message (http.ws/normalize-message
{:type "subscribe-file" :file-id "invalid"})]
(t/is (= :subscribe-file (:type message)))
(t/is (nil? (:file-id message)))))
(t/deftest normalize-message-team-id
(let [team-id (uuid/next)
message (http.ws/normalize-message
{:type "subscribe-team" :team-id (str team-id)})]
(t/is (= :subscribe-team (:type message)))
(t/is (= team-id (:team-id message)))))
(defn- ws-request
[{:keys [headers params]}]
{:params (merge {:session-id (str (uuid/next))} params)
::session/profile-id (uuid/next)
:headers (merge {"upgrade" "websocket"
"connection" "Upgrade"}
headers)})
(defn- listener-options
[request]
(let [response (http.ws/http-handler {} request)
listener (::yws/listener response)]
(t/is (map? listener))
(t/is (fn? (:on-open listener)))
(t/is (fn? (:on-close listener)))
(t/is (fn? (:on-error listener)))
(t/is (fn? (:on-message listener)))
(t/is (fn? (:on-pong listener)))
(::ws/options (meta listener))))
(t/deftest http-handler-negotiates-json-codecs
(let [request (ws-request {:headers {"accept" "application/json"}})
options (listener-options request)
encode (::ws/encode-fn options)
decode (::ws/decode-fn options)
file-id (uuid/next)
raw (str "{\"type\":\"subscribe-file\",\"requestId\":42,"
"\"fileId\":\"" file-id "\"}")]
(let [decoded (decode raw)]
(t/is (= "subscribe-file" (:type decoded)))
(t/is (= 42 (:request-id decoded)))
(t/is (= (str file-id) (:file-id decoded))))
(let [encoded (encode {:type :join-file
:file-id file-id
:request-id 42})]
(t/is (string? encoded))
(t/is (= "join-file" (get (json/decode encoded) "type")))
(t/is (= 42 (get (json/decode encoded) "requestId")))
(t/is (= (str file-id) (get (json/decode encoded) "fileId"))))))
(t/deftest http-handler-negotiates-json-codecs-from-fmt-param
(let [request (-> (ws-request {:headers {"accept" "application/transit+json"}})
(assoc :query-params {:_fmt "json"})
(assoc-in [:params :_fmt] "json"))
options (listener-options request)
encoded ((::ws/encode-fn options) {:type :join-file})]
(t/is (string? encoded))
(t/is (= "join-file" (get (json/decode encoded) "type")))))
(t/deftest http-handler-defaults-to-transit-codecs
(let [request (ws-request {})
options (listener-options request)
encode (::ws/encode-fn options)
decode (::ws/decode-fn options)
msg {:type :join-file :file-id (uuid/next)}]
(t/is (= msg (decode (encode msg))))))