下面是一个完整的工作示例，它基于您仅提供的一点单元测试。我根本没有使用您的视图模型代码，我只是编写了一个通过测试的视图模型。
我使用了我在这里提供的测试工具：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)
    }
}

下面是使上述工作的生产代码：

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 { }