SWIFT-1326 Run load balancer spec tests

Sources/MongoSwift/Operations/ListCollectionsOperation.swift

@@ -72,7 +72,7 @@ public struct CollectionSpecification: Codable {
 }
 
 /// Options to use when executing a `listCollections` command on a `MongoDatabase`.
-public struct ListCollectionsOptions: Encodable {
+public struct ListCollectionsOptions: Codable {
     /// The batchSize for the returned cursor.
     public var batchSize: Int?
 

Sources/TestsCommon/APMUtils.swift

@@ -31,7 +31,7 @@ public class TestCommandMonitor: CommandEventHandler {
 
     /// Retrieve all the events seen so far that match the optionally provided filters, clearing the event cache.
     public func events(
-        withEventTypes typeFilter: [CommandEvent.EventType]? = nil,
+        withEventTypes typeFilter: [EventType]? = nil,
         withNames nameFilter: [String]? = nil
     ) -> [CommandEvent] {
         defer { self.events.removeAll() }
@@ -58,22 +58,22 @@ public class TestCommandMonitor: CommandEventHandler {
     }
 }
 
-extension CommandEvent {
-    public enum EventType: String, Decodable {
-        case commandStarted = "commandStartedEvent"
-        case commandSucceeded = "commandSucceededEvent"
-        case commandFailed = "commandFailedEvent"
-    }
+public enum EventType: String, Decodable {
+    case commandStartedEvent, commandSucceededEvent, commandFailedEvent,
+         connectionCreatedEvent, connectionReadyEvent, connectionClosedEvent,
+         connectionCheckedInEvent, connectionCheckedOutEvent, connectionCheckOutFailedEvent,
+         poolCreatedEvent, poolReadyEvent, poolClearedEvent, poolClosedEvent
+}
 
-    /// The "type" of this event. Used for filtering events by their type.
+extension CommandEvent {
     public var type: EventType {
         switch self {
         case .started:
-            return .commandStarted
+            return .commandStartedEvent
         case .failed:
-            return .commandFailed
+            return .commandFailedEvent
         case .succeeded:
-            return .commandSucceeded
+            return .commandSucceededEvent
         }
     }
 

Tests/LinuxMain.swift

@@ -111,6 +111,12 @@ extension EventLoopBoundMongoClientTests {
     ]
 }
 
+extension LoadBalancerTests {
+    static var allTests = [
+        ("testLoadBalancers", testLoadBalancers),
+    ]
+}
+
 extension MongoClientTests {
     static var allTests = [
         ("testUsingClosedClient", testUsingClosedClient),
@@ -397,6 +403,7 @@ XCTMain([
     testCase(CrudTests.allTests),
     testCase(DNSSeedlistTests.allTests),
     testCase(EventLoopBoundMongoClientTests.allTests),
+    testCase(LoadBalancerTests.allTests),
     testCase(MongoClientTests.allTests),
     testCase(MongoCollectionTests.allTests),
     testCase(MongoCollection_BulkWriteTests.allTests),

Tests/MongoSwiftSyncTests/LoadBalancerTests.swift

@@ -0,0 +1,65 @@
+import MongoSwiftSync
+import Nimble
+import TestsCommon
+
+final class LoadBalancerTests: MongoSwiftTestCase {
+    let skipFiles: [String] = [
+        // We don't support this option.
+        "wait-queue-timeouts.json"
+    ]
+
+    func testLoadBalancers() throws {
+        let tests = try retrieveSpecTestFiles(
+            specName: "load-balancers",
+            excludeFiles: skipFiles,
+            asType: UnifiedTestFile.self
+        ).map { $0.1 }
+
+        let skipTests = [
+            // The sessions spec requires that sessions can only be used with the MongoClient that created them.
+            // Consequently, libmongoc enforces that a `mongoc_client_session_t` may only be used with the
+            // `mongoc_client_t` that created it. In Swift, this translates to a requirement that we always pin
+            // `Connection`s to `ClientSession`s from the time the session is first used until it is closed/
+            // deinitialized. Since all of these tests use a session entity that is created before the tests are run
+            // and closed after they complete, the session always has the connection pinned to it, and we never release
+            // the connection as these tests expect.
+            "transactions are correctly pinned to connections for load-balanced clusters": [
+                "pinned connection is released after a non-transient abort error",
+                "pinned connection is released after a transient non-network CRUD error",
+                "pinned connection is released after a transient network CRUD error",
+                "pinned connection is released after a transient non-network commit error",
+                "pinned connection is released after a transient network commit error",
+                "pinned connection is released after a transient non-network abort error",
+                "pinned connection is released after a transient network abort error",
+                "pinned connection is released on successful abort",
+                "pinned connection is returned when a new transaction is started",
+                "pinned connection is returned when a non-transaction operation uses the session",
+                "a connection can be shared by a transaction and a cursor"
+            ],
+            "cursors are correctly pinned to connections for load-balanced clusters": [
+                // This test assumes that we release a cursor's pinned connection as soon as the cursor is exhausted
+                // server-side, regardless of if it has been fully iterated. However, we do not release connections
+                // until the first iteration attempt after the last document in the cursor.
+                "no connection is pinned if all documents are returned in the initial batch",
+                // This test assumes that we release a cursor's pinned connection after the last document is returned.
+                // However, currently there is no way for us to reliably tell a libmongoc cursor is at its end without
+                // attempting to iterate it first. To avoid having to implement some sort of caching mechanism we do
+                // not close cursors until we attempt to iterate and get nil back, so this cursor is not closed/its
+                // connection is not released after 3 iterations, as the test expects.
+                "pinned connections are returned when the cursor is drained",
+                // These tests assume we do not automatically close the cursor when we encounter such errors, however
+                // we close cursors on all errors besides decoding errors, so the connections do get returned.
+                "pinned connections are not returned after an network error during getMore",
+                "pinned connections are not returned to the pool after a non-network error on getMore",
+                // TODO: SWIFT-1322 Unskip.
+                "listCollections pins the cursor to a connection",
+                // Skipping as we do not support a batchSize for listIndexes. We closed SWIFT-1325 as "won't fix", but
+                // should we ever decide to do it we could unskip this test then.
+                "listIndexes pins the cursor to a connection"
+            ]
+        ]
+
+        let runner = try UnifiedTestRunner()
+        try runner.runFiles(tests, skipTests: skipTests)
+    }
+}

Tests/MongoSwiftSyncTests/SpecTestRunner/CodableExtensions.swift

@@ -142,7 +142,7 @@ extension MongoClientOptions: StrictDecodable {
     internal typealias CodingKeysType = CodingKeys
 
     internal enum CodingKeys: String, CodingKey, CaseIterable {
-        case retryReads, retryWrites, w, readConcernLevel, readPreference, heartbeatFrequencyMS
+        case retryReads, retryWrites, w, readConcernLevel, readPreference, heartbeatFrequencyMS, loadBalanced, appname
     }
 
     public init(from decoder: Decoder) throws {
@@ -155,8 +155,12 @@ extension MongoClientOptions: StrictDecodable {
         let retryWrites = try container.decodeIfPresent(Bool.self, forKey: .retryWrites)
         let writeConcern = try? WriteConcern(w: container.decode(WriteConcern.W.self, forKey: .w))
         let heartbeatFrequencyMS = try container.decodeIfPresent(Int.self, forKey: .heartbeatFrequencyMS)
+        let loadBalanced = try container.decodeIfPresent(Bool.self, forKey: .loadBalanced)
+        let appName = try container.decodeIfPresent(String.self, forKey: .appname)
         self.init(
+            appName: appName,
             heartbeatFrequencyMS: heartbeatFrequencyMS,
+            loadBalanced: loadBalanced,
             readConcern: readConcern,
             readPreference: readPreference,
             retryReads: retryReads,

Tests/MongoSwiftSyncTests/SyncChangeStreamTests.swift

@@ -443,7 +443,10 @@ final class SyncChangeStreamTests: MongoSwiftTestCase {
             return
         }
 
-        let events = try captureCommandEvents(eventTypes: [.commandStarted], commandNames: ["aggregate"]) { client in
+        let events = try captureCommandEvents(
+            eventTypes: [.commandStartedEvent],
+            commandNames: ["aggregate"]
+        ) { client in
             try withTestNamespace(client: client) { _, coll in
                 let options = ChangeStreamOptions(
                     batchSize: 123,

Tests/MongoSwiftSyncTests/SyncTestUtils.swift

@@ -141,7 +141,7 @@ extension MongoClient {
 /// Captures any command monitoring events filtered by type and name that are emitted during the execution of the
 /// provided closure. A client pre-configured for command monitoring is passed into the closure.
 internal func captureCommandEvents(
-    eventTypes: [CommandEvent.EventType]? = nil,
+    eventTypes: [EventType]? = nil,
     commandNames: [String]? = nil,
     f: (MongoClient) throws -> Void
 ) throws -> [CommandEvent] {

Tests/MongoSwiftSyncTests/UnifiedTestRunner/EntityDescription.swift

@@ -32,7 +32,7 @@ enum EntityDescription: Decodable {
 
         /// Types of events that can be observed for this client. Unspecified event types MUST be ignored by this
         /// client's event listeners.
-        let observeEvents: [CommandEvent.EventType]?
+        let observeEvents: [EventType]?
 
         /// Command names for which the test runner MUST ignore any observed command monitoring events. The command(s)
         /// will be ignored in addition to configureFailPoint and any commands containing sensitive information (per
@@ -166,10 +166,10 @@ struct UnifiedTestClient {
 class UnifiedTestCommandMonitor: CommandEventHandler {
     private var monitoring: Bool
     var events: [CommandEvent]
-    private let observeEvents: [CommandEvent.EventType]
+    private let observeEvents: [EventType]
     private let ignoreEvents: [String]
 
-    public init(observeEvents: [CommandEvent.EventType]?, ignoreEvents: [String]?) {
+    public init(observeEvents: [EventType]?, ignoreEvents: [String]?) {
         self.events = []
         self.monitoring = false
         self.observeEvents = observeEvents ?? []

Tests/MongoSwiftSyncTests/UnifiedTestRunner/ExpectedEventsForClient.swift

@@ -6,10 +6,35 @@ struct ExpectedEventsForClient: Decodable {
     /// Client entity on which the events are expected to be observed.
     let client: String
 
+    enum EventType: String, Decodable {
+        case command, cmap
+    }
+
+    /// The type of events to be observed.
+    let eventType: EventType
+
     /// List of events, which are expected to be observed (in this order) on the corresponding client while executing
     /// operations. If the array is empty, the test runner MUST assert that no events were observed on the client
     /// (excluding ignored events).
     let events: [ExpectedEvent]
+
+    enum CodingKeys: String, CodingKey {
+        case client, eventType, events
+    }
+
+    init(from decoder: Decoder) throws {
+        let container = try decoder.container(keyedBy: CodingKeys.self)
+        self.client = try container.decode(String.self, forKey: .client)
+        // Defaults to command if omitted.
+        self.eventType = try container.decodeIfPresent(EventType.self, forKey: .eventType) ?? .command
+        switch self.eventType {
+        case .command:
+            self.events = try container.decode([ExpectedEvent].self, forKey: .events)
+        case .cmap:
+            // TODO: SWIFT-1321 actually parse these out.
+            self.events = []
+        }
+    }
 }
 
 /// Describes expected events.

Tests/MongoSwiftSyncTests/UnifiedTestRunner/HasListableProperties.swift

@@ -23,3 +23,4 @@ extension UpdateModelOptions: HasListableProperties {}
 extension UpdateOptions: HasListableProperties {}
 extension ReplaceOneModelOptions: HasListableProperties {}
 extension ChangeStreamOptions: HasListableProperties {}
+extension ListCollectionsOptions: HasListableProperties {}

Tests/MongoSwiftSyncTests/UnifiedTestRunner/UnifiedClientOperations.swift

@@ -1,4 +1,28 @@
-import MongoSwiftSync
+@testable import MongoSwift
+@testable import MongoSwiftSync
+import Nimble
+
+struct AssertNumberConnectionsCheckedOut: UnifiedOperationProtocol {
+    /// The name of the client entity to perform the assertion on.
+    let client: String
+
+    /// The number of connections expected to be checked out.
+    let connections: Int
+
+    static var knownArguments: Set<String> {
+        ["client", "connections"]
+    }
+
+    func execute(on _: UnifiedOperation.Object, context: Context) throws -> UnifiedOperationResult {
+        let testClient = try context.entities.getEntity(id: self.client).asTestClient()
+        let actualCheckedOut = testClient.client.asyncClient.connectionPool.checkedOutConnections
+        expect(actualCheckedOut).to(
+            equal(self.connections),
+            description: "Number of checked out connections did not match expected. Path: \(context.path)"
+        )
+        return .none
+    }
+}
 
 struct UnifiedListDatabases: UnifiedOperationProtocol {
     static var knownArguments: Set<String> { [] }

Tests/MongoSwiftSyncTests/UnifiedTestRunner/UnifiedCollectionOperations.swift

@@ -75,6 +75,44 @@ struct UnifiedCreateIndex: UnifiedOperationProtocol {
     }
 }
 
+struct UnifiedListIndexes: UnifiedOperationProtocol {
+    /// Optional identifier for a session entity to use.
+    let session: String?
+
+    /// We consider this a known argument and decode it even though we don't support it, because a load balancer test
+    /// file uses this option and we could not decode/run the entire file otherwise.
+    let batchSize: Int?
+
+    private enum CodingKeys: String, CodingKey, CaseIterable {
+        case session, batchSize
+    }
+
+    static var knownArguments: Set<String> {
+        Set(
+            CodingKeys.allCases.map { $0.rawValue }
+        )
+    }
+
+    init(from decoder: Decoder) throws {
+        let container = try decoder.container(keyedBy: CodingKeys.self)
+        self.session = try container.decodeIfPresent(String.self, forKey: .session)
+        self.batchSize = try container.decodeIfPresent(Int.self, forKey: .batchSize)
+    }
+
+    func execute(on object: UnifiedOperation.Object, context: Context) throws -> UnifiedOperationResult {
+        guard self.batchSize == nil else {
+            throw TestError(
+                message: "listIndexes operation specifies a batchSize, but we do not support the option -- you may " +
+                    "need to skip this test. Path: \(context.path)"
+            )
+        }
+        let collection = try context.entities.getEntity(from: object).asCollection()
+        let session = try context.entities.resolveSession(id: self.session)
+        let results = try collection.listIndexes(session: session)
+        return .rootDocumentArray(try results.map { try $0.get() }.map { try BSONEncoder().encode($0) })
+    }
+}
+
 struct UnifiedBulkWrite: UnifiedOperationProtocol {
     /// Writes to perform.
     let requests: [WriteModel<BSONDocument>]

Tests/MongoSwiftSyncTests/UnifiedTestRunner/UnifiedDatabaseOperations.swift

@@ -1,3 +1,4 @@
+import MongoSwiftSync
 import SwiftBSON
 struct UnifiedCreateCollection: UnifiedOperationProtocol {
     /// The collection to create.
@@ -63,3 +64,38 @@ struct UnifiedRunCommand: UnifiedOperationProtocol {
         return .none
     }
 }
+
+struct UnifiedListCollections: UnifiedOperationProtocol {
+    /// Filter to use for the command.
+    let filter: BSONDocument
+
+    /// Optional identifier for a session entity to use.
+    let session: String?
+
+    let options: ListCollectionsOptions
+
+    enum CodingKeys: String, CodingKey, CaseIterable {
+        case filter, session
+    }
+
+    init(from decoder: Decoder) throws {
+        self.options = try decoder.singleValueContainer().decode(ListCollectionsOptions.self)
+        let container = try decoder.container(keyedBy: CodingKeys.self)
+        self.session = try container.decodeIfPresent(String.self, forKey: .session)
+        self.filter = try container.decode(BSONDocument.self, forKey: .filter)
+    }
+
+    static var knownArguments: Set<String> {
+        Set(
+            CodingKeys.allCases.map { $0.rawValue } +
+                ListCollectionsOptions().propertyNames
+        )
+    }
+
+    func execute(on object: UnifiedOperation.Object, context: Context) throws -> UnifiedOperationResult {
+        let db = try context.entities.getEntity(from: object).asDatabase()
+        let session = try context.entities.resolveSession(id: self.session)
+        let results = try db.listCollections(self.filter, options: self.options, session: session)
+        return .rootDocumentArray(try results.map { try $0.get() }.map { try BSONEncoder().encode($0) })
+    }
+}

Tests/MongoSwiftSyncTests/UnifiedTestRunner/UnifiedOperation.swift

@@ -154,6 +154,8 @@ struct UnifiedOperation: Decodable {
             self.operation = try container.decode(UnifiedAssertIndexNotExists.self, forKey: .arguments)
         case "assertDifferentLsidOnLastTwoCommands":
             self.operation = try container.decode(AssertDifferentLsidOnLastTwoCommands.self, forKey: .arguments)
+        case "assertNumberConnectionsCheckedOut":
+            self.operation = try container.decode(AssertNumberConnectionsCheckedOut.self, forKey: .arguments)
         case "assertSameLsidOnLastTwoCommands":
             self.operation = try container.decode(AssertSameLsidOnLastTwoCommands.self, forKey: .arguments)
         case "assertSessionDirty":
@@ -214,8 +216,12 @@ struct UnifiedOperation: Decodable {
             self.operation = try container.decode(UnifiedInsertMany.self, forKey: .arguments)
         case "iterateUntilDocumentOrError":
             self.operation = IterateUntilDocumentOrError()
+        case "listCollections":
+            self.operation = try container.decode(UnifiedListCollections.self, forKey: .arguments)
         case "listDatabases":
             self.operation = UnifiedListDatabases()
+        case "listIndexes":
+            self.operation = try container.decode(UnifiedListIndexes.self, forKey: .arguments)
         case "replaceOne":
             self.operation = try container.decode(UnifiedReplaceOne.self, forKey: .arguments)
         case "runCommand":

Tests/MongoSwiftSyncTests/UnifiedTestRunner/UnifiedTestRunner.swift

@@ -207,7 +207,8 @@ struct UnifiedTestRunner {
                     }
 
                     if let expectEvents = test.expectEvents {
-                        for expectedEventList in expectEvents {
+                        // TODO: SWIFT-1321 don't skip CMAP event assertions here.
+                        for expectedEventList in expectEvents where expectedEventList.eventType != .cmap {
                             let clientId = expectedEventList.client
 
                             guard let actualEvents = clientEvents[clientId] else {