Fix Realtime tests

Supabase.xctestplan

@@ -60,6 +60,13 @@
         "identifier" : "AuthTests",
         "name" : "AuthTests"
       }
+    },
+    {
+      "target" : {
+        "containerPath" : "container:",
+        "identifier" : "_HelpersTests",
+        "name" : "_HelpersTests"
+      }
     }
   ],
   "version" : 1

Tests/RealtimeTests/CallbackManagerTests.swift

@@ -5,6 +5,7 @@
 //  Created by Guilherme Souza on 26/12/23.
 //
 
+import ConcurrencyExtras
 import CustomDump
 @testable import Realtime
 import XCTest
@@ -98,22 +99,22 @@ final class CallbackManagerTests: XCTestCase {
       deleteSpecificUserFilter,
     ])
 
-    var receivedActions: [AnyAction] = []
+    let receivedActions = LockIsolated<[AnyAction]>([])
     let updateUsersId = callbackManager.addPostgresCallback(filter: updateUsersFilter) { action in
-      receivedActions.append(action)
+      receivedActions.withValue { $0.append(action) }
     }
 
     let insertUsersId = callbackManager.addPostgresCallback(filter: insertUsersFilter) { action in
-      receivedActions.append(action)
+      receivedActions.withValue { $0.append(action) }
     }
 
     let anyUsersId = callbackManager.addPostgresCallback(filter: anyUsersFilter) { action in
-      receivedActions.append(action)
+      receivedActions.withValue { $0.append(action) }
     }
 
     let deleteSpecificUserId = callbackManager
       .addPostgresCallback(filter: deleteSpecificUserFilter) { action in
-        receivedActions.append(action)
+        receivedActions.withValue { $0.append(action) }
       }
 
     let currentDate = Date()
@@ -122,14 +123,16 @@ final class CallbackManagerTests: XCTestCase {
       columns: [],
       commitTimestamp: currentDate,
       record: ["email": .string("[email protected]")],
-      oldRecord: ["email": .string("[email protected]")]
+      oldRecord: ["email": .string("[email protected]")],
+      rawMessage: RealtimeMessageV2(joinRef: nil, ref: nil, topic: "", event: "", payload: [:])
     )
     callbackManager.triggerPostgresChanges(ids: [updateUsersId], data: .update(updateUserAction))
 
     let insertUserAction = InsertAction(
       columns: [],
       commitTimestamp: currentDate,
-      record: ["email": .string("[email protected]")]
+      record: ["email": .string("[email protected]")],
+      rawMessage: RealtimeMessageV2(joinRef: nil, ref: nil, topic: "", event: "", payload: [:])
     )
     callbackManager.triggerPostgresChanges(ids: [insertUsersId], data: .insert(insertUserAction))
 
