SWIFT-1610, SWIFT-1378, SWIFT-1632: Spec tests sync + expose clusterTime on ChangeStreamEvent (#775)

Author: kmahar

Sources/MongoSwift/ChangeStreamEvent.swift

@@ -122,6 +122,9 @@ public struct ChangeStreamEvent<T: Codable>: Codable {
     /// Only present for server versions 6.0 and above.
     public let wallTime: Date?
 
+    /// The cluster time at which the change occurred. Only present for server versions 4.0 and above.
+    public let clusterTime: BSONTimestamp?
+
     /**
      * Always present for operations of type `insert` and `replace`. Also present for operations of type `update` if
      * the user has specified `.updateLookup` for the `fullDocument` option in the `ChangeStreamOptions` used to create
@@ -136,7 +139,7 @@ public struct ChangeStreamEvent<T: Codable>: Codable {
     public let fullDocument: T?
 
     private enum CodingKeys: String, CodingKey {
-        case operationType, _id, ns, to, documentKey, updateDescription, wallTime, fullDocument
+        case operationType, _id, ns, to, documentKey, updateDescription, wallTime, clusterTime, fullDocument
     }
 
     // Custom decode method to work around the fact that `invalidate` events do not have an `ns` field in the raw
@@ -164,6 +167,7 @@ public struct ChangeStreamEvent<T: Codable>: Codable {
 
         self.documentKey = try container.decodeIfPresent(BSONDocument.self, forKey: .documentKey)
         self.wallTime = try container.decodeIfPresent(Date.self, forKey: .wallTime)
+        self.clusterTime = try container.decodeIfPresent(BSONTimestamp.self, forKey: .clusterTime)
         self.updateDescription = try container.decodeIfPresent(UpdateDescription.self, forKey: .updateDescription)
         self.fullDocument = try container.decodeIfPresent(T.self, forKey: .fullDocument)
     }

Sources/TestsCommon/APMUtils.swift

@@ -81,13 +81,11 @@ public class TestCommandMonitor: CommandEventHandler {
 }
 
 public enum EventType: String, Decodable {
-    case commandStartedEvent, commandSucceededEvent, commandFailedEvent,
-         connectionCreatedEvent, connectionReadyEvent, connectionClosedEvent,
-         connectionCheckedInEvent, connectionCheckedOutEvent, connectionCheckOutFailedEvent,
-         poolCreatedEvent, poolReadyEvent, poolClearedEvent, poolClosedEvent,
-         topologyDescriptionChanged, topologyOpening, topologyClosed, serverDescriptionChanged,
-         serverOpening, serverClosed, serverHeartbeatStarted, serverHeartbeatSucceeded,
-         serverHeartbeatFailed
+    case commandStartedEvent, commandSucceededEvent, commandFailedEvent, connectionCreatedEvent, connectionReadyEvent,
+         connectionClosedEvent, connectionCheckedInEvent, connectionCheckOutStartedEvent, connectionCheckedOutEvent,
+         connectionCheckOutFailedEvent, poolCreatedEvent, poolReadyEvent, poolClearedEvent, poolClosedEvent,
+         topologyDescriptionChanged, topologyOpening, topologyClosed, serverDescriptionChanged, serverOpening,
+         serverClosed, serverHeartbeatStarted, serverHeartbeatSucceeded, serverHeartbeatFailed
 }
 
 extension CommandEvent {

Tests/MongoSwiftTests/AsyncAwaitTestUtils.swift

@@ -165,9 +165,9 @@ extension MongoClient {
     }
 
     internal func getUnmetRequirement(_ testRequirement: TestRequirement) async throws -> UnmetRequirement? {
-        async let topologyType = try self.topologyType()
-        async let serverVersion = try self.serverVersion()
-        async let params = try self.serverParameters()
+        async let topologyType = try await self.topologyType()
+        async let serverVersion = try await self.serverVersion()
+        async let params = try await self.serverParameters()
         return try await testRequirement.getUnmetRequirement(givenCurrent: serverVersion, topologyType, params)
     }
 

Tests/MongoSwiftTests/AsyncAwaitTests.swift

@@ -234,7 +234,7 @@ final class MongoCursorAsyncAwaitTests: MongoSwiftTestCase {
         }
 
         testAsync {
-            let opts = CreateCollectionOptions(capped: true, size: 5)
+            let opts = CreateCollectionOptions(capped: true, max: 5, size: 100_000)
             try await self.withTestNamespace(collectionOptions: opts) { _, _, coll in
                 try await coll.insertMany([["x": 1], ["x": 2], ["x": 3]])
 

Tests/MongoSwiftTests/ChangeStreamTests.swift

@@ -171,7 +171,9 @@ final class ChangeStreamTests: MongoSwiftTestCase {
             // TODO: SWIFT-1458 Unskip.
             "change-streams-showExpandedEvents.json",
             // TODO: SWIFT-1472 Unskip.
-            "change-streams-pre_and_post_images.json"
+            "change-streams-pre_and_post_images.json",
+            // TODO: SWIFT-1614 Unskip.
+            "change-streams-disambiguatedPaths.json"
         ]
         let tests = try retrieveSpecTestFiles(
             specName: "change-streams",
@@ -182,5 +184,29 @@ final class ChangeStreamTests: MongoSwiftTestCase {
         let testRunner = try await UnifiedTestRunner()
         try await testRunner.runFiles(tests)
     }
+
+    func testClusterTime() async throws {
+        try await self.withTestClient { client in
+            // cluster time is only included as of 4.0.
+            let requirement = TestRequirement(
+                minServerVersion: ServerVersion(major: 4, minor: 0, patch: 0),
+                acceptableTopologies: [.replicaSet, .sharded, .shardedReplicaSet, .loadBalanced]
+            )
+            let unmetRequirement = try await client.getUnmetRequirement(requirement)
+            guard unmetRequirement == nil else {
+                printSkipMessage(testName: self.name, unmetRequirement: unmetRequirement!)
+                return
+            }
+            let db = client.db(Self.testDatabase)
+            try await db.collection(self.getCollectionName()).drop()
+            let coll = try await db.createCollection(self.getCollectionName())
+
+            let stream = try await coll.watch()
+
+            _ = try await coll.insertOne(["x": 1])
+            let event = try await stream.next()
+            expect(event?.clusterTime).toNot(beNil())
+        }
+    }
 }
 #endif

Tests/MongoSwiftTests/Failpoint.swift

@@ -29,9 +29,10 @@ extension FailPoint {
     /// Enables the failpoint, and returns a `EnabledFailpoint` which can handle disabling when needed
     internal func enable(
         using client: MongoClient,
-        options: RunCommandOptions? = nil
+        options: RunCommandOptions? = nil,
+        session: ClientSession? = nil
     ) async throws -> EnabledFailpoint {
-        try await client.db("admin").runCommand(self.failPoint, options: options)
+        try await client.db("admin").runCommand(self.failPoint, options: options, session: session)
         return EnabledFailpoint(failPoint: self, client: client)
     }
 

Tests/MongoSwiftTests/UnifiedTestRunner/SpecialTestOperations.swift

@@ -12,14 +12,18 @@ struct UnifiedFailPoint: UnifiedOperationProtocol {
     /// The client entity to use for setting the failpoint.
     let client: String
 
+    /// Optional name of a session entity to use for setting the failpoint.
+    let session: String?
+
     static var knownArguments: Set<String> {
-        ["failPoint", "client"]
+        ["failPoint", "client", "session"]
     }
 
     func execute(on _: UnifiedOperation.Object, context: Context) async throws -> UnifiedOperationResult {
         let testClient = try context.entities.getEntity(id: self.client).asTestClient()
+        let session = try context.entities.resolveSession(id: self.session)
         let opts = RunCommandOptions(readPreference: .primary)
-        let fpGuard = try await self.failPoint.enable(using: testClient.client, options: opts)
+        let fpGuard = try await self.failPoint.enable(using: testClient.client, options: opts, session: session)
         context.enabledFailPoints.append(fpGuard)
         return .none
     }

Tests/Specs/change-streams/tests/unified/change-streams-clusterTime.json

@@ -0,0 +1,81 @@
+{
+  "description": "change-streams-clusterTime",
+  "schemaVersion": "1.3",
+  "createEntities": [
+    {
+      "client": {
+        "id": "client0",
+        "useMultipleMongoses": false
+      }
+    },
+    {
+      "database": {
+        "id": "database0",
+        "client": "client0",
+        "databaseName": "database0"
+      }
+    },
+    {
+      "collection": {
+        "id": "collection0",
+        "database": "database0",
+        "collectionName": "collection0"
+      }
+    }
+  ],
+  "runOnRequirements": [
+    {
+      "minServerVersion": "4.0.0",
+      "topologies": [
+        "replicaset",
+        "sharded-replicaset",
+        "load-balanced",
+        "sharded"
+      ]
+    }
+  ],
+  "initialData": [
+    {
+      "collectionName": "collection0",
+      "databaseName": "database0",
+      "documents": []
+    }
+  ],
+  "tests": [
+    {
+      "description": "clusterTime is present",
+      "operations": [
+        {
+          "name": "createChangeStream",
+          "object": "collection0",
+          "arguments": {
+            "pipeline": []
+          },
+          "saveResultAsEntity": "changeStream0"
+        },
+        {
+          "name": "insertOne",
+          "object": "collection0",
+          "arguments": {
+            "document": {
+              "_id": 1
+            }
+          }
+        },
+        {
+          "name": "iterateUntilDocumentOrError",
+          "object": "changeStream0",
+          "expectResult": {
+            "ns": {
+              "db": "database0",
+              "coll": "collection0"
+            },
+            "clusterTime": {
+              "$$exists": true
+            }
+          }
+        }
+      ]
+    }
+  ]
+}

Tests/Specs/change-streams/tests/unified/change-streams-clusterTime.yml

@@ -0,0 +1,40 @@
+description: "change-streams-clusterTime"
+schemaVersion: "1.3"
+createEntities:
+  - client:
+      id: &client0 client0
+      useMultipleMongoses: false
+  - database:
+      id: &database0 database0
+      client: *client0
+      databaseName: *database0
+  - collection:
+      id: &collection0 collection0
+      database: *database0
+      collectionName: *collection0
+
+runOnRequirements:
+  - minServerVersion: "4.0.0"
+    topologies: [ replicaset, sharded-replicaset, load-balanced, sharded ]
+
+initialData:
+  - collectionName: *collection0
+    databaseName: *database0
+    documents: []
+
+tests:
+  - description: "clusterTime is present"
+    operations:
+      - name: createChangeStream
+        object: *collection0
+        arguments: { pipeline: [] }
+        saveResultAsEntity: &changeStream0 changeStream0
+      - name: insertOne
+        object: *collection0
+        arguments:
+          document: { _id: 1 }
+      - name: iterateUntilDocumentOrError
+        object: *changeStream0
+        expectResult:
+          ns: { db: *database0, coll: *collection0 }
+          clusterTime: { $$exists: true }