Как использовать JUnit для тестирования асинхронных процессов

Как вы тестируете методы, которые запускают асинхронные процессы с JUnit?

Я не знаю, как заставить мой тест ждать завершения процесса (это не совсем модульный тест, это больше похоже на интеграционный тест, поскольку он включает в себя несколько классов, а не только один).

ИМХО, это плохая практика - создавать модульные тесты или ждать их в потоках и т. Д. Вы хотели бы, чтобы эти тесты выполнялись за доли секунды. Вот почему я хотел бы предложить двухэтапный подход к тестированию асинхронных процессов.

1. Проверьте, что ваш асинхронный процесс отправлен правильно. Вы можете смоделировать объект, который принимает ваши асинхронные запросы, и убедиться, что отправленное задание имеет правильные свойства и т. Д.
2. Проверьте, что ваши асинхронные обратные вызовы делают правильные вещи. Здесь вы можете смоделировать изначально отправленное задание и предположить, что оно правильно инициализировано, и убедиться, что ваши обратные вызовы верны.

Альтернативой является использование класса CountDownLatch .

public class DatabaseTest {

    /**
     * Data limit
     */
    private static final int DATA_LIMIT = 5;

    /**
     * Countdown latch
     */
    private CountDownLatch lock = new CountDownLatch(1);

    /**
     * Received data
     */
    private List<Data> receiveddata;

    @Test
    public void testDataRetrieval() throws Exception {
        Database db = new MockDatabaseImpl();
        db.getData(DATA_LIMIT, new DataCallback() {
            @Override
            public void onSuccess(List<Data> data) {
                receiveddata = data;
                lock.countDown();
            }
        });

        lock.await(2000, TimeUnit.MILLISECONDS);

        assertNotNull(receiveddata);
        assertEquals(DATA_LIMIT, receiveddata.size());
    }
}

ПРИМЕЧАНИЕ: вы не можете просто использовать синхронизированный с обычным объектом в качестве блокировки, так как быстрые обратные вызовы могут снять блокировку до вызова метода ожидания блокировки. Смотрите это сообщение в блоге Джо Уолнесом.

РЕДАКТИРОВАТЬ Удалены синхронизированные блоки вокруг CountDownLatch благодаря комментариям @jtahlborn и @Ring

Вы можете попробовать использовать библиотеку Awaitility . Это облегчает тестирование систем, о которых вы говорите.

Если вы используете CompletableFuture (представленный в Java 8) или SettableFuture (из Google Guava ), вы можете завершить свой тест, как только он будет сделан, вместо того, чтобы ждать предварительно установленное количество времени. Ваш тест будет выглядеть примерно так:

CompletableFuture<String> future = new CompletableFuture<>();
executorService.submit(new Runnable() {         
    @Override
    public void run() {
        future.complete("Hello World!");                
    }
});
assertEquals("Hello World!", future.get());

Запустите процесс и дождитесь результата, используя Future.

Один метод, который я нашел довольно полезным для тестирования асинхронных методов, - это внедрение Executorэкземпляра в конструктор объекта для теста. В производственном процессе экземпляр executor настроен на асинхронную работу, в то время как в тесте его можно смоделировать для синхронного запуска.

Итак, предположим, что я пытаюсь проверить асинхронный метод Foo#doAsync(Callback c),

class Foo {
  private final Executor executor;
  public Foo(Executor executor) {
    this.executor = executor;
  }

  public void doAsync(Callback c) {
    executor.execute(new Runnable() {
      @Override public void run() {
        // Do stuff here
        c.onComplete(data);
      }
    });
  }
}

В производстве, я бы построить Fooс Executors.newSingleThreadExecutor()экземпляром Исполнителя в то время как в тесте , я бы , наверное , построить его с синхронным исполнителем , который делает следующее -

class SynchronousExecutor implements Executor {
  @Override public void execute(Runnable r) {
    r.run();
  }
}

Теперь мой тест JUnit асинхронного метода довольно чистый -

@Test public void testDoAsync() {
  Executor executor = new SynchronousExecutor();
  Foo objectToTest = new Foo(executor);

  Callback callback = mock(Callback.class);
  objectToTest.doAsync(callback);

  // Verify that Callback#onComplete was called using Mockito.
  verify(callback).onComplete(any(Data.class));

  // Assert that we got back the data that we expected.
  assertEquals(expectedData, callback.getData());
}

В тестировании многопоточного / асинхронного кода нет ничего плохого, особенно если многопоточность является точкой кода, который вы тестируете. Общий подход к тестированию этого материала заключается в следующем:

• Заблокировать основной тестовый поток
• Захват неудачных утверждений из других потоков
• Разблокировать основной тестовый поток
• Отбрось любые неудачи