@@ -139,15 +142,16 @@ final class CallbackManagerTests: XCTestCase {
     let deleteSpecificUserAction = DeleteAction(
       columns: [],
       commitTimestamp: currentDate,
-      oldRecord: ["id": .string("1234")]
+      oldRecord: ["id": .string("1234")],
+      rawMessage: RealtimeMessageV2(joinRef: nil, ref: nil, topic: "", event: "", payload: [:])
     )
     callbackManager.triggerPostgresChanges(
       ids: [deleteSpecificUserId],
       data: .delete(deleteSpecificUserAction)
     )
 
     XCTAssertNoDifference(
-      receivedActions,
+      receivedActions.value,
       [
         .update(updateUserAction),
         anyUserAction,
@@ -162,16 +166,22 @@ final class CallbackManagerTests: XCTestCase {
   func testTriggerBroadcast() {
     let callbackManager = CallbackManager()
     let event = "new_user"
-    let json = AnyJSON.object(["email": .string("[email protected]")])
+    let message = RealtimeMessageV2(
+      joinRef: nil,
+      ref: nil,
+      topic: "realtime:users",
+      event: event,
+      payload: ["email": "[email protected]"]
+    )
 
-    var receivedJSON: AnyJSON?
+    let receivedMessage = LockIsolated(RealtimeMessageV2?.none)
     callbackManager.addBroadcastCallback(event: event) {
-      receivedJSON = $0
+      receivedMessage.setValue($0)
     }
 
-    callbackManager.triggerBroadcast(event: event, json: json)
+    callbackManager.triggerBroadcast(event: event, message: message)
 
-    XCTAssertEqual(receivedJSON, json)
+    XCTAssertEqual(receivedMessage.value, message)
   }
 
   func testTriggerPresenceDiffs() {
@@ -183,18 +193,22 @@ final class CallbackManagerTests: XCTestCase {
     let joins = ["user1": Presence(channel: channel)]
     let leaves = ["user2": Presence(channel: channel)]
 
-    var receivedAction: PresenceAction?
+    let receivedAction = LockIsolated(PresenceAction?.none)
 
     callbackManager.addPresenceCallback {
-      receivedAction = $0
+      receivedAction.setValue($0)
     }
 
-    callbackManager.triggerPresenceDiffs(joins: joins, leaves: leaves)
+    callbackManager.triggerPresenceDiffs(
+      joins: joins,
+      leaves: leaves,
+      rawMessage: RealtimeMessageV2(joinRef: nil, ref: nil, topic: "", event: "", payload: [:])
+    )
 
-    XCTAssertIdentical(receivedAction?.joins["user1"], joins["user1"])
-    XCTAssertIdentical(receivedAction?.leaves["user2"], leaves["user2"])
+    XCTAssertIdentical(receivedAction.value?.joins["user1"], joins["user1"])
+    XCTAssertIdentical(receivedAction.value?.leaves["user2"], leaves["user2"])
 
-    XCTAssertEqual(receivedAction?.joins.count, 1)
-    XCTAssertEqual(receivedAction?.leaves.count, 1)
+    XCTAssertEqual(receivedAction.value?.joins.count, 1)
+    XCTAssertEqual(receivedAction.value?.leaves.count, 1)
   }
 }

Tests/RealtimeTests/MockWebSocketClient.swift

@@ -0,0 +1,65 @@
+//
+//  MockWebSocketClient.swift
+//
+//
+//  Created by Guilherme Souza on 29/12/23.
+//
+
+import ConcurrencyExtras
+import Foundation
+@testable import Realtime
+
+final class MockWebSocketClient: WebSocketClientProtocol {
+  struct MutableState {
+    var sentMessages: [RealtimeMessageV2] = []
+    var responsesHandlers: [(RealtimeMessageV2) -> RealtimeMessageV2?] = []
+    var receiveContinuation: AsyncThrowingStream<RealtimeMessageV2, Error>.Continuation?
+  }
+
+  let status: [Result<WebSocketClient.ConnectionStatus, Error>]
+  let mutableState = LockIsolated(MutableState())
+
+  init(status: [Result<WebSocketClient.ConnectionStatus, Error>]) {
+    self.status = status
+  }
+
+  func connect() -> AsyncThrowingStream<WebSocketClient.ConnectionStatus, Error> {
+    AsyncThrowingStream {
+      for result in status {
+        $0.yield(with: result)
+      }
+    }
+  }
+
+  func send(_ message: RealtimeMessageV2) async throws {
+    mutableState.withValue {
+      $0.sentMessages.append(message)
+
+      if let response = $0.responsesHandlers.lazy.compactMap({ $0(message) }).first {
+        $0.receiveContinuation?.yield(response)
+      }
+    }
+  }
+
+  func receive() -> AsyncThrowingStream<RealtimeMessageV2, Error> {
+    mutableState.withValue {
+      let (stream, continuation) = AsyncThrowingStream<RealtimeMessageV2, Error>.makeStream()
+      $0.receiveContinuation = continuation
+      return stream
+    }
+  }
+
+  func cancel() {
+    mutableState.receiveContinuation?.finish()
+  }
+
+  func when(_ handler: @escaping (RealtimeMessageV2) -> RealtimeMessageV2?) {
+    mutableState.withValue {
+      $0.responsesHandlers.append(handler)
+    }
+  }
+
+  func mockReceive(_ message: RealtimeMessageV2) {
+    mutableState.receiveContinuation?.yield(message)
+  }
+}