✨ 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

@@ -314,7 +314,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)]

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]

backend/test/backend_tests/http_websocket_test.clj

@@ -9,8 +9,11 @@
    [app.common.json :as json]
    [app.common.transit :as tr]
    [app.common.uuid :as uuid]
+   [app.http.session :as-alias session]
    [app.http.websocket :as http.ws]
-   [clojure.test :as t]))
+   [app.util.websocket :as ws]
+   [clojure.test :as t]
+   [yetti.websocket :as yws]))
 
 (t/deftest resolve-encoder-defaults-to-transit
   (let [encode (http.ws/resolve-encoder :transit)
@@ -76,3 +79,60 @@
                  {: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))))))