Skip to content

Commit 91da0bb

Browse files
EvanthxEvanthx
committed
Added some tests
1 parent 7daf427 commit 91da0bb

2 files changed

Lines changed: 257 additions & 0 deletions

File tree

‎temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerLinkPropagationTest.java‎

Lines changed: 177 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,177 @@
1+
package io.temporal.internal.client;
2+
3+
import static org.mockito.ArgumentMatchers.any;
4+
import static org.mockito.Mockito.mock;
5+
import static org.mockito.Mockito.when;
6+
7+
import com.uber.m3.tally.RootScopeBuilder;
8+
import com.uber.m3.tally.Scope;
9+
import io.temporal.api.common.v1.Link;
10+
import io.temporal.api.common.v1.WorkflowExecution;
11+
import io.temporal.api.enums.v1.EventType;
12+
import io.temporal.api.workflowservice.v1.SignalWorkflowExecutionRequest;
13+
import io.temporal.api.workflowservice.v1.SignalWorkflowExecutionResponse;
14+
import io.temporal.client.WorkflowClient;
15+
import io.temporal.client.WorkflowClientOptions;
16+
import io.temporal.common.interceptors.Header;
17+
import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.WorkflowSignalInput;
18+
import io.temporal.internal.client.external.GenericWorkflowClient;
19+
import io.temporal.internal.nexus.CurrentNexusOperationContext;
20+
import io.temporal.internal.nexus.InternalNexusOperationContext;
21+
import java.util.Arrays;
22+
import java.util.Collections;
23+
import java.util.List;
24+
import org.junit.After;
25+
import org.junit.Assert;
26+
import org.junit.Before;
27+
import org.junit.Test;
28+
import org.mockito.ArgumentCaptor;
29+
30+
/**
31+
* Unit tests for {@link RootWorkflowClientInvoker#signal} link propagation in and out of the Nexus
32+
* operation context. These run against mocked dependencies and exercise the code paths that the
33+
* integration tests in {@code SignalOperationLinkingTest} can only cover when a real flag-enabled
34+
* server is available.
35+
*/
36+
public class RootWorkflowClientInvokerLinkPropagationTest {
37+
38+
private static final String NAMESPACE = "test-namespace";
39+
private static final String WORKFLOW_ID = "wf-target";
40+
41+
private GenericWorkflowClient genericClient;
42+
private RootWorkflowClientInvoker invoker;
43+
private InternalNexusOperationContext nexusCtx;
44+
45+
@Before
46+
public void setUp() {
47+
genericClient = mock(GenericWorkflowClient.class);
48+
invoker =
49+
new RootWorkflowClientInvoker(
50+
genericClient,
51+
WorkflowClientOptions.newBuilder()
52+
.setNamespace(NAMESPACE)
53+
.validateAndBuildWithDefaults(),
54+
new WorkerFactoryRegistry());
55+
Scope metricsScope = new RootScopeBuilder().reportEvery(com.uber.m3.util.Duration.ofMillis(10));
56+
nexusCtx =
57+
new InternalNexusOperationContext(
58+
NAMESPACE, "tq", "endpoint", metricsScope, mock(WorkflowClient.class));
59+
CurrentNexusOperationContext.set(nexusCtx);
60+
}
61+
62+
@After
63+
public void tearDown() {
64+
CurrentNexusOperationContext.unset();
65+
}
66+
67+
/**
68+
* Happy path against a flag-enabled server: inbound nexus links are forwarded onto the
69+
* SignalWorkflowExecutionRequest, and the response's backlink is captured back onto the operation
70+
* context.
71+
*/
72+
@Test
73+
public void signalForwardsInboundLinksAndCapturesResponseBacklink() {
74+
Link inboundLink =
75+
workflowEventLink(
76+
"caller-wf", "caller-run", EventType.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED);
77+
nexusCtx.setNexusOperationLinks(Collections.singletonList(inboundLink));
78+
79+
Link responseLink =
80+
workflowEventLink(
81+
WORKFLOW_ID, "target-run", EventType.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED);
82+
SignalWorkflowExecutionResponse response =
83+
SignalWorkflowExecutionResponse.newBuilder().setLink(responseLink).build();
84+
when(genericClient.signal(any(SignalWorkflowExecutionRequest.class))).thenReturn(response);
85+
86+
invoker.signal(newSignalInput());
87+
88+
// Forward direction: the request the SDK sent carries the inbound link.
89+
ArgumentCaptor<SignalWorkflowExecutionRequest> captor =
90+
ArgumentCaptor.forClass(SignalWorkflowExecutionRequest.class);
91+
org.mockito.Mockito.verify(genericClient).signal(captor.capture());
92+
SignalWorkflowExecutionRequest sent = captor.getValue();
93+
Assert.assertEquals("request should carry the single inbound link", 1, sent.getLinksCount());
94+
Assert.assertEquals(inboundLink, sent.getLinks(0));
95+
96+
// Backward direction: the response's link is now on the context for the task handler to read.
97+
List<Link> captured = nexusCtx.getSignalWorkflowResponseLinks();
98+
Assert.assertEquals("expected one captured backlink", 1, captured.size());
99+
Assert.assertEquals(responseLink, captured.get(0));
100+
}
101+
102+
/**
103+
* Older-server compatibility: the server returns a response without {@code link} set. The SDK
104+
* must not crash and must leave the operation context's backlink list empty.
105+
*/
106+
@Test
107+
public void signalAgainstOlderServerCapturesNoBacklink() {
108+
Link inboundLink =
109+
workflowEventLink(
110+
"caller-wf", "caller-run", EventType.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED);
111+
nexusCtx.setNexusOperationLinks(Collections.singletonList(inboundLink));
112+
113+
// Pre-1.31 server / flag-off server: response has no link.
114+
SignalWorkflowExecutionResponse response = SignalWorkflowExecutionResponse.getDefaultInstance();
115+
when(genericClient.signal(any(SignalWorkflowExecutionRequest.class))).thenReturn(response);
116+
117+
invoker.signal(newSignalInput());
118+
119+
// Forward direction still works regardless of server version.
120+
ArgumentCaptor<SignalWorkflowExecutionRequest> captor =
121+
ArgumentCaptor.forClass(SignalWorkflowExecutionRequest.class);
122+
org.mockito.Mockito.verify(genericClient).signal(captor.capture());
123+
Assert.assertEquals(1, captor.getValue().getLinksCount());
124+
125+
// Backward direction: no backlink captured because the server didn't send one.
126+
Assert.assertTrue(
127+
"expected no captured backlink when server returned no link",
128+
nexusCtx.getSignalWorkflowResponseLinks().isEmpty());
129+
}
130+
131+
/**
132+
* Multi-signal: two signal RPCs in a row each contribute a backlink; both must be captured in
133+
* order on the context, ready for the task handler to drain into the operation response.
134+
*/
135+
@Test
136+
public void multipleSignalsAccumulateAllBacklinks() {
137+
Link firstResponseLink =
138+
workflowEventLink("callee-a", "run-a", EventType.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED);
139+
Link secondResponseLink =
140+
workflowEventLink("callee-b", "run-b", EventType.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED);
141+
when(genericClient.signal(any(SignalWorkflowExecutionRequest.class)))
142+
.thenReturn(SignalWorkflowExecutionResponse.newBuilder().setLink(firstResponseLink).build())
143+
.thenReturn(
144+
SignalWorkflowExecutionResponse.newBuilder().setLink(secondResponseLink).build());
145+
146+
invoker.signal(newSignalInput());
147+
invoker.signal(newSignalInput());
148+
149+
List<Link> captured = nexusCtx.getSignalWorkflowResponseLinks();
150+
Assert.assertEquals(
151+
"expected one backlink per signal call",
152+
Arrays.asList(firstResponseLink, secondResponseLink),
153+
captured);
154+
}
155+
156+
// ── helpers ──────────────────────────────────────────────────────────────────────────────
157+
158+
private static WorkflowSignalInput newSignalInput() {
159+
return new WorkflowSignalInput(
160+
WorkflowExecution.newBuilder().setWorkflowId(WORKFLOW_ID).build(),
161+
"test-signal",
162+
Header.empty(),
163+
new Object[] {"payload"});
164+
}
165+
166+
private static Link workflowEventLink(String workflowId, String runId, EventType eventType) {
167+
return Link.newBuilder()
168+
.setWorkflowEvent(
169+
Link.WorkflowEvent.newBuilder()
170+
.setNamespace(NAMESPACE)
171+
.setWorkflowId(workflowId)
172+
.setRunId(runId)
173+
.setEventRef(
174+
Link.WorkflowEvent.EventReference.newBuilder().setEventType(eventType)))
175+
.build();
176+
}
177+
}

