Skip to content

Commit

Permalink
Add cancel async operation tests (temporalio#2250)
Browse files Browse the repository at this point in the history
Add cancel async operation tests
  • Loading branch information
Quinn-With-Two-Ns committed Oct 10, 2024
1 parent e60ace9 commit 1c509d6
Show file tree
Hide file tree
Showing 4 changed files with 118 additions and 30 deletions.
Original file line number Diff line number Diff line change
@@ -0,0 +1,112 @@
/*
* Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
*
* Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Modifications copyright (C) 2017 Uber Technologies, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this material except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package io.temporal.workflow.nexus;

import io.nexusrpc.handler.OperationHandler;
import io.nexusrpc.handler.OperationImpl;
import io.nexusrpc.handler.ServiceImpl;
import io.temporal.client.WorkflowFailedException;
import io.temporal.client.WorkflowOptions;
import io.temporal.failure.CanceledFailure;
import io.temporal.failure.NexusOperationFailure;
import io.temporal.nexus.WorkflowClientOperationHandlers;
import io.temporal.testing.internal.SDKTestWorkflowRule;
import io.temporal.workflow.*;
import io.temporal.workflow.shared.TestNexusServices;
import io.temporal.workflow.shared.TestWorkflows;
import java.time.Duration;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;

public class CancelAsyncOperationTest {
@Rule
public SDKTestWorkflowRule testWorkflowRule =
SDKTestWorkflowRule.newBuilder()
.setWorkflowTypes(TestNexus.class, AsyncWorkflowOperationTest.TestOperationWorkflow.class)
.setNexusServiceImplementation(new TestNexusServiceImpl())
.build();

@Test
public void asyncOperationImmediatelyCancelled() {
TestWorkflows.TestWorkflow1 workflowStub =
testWorkflowRule.newWorkflowStubTimeoutOptions(TestWorkflows.TestWorkflow1.class);
WorkflowFailedException exception =
Assert.assertThrows(
WorkflowFailedException.class, () -> workflowStub.execute("immediately"));
Assert.assertTrue(exception.getCause() instanceof NexusOperationFailure);
NexusOperationFailure nexusFailure = (NexusOperationFailure) exception.getCause();
Assert.assertTrue(nexusFailure.getCause() instanceof CanceledFailure);
CanceledFailure canceledFailure = (CanceledFailure) nexusFailure.getCause();
Assert.assertEquals(
"operation canceled before it was started", canceledFailure.getOriginalMessage());
}

@Test
public void asyncOperationCancelled() {
TestWorkflows.TestWorkflow1 workflowStub =
testWorkflowRule.newWorkflowStubTimeoutOptions(TestWorkflows.TestWorkflow1.class);
WorkflowFailedException exception =
Assert.assertThrows(WorkflowFailedException.class, () -> workflowStub.execute(""));
Assert.assertTrue(exception.getCause() instanceof NexusOperationFailure);
NexusOperationFailure nexusFailure = (NexusOperationFailure) exception.getCause();
Assert.assertTrue(nexusFailure.getCause() instanceof CanceledFailure);
}

public static class TestNexus implements TestWorkflows.TestWorkflow1 {
@Override
public String execute(String input) {
NexusOperationOptions options =
NexusOperationOptions.newBuilder()
.setScheduleToCloseTimeout(Duration.ofSeconds(10))
.build();
NexusServiceOptions serviceOptions =
NexusServiceOptions.newBuilder().setOperationOptions(options).build();
TestNexusServices.TestNexusService1 serviceStub =
Workflow.newNexusServiceStub(TestNexusServices.TestNexusService1.class, serviceOptions);
Workflow.newCancellationScope(
() -> {
NexusOperationHandle<String> handle =
Workflow.startNexusOperation(serviceStub::operation, "block");
if (input.isEmpty()) {
handle.getExecution().get();
}
CancellationScope.current().cancel();
handle.getResult().get();
})
.run();
return "Should not get here";
}
}

@ServiceImpl(service = TestNexusServices.TestNexusService1.class)
public class TestNexusServiceImpl {
@OperationImpl
public OperationHandler<String, String> operation() {
return WorkflowClientOperationHandlers.fromWorkflowMethod(
(context, details, client, input) ->
client.newWorkflowStub(
AsyncWorkflowOperationTest.OperationWorkflow.class,
WorkflowOptions.newBuilder().setWorkflowId(details.getRequestId()).build())
::execute);
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@
import org.junit.Rule;
import org.junit.Test;

public class SyncClientOperationTest extends BaseNexusTest {
public class SyncClientOperationTest {
private final TestStatsReporter reporter = new TestStatsReporter();

@Rule
Expand All @@ -56,11 +56,6 @@ public class SyncClientOperationTest extends BaseNexusTest {
.setNexusServiceImplementation(new TestNexusServiceImpl())
.build();

@Override
protected SDKTestWorkflowRule getTestWorkflowRule() {
return testWorkflowRule;
}

@Test
public void syncClientOperation() {
TestUpdatedWorkflow workflowStub =
Expand Down Expand Up @@ -105,10 +100,7 @@ public String execute() {
.setScheduleToCloseTimeout(Duration.ofSeconds(1))
.build();
NexusServiceOptions serviceOptions =
NexusServiceOptions.newBuilder()
.setEndpoint(getEndpointName())
.setOperationOptions(options)
.build();
NexusServiceOptions.newBuilder().setOperationOptions(options).build();
// Try to call a synchronous operation in a blocking way
TestNexusServices.TestNexusService1 serviceStub =
Workflow.newNexusServiceStub(TestNexusServices.TestNexusService1.class, serviceOptions);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@
import org.junit.Rule;
import org.junit.Test;

public class WorkflowHandleFuncTest extends BaseNexusTest {
public class WorkflowHandleFuncTest {
@Rule
public SDKTestWorkflowRule testWorkflowRule =
SDKTestWorkflowRule.newBuilder()
Expand All @@ -46,11 +46,6 @@ public class WorkflowHandleFuncTest extends BaseNexusTest {
.setNexusServiceImplementation(new TestNexusServiceFuncImpl())
.build();

@Override
protected SDKTestWorkflowRule getTestWorkflowRule() {
return testWorkflowRule;
}

@Test
public void handleTests() {
TestWorkflows.TestWorkflow1 workflowStub =
Expand All @@ -67,10 +62,7 @@ public String execute(String input) {
.setScheduleToCloseTimeout(Duration.ofSeconds(10))
.build();
NexusServiceOptions serviceOptions =
NexusServiceOptions.newBuilder()
.setEndpoint(getEndpointName())
.setOperationOptions(options)
.build();
NexusServiceOptions.newBuilder().setOperationOptions(options).build();

TestNexusServiceFunc serviceStub =
Workflow.newNexusServiceStub(TestNexusServiceFunc.class, serviceOptions);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@
import org.junit.Rule;
import org.junit.Test;

public class WorkflowHandleProcTest extends BaseNexusTest {
public class WorkflowHandleProcTest {
@Rule
public SDKTestWorkflowRule testWorkflowRule =
SDKTestWorkflowRule.newBuilder()
Expand All @@ -48,11 +48,6 @@ public class WorkflowHandleProcTest extends BaseNexusTest {
.setNexusServiceImplementation(new TestNexusServiceFuncImpl())
.build();

@Override
protected SDKTestWorkflowRule getTestWorkflowRule() {
return testWorkflowRule;
}

@Test
public void handleTests() {
TestWorkflows.TestWorkflow1 workflowStub =
Expand All @@ -69,10 +64,7 @@ public String execute(String input) {
.setScheduleToCloseTimeout(Duration.ofSeconds(10))
.build();
NexusServiceOptions serviceOptions =
NexusServiceOptions.newBuilder()
.setEndpoint(getEndpointName())
.setOperationOptions(options)
.build();
NexusServiceOptions.newBuilder().setOperationOptions(options).build();

TestNexusServiceProc serviceStub =
Workflow.newNexusServiceStub(TestNexusServiceProc.class, serviceOptions);
Expand Down

0 comments on commit 1c509d6

Please sign in to comment.