Interface TestWorkflowEnvironment

All Superinterfaces:
AutoCloseable, Closeable
All Known Implementing Classes:
TestWorkflowEnvironmentInternal
public interface TestWorkflowEnvironment extends Closeable
TestWorkflowEnvironment provides workflow unit testing capabilities.

Testing the workflow code is hard as it might be potentially very long-running. The included in-memory implementation of the Temporal service supports an automatic time skipping. Anytime a workflow under the test as well as the unit test code are waiting on a timer (or sleep) the internal service time is automatically advanced to the nearest time that unblocks one of the waiting threads. This way a workflow that runs in production for months is unit tested in milliseconds. Here is an example of a test that executes in a few milliseconds instead of over two hours that are needed for the workflow to complete: 


   public class SignaledWorkflowImpl implements SignaledWorkflow {
     private String signalInput;

    @Override
     public String workflow1(String input) {
       Workflow.sleep(Duration.ofHours(1));
       Workflow.await(() -> signalInput != null);
       Workflow.sleep(Duration.ofHours(1));
       return signalInput + "-" + input;
     }

    @Override
     public void processSignal(String input) {
       signalInput = input;
    }
  }

 @Test
  public void testSignal() throws ExecutionException, InterruptedException {
    TestWorkflowEnvironment testEnvironment = TestWorkflowEnvironment.newInstance();

    // Creates a worker that polls tasks from the service owned by the testEnvironment.
    Worker worker = testEnvironment.newWorker(TASK_QUEUE);
    worker.registerWorkflowImplementationTypes(SignaledWorkflowImpl.class);
    worker.start();

    // Creates a WorkflowClient that interacts with the server owned by the testEnvironment.
    WorkflowClient client = testEnvironment.getWorkflowClient();
    SignaledWorkflow workflow = client.newWorkflowStub(SignaledWorkflow.class);

    // Starts a workflow execution
    CompletableFuture result = WorkflowClient.execute(workflow::workflow1, "input1");

    // The sleep forwards the service clock for 65 minutes without blocking.
    // This ensures that the signal is sent after the one hour sleep in the workflow code.
    testEnvironment.sleep(Duration.ofMinutes(65));
    workflow.processSignal("signalInput");

    // Blocks until workflow is complete. Workflow sleep forwards clock for one hour and
    // this call returns almost immediately.
    assertEquals("signalInput-input1", result.get());

    // Closes workers and releases in-memory service.
    testEnvironment.close();
  }

  • Method Details

    • newInstance

      static TestWorkflowEnvironment newInstance()
      Creates the environment with default options.

    • newInstance

      static TestWorkflowEnvironment newInstance(TestEnvironmentOptions options)
      Creates the environment with supplied options.

    • startLocal

      @Experimental static TestWorkflowEnvironment startLocal()
      Starts a local Temporal dev server and returns an environment that owns it.

      Unlike the in-memory test server, a local dev-server environment does not support time skipping. 

      
 try (TestWorkflowEnvironment environment = TestWorkflowEnvironment.startLocal()) {
   Worker worker = environment.newWorker("test-task-queue");
   // Register implementations and run workflows against the local dev server.
 }

    • startLocal

      @Experimental static TestWorkflowEnvironment startLocal(@Nullable TestEnvironmentOptions testOptions)
      Starts a local Temporal dev server using the environment namespace and returns an environment that owns it. Local dev-server environments do not support time skipping.

    • startLocal

      @Experimental static TestWorkflowEnvironment startLocal(@Nonnull TemporalDevServerOptions serverOptions)
      Starts a local Temporal dev server with the supplied server options. Local dev-server environments do not support time skipping.

    • startLocal

      @Experimental static TestWorkflowEnvironment startLocal(@Nullable TestEnvironmentOptions testOptions, @Nonnull TemporalDevServerOptions serverOptions)
      Starts a local Temporal dev server and returns an environment that owns it.

      The namespace in testOptions is authoritative and is created by the dev server. Local dev-server environments do not support time skipping.

    • newWorker

      io.temporal.worker.Worker newWorker(String taskQueue)
      Creates a new Worker instance that is connected to the in-memory test Temporal service.
      Parameters:
      taskQueue - task queue to poll.

    • newWorker

      io.temporal.worker.Worker newWorker(String taskQueue, io.temporal.worker.WorkerOptions options)
      Creates a new Worker instance that is connected to the in-memory test Temporal service.
      Parameters:
      taskQueue - task queue to poll.

    • getWorkflowClient

      io.temporal.client.WorkflowClient getWorkflowClient()
      Creates a WorkflowClient that is connected to the in-memory test Temporal service.

    • getActivityClient

      io.temporal.client.ActivityClient getActivityClient()
      Creates an ActivityClient that is connected to the in-memory test Temporal service.

    • currentTimeMillis

      long currentTimeMillis()
      This time might not be equal to System.currentTimeMillis() due to time skipping.
      Returns:
      the current in-memory test Temporal service time in milliseconds or System.currentTimeMillis() if an external service without time skipping support is used

    • sleep

      void sleep(Duration duration)
      Wait until internal test Temporal service time passes the specified duration. This call also indicates that workflow time might jump forward (if none of the activities are running) up to the specified duration.

      This method falls back to Thread.sleep(long) if an external service without time skipping support is used

    • registerDelayedCallback

      void registerDelayedCallback(Duration delay, Runnable r)
      Registers a callback to run after the specified delay according to the test Temporal service internal clock.

    • registerSearchAttribute

      boolean registerSearchAttribute(String name, io.temporal.api.enums.v1.IndexedValueType type)
      Register a Search Attribute with the server.
      Parameters:
      name - Search Attribute name
      type - Search Attribute type to be used for an elastic search index
      Returns:
      true if the search attribute was registered, false if it was registered already
      See Also:

    • createNexusEndpoint

      io.temporal.api.nexus.v1.Endpoint createNexusEndpoint(String name, String taskQueue)
      Register a Nexus Endpoint with the server.
      Parameters:
      name - Nexus Endpoint name
      taskQueue - Task Queue to be used for the endpoint
      Returns:
      Endpoint object

    • deleteNexusEndpoint

      void deleteNexusEndpoint(io.temporal.api.nexus.v1.Endpoint endpoint)
      Delete a Nexus Endpoint on the server.
      Parameters:
      endpoint - current endpoint to be deleted

    • getWorkflowService

      @Deprecated io.temporal.serviceclient.WorkflowServiceStubs getWorkflowService()
      Deprecated.
      use {getWorkflowServiceStubs()
      Returns:
      the in-memory test Temporal service that is owned by this.

    • getWorkflowServiceStubs

      io.temporal.serviceclient.WorkflowServiceStubs getWorkflowServiceStubs()
      Returns:
      WorkflowServiceStubs connected to the test server (in-memory or external)

    • getOperatorServiceStubs

      io.temporal.serviceclient.OperatorServiceStubs getOperatorServiceStubs()
      Returns:
      OperatorServiceStubs connected to the test server

    • getNamespace

      String getNamespace()

    • getDiagnostics

      String getDiagnostics()
      Currently prints histories of all workflow instances stored in the service. This is useful information to print in the case of a unit test failure. A convenient way to achieve this is to add the following Rule to a unit test: 
      
  @Rule
   public TestWatcher watchman =
       new TestWatcher() {
        @Override
         protected void failed(Throwable e, Description description) {
           System.err.println(testEnvironment.getDiagnostics());
           testEnvironment.close();
         }
       };
      Returns:
      the diagnostic data about the internal service state.

    • getWorkflowExecutionHistory

      @Deprecated io.temporal.common.WorkflowExecutionHistory getWorkflowExecutionHistory(@Nonnull io.temporal.api.common.v1.WorkflowExecution execution)
      Deprecated.
      use WorkflowClient.fetchHistory(String, String)
      Parameters:
      execution - identifies the workflowId and runId (optionally) to reach the history for
      Returns:
      history of the execution

    • close

      void close()
      Calls shutdownNow() and awaitTermination(long, TimeUnit).
      Specified by:
      close in interface AutoCloseable
      Specified by:
      close in interface Closeable

    • getWorkerFactory

      io.temporal.worker.WorkerFactory getWorkerFactory()

    • start

      void start()
      Start all workers created by this factory.

    • isStarted

      boolean isStarted()
      Was start() called?

    • isShutdown

      boolean isShutdown()
      Was shutdownNow() or shutdown() called?

    • isTerminated

      boolean isTerminated()
      Are all tasks done after shutdownNow() or shutdown()?

    • shutdownTestService

      @Deprecated void shutdownTestService()
      Deprecated.
      Initiates Test Service shutdown. This method is temporarily exposed to solve long poll thread shutdown for io.temporal.workflow.interceptorsTests.InterceptorExceptionTests#testExceptionOnStart(). See issue: https://github.com/temporalio/sdk-java/issues/608

    • shutdown

      void shutdown()
      Initiates an orderly shutdown in which polls are stopped and already received workflow and activity tasks are executed. After the shutdown calls to ActivityExecutionContext.heartbeat(Object) start throwing ActivityWorkerShutdownException. Invocation has no additional effect if already shut down. This method does not wait for previously received tasks to complete execution. Use awaitTermination(long, TimeUnit) to do that.

    • shutdownNow

      void shutdownNow()
      Initiates an orderly shutdown in which polls are stopped and already received workflow and activity tasks are attempted to be stopped. This implementation cancels tasks via Thread.interrupt(), so any task that fails to respond to interrupts may never terminate. Also, after the shutdownNow calls to ActivityExecutionContext.heartbeat(Object) start throwing ActivityWorkerShutdownException. Invocation has no additional effect if already shut down. This method does not wait for previously received tasks to complete execution. Use awaitTermination(long, TimeUnit) to do that.

    • awaitTermination

      void awaitTermination(long timeout, TimeUnit unit)
      Blocks until all tasks have completed execution after a shutdown request, or the timeout occurs, or the current thread is interrupted, whichever happens first.