So verwenden Sie JUnit zum Testen asynchroner Prozesse

Wie testen Sie Methoden, die asynchrone Prozesse mit JUnit auslösen?

Ich weiß nicht, wie ich meinen Test auf das Ende des Prozesses warten lassen soll (es ist nicht gerade ein Komponententest, es ist eher ein Integrationstest, da es mehrere Klassen und nicht nur eine umfasst).

IMHO ist es eine schlechte Praxis, Unit-Tests erstellen zu lassen oder auf Threads usw. zu warten. Sie möchten, dass diese Tests in Sekundenbruchteilen ausgeführt werden. Aus diesem Grund möchte ich einen zweistufigen Ansatz zum Testen von asynchronen Prozessen vorschlagen.

1. Testen Sie, ob Ihr asynchroner Prozess ordnungsgemäß gesendet wurde. Sie können das Objekt verspotten, das Ihre asynchronen Anforderungen akzeptiert, und sicherstellen, dass der übergebene Job die richtigen Eigenschaften usw. hat.
2. Testen Sie, ob Ihre asynchronen Rückrufe die richtigen Dinge tun. Hier können Sie den ursprünglich übergebenen Job verspotten und davon ausgehen, dass er ordnungsgemäß initialisiert wurde, und überprüfen, ob Ihre Rückrufe korrekt sind.

Eine Alternative ist die Verwendung der CountDownLatch- Klasse.

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

HINWEIS Sie können nicht einfach mit einem regulären Objekt synchronisiert als Sperre verwenden, da schnelle Rückrufe die Sperre aufheben können, bevor die Wartemethode der Sperre aufgerufen wird. Siehe diesen Blog-Beitrag von Joe Walnes.

BEARBEITEN Synchronisierte Blöcke um CountDownLatch wurden dank Kommentaren von @jtahlborn und @Ring entfernt

Sie können versuchen, die Awaitility- Bibliothek zu verwenden. Es macht es einfach, die Systeme zu testen, über die Sie sprechen.

Wenn Sie eine CompletableFuture (in Java 8 eingeführt) oder eine SettableFuture (von Google Guava ) verwenden, können Sie Ihren Test sofort beenden, anstatt eine voreingestellte Zeit zu warten. Ihr Test würde ungefähr so ​​aussehen:

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

Starten Sie den Prozess und warten Sie mit a auf das Ergebnis Future.

Eine Methode, die ich zum Testen asynchroner Methoden als sehr nützlich empfunden habe, ist das Einfügen einer ExecutorInstanz in den Konstruktor des zu testenden Objekts. In der Produktion ist die Executor-Instanz so konfiguriert, dass sie asynchron ausgeführt wird, während sie im Test verspottet werden kann, um synchron ausgeführt zu werden.

Angenommen, ich versuche, die asynchrone Methode zu testen 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);
      }
    });
  }
}

In der Produktion würde ich Foomit einer Executors.newSingleThreadExecutor()Executor-Instanz konstruieren, während ich sie im Test wahrscheinlich mit einem synchronen Executor konstruieren würde, der Folgendes ausführt:

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

Jetzt ist mein JUnit-Test der asynchronen Methode ziemlich sauber -

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

Es ist an sich nichts Falsches daran, Threaded / Async-Code zu testen, insbesondere wenn Threading der Punkt des Codes ist, den Sie testen. Der allgemeine Ansatz zum Testen dieses Materials ist:

• Blockieren Sie den Haupttest-Thread
• Erfassen Sie fehlgeschlagene Zusicherungen von anderen Threads
• Entsperren Sie den Haupttest-Thread
• Fehler erneut auslösen

Aber das ist eine Menge Boilerplate für einen Test. Ein besserer / einfacherer Ansatz ist die Verwendung von ConcurrentUnit :

  final Waiter waiter = new Waiter();

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

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

Der Vorteil gegenüber dem CountdownLatchAnsatz besteht darin, dass er weniger ausführlich ist, da Assertionsfehler, die in einem Thread auftreten, ordnungsgemäß an den Hauptthread gemeldet werden, was bedeutet, dass der Test fehlschlägt, wenn er sollte. Eine Beschreibung, die den CountdownLatchAnsatz von ConcurrentUnit vergleicht , finden Sie hier .

