[코틀린 코루틴] 플로우 테스트하기

변환 함수

 

Flow를 반환하는 함수들은 대부분 다른 Flow를 만들어내는 함수를 호출해 동작한다.

예를 들어, 다음과 같은 서비스가 있다고 가정해본다.

class ObserveAppointmentsService(
    private val appointmentRepository: AppointmentRepository
) {
    fun observeAppointments(): Flow<List<Appointment>> =
        appointmentRepository
            .observeAppointments()
            .filterIsInstance<AppointmentsUpdate>()
            .map { it.appointments }
            .distinctUntilChanged()
            .retry {
                it is ApiException && it.code in 500..599
            }
}

observeAppointment 함수는 원소 필터링, 매핑, 반복되는 원소의 제거, 특정 예외의 경우 재시도와 같은 연산으로 AppointmentRepository의 observeAppointment를 데코레이트 한다.
이 함수를 테스트하려면, 이러한 기능들이 각각의 조건에 맞게 작동하는지 테스트하는 단위 테스트가 필요하다.

 

• 갱신된 예약만 유지: 예약 정보가 업데이트될 때만 그 값을 방출하고, 변경되지 않은 약속은 무시한다.
• 이전 원소와 동일한 원소는 제거: 같은 예약 목록이 연속적으로 전달되지 않도록, 이전과 동일한 값을 걸러낸다.
• 5xx 에러코드를 가진 API 예외가 발생하면 재시도: 서버 오류(500번대)가 발생하면, 해당 오류를 감지하고 자동으로 재시도한다.

 

먼저, 테스트를 위해 AppointmentRepository를 가짜로 만들거나(fake), 모킹(mock)해야 하며, observeAppointment 함수에서 flowOf를 사용해 한정된 플로우를 소스플로우로 정의하면 된다.

테스트하는 함수에서 시간이 중요하지 않다면, toList 함수를 사용해 결과를 리스트로 변환하고 단언문(assertion)에서 비교하기만 하면 된다.

class FakeAppointmentRepository(
    private val flow: Flow<AppointmentsEvent>
) : AppointmentRepository {
    override fun observeAppointments() = flow
}

data class Appointment(val t1: String, val t2: Instant)
class ObserveAppointmentsServiceTest {
    val aData1 = Instant.parse("2020-08-30T18:43:00Z")
    val anAppointment1 = Appointment("APP1", aData1)
    val aData2 = Instant.parse("2020-08-31T18:43:00Z")
    val anAppointment2 = Appointment("APP2", aData2)
    
    @Test
    fun `should keep only appoointments from...`() = runTest {
        // given
        val repo = FakeAppointmentRepository(
            flowOf(
                AppointmentsConfirmed,
                AppointmentUpdate(listOf(anAppointment1)),
                AppointmentUpdate(listOf(anAppointment2)),
                AppointmentsConfirmed,
            )
        )
        val service = ObserveAppointmentsService(repo)
        
        // when
        val result = service.observeAppointments().toList()
        
        // then
        assertEquals(
            listOf(
                listOf(anAppointment1),
                listOf(anAppointment2),
            ),
            result
        )
    }
}

 

위와 같은 테스트는 원소가 지연 없이 전달되는지는 확인할 수 없다.

시간 의존성을 확인하는 테스트를 구현하려면, runTest를 사용하고 데이터를 만드는 플로우의 람다식에 delay를 추가해야 한다.

변환된 플로우에서는 언제 원소가 방출되었는지에 대한 정보를 저장해야 하며, 단언문에서  그 결과를 확인하면 된다.

@Test
fun `should eliminate elements that are...`() = runTest {
    // given
    val repo = FakeAppointmentRepository(
        flow {
            delay(1000)
            emit(AppointmentUpdate(listOf(anAppointment1)))
            emit(AppointmentUpdate(listOf(anAppointment1)))
            delay(1000)
            emit(AppointmentUpdate(listOf(anAppointment2)))
            delay(1000)
            emit(AppointmentUpdate(listOf(anAppointment2)))
            emit(AppointmentUpdate(listOf(anAppointment1)))
        }
    )
    val service = ObserveAppointmentsService(repo)

    // when
    val result = service.observeAppointments()
        .map { curruentTime to it }
        .toList()

    // then
    assertEquals(
        listOf(
            1000L to listOf(anAppointment1),
            2000L to listOf(anAppointment2),
            3000L to listOf(anAppointment1),
        ),
        result
    )
}

