如何使用 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()
            });
    }
}

这是我的 JUnit 测试 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);
    }
}

如果我运行它喜欢最后一次检查状态 ENDED 的验证失败。如果我添加 Thread.sleep(3l); 它会通过。有没有更好或更安全的方法来通过测试而不添加 sleep() 方法？

Answer 1

我什至不尝试在单元测试中管理其他线程。我不确定 executor 在您的示例中来自哪里，但我会在您的测试中注入一个在当前线程上执行的版本。 Guava returns 一个来自 com.google.common.util.concurrent.MoreExecutors::directExecutor.

Answer 2

两种可行的选择

1 - 在 Dispatcher

中添加 shutdown 和 await 方法

您可以在 Dispatcher Class 中添加 shutdown 和 awaitTermination 方法。（如果需要你可以统一它们）

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.

在您的 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);
}

2 - 添加一个 getExecutor 方法并在测试中调用 shutdown/await

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

在你的测试中：

@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 3

测试 Failsafe 的策略会很困难，例如：您将如何模拟这样的测试：“我将重试 3 次，第 4 次我会成功。”，因为您确实拥有 withMaxRetries(5)。为此，我建议您阅读 Mockito 中的 thenAnswer，而不是 thenReturn。至少这是我们测试使用 Failsafe.

的代码的方式

尽管你有疑问，只需使用 dependency injection（人们认为它只存在于 Spring 中）。因此，将您的代码重新编写为：

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

请注意，您的 ExecutorService 现在已传入构造函数。现在创建一个 DispatcherTask 的真实实例，并模拟所有依赖项：

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

什么是someExecutorService？好吧，它是一个 ExecutorService，它将在运行单元测试的同一线程中执行您的代码。选择任意 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.

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

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

java

multithreading

junit

completable-future

junit-jupiter

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

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

java

multithreading

junit

completable-future

junit-jupiter

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