test(datastore): enable datastore base integration test (#3014)

Author: Di Wu

.github/workflows/integ_test_datastore.yml

@@ -30,7 +30,6 @@ jobs:
           aws_s3_bucket: ${{ secrets.AWS_S3_BUCKET_INTEG }}
 
   datastore-base-test:
-    if: ${{ false }}
     needs: prepare-for-test
     uses: ./.github/workflows/integ_test_datastore_base.yml
     secrets: inherit
@@ -51,7 +50,6 @@ jobs:
     secrets: inherit
 
   datastore-multi-auth-test:
-    if: ${{ false }}
     needs: prepare-for-test
     uses: ./.github/workflows/integ_test_datastore_multi_auth.yml
     secrets: inherit

.github/workflows/integ_test_datastore_base.yml

@@ -9,7 +9,7 @@ permissions:
 
 jobs:
   datastore-integration-test-base:
-    timeout-minutes: 30
+    timeout-minutes: 45
     runs-on: macos-13
     environment: IntegrationTest
     steps:

.github/workflows/integ_test_datastore_lazy_load.yml

@@ -9,8 +9,8 @@ permissions:
 
 jobs:
   datastore-integration-lazy-load-test:
-    timeout-minutes: 30
-    runs-on: macos-12
+    timeout-minutes: 45
+    runs-on: macos-13
     environment: IntegrationTest
     steps:
       - uses: actions/checkout@2541b1294d2704b0964813337f33b291d3f8596b
@@ -33,3 +33,5 @@ jobs:
         with:
           project_path: ./AmplifyPlugins/DataStore/Tests/DataStoreHostApp
           scheme: AWSDataStorePluginLazyLoadTests
+          destination: 'platform=iOS Simulator,name=iPhone 14,OS=latest'
+          xcode_path: '/Applications/Xcode_14.3.app'

AmplifyPlugins/DataStore/Sources/AWSDataStorePlugin/AWSDataStorePlugin.swift

@@ -236,6 +236,7 @@ final public class AWSDataStorePlugin: DataStoreCategoryPlugin {
 
     public func reset() async {
         dispatchedModelSyncedEvents = [:]
+        dataStorePublisher?.sendFinished()
         if let resettable = storageEngine as? Resettable {
             log.verbose("Resetting storageEngine")
             await resettable.reset()

AmplifyPlugins/DataStore/Tests/DataStoreHostApp/AWSDataStorePluginAuthCognitoTests/AWSDataStoreAuthBaseTest.swift

@@ -64,8 +64,6 @@ class AuthRecorderInterceptor: URLRequestInterceptor {
 }
 
 class AWSDataStoreAuthBaseTest: XCTestCase {
-    var requests: Set<AnyCancellable> = []
-
     var amplifyConfig: AmplifyConfiguration!
     var user1: TestUser?
     var user2: TestUser?
@@ -77,9 +75,9 @@ class AWSDataStoreAuthBaseTest: XCTestCase {
 
     override func tearDown() async throws {
         try await clearDataStore()
-        requests = []
         try await signOut()
         await Amplify.reset()
+        try await Task.sleep(seconds: 1)
     }
 
     // MARK: - Test Helpers
@@ -302,8 +300,11 @@ extension AWSDataStoreAuthBaseTest {
 
     /// Asserts that DataStore is in a ready state and subscriptions are established
     /// - Parameter events: DataStore Hub events
-    func assertDataStoreReady(_ expectations: AuthTestExpectations,
-                              expectedModelSynced: Int = 1) async throws {
+    func assertDataStoreReady(
+        _ expectations: AuthTestExpectations,
+        expectedModelSynced: Int = 1
+    ) async throws {
+        var requests: Set<AnyCancellable> = []
         var modelSyncedCount = 0
         let dataStoreEvents = HubPayload.EventName.DataStore.self
         Amplify
@@ -341,9 +342,12 @@ extension AWSDataStoreAuthBaseTest {
     ///   - model: model instance saved and then deleted
     ///   - expectations: test expectations
     ///   - onFailure: failure callback
-    func assertMutations<M: Model>(model: M,
-                                   _ expectations: AuthTestExpectations,
-                                   onFailure: @escaping (_ error: DataStoreError) -> Void) async throws {
+    func assertMutations<M: Model>(
+        model: M,
+        _ expectations: AuthTestExpectations,
+        onFailure: @escaping (_ error: DataStoreError) -> Void
+    ) async throws {
+        var requests: Set<AnyCancellable> = []
         Amplify
             .Hub
             .publisher(for: .dataStore)

AmplifyPlugins/DataStore/Tests/DataStoreHostApp/AWSDataStorePluginIntegrationTests/Connection/DataStoreConnectionScenario1Tests.swift

@@ -64,7 +64,7 @@ class DataStoreConnectionScenario1Tests: SyncEngineIntegrationTestBase {
         }
 
         _ = try await Amplify.DataStore.save(team)
-        await waitForExpectations(timeout: networkTimeout)
+        await fulfillment(of: [syncedTeamReceived], timeout: networkTimeout)
 
         let syncProjectReceived = expectation(description: "received project from sync path")
         hubListener = Amplify.Hub.listen(to: .dataStore,
@@ -84,7 +84,7 @@ class DataStoreConnectionScenario1Tests: SyncEngineIntegrationTestBase {
             return
         }
         _ = try await Amplify.DataStore.save(project)
-        await waitForExpectations(timeout: networkTimeout)
+        await fulfillment(of: [syncProjectReceived], timeout: networkTimeout)
 
         let queriedProjectOptional = try await Amplify.DataStore.query(Project1.self, byId: project.id)
         guard let queriedProject = queriedProjectOptional else {
@@ -126,7 +126,7 @@ class DataStoreConnectionScenario1Tests: SyncEngineIntegrationTestBase {
         }
         project.team = anotherTeam
         _ = try await Amplify.DataStore.save(project)
-        await waitForExpectations(timeout: networkTimeout)
+        await fulfillment(of: [syncUpdatedProjectReceived], timeout: networkTimeout)
 
         let queriedProjectOptional = try await Amplify.DataStore.query(Project1.self, byId: project.id)
         XCTAssertNotNil(queriedProjectOptional)
@@ -173,8 +173,7 @@ class DataStoreConnectionScenario1Tests: SyncEngineIntegrationTestBase {
 
         _ = try await Amplify.DataStore.save(team)
         _ = try await Amplify.DataStore.save(project)
-        
-        await waitForExpectations(timeout: TestCommonConstants.networkTimeout)
+        await fulfillment(of: [createReceived], timeout: TestCommonConstants.networkTimeout)
 
         let deleteReceived = expectation(description: "Delete notification received")
         deleteReceived.expectedFulfillmentCount = 2 // 1 project and 1 team
@@ -206,7 +205,7 @@ class DataStoreConnectionScenario1Tests: SyncEngineIntegrationTestBase {
         // TODO: Delete Team should not be necessary, cascade delete should delete the team when deleting the project.
         // Once cascade works for hasOne, the following code can be removed.
         _ = try await Amplify.DataStore.delete(team)
-        await waitForExpectations(timeout: TestCommonConstants.networkTimeout)
+        await fulfillment(of: [deleteReceived], timeout: TestCommonConstants.networkTimeout)
 
         let queriedProject = try await Amplify.DataStore.query(Project1.self, byId: project.id)
         XCTAssertNil(queriedProject)

AmplifyPlugins/DataStore/Tests/DataStoreHostApp/AWSDataStorePluginIntegrationTests/DataStoreEndToEndTests.swift

@@ -36,7 +36,7 @@ class DataStoreEndToEndTests: SyncEngineIntegrationTestBase {
 
     func testCreate() async throws {
         await setUp(withModels: TestModelRegistration())
-        try await startAmplifyAndWaitForSync()
+        try await startAmplifyAndWaitForReady()
         var cancellables = Set<AnyCancellable>()
         let date = Temporal.DateTime.now()
         let newPost = Post(
@@ -55,20 +55,25 @@ class DataStoreEndToEndTests: SyncEngineIntegrationTestBase {
         let syncReceived = expectation(description: "SyncReceived(MutationEvent(version: 1))")
         let localEventReceived = expectation(description: "received mutation event with version nil")
         let remoteEventReceived = expectation(description: "received mutation event with version 1")
+
         Amplify.Hub.publisher(for: .dataStore)
             .sink { payload in
                 let event = DataStoreHubEvent(payload: payload)
                 switch event {
-                case .outboxMutationEnqueued:
-                    outboxMutationEnqueued.fulfill()
+                case .outboxMutationEnqueued(let event):
+                    if event.element.model.identifier == newPost.identifier {
+                        outboxMutationEnqueued.fulfill()
+                    }
                 case .outboxStatus(let status):
                     if !status.isEmpty {
                         outboxIsNotEmptyReceived.fulfill()
                     } else {
                         outboxIsEmptyReceived.fulfill()
                     }
-                case .outboxMutationProcessed:
-                    outboxMutationProcessed.fulfill()
+                case .outboxMutationProcessed(let event):
+                    if event.element.model.identifier == newPost.identifier {
+                        outboxMutationProcessed.fulfill()
+                    }
                 case .syncReceived(let mutationEvent):
                     guard let post = try? mutationEvent.decodeModel() as? Post, post.id == newPost.id else {
                         return
@@ -77,14 +82,13 @@ class DataStoreEndToEndTests: SyncEngineIntegrationTestBase {
                         XCTAssertEqual(post.content, newPost.content)
                         XCTAssertEqual(mutationEvent.version, 1)
                         syncReceived.fulfill()
-                        return
                     }
                 default:
                     break
                 }
             }.store(in: &cancellables)
 
-        let task = Task {
+        Task {
             let mutationEvents = Amplify.DataStore.observe(Post.self)
             do {
                 for try await mutationEvent in mutationEvents {
@@ -109,7 +113,7 @@ class DataStoreEndToEndTests: SyncEngineIntegrationTestBase {
             syncReceived,
             localEventReceived,
             remoteEventReceived
-        ], timeout: 10)
+        ], timeout: 30)
     }
 
     func testCreateMutateDelete() async throws {
@@ -530,8 +534,8 @@ class DataStoreEndToEndTests: SyncEngineIntegrationTestBase {
         try await startAmplifyAndWaitForSync()
 
         let parallelSize = 100
-        let initExpectation = asyncExpectation(description: "expect MutationEventPublisher works fine")
-        let parallelExpectation = asyncExpectation(description: "expect parallel processing no data loss")
+        let initExpectation = expectation(description: "expect MutationEventPublisher works fine")
+        let parallelExpectation = expectation(description: "expect parallel processing no data loss")
 
         let newPost = Post(title: UUID().uuidString, content: UUID().uuidString, createdAt: .now())
 
@@ -553,27 +557,27 @@ class DataStoreEndToEndTests: SyncEngineIntegrationTestBase {
             .compactMap(extractPost)
             .sink { post in
                 if post.title == newPost.title {
-                    Task { await initExpectation.fulfill() }
+                    initExpectation.fulfill()
                 }
 
                 if post.title.hasPrefix(titlePrefix) {
                     expectedResult.insert(post.title)
                 }
 
                 if expectedResult.count == parallelSize {
-                    Task { await parallelExpectation.fulfill() }
+                    parallelExpectation.fulfill()
                 }
             }
         try await Amplify.DataStore.save(newPost)
-        await AsyncTesting.waitForExpectations([initExpectation], timeout: 10)
+        await fulfillment(of: [initExpectation], timeout: 10)
         try await Task.sleep(seconds: 1)
 
         for post in posts {
             Task {
                 try? await Amplify.DataStore.save(post)
             }
         }
-        await AsyncTesting.waitForExpectations([parallelExpectation], timeout: Double(parallelSize))
+        await fulfillment(of: [parallelExpectation], timeout: Double(parallelSize))
         cancellable.cancel()
         XCTAssertEqual(expectedResult, Set(posts.map(\.title)))
     }

AmplifyPlugins/DataStore/Tests/DataStoreHostApp/AWSDataStorePluginIntegrationTests/DataStoreHubEventsTests.swift

@@ -41,6 +41,7 @@ class DataStoreHubEventTests: HubEventsIntegrationTestBase {
         try configureAmplify(withModels: TestModelRegistration())
         try await Amplify.DataStore.clear()
         await Amplify.reset()
+        try await Task.sleep(seconds: 1)
 
         let networkStatusReceived = expectation(description: "networkStatus received")
         var networkStatusActive = false

AmplifyPlugins/DataStore/Tests/DataStoreHostApp/AWSDataStorePluginIntegrationTests/DataStoreObserveQueryTests.swift

@@ -476,8 +476,8 @@ class DataStoreObserveQueryTests: SyncEngineIntegrationTestBase {
 
         let testId = UUID().uuidString
         var snapshotCount = 0
-        let observeQueryReadyForTest = asyncExpectation(description: "observeQuery initial query completed")
-        let allSnapshotsReceived = asyncExpectation(description: "query snapshots received")
+        let observeQueryReadyForTest = expectation(description: "observeQuery initial query completed")
+        let allSnapshotsReceived = expectation(description: "query snapshots received")
         let sink = Amplify.Publisher.create(Amplify.DataStore.observeQuery(for: Post.self,
                                                      where: Post.keys.content == testId,
                                                      sort: .ascending(Post.keys.title)))
@@ -495,7 +495,7 @@ class DataStoreObserveQueryTests: SyncEngineIntegrationTestBase {
                 switch snapshotCount {
                 case 1:
                     XCTAssertEqual(items.count, 0)
-                    Task { await observeQueryReadyForTest.fulfill() }
+                    observeQueryReadyForTest.fulfill()
                 case 2, 3:
                     XCTAssertEqual(items.count, 1)
                     XCTAssertEqual(items[0].title, "a")
@@ -527,14 +527,14 @@ class DataStoreObserveQueryTests: SyncEngineIntegrationTestBase {
                     XCTAssertEqual(items[0].title, "b")
                 case 14:
                     XCTAssertEqual(items.count, 0)
-                    Task { await allSnapshotsReceived.fulfill() }
+                    allSnapshotsReceived.fulfill()
                 default:
                     break
                 }
 
             }
 
-        await waitForExpectations([observeQueryReadyForTest], timeout: 10)
+        await fulfillment(of: [observeQueryReadyForTest], timeout: 10)
         // (1) Add several models
         let postA = Post(title: "a", content: testId, createdAt: .now())
         try await savePostAndWaitForSync(postA)
@@ -559,7 +559,7 @@ class DataStoreObserveQueryTests: SyncEngineIntegrationTestBase {
         try await deletePostAndWaitForSync(postNFromA)
         try await deletePostAndWaitForSync(postBFromZ)
 
-        await waitForExpectations([allSnapshotsReceived], timeout: 100)
+        await fulfillment(of: [allSnapshotsReceived], timeout: 100)
 
         sink.cancel()
     }