How I can test what paged list return error during response?

I am trying to test error in PagedList's data source (when load new group of data). My DataSource looks like that:

package com.ps.superheroapp.ui.character_screen.list

import androidx.paging.PositionalDataSource
import com.ps.superheroapp.api.MarvelApiService
import com.ps.superheroapp.objects.SchedulerNames
import io.reactivex.Scheduler
import io.reactivex.disposables.CompositeDisposable
import javax.inject.Named

class CharactersDataSource(
    private val compositeDisposable: CompositeDisposable,
    private val marvelApi: MarvelApiService,
    @Named(SchedulerNames.MAIN) private val scheduler: Scheduler,
    private val filter: Filter
) : PositionalDataSource<Character>() {

    var events: SourcedDataEventsHandler? = null

    override fun loadInitial(params: LoadInitialParams, callback: LoadInitialCallback<Character>) {
        compositeDisposable.add(marvelApi.searchCharacter(params.pageSize, 0, filter.searchQuery)
            .observeOn(scheduler)
            .doOnSubscribe {
                events?.onLoadStarted()
            }
            .subscribe({
                callback.onResult(it.data.results ?: arrayListOf(), 0)
                events?.onLoadFinishedSuccessfully()
            }, {
                events?.onLoadFinishedWithError(it)
            })
        )
    }

    override fun loadRange(params: LoadRangeParams, callback: LoadRangeCallback<Character>) {
        compositeDisposable.add(
            marvelApi.searchCharacter(params.loadSize, params.startPosition, filter.searchQuery)
                .observeOn(scheduler)
                .doOnSubscribe {
                    events?.onLoadStarted()
                }
                .subscribe({
                    callback.onResult(it.data.results ?: arrayListOf())
                    events?.onLoadFinishedSuccessfully()
                }, {
                    events?.onLoadFinishedWithError(it)
                })
        )
    }
}

If I will just mock events handler class and during test will call methods from it, this test will not test anything.

I searched ways or best practice to test this type of behaviour but I didn't found something.

My test looks like that:

@Test
fun should_show_network_error_when_screen_data_cannot_be_loaded_because_of_internet_connection() {
`when`(connectivityChecker.isOffline()).thenReturn(true)
    //logic to imitate error during loading from data source

    vm.fetchCharacters()

    Assert.assertEquals(ErrorType.NETWORK, vm.error.get())
}

Could you please give me some advice or architectural example, or example of unit test to test this. Thanks in advance

I found next solution which is work:

I changed callback interface on PublishSubject in Data Source:

class CharactersDataSource(
    private val compositeDisposable: CompositeDisposable,
    private val marvelApi: MarvelApiService,
    @Named(SchedulerNames.MAIN) private val scheduler: Scheduler,
    val filter: Filter
) : PositionalDataSource<Character>() {

    private val onEvent = PublishSubject.create<CharacterLoadEvent>()

    override fun loadInitial(params: LoadInitialParams, callback: LoadInitialCallback<Character>) {
        compositeDisposable.add(marvelApi.searchCharacter(params.pageSize, 0, filter.searchQuery)
            .observeOn(scheduler)
            .doOnSubscribe {
                onEvent.onNext(CharacterLoadEvent.LOAD_STARTED)
            }
            .subscribe({
                callback.onResult(it.data.results ?: arrayListOf(), 0)
                onEvent.onNext(CharacterLoadEvent.LOADED)
            }, {
                onEvent.onNext(CharacterLoadEvent.ERROR)
            })
        )
    }

    override fun loadRange(params: LoadRangeParams, callback: LoadRangeCallback<Character>) {
        compositeDisposable.add(
            marvelApi.searchCharacter(params.loadSize, params.startPosition, filter.searchQuery)
                .observeOn(scheduler)
                .doOnSubscribe {
                    onEvent.onNext(CharacterLoadEvent.LOAD_STARTED)
                }
                .subscribe({
                    callback.onResult(it.data.results ?: arrayListOf())
                    onEvent.onNext(CharacterLoadEvent.LOADED)
                }, {
                    onEvent.onNext(CharacterLoadEvent.ERROR)
                })
        )
    }

    fun observeCharactersLoadEvents(): Observable<CharacterLoadEvent> = onEvent
}

and in tests mock behaviour of this publisher. Couple of tests looks like:

@Test
    fun should_show_network_error_when_screen_data_cannot_be_loaded_because_of_internet_connection() {
        `when`(connectivityChecker.isOffline()).thenReturn(true)
        `when`(interactor.observeCharactersLoadEvents()).thenReturn(Observable.just(CharacterLoadEvent.ERROR))
        `when`(interactor.getCharacters()).then {
            TestPageList.get<Character>(listOf())
        }

        vm.fetchCharacters()

        Assert.assertEquals(ErrorType.NETWORK, vm.error.get())
    }

    @Test
    fun should_show_general_error_when_screen_data_cannot_be_loaded_because_of_unknown_error() {
        `when`(connectivityChecker.isOffline()).thenReturn(false)
        `when`(interactor.observeCharactersLoadEvents()).thenReturn(Observable.just(CharacterLoadEvent.ERROR))

        vm.fetchCharacters()
        Assert.assertEquals(ErrorType.GENERAL, vm.error.get())
    }

    @Test
    fun should_filter_character_list_when_user_enter_text_in_search_field() {
        `when`(interactor.observeCharactersLoadEvents()).thenReturn(Observable.just(CharacterLoadEvent.LOAD_STARTED))
        `when`(interactor.getCharacters()).then {
            TestPageList.get<Character>(listOf())
        }

        vm.searchQuery.value = "Hulk"

        Mockito.verify(interactor, times(1)).getCharacters("Hulk")
    }

This can help to test behaviour of DataSource and it works for me. I will also be very appreciate for any idea how to improve this Architecture/Approach/Test