Но это очень много для одного теста. Лучший / более простой подход - просто использовать ConcurrentUnit :

  final Waiter waiter = new Waiter();

  new Thread(() -> {
    doSomeWork();
    waiter.assertTrue(true);
    waiter.resume();
  }).start();

  // Wait for resume() to be called
  waiter.await(1000);

Преимущество этого CountdownLatchподхода в том, что он менее многословен, поскольку ошибки утверждений, возникающие в любом потоке, должным образом передаются в основной поток, что означает, что тест завершается неудачей, когда это необходимо. Запись, которая сравнивает CountdownLatchподход к ConcurrentUnit находится здесь .

Я также написал пост в блоге на эту тему для тех, кто хочет узнать немного подробнее.

Как насчет вызова SomeObject.waitи notifyAllкак описано здесь ИЛИ используя метод Роботиума Solo.waitForCondition(...) ИЛИ используйте класс, который я написал, чтобы сделать это (см. Комментарии и тестовый класс для использования)

Я нахожу библиотеку socket.io для проверки асинхронной логики. Это выглядит просто и кратко с помощью LinkedBlockingQueue . Вот пример :

    @Test(timeout = TIMEOUT)
public void message() throws URISyntaxException, InterruptedException {
    final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();

    socket = client();
    socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
        @Override
        public void call(Object... objects) {
            socket.send("foo", "bar");
        }
    }).on(Socket.EVENT_MESSAGE, new Emitter.Listener() {
        @Override
        public void call(Object... args) {
            values.offer(args);
        }
    });
    socket.connect();

    assertThat((Object[])values.take(), is(new Object[] {"hello client"}));
    assertThat((Object[])values.take(), is(new Object[] {"foo", "bar"}));
    socket.disconnect();
}

Используя LinkedBlockingQueue, возьмите API для блокировки, пока не получите результат, как в синхронном режиме. И установите таймаут, чтобы не тратить слишком много времени на ожидание результата.

Стоит отметить, что Testing Concurrent Programsв « Параллельности на практике» есть очень полезная глава, которая описывает некоторые подходы модульного тестирования и дает решения для проблем.

Это то, что я использую в настоящее время, если результат теста генерируется асинхронно.

public class TestUtil {

    public static <R> R await(Consumer<CompletableFuture<R>> completer) {
        return await(20, TimeUnit.SECONDS, completer);
    }

    public static <R> R await(int time, TimeUnit unit, Consumer<CompletableFuture<R>> completer) {
        CompletableFuture<R> f = new CompletableFuture<>();
        completer.accept(f);
        try {
            return f.get(time, unit);
        } catch (InterruptedException | TimeoutException e) {
            throw new RuntimeException("Future timed out", e);
        } catch (ExecutionException e) {
            throw new RuntimeException("Future failed", e.getCause());
        }
    }
}

При использовании статического импорта тест выглядит довольно хорошо. (обратите внимание, в этом примере я начинаю тему, чтобы проиллюстрировать идею)

    @Test
    public void testAsync() {
        String result = await(f -> {
            new Thread(() -> f.complete("My Result")).start();
        });
        assertEquals("My Result", result);
    }

Если f.completeне вызывается, тест не пройден по истечении времени ожидания. Вы также можете использовать, f.completeExceptionallyчтобы потерпеть неудачу рано.

Здесь есть много ответов, но простой - просто создать законченное CompletableFuture и использовать его:

CompletableFuture.completedFuture("donzo")

Итак, в моем тесте:

this.exactly(2).of(mockEventHubClientWrapper).sendASync(with(any(LinkedList.class)));
this.will(returnValue(new CompletableFuture<>().completedFuture("donzo")));

Я просто следю за тем, чтобы все это называлось так или иначе. Этот метод работает, если вы используете этот код:

CompletableFuture.allOf(calls.toArray(new CompletableFuture[0])).join();

Это прыгнет прямо через это, поскольку все CompletableFutures закончены!

Избегайте тестирования с параллельными потоками всякий раз, когда вы можете (что происходит в большинстве случаев). Это только сделает ваши тесты нестабильными (иногда проходит, иногда не проходит).

Только когда вам нужно вызвать какую-то другую библиотеку / систему, вам, возможно, придется ждать в других потоках, в этом случае всегда используйте библиотеку Awaitility вместо Thread.sleep().

Никогда не звоните get()или не проводите join()тесты, иначе ваши тесты могут работать вечно на вашем CI-сервере, если будущее никогда не закончится. Всегда утверждайте isDone()сначала в своих тестах перед вызовом get(). Для CompletionStage это .toCompletableFuture().isDone().

