From a8ae691875fa27f3a16959b61ded5d23cf7b4adc Mon Sep 17 00:00:00 2001 From: Andrey Antukh Date: Fri, 28 Aug 2026 10:16:02 +0000 Subject: [PATCH] :sparkles: Add test hooks for websocket listener options 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 --- backend/src/app/http/websocket.clj | 5 +- backend/src/app/util/websocket.clj | 62 ++++--- .../backend_tests/http_websocket_test.clj | 160 ++++++++++++++++-- 3 files changed, 180 insertions(+), 47 deletions(-) diff --git a/backend/src/app/http/websocket.clj b/backend/src/app/http/websocket.clj index 760818b086..8724ad4778 100644 --- a/backend/src/app/http/websocket.clj +++ b/backend/src/app/http/websocket.clj @@ -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)] diff --git a/backend/src/app/util/websocket.clj b/backend/src/app/util/websocket.clj index 6e7dd9a3a1..ee5af1329c 100644 --- a/backend/src/app/util/websocket.clj +++ b/backend/src/app/util/websocket.clj @@ -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] diff --git a/backend/test/backend_tests/http_websocket_test.clj b/backend/test/backend_tests/http_websocket_test.clj index 9a8e9ba0ab..d19d965015 100644 --- a/backend/test/backend_tests/http_websocket_test.clj +++ b/backend/test/backend_tests/http_websocket_test.clj @@ -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))))))