‎temporal-sdk/src/test/java/io/temporal/internal/nexus/NexusTaskHandlerImplTest.java‎

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,9 +8,12 @@
88
import com.uber.m3.util.Duration;
99
import io.nexusrpc.Header;
1010
import io.nexusrpc.handler.*;
11+
import io.temporal.api.common.v1.Link;
1112
import io.temporal.api.common.v1.Payload;
13+
import io.temporal.api.enums.v1.EventType;
1214
import io.temporal.api.nexus.v1.Request;
1315
import io.temporal.api.nexus.v1.StartOperationRequest;
16+
import io.temporal.api.nexus.v1.StartOperationResponse;
1417
import io.temporal.api.workflowservice.v1.PollNexusTaskQueueResponse;
1518
import io.temporal.client.WorkflowClient;
1619
import io.temporal.common.converter.DataConverter;
@@ -157,6 +160,83 @@ public void startAsyncSyncOperation() throws TimeoutException {
157160
"test id", result.getResponse().getStartOperation().getAsyncSuccess().getOperationToken());
158161
}
159162

163+
/**
164+
* Verify that signal-response backlinks stashed on the {@link InternalNexusOperationContext}
165+
* during a handler invocation are merged into the resulting {@code StartOperationResponse.Async}
166+
* via {@link io.temporal.internal.common.LinkConverter}. No server required.
167+
*/
168+
@Test
169+
public void asyncResponseIncludesSignalBacklinks() throws TimeoutException {
170+
WorkflowClient client = mock(WorkflowClient.class);
171+
NexusTaskHandlerImpl nexusTaskHandlerImpl =
172+
new NexusTaskHandlerImpl(
173+
client, NAMESPACE, TASK_QUEUE, dataConverter, new WorkerInterceptor[] {});
174+
nexusTaskHandlerImpl.registerNexusServiceImplementations(
175+
new Object[] {new BacklinkStashingAsyncServiceImpl()});
176+
nexusTaskHandlerImpl.start();
177+
178+
PollNexusTaskQueueResponse.Builder task =
179+
PollNexusTaskQueueResponse.newBuilder()
180+
.setRequest(
181+
Request.newBuilder()
182+
.setStartOperation(
183+
StartOperationRequest.newBuilder()
184+
.setOperation("operation")
185+
.setService("TestNexusService1")
186+
.setPayload(dataConverter.toPayload("op-token").get())
187+
.build()));
188+
189+
NexusTaskHandler.Result result =
190+
nexusTaskHandlerImpl.handle(new NexusTask(task, null, null), metricsScope);
191+
192+
Assert.assertNull(result.getHandlerException());
193+
StartOperationResponse.Async async = result.getResponse().getStartOperation().getAsyncSuccess();
194+
Assert.assertEquals("op-token", async.getOperationToken());
195+
Assert.assertEquals(
196+
"expected one signal backlink on the async response", 1, async.getLinksCount());
197+
// The backlink was stashed as a WorkflowEvent for callee workflowId "callee-wf"; the response
198+
// should contain a temporal:// URL referencing that workflow.
199+
Assert.assertTrue(
200+
"expected backlink URL to reference the callee workflow, got: "
201+
+ async.getLinks(0).getUrl(),
202+
async.getLinks(0).getUrl().contains("callee-wf"));
203+
}
204+
205+
/**
206+
* Handler that simulates what a real Nexus operation would do after issuing a signal: stash a
207+
* backlink on the operation context, then return an async result. Lets us exercise the
208+
* async-response link merge in {@link NexusTaskHandlerImpl} without standing up a real signal
209+
* RPC.
210+
*/
211+
@ServiceImpl(service = TestNexusServices.TestNexusService1.class)
212+
public class BacklinkStashingAsyncServiceImpl {
213+
@OperationImpl
214+
public OperationHandler<String, String> operation() {
215+
return new OperationHandler<String, String>() {
216+
@Override
217+
public OperationStartResult<String> start(
218+
OperationContext ctx, OperationStartDetails details, @Nullable String token) {
219+
Link backlink =
220+
Link.newBuilder()
221+
.setWorkflowEvent(
222+
Link.WorkflowEvent.newBuilder()
223+
.setNamespace(NAMESPACE)
224+
.setWorkflowId("callee-wf")
225+
.setRunId("callee-run-id")
226+
.setEventRef(
227+
Link.WorkflowEvent.EventReference.newBuilder()
228+
.setEventType(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED)))
229+
.build();
230+
CurrentNexusOperationContext.get().addSignalWorkflowResponseLink(backlink);
231+
return OperationStartResult.async(token);
232+
}
233+
234+
@Override
235+
public void cancel(OperationContext ctx, OperationCancelDetails details) {}
236+
};
237+
}
238+
}
239+
160240
@ServiceImpl(service = TestNexusServices.TestNexusService1.class)
161241
public class TestNexusServiceImpl {
162242
@OperationImpl

0 commit comments

Comments
 (0)