Когда вы тестируете неблокирующий метод, подобный этому:

public static CompletionStage<String> createGreeting(CompletableFuture<String> future) {
    return future.thenApply(result -> "Hello " + result);
}

тогда вам следует не просто проверить результат, пройдя завершенное Future в тесте, вы также должны убедиться, что ваш метод doSomething()не блокируется с помощью вызова join()или get(). Это особенно важно, если вы используете неблокирующую платформу.

Для этого выполните тестирование с незавершенным будущим, которое вы установили как завершенное вручную:

@Test
public void testDoSomething() throws Exception {
    CompletableFuture<String> innerFuture = new CompletableFuture<>();
    CompletableFuture<String> futureResult = createGreeting(innerFuture).toCompletableFuture();
    assertFalse(futureResult.isDone());

    // this triggers the future to complete
    innerFuture.complete("world");
    assertTrue(futureResult.isDone());

    // futher asserts about fooResult here
    assertEquals(futureResult.get(), "Hello world");
}

Таким образом, если вы добавите future.join()doSomething (), тест не пройден.

Если ваша служба использует ExecutorService, например, in thenApplyAsync(..., executorService), то в ваших тестах внедряется однопотоковая служба ExecutorService, например, из guava:

ExecutorService executorService = Executors.newSingleThreadExecutor();

Если в вашем коде используется forkJoinPool, например, перепишите thenApplyAsync(...)код для использования ExecutorService (есть много веских причин) или используйте Awaitility.

Чтобы сократить пример, я сделал BarService аргументом метода, реализованным как лямбда-код Java8 в тесте, обычно это была бы вставленная ссылка, которую вы бы высмеяли.

Я предпочитаю использовать ожидание и уведомление. Это просто и понятно.

@Test
public void test() throws Throwable {
    final boolean[] asyncExecuted = {false};
    final Throwable[] asyncThrowable= {null};

    // do anything async
    new Thread(new Runnable() {
        @Override
        public void run() {
            try {
                // Put your test here.
                fail(); 
            }
            // lets inform the test thread that there is an error.
            catch (Throwable throwable){
                asyncThrowable[0] = throwable;
            }
            // ensure to release asyncExecuted in case of error.
            finally {
                synchronized (asyncExecuted){
                    asyncExecuted[0] = true;
                    asyncExecuted.notify();
                }
            }
        }
    }).start();

    // Waiting for the test is complete
    synchronized (asyncExecuted){
        while(!asyncExecuted[0]){
            asyncExecuted.wait();
        }
    }

    // get any async error, including exceptions and assertationErrors
    if(asyncThrowable[0] != null){
        throw asyncThrowable[0];
    }
}

По сути, нам нужно создать окончательную ссылку на массив, которая будет использоваться внутри анонимного внутреннего класса. Я предпочел бы создать логическое значение [], потому что я могу поставить значение для управления, если нам нужно ждать (). Когда все сделано, мы просто выпускаем asyncExecuted.

Для всех пользователей Spring в настоящее время я обычно делаю свои интеграционные тесты, где задействовано асинхронное поведение:

Вызвать событие приложения в производственном коде, когда асинхронная задача (например, вызов ввода-вывода) завершена. В большинстве случаев это событие необходимо в любом случае для обработки ответа асинхронной операции в рабочей среде.

После этого события вы можете использовать следующую стратегию в своем тестовом примере:

1. Выполнить тестируемую систему
2. Прослушайте событие и убедитесь, что событие сработало
3. Делай свои утверждения

Чтобы это сломать, сначала нужно запустить некое доменное событие. Я использую UUID здесь, чтобы определить задачу, которая завершена, но вы, конечно, можете использовать что-то еще, если оно уникально.

(Обратите внимание, что следующие фрагменты кода также используют аннотации Lombok, чтобы избавиться от кода котельной пластины)

@RequiredArgsConstructor
class TaskCompletedEvent() {
  private final UUID taskId;
  // add more fields containing the result of the task if required
}

Сам производственный код обычно выглядит так:

@Component
@RequiredArgsConstructor
class Production {

  private final ApplicationEventPublisher eventPublisher;

  void doSomeTask(UUID taskId) {
    // do something like calling a REST endpoint asynchronously
    eventPublisher.publishEvent(new TaskCompletedEvent(taskId));
  }

}

Затем я могу использовать Spring, @EventListenerчтобы поймать опубликованное событие в тестовом коде. Слушатель событий немного более вовлечен, потому что он должен обрабатывать два случая потокобезопасным способом:

1. Рабочий код работает быстрее, чем тестовый пример, и событие уже сработало до того, как тестовый случай проверит событие, или
2. Тестовый случай быстрее, чем производственный код, и тестовый случай должен ждать события.