'5XX 에러코드를 가진 API 예외가 발생하면 재시도해야 한다'라는 기능에 대한 단위 테스트를 만들때, 5XX ApiException을 방출하는 소스플로우를 구현하여 테스트하면 테스트하는 함수가 무한정 재시도하게 되며 끝나지 않는다.

이런 경우 take를 사용해 원소의 수를 제한하면 된다.

@Test
fun `should retry when API exception...`() = runTest {
    // given
    val repo = FakeAppointmentRepository(
        flow {
            emit(AppointmentUpdate(listOf(anAppointment1)))
            throw ApiException(502, "Some message")
        }
    )
    val service = ObserveAppointmentsService(repo)

    // when
    val result = service.observeAppointments()
        .take(3)
        .toList()

    // then
    assertEquals(
        listOf(
            listOf(anAppointment1),
            listOf(anAppointment1),
            listOf(anAppointment1),
        ),
        result
    )
}

 

 

끝나지 않는 플로우 테스트하기

 

상태플로우와 공유플로우는 스코프가 취소될 때까지 완료되지 않으며, runTest를 사용해 테스트할 경우 스코프는 backgroundScope가 되므로 테스트에서 스코프가 끝나는 것을 기다릴 수는 없다.

특정 사용자로부터 온 메시지를 감지하는 데 사용되는 서비스를 통해 예제를 구현해 보면 다음과 같다.

class MessageService(
    messageSource: Flow<Message>,
    scope: CoroutineScope
) {
    private val source = messagesSource
        .shareIn(
            scope = scope,
            started = SharingStarted.WhileSubscribed()
        )

    fun observeMessages(fromUserId: String) = source
        .filter { it.fromUserId == fromUserId }
}

class MessagesServiceTest {
    @Test
    fun `should emit messages from user`() = runTest {
        // given
        val source = flowOf(
            Message(fromUserId = "0", text = "A"),
            Message(fromUserId = "1", text = "B"),
            Message(fromUserId = "0", text = "C"),
        )
        val service = MessageService(
            messageSource = source,
            scope = backgroundScope,
        )
        
        // when
        val result = service.observeMessages("0")
            .toList() // 여기서 영원히 기다리게 된다.
        
        // then
        assertEquals(
            listOf(
                Message(fromUserId = "0", text = "A"),
                Message(fromUserId = "0", text = "C"),
            ), result
        )
    }
}

위의 테스트는 toList에서 영원히 중단된다.

기대되는 원소의 개수를 특정하여 take를 호출하면 해결되지만, 많은 정보를 잃게 된다.

// when
val result = service.observeMessages("0")
    .take(2)
    .toList()

다음은 backgroundScope에서 플로우를 시작하고 플로우가 방출하는 모든 원소를 컬렉션에 저장하는 것이다.

class MessagesServiceTest {
    @Test
    fun `should emit messages from user`() = runTest {
        // given
        val source = flowOf(
            Message(fromUserId = "0", text = "A"),
            delay(1000)
            Message(fromUserId = "1", text = "B"),
            Message(fromUserId = "0", text = "C"),
        )
        val service = MessageService(
            messageSource = source,
            scope = backgroundScope,
        )

        // when
        val emittedMessages = mutableListOf<Message>()
        service.observeMessages("0")
            .onEach { emittedMessages.add(it) }
            .launchIn(backgroundScope)
        delay(1)

        // then
        assertEquals(
            listOf(
                Message(fromUserId = "0", text = "A"),
            ), emittedMessages
        )
        
        // when
        delay(1000)

        // then
        assertEquals(
            listOf(
                Message(fromUserId = "0", text = "A"),
                Message(fromUserId = "0", text = "C"),
            ), emittedMessages
        )
    }
}

 

다른 방법으로는, 짧은 시간 동안만 감지할 수 있는 toList 함수를 사용하는 것이다.

suspend fun <T> Flow<T>.toListDuring(
    duration: Duration
): List<T> = coroutineScope {
    val result = mutableListOf<T>()
    val job = launch {
        this@toListDuring.collect(result::add)
    }
    delay(duration)
    job.cancel()
    return@coroutineScope result
}

