How to test multiple events in RxTest

Question

We are currently implementing unit tests for our ViewModel.

When a requestInAppPayment Input request comes in from the view model, subscription is implemented only when the instance property isPurchasing filter is false (to prevent multiple touches on the purchase button).

I want to write a test case for the filter when the purchase button is pressed while the purchase is in progress, but I don't know how to do it.

I want to write a test case that filters when the purchase button is clicked while the purchase is in progress.

in ViewMode

requestInAppPayment
            .filter { [weak self] _ in
                self?.isPurchasing == false
            }
            .subscribe(with: self, onNext: { owner, inAppPayment in
                owner.isPurchasing = true
                owner.showIndicator.onNext(())
                owner.requestInAppPurchaseUseCase.execute(productID: inAppPayment.productID)
            })
            .disposed(by: disposeBag)

 requestInAppPurchaseUseCase.purchaseSuccess
            .do(onNext: { [weak self] _ in
                self?.isPurchasing = false
                self?.hideIndicator.onNext(())
            })

in Unit Test

       let observer = scheduler.createObserver(String.self)
       let stubPaymentItem = InAppPayment()
        
        scheduler.createHotObservable([
            .next(10, stubPaymentItem),
            .next(20, stubPaymentItem)
        ])
        .bind(to: viewModel.input.requestInAppPurchase)
        .disposed(by: disposeBag)
        
        viewModel.output.showPaymentResultPage
            .map { "success" }
            .drive(observer)
            .disposed(by: disposeBag)
        
        scheduler.start()
        
// result = [.next(10, "success"), .next(20, "success")],
// The logic I think is [.next(10, "success")]

    XCTAssertEqual(observer.events, [
          .next(10, "success")
        ])

Answer 1

Here is a fully working example based on what little you provided of the unit test alone. I did not use your view model code at all, I just wrote a view model that passes the test.

I used the test harness I have available here: https://gist.github.com/danielt1263/bd449100764e3166644f7a38bca86c96

class ViewModelTests: XCTestCase {
    func test() {
        let scheduler = TestScheduler(initialClock: 0)
        let disposeBag = DisposeBag()
        let iapObserver = scheduler.createObserver(InAppPayment.self)
        let mockUseCase = MockRIAP(scheduler: scheduler)
        let viewModel = ViewModel(requestInAppPurchaseUseCase: mockUseCase)
        // I had to add all of the above to make the test compile.
        let observer = scheduler.createObserver(String.self)
        let stubPaymentItem = InAppPayment(id: 25) // I had to give the object an ID

        scheduler.createObservable(timeline: "-AA", values: ["A": stubPaymentItem]) // waits one second then emits two stubPaymentItems one second apart.
        .bind(to: viewModel.input.requestInAppPurchase)
        .disposed(by: disposeBag)

        viewModel.output.showPaymentResultPage
            .map { "success" }
            .drive(observer)
            .disposed(by: disposeBag)

        scheduler.start()

        XCTAssertEqual(observer.events, [
            .next(2, "success") // success emits at the 2 second mark because the trigger waits a second and then the mock waits a second. The second trigger is ignored.
        ])
    }
}

class MockRIAP: RequestInAppPurchaseUseCase {
    let args: TestableObserver<InAppPayment.ID>
    let _execute: (InAppPayment.ID) -> Observable<IAPResponse>

    init(scheduler: TestScheduler) {
        args = scheduler.createObserver(InAppPayment.ID.self)
        _execute = scheduler.mock(args: args, values: ["A": IAPResponse()], timelineSelector: { _ in "-A|" }) // waits one second then emits an IAPResponse
    }
    func execute(id: InAppPayment.ID) -> RxSwift.Observable<IAPResponse> {
        _execute(id)
    }
}

Here is the production code to make the above work:

protocol RequestInAppPurchaseUseCase {
    func execute(id: InAppPayment.ID) -> Observable<IAPResponse>
}

struct ViewModel {
    struct Input {
        let requestInAppPurchase: AnyObserver<InAppPayment>
    }
    struct Output {
        let showPaymentResultPage: Driver<Void> // a Driver<Void> doesn't make sense. Why would you want to emit the previous result? Better would be a Signal<Void>.
    }
    let input: Input
    let output: Output
}

extension ViewModel {
    init(requestInAppPurchaseUseCase: RequestInAppPurchaseUseCase) {
        let _requestInAppPurchase = PublishSubject<InAppPayment>()
        let showResults = _requestInAppPurchase
            .flatMapFirst { purchase in
                requestInAppPurchaseUseCase.execute(id: purchase.id)
            }
            .map { _ in }
            .asDriver(onErrorDriveWith: .empty())

        input = Input(requestInAppPurchase: _requestInAppPurchase.asObserver())
        output = Output(showPaymentResultPage: showResults)
    }
}

struct InAppPayment: Identifiable {
    let id: Int
}

struct IAPResponse { }