А CountDownLatchиспользуется для второго случая, как указано в других ответах здесь. Также обратите внимание, что @Orderаннотация к методу обработчика событий гарантирует, что этот метод обработчика событий вызывается после любых других прослушивателей событий, используемых в производстве.

@Component
class TaskCompletionEventListener {

  private Map<UUID, CountDownLatch> waitLatches = new ConcurrentHashMap<>();
  private List<UUID> eventsReceived = new ArrayList<>();

  void waitForCompletion(UUID taskId) {
    synchronized (this) {
      if (eventAlreadyReceived(taskId)) {
        return;
      }
      checkNobodyIsWaiting(taskId);
      createLatch(taskId);
    }
    waitForEvent(taskId);
  }

  private void checkNobodyIsWaiting(UUID taskId) {
    if (waitLatches.containsKey(taskId)) {
      throw new IllegalArgumentException("Only one waiting test per task ID supported, but another test is already waiting for " + taskId + " to complete.");
    }
  }

  private boolean eventAlreadyReceived(UUID taskId) {
    return eventsReceived.remove(taskId);
  }

  private void createLatch(UUID taskId) {
    waitLatches.put(taskId, new CountDownLatch(1));
  }

  @SneakyThrows
  private void waitForEvent(UUID taskId) {
    var latch = waitLatches.get(taskId);
    latch.await();
  }

  @EventListener
  @Order
  void eventReceived(TaskCompletedEvent event) {
    var taskId = event.getTaskId();
    synchronized (this) {
      if (isSomebodyWaiting(taskId)) {
        notifyWaitingTest(taskId);
      } else {
        eventsReceived.add(taskId);
      }
    }
  }

  private boolean isSomebodyWaiting(UUID taskId) {
    return waitLatches.containsKey(taskId);
  }

  private void notifyWaitingTest(UUID taskId) {
    var latch = waitLatches.remove(taskId);
    latch.countDown();
  }

}

Последний шаг - выполнить тестируемую систему в тестовом примере. Я использую тест SpringBoot с JUnit 5 здесь, но это должно работать одинаково для всех тестов, использующих контекст Spring.

@SpringBootTest
class ProductionIntegrationTest {

  @Autowired
  private Production sut;

  @Autowired
  private TaskCompletionEventListener listener;

  @Test
  void thatTaskCompletesSuccessfully() {
    var taskId = UUID.randomUUID();
    sut.doSomeTask(taskId);
    listener.waitForCompletion(taskId);
    // do some assertions like looking into the DB if value was stored successfully
  }

}

Обратите внимание, что в отличие от других ответов здесь, это решение также будет работать, если вы выполняете свои тесты параллельно и несколько потоков одновременно выполняют асинхронный код.

Если вы хотите проверить логику, просто не проверяйте ее асинхронно.

Например, чтобы протестировать этот код, который работает на результатах асинхронного метода.

public class Example {
    private Dependency dependency;

    public Example(Dependency dependency) {
        this.dependency = dependency;            
    }

    public CompletableFuture<String> someAsyncMethod(){
        return dependency.asyncMethod()
                .handle((r,ex) -> {
                    if(ex != null) {
                        return "got exception";
                    } else {
                        return r.toString();
                    }
                });
    }
}

public class Dependency {
    public CompletableFuture<Integer> asyncMethod() {
        // do some async stuff       
    }
}

В тесте макет зависимости с синхронной реализацией. Модульное тестирование полностью синхронно и выполняется за 150 мс.

public class DependencyTest {
    private Example sut;
    private Dependency dependency;

    public void setup() {
        dependency = Mockito.mock(Dependency.class);;
        sut = new Example(dependency);
    }

    @Test public void success() throws InterruptedException, ExecutionException {
        when(dependency.asyncMethod()).thenReturn(CompletableFuture.completedFuture(5));

        // When
        CompletableFuture<String> result = sut.someAsyncMethod();

        // Then
        assertThat(result.isCompletedExceptionally(), is(equalTo(false)));
        String value = result.get();
        assertThat(value, is(equalTo("5")));
    }

    @Test public void failed() throws InterruptedException, ExecutionException {
        // Given
        CompletableFuture<Integer> c = new CompletableFuture<Integer>();
        c.completeExceptionally(new RuntimeException("failed"));
        when(dependency.asyncMethod()).thenReturn(c);

        // When
        CompletableFuture<String> result = sut.someAsyncMethod();

        // Then
        assertThat(result.isCompletedExceptionally(), is(equalTo(false)));
        String value = result.get();
        assertThat(value, is(equalTo("got exception")));
    }
}

Вы не проверяете асинхронное поведение, но можете проверить правильность логики.