class MessagesServiceTest {
    @Test
    fun `should emit messages from user`() = runTest {
        // given
        val source = flowOf(
            Message(fromUserId = "0", text = "A"),
            Message(fromUserId = "1", text = "B"),
            Message(fromUserId = "0", text = "C"),
        )
        val service = MessageService(
            messageSource = source,
            scope = backgroundScope,
        )

        // when
        val emittedMessages = service.observeMessages("0")
            .toListDuring(1.milliseconds) // 여기서 영원히 기다리게 된다.

        // then
        assertEquals(
            listOf(
                Message(fromUserId = "0", text = "A"),
                Message(fromUserId = "0", text = "C"),
            ), emittedMessages
        )
    }
}

 

 

개방할 연결 개수 정하기

 

위에서 사용한 MessageService는 한 개의 관찰자만 연결할 수 있었으나, 다음과 같이 여러 개의 관찰자를 연결할 수도 있다.

// 데이터 소스와 최대 한 개의 연결을 맺을 수 있다.
class MessageService(
    messageSource: Flow<Message>,
    scope: CoroutineScope
) {
    private val source = messagesSource
        .shareIn(
            scope = scope,
            started = SharingStarted.WhileSubscribed()
        )

    fun observeMessages(fromUserId: String) = source
        .filter { it.fromUserId == fromUserId }
}

// 데이터 소스와 여러 개의 연결을 맺을 수 있다.
class MessageService(
    messageSource: Flow<Message>
) {
    fun observeMessages(fromUserId: String) = messageSource
        .filter { it.fromUserId == fromUserId }
}

실제로 연결이 몇 개나 있는지 확인하는 방법은 구독자의 수를 세는 플로우를 만드는 것으로, onStart할 때 카운터의 수를 1씩 증가시키고 onCompletion할 때 1씩 감소시키면 된다.

@Test
fun `should start at most one connection`() = runTest {
    // given
    var connectionsCounter = 0
    val source = infiniteFlow
        .onStart { connectionsCounter++ }
        .onCompletion { connectionsCounter-- }
    val service = MessageService(
        messageSource = source,
        scope = backgroundScope,
    )

    // when
    service.observeMessages("0")
        .launchIn(backgroundScope)
    service.observeMessages("1")
        .launchIn(backgroundScope)
    service.observeMessages("0")
        .launchIn(backgroundScope)
    service.observeMessages("2")
        .launchIn(backgroundScope)
    delay(1000)

    // then
    assertEquals(
        1, connectionsCounter
    )
}

 

 

뷰 모델 테스트하기

 

플로우 빌더는 원소가 소스로부터 어떻게 방출되어야 하는지 지정하는 간단하고 강력한 메서드이다.

하지만 SharedFlow를 소스로 사용하고 테스트에서 원소를 방출하는 방법도 있으며, 뷰 모델을 테스트할 때 이 방식이 유용하다.

class ChatViewModel(
    private val messagesService: MessagesService,
) : ViewModel() {
    private val _lastMessage = MutableStateFlow<String?>(null)
    val lastMessage: StateFlow<String?> = _lastMessage
    
    private val _messages = MutableStateFlow(emptyList<String>())
    val messages: StateFlow<List<String>?> = _messages
    
    fun start(fromUserId: String) {
        messagesService.observeMessages(fromUserId)
            .onEach {
                val text = it.text
                _lastMessage.value = text
                _messages.value = _messages.value + text
            }
            .launchIn(viewModelScope)
    }
}

class ChatViewModelTest {
    @Test
    fun `should expose messages from user`() = runTest {
        // given
        val source = MutableSharedFlow<Message>()
        
        // when
        val viewModel = ChatViewModel(
            messgesService = FakeMessagesService(source)
        )
        viewModel.start("0")
        
        // then
        assertEquals(null, viewModel.lastMessage.value)
        assertEquals(emptyList(), viewModel.messages.value)
        
        // when
        source.emit(Message(fromUserId = "0", text = "ABC"))
        
        // then
        assertEquals("ABC", viewModel.lastMessage.value)
        assertEquals(listOf("ABC"), viewModel.messages.value)
        
        // when
        source.emit(Message(fromUserId = "0", text = "DEF"))
        source.emit(Message(fromUserId = "1", text = "GHI"))
        
        // then
        assertEquals("DEF", viewModel.lastMessage.value)
        assertEquals(listOf("ABC", "DEF"), viewModel.messages.value)
    }
}