How to test with mockito a method that is running an async thread

Question

I have the following code which I want to test:

@Slf4j
@Component
public class DispatcherTask {

    private final MyClassService myClassService;
    private final SreamingService streamingService;
    private ExecutorService executor = Executors.newCachedThreadPool();
    private final Set < String > dispatching = new ConcurrentSkipListSet < > ();
    private final RetryPolicy < Object > httpRetryPolicy;

    public DispatcherTask(MyClassService myClassService, SreamingService streamingService) {
        this.myClassService = myClassService;
        this.streamingService = streamingService;
        this.httpRetryPolicy = new RetryPolicy < > ()
            .handle(HttpClientErrorException.class)
            .onRetriesExceeded(e - > log.warn("Max retries have been exceeded for http exception"))
            .withBackoff(2, 3, ChronoUnit.SECONDS)
            .withMaxRetries(5);
    }

    public void initMethod(MyClass myclass) {
        Failsafe.with(httpRetryPolicy).with(executor)
            .onFailure((e) - > {
                // TODO log.error();
                myClassService.updateStatus(myclass.getId(), Status.FAILED);
            })
            .onSuccess((o) - > {
                MyClass updatedMyClass = myClassService.updateStatus(myclass.getId(), GameStatus.ENDED);
                streamingService.storeData(updateMyClass);
                dispatching.remove(myclass.getPlatformId());
                //TODO log.info()
            })
            .runAsync(() - > {
                MyClass updatedMyClass =
                myClassService.updateStatus(myclass.getId(), Status.STREAMING);
                streamingService.polling(StreamedClass.builder().myclass(updatedMyClass).build());
                // TODO log.info()
            });
    }
}

This is my JUnit test with Mockito:

@ExtendWith(MockitoExtension.class)
public class DispatcherTaskTest {

    @Mock private MyClassService myClassService;
    @Mock private StreamingService streamingService;
    @InjectMocks private DispatcherTask dispatcherTask;

    @Test
    void test() throws InterruptedException {
        //given
        MyClass myclass = new MyClass().setId(1L).setName("name").setStatus(Status.CREATED);
        when(myClassService.updateStatus(myclass.getId(), Status.STREAMING)).thenReturn(myclass);
        when(myClassService.updateStatus(myclass.getId(), Status.ENDED)).thenReturn(myclass);
        doNothing().when(streamingService).polling(any());
        doNothing().when(streamingService).storeData(any());
        //when
        dispatcherTask.initMethod(myclass)
        //then
        verify(myClassService, times(1)).updateStatus(myclass.getId(), Status.STREAMING);
        verify(myClassService, times(1)).updateStatus(myclass.getId(), Status.ENDED);
    }
}

If I run it like that the last verify that checks the status ENDED fails. If I add a Thread.sleep(3l); it passes. Is there a better or more secure way to pass the test without adding the sleep() method?

Answer 1

Two alternatives that may work

1 - Add shutdown and await methods in the Dispatcher

You could add the shutdown and awaitTermination methods in your Dispatcher Class. (You could unify them if wanted)

public class DispatcherTask 
{    
  //...
  public void shutdownExecutor()
  {
     executor.shutdown();
  }
  public void waitToFinish(long seconds)
  {
     executor.awaitTermination(seconds,TimeUnit.SECONDS);
  }
  //...
}

boolean awaitTermination

Blocks until all tasks have completed execution after a shutdown request, or the timeout occurs, or the current thread is interrupted, whichever happens first.

---

With this method in your Dispatcher, the test could be altered so you can:

@Test
void test() throws InterruptedException {
    //...
    dispatcherTask.initMethod(myclass);
    verify(myClassService, times(1)).updateStatus(myclass.getId(), Status.STREAMING);
   
    dispatcherTask.shutdownExecutor();
    dispatcherTask.waitToFinish(3L); //this will block without manually sleeping
    verify(myClassService, times(1)).updateStatus(myclass.getId(), Status.ENDED);
}

---

2 - Add a getExecutor method and call shutdown/await in the tests

public class DispatcherTask 
{    
  //...
  public ExecutorService getExecutor()
  {
     return executor;
  }
  //...
}

And in your test:

@Test
void test() throws InterruptedException {
    //...
     dispatcherTask.initMethod(myclass);
     verify(myClassService, times(1)).updateStatus(myclass.getId(), Status.STREAMING);
    
     //create a variable in the test or just invoke getExecutor
     dispatcherTask.getExecutor().shutdown();
     dispatcherTask.getExecutor().awaitTermination(3L, TimeUnit.SECONDS); 
     verify(myClassService, times(1)).updateStatus(myclass.getId(), Status.ENDED);
}

Answer 2

It is going to be difficult to test the policies of Failsafe, for example: how will you emulate a test that says : "I will retry 3 times and on the 4-th one I will succeed.", because you do have that withMaxRetries(5). For that I would suggest you read what thenAnswer in Mockito does, instead of thenReturn. At least this is how we test our code that uses Failsafe.

To you question though, just use dependency injection (which people think is only present in Spring). So re-write your code to:

 public DispatcherTask(MyClassService myClassService, StreamingService streamingService, ExecutorService executor ) {

notice that your ExecutorService is now passed in the constructor. And now create a real instance of DispatcherTask, with all it's dependencies mocked:

 DispatcherTask dt = new DispatcherTask(
       myClassService,       // this one is a Mock
       streamingService,     // this one is a Mock
       someExecutorService   // and what is this?
 );

what is someExecutorService? Well it is an ExecutorService that will execute your code in the same thread that runs the unit test. Choose any from here, as an example. Remember that ExecutorService is an interface, so creating an implementation that runs in the same thread is trivial, like this answer did.

Answer 3

I don't even try to manage other threads in a unit test. I'm not sure where executor comes from in your example but I would inject a version in your test that executes on the current thread. Guava returns one from com.google.common.util.concurrent.MoreExecutors::directExecutor.