How do you test Combine pipelines, and what is a TestScheduler pattern?

11.6.2 · L3 · Interview

Predict first

A pipeline is tested synchronously with a subject and a collected array. What is the result?

import Combine

var bag = Set<AnyCancellable>()
let input = PassthroughSubject<Int, Never>()
var result: [Int] = []
input.map { $0 * 2 }.sink { result.append($0) }.store(in: &bag)
input.send(1); input.send(2)
print(result)

[]
[2, 4]
[1, 2]

Predict first

Two values are collected in batches of two from five:

import Combine

var bag = Set<AnyCancellable>()
var col: [[Int]] = []
(1...5).publisher.collect(2).sink { col.append($0) }.store(in: &bag)
print(col)

[[1, 2], [3, 4], [5]]
[[1, 2], [3, 4]]
[[1], [2], [3], [4], [5]]

Mental model. A test scheduler is a stopwatch you can wind forward by hand. Instead of waiting two real seconds to see a debounce fire, you turn the knob two seconds and it fires at once.

A pipeline is easiest to test when its inputs and its scheduler are injected. The simplest tests are synchronous: feed a PassthroughSubject or an array publisher, subscribe with sink, collect values into an array and assert on it, as long as nothing in the pipeline hops threads. For asynchronous or timed pipelines you have three options. 1) Wait: use an XCTestExpectation (fulfilled in the sink or completion) and wait(for:timeout:), or, in modern tests, the async form for await over publisher.values (with a timeout) so a test reads linearly. This is the least intrusive but is slow for time-based operators. 2) Replace the scheduler with ImmediateScheduler.shared: a pipeline that takes a Scheduler generic (func make<S: Scheduler>(on scheduler: S)) or an AnySchedulerOf<DispatchQueue> runs synchronously in tests, so receive(on:) hops become no-ops. 3) A controllable test scheduler: Combine ships no virtual clock, but the common TestScheduler pattern (from the open-source combine-schedulers package) is a scheduler whose time you advance by hand: scheduler.advance(by: .seconds(1)) fires every debounce or delay scheduled within that window instantly and deterministically, so a debounce test takes microseconds and never flakes.

Practical guidance. Inject the scheduler (and any clock or request publisher) through the initialiser; do not hard-code DispatchQueue.main in the pipeline. Use .first() or .collect() in tests to turn a stream into a single value you can wait on, and XCTUnwrap or Result to assert failures. Test the behaviours that matter: debounce coalescing, that older requests are ignored (switchToLatest), that an error does not end the stream, and that cancellation stops work. Keep cancellables in the test so the pipeline stays alive until you assert. Avoid sleeping for fixed times; they are slow and fragile. In async/await code, none of this machinery is needed: a test just awaits the function (with Swift’s Clock protocol you can inject a test clock for time).

With no thread hop in the pipeline, the values are delivered synchronously, so the test can assert immediately.

collect(2) groups values in twos and emits the remainder on completion.

Common traps

Testing with sleep. Slow, and it flakes under load; inject a scheduler.
Letting the cancellable go before asserting. The pipeline is cancelled and the test asserts on nothing.

They might follow up with

How do you make a receive(on: .main) synchronous in a test? Inject the scheduler and use ImmediateScheduler.shared.
What does advance(by:) do? It moves a virtual clock forward and fires everything due in that window.

How do you test Combine pipelines, and what is a TestScheduler pattern?

Predict first

Predict first

Common traps

They might follow up with

Related questions