Ich habe auch einen Blog-Beitrag zum Thema für diejenigen geschrieben, die etwas mehr Details erfahren möchten.

Wie wäre es mit einem Aufruf SomeObject.waitund notifyAllwie hier beschrieben ODER mit der Robotiums- Solo.waitForCondition(...) Methode ODER mit einer Klasse, die ich dazu geschrieben habe ( Informationen zur Verwendung finden Sie in den Kommentaren und in der Testklasse )

Ich finde eine Bibliothek socket.io zum Testen der asynchronen Logik. Mit LinkedBlockingQueue sieht es einfach und kurz aus . Hier ist ein Beispiel :

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

Wenn Sie LinkedBlockingQueue verwenden, blockieren Sie die API, bis Sie das Ergebnis wie auf synchrone Weise erhalten. Stellen Sie eine Zeitüberschreitung ein, um zu vermeiden, dass zu viel Zeit für das Warten auf das Ergebnis benötigt wird.

Es ist erwähnenswert, dass es Testing Concurrent Programsin Concurrency in Practice ein sehr nützliches Kapitel gibt, in dem einige Unit-Test-Ansätze beschrieben und Lösungen für Probleme angegeben werden.

Dies ist, was ich heutzutage verwende, wenn das Testergebnis asynchron erzeugt wird.

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

Mit statischen Importen liest sich der Test ein bisschen nett. (Beachten Sie, dass ich in diesem Beispiel einen Thread beginne, um die Idee zu veranschaulichen.)

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

Wenn f.completenicht aufgerufen, schlägt der Test nach einer Zeitüberschreitung fehl. Sie können auch verwenden, f.completeExceptionallyum früh zu scheitern.

Hier gibt es viele Antworten, aber eine einfache besteht darin, einfach eine fertige CompletableFuture zu erstellen und diese zu verwenden:

CompletableFuture.completedFuture("donzo")

Also in meinem Test:

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

Ich stelle nur sicher, dass all dieses Zeug trotzdem angerufen wird. Diese Technik funktioniert, wenn Sie diesen Code verwenden:

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

Es wird direkt durchlaufen, wenn alle CompletableFutures fertig sind!

Vermeiden Sie es, mit parallelen Threads zu testen, wann immer Sie können (was meistens der Fall ist). Dadurch werden Ihre Tests nur schuppig (manchmal bestanden, manchmal nicht bestanden).

Nur wenn Sie eine andere Bibliothek / ein anderes System aufrufen müssen, müssen Sie möglicherweise auf andere Threads warten. Verwenden Sie in diesem Fall immer die Awaitility- Bibliothek anstelle von Thread.sleep().

Rufen Sie niemals einfach an get()oder nehmen Sie join()an Ihren Tests teil, sonst werden Ihre Tests möglicherweise für immer auf Ihrem CI-Server ausgeführt, falls die Zukunft nie abgeschlossen wird. Setzen Sie isDone()Ihre Tests immer zuerst durch, bevor Sie anrufen get(). Für CompletionStage also .toCompletableFuture().isDone().

Wenn Sie eine nicht blockierende Methode wie diese testen:

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

dann sollten Sie nicht nur das Ergebnis testen , indem Sie ein ausgefülltes Zukunft in dem Test bestanden, sollten Sie auch sicherstellen , dass Ihre Methode doSomething()blockiert nicht durch den Aufruf join()oder get(). Dies ist insbesondere wichtig, wenn Sie ein nicht blockierendes Framework verwenden.

Testen Sie dazu mit einer nicht abgeschlossenen Zukunft, die Sie manuell abgeschlossen haben:

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

Auf diese Weise future.join()schlägt der Test fehl , wenn Sie doSomething () hinzufügen .

Wenn Ihr Dienst einen ExecutorService wie in verwendet thenApplyAsync(..., executorService), fügen Sie in Ihren Tests einen ExecutorService mit einem Thread wie den von guava ein:

