AsyncStream and AsyncThrowingStream testing utility

AsyncThrowingStreamTestReducer.swift

/// MIT License
///
/// Copyright (c) 2023 Dariusz Rybicki Darrarski
///
/// Permission is hereby granted, free of charge, to any person obtaining a copy
/// of this software and associated documentation files (the "Software"), to deal
/// in the Software without restriction, including without limitation the rights
/// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
/// copies of the Software, and to permit persons to whom the Software is
/// furnished to do so, subject to the following conditions:
///
/// The above copyright notice and this permission notice shall be included in all
/// copies or substantial portions of the Software.
///
/// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
/// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
/// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
/// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
/// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
/// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
/// SOFTWARE.

import ComposableArchitecture

public struct _AsyncThrowingStreamTestReducer<Element, Failure>: ReducerProtocol
where Element: Sendable,
      Element: Equatable,
      Failure: Error
{
  public struct State: Equatable, Sendable {}
  public struct Finish: Equatable, Sendable {}
  public enum EffectId {}
  public enum Action: Equatable, Sendable {
    case start
    case stop
    case element(Element)
    case completion(TaskResult<Finish>)
    public static var finish: Action { .completion(.success(.init())) }
    public static func failure(_ error: Error) -> Action { .completion(.failure(error)) }
  }

  let stream: AsyncThrowingStream<Element, Failure>

  public func reduce(into state: inout State, action: Action) -> EffectTask<Action> {
    switch action {
    case .start:
      return .run { [stream] send in
        do {
          for try await element in stream {
            await send(.element(element))
          }
          await send(.completion(.success(.init())))
        } catch {
          await send(.completion(.failure(error)))
        }
      }
      .cancellable(id: EffectId.self, cancelInFlight: true)

    case .stop:
      return .cancel(id: EffectId.self)

    case .element(_), .completion(_):
      return .none
    }
  }
}

extension TestStore where Environment == Void {
  /// Creates TestStore with AsyncThrowingStreamTestReducer and provided stream.
  ///
  /// Testing stream:
  ///
  /// ```swift
  /// let stream: AsyncStream<Int> = [1, 2, 3].async.eraseToStream()
  /// let store = TestStore(stream: stream.eraseToThrowingStream())
  /// await store.send(.start)
  /// await store.receive(.element(1))
  /// await store.receive(.element(2))
  /// await store.receive(.element(3))
  /// await store.receive(.finish)
  /// ```
  ///
  /// Testing throwing stream:
  ///
  /// ```swift
  /// struct Failure: Error, Equatable {}
  /// let stream: AsyncThrowingStream<Int, Error> = .init { throw Failure() }
  /// let store = TestStore(stream: stream)
  /// await store.send(.start)
  /// await store.receive(.failure(Failure()))
  /// ```
  ///
  /// - Parameter stream: Stream
  public convenience init<Element, Failure>(stream: AsyncThrowingStream<Element, Failure>)
  where Element: Equatable, Failure: Error,
  State == _AsyncThrowingStreamTestReducer<Element, Failure>.State,
  Action == _AsyncThrowingStreamTestReducer<Element, Failure>.Action,
  ScopedState == _AsyncThrowingStreamTestReducer<Element, Failure>.State,
  ScopedAction == _AsyncThrowingStreamTestReducer<Element, Failure>.Action
  {
    self.init(
      initialState: _AsyncThrowingStreamTestReducer<Element, Failure>.State(),
      reducer: _AsyncThrowingStreamTestReducer<Element, Failure>(stream: stream)
    )
  }
}