如何用mockito测试正在运行异步线程的方法

Question

我有以下代码要测试：

@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()
            });
    }
}

这是我使用mockito进行的JUnit测试：

@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);
    }
}

如果我这样运行它，则检查状态Ended的最后一次验证将失败。如果添加thread.sleep(3L)；，它就会通过。是否有更好或更安全的方法通过测试，而不添加sleep()方法？

Answer 1

您可以在Dispatcher类中添加shutdown和awaittermination方法。（如果需要，您可以统一它们）

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

布尔等待终止

阻塞，直到在关闭请求后所有任务都已完成执行，或发生超时，或当前线程中断（以先发生者为准）。

在Dispatcher中使用此方法，可以更改测试，以便您可以：

@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);
}

Answer 2

我甚至不尝试在单元测试中管理其他线程。我不确定示例中的executor从何而来，但我会在测试中注入一个在当前线程上执行的版本。Guava从com.google.common.util.concurrent.MoreExecutors：：DirectExecutor返回一个。