ExecutorService executorService = Executors.newSingleThreadExecutor();

Wenn Ihr Code den forkJoinPool verwendet, z. B. thenApplyAsync(...), schreiben Sie den Code neu, um einen ExecutorService zu verwenden (es gibt viele gute Gründe), oder verwenden Sie Awaitility.

Um das Beispiel zu verkürzen, habe ich BarService zu einem Methodenargument gemacht, das im Test als Java8-Lambda implementiert wurde. In der Regel handelt es sich um eine injizierte Referenz, die Sie verspotten würden.

Ich bevorzuge es zu warten und zu benachrichtigen. Es ist einfach und klar.

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

Grundsätzlich müssen wir eine endgültige Array-Referenz erstellen, die innerhalb einer anonymen inneren Klasse verwendet werden soll. Ich würde lieber einen Booleschen Wert [] erstellen, da ich einen Wert zur Steuerung setzen kann, wenn wir warten müssen (). Wenn alles erledigt ist, geben wir einfach asyncExecuted frei.

Für alle Spring-Benutzer da draußen mache ich heutzutage normalerweise meine Integrationstests, bei denen es um asynchrones Verhalten geht:

Lösen Sie ein Anwendungsereignis im Produktionscode aus, wenn eine asynchrone Aufgabe (z. B. ein E / A-Aufruf) abgeschlossen ist. Meistens ist dieses Ereignis ohnehin erforderlich, um die Reaktion des asynchronen Vorgangs in der Produktion zu verarbeiten.

Mit diesem Ereignis können Sie in Ihrem Testfall die folgende Strategie anwenden:

1. Führen Sie das zu testende System aus
2. Hören Sie auf das Ereignis und stellen Sie sicher, dass das Ereignis ausgelöst wurde
3. Mach deine Behauptungen

Um dies zu beheben, benötigen Sie zunächst eine Art Domain-Ereignis, um ausgelöst zu werden. Ich verwende hier eine UUID, um die abgeschlossene Aufgabe zu identifizieren, aber Sie können natürlich auch etwas anderes verwenden, solange es eindeutig ist.

(Beachten Sie, dass die folgenden Codefragmente auch Lombok- Anmerkungen verwenden, um den Kesselplattencode zu entfernen.)

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

Der Produktionscode selbst sieht dann normalerweise so aus:

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

}

Ich kann dann eine Feder verwenden @EventListener, um das veröffentlichte Ereignis im Testcode abzufangen. Der Ereignis-Listener ist etwas komplizierter, da er zwei Fälle threadsicher behandeln muss:

1. Der Produktionscode ist schneller als der Testfall und das Ereignis wurde bereits ausgelöst, bevor der Testfall nach dem Ereignis sucht, oder
2. Der Testfall ist schneller als der Produktionscode und der Testfall muss auf das Ereignis warten.

A CountDownLatchwird für den zweiten Fall verwendet, wie in anderen Antworten hier erwähnt. Beachten Sie außerdem, dass die @OrderAnmerkung zur Ereignishandlermethode sicherstellt, dass diese Ereignishandlermethode nach allen anderen in der Produktion verwendeten Ereignislistenern aufgerufen wird.

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

}

Der letzte Schritt besteht darin, das zu testende System in einem Testfall auszuführen. Ich verwende hier einen SpringBoot-Test mit JUnit 5, aber dies sollte für alle Tests mit einem Spring-Kontext gleich funktionieren.

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

}

Beachten Sie, dass diese Lösung im Gegensatz zu anderen Antworten hier auch funktioniert, wenn Sie Ihre Tests parallel ausführen und mehrere Threads gleichzeitig den asynchronen Code ausführen.

Wenn Sie die Logik testen möchten, testen Sie sie nicht asynchron.

Zum Beispiel, um diesen Code zu testen, der mit Ergebnissen einer asynchronen Methode arbeitet.

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       
    }
}

Im Test verspotten Sie die Abhängigkeit mit synchroner Implementierung. Der Unit-Test ist vollständig synchron und läuft in 150ms.

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

Sie testen das asynchrone Verhalten nicht, können aber testen, ob die Logik korrekt ist.