You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@texera.apache.org by gi...@apache.org on 2026/07/13 04:53:24 UTC
(texera) branch gh-readonly-queue/main/pr-6371-cda063b1bc55794a07dcb0836acc707153a1f1ff created (now 4876972466)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-6371-cda063b1bc55794a07dcb0836acc707153a1f1ff
in repository https://gitbox.apache.org/repos/asf/texera.git
at 4876972466 test(frontend): extend ExecuteWorkflowService coverage (#6371)
This branch includes the following new commits:
new 4876972466 test(frontend): extend ExecuteWorkflowService coverage (#6371)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(texera) 01/01: test(frontend): extend ExecuteWorkflowService coverage (#6371)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6371-cda063b1bc55794a07dcb0836acc707153a1f1ff
in repository https://gitbox.apache.org/repos/asf/texera.git
commit 4876972466b4cdcf2947e6ad14fce576146784ce
Author: Xinyuan Lin <xi...@uci.edu>
AuthorDate: Sun Jul 12 21:53:03 2026 -0700
test(frontend): extend ExecuteWorkflowService coverage (#6371)
### What changes were proposed in this PR?
Extends the `ExecuteWorkflowService` spec (was ~49% covered): the
websocket-event switch (region/region-state/worker-assignment/default),
reconfiguration-event notifications, execution-state transitions, and
request-builder helpers.
### Any related issues, documentation, discussions?
Closes #6368.
### How was this PR tested?
Added 24 tests (now 35), driving the faked WorkflowWebsocketService
event subject; clears sessionStorage in afterEach. Verified locally with
Vitest and CI's `format:ci` gate clean. Test-only change — no production
code is modified.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8 [1M context])
---
.../execute-workflow.service.spec.ts | 308 ++++++++++++++++++++-
1 file changed, 307 insertions(+), 1 deletion(-)
diff --git a/frontend/src/app/workspace/service/execute-workflow/execute-workflow.service.spec.ts b/frontend/src/app/workspace/service/execute-workflow/execute-workflow.service.spec.ts
index e1ff418abb..df43f9ef6d 100644
--- a/frontend/src/app/workspace/service/execute-workflow/execute-workflow.service.spec.ts
+++ b/frontend/src/app/workspace/service/execute-workflow/execute-workflow.service.spec.ts
@@ -30,7 +30,8 @@ import { UndoRedoService } from "../undo-redo/undo-redo.service";
import { OperatorMetadataService } from "../operator-metadata/operator-metadata.service";
import { StubOperatorMetadataService } from "../operator-metadata/stub-operator-metadata.service";
import { JointUIService } from "../joint-ui/joint-ui.service";
-import { of } from "rxjs";
+import { of, Subject } from "rxjs";
+import { WorkflowWebsocketService } from "../workflow-websocket/workflow-websocket.service";
import { mockLogicalPlan_scan_result, mockWorkflowPlan_scan_result } from "./mock-workflow-plan";
import { HttpClientTestingModule } from "@angular/common/http/testing";
@@ -45,6 +46,15 @@ import { UserService } from "src/app/common/service/user/user.service";
import { StubUserService } from "src/app/common/service/user/stub-user.service";
import { MockComputingUnitStatusService } from "../../../common/service/computing-unit/computing-unit-status/mock-computing-unit-status.service";
import { commonTestProviders } from "../../../common/testing/test-utils";
+import {
+ RegionStateEvent,
+ RegionUpdateEvent,
+ ReplayExecutionInfo,
+ TexeraWebsocketEvent,
+} from "../../types/workflow-websocket.interface";
+import { mockScanPredicate } from "../workflow-graph/model/mock-workflow-data";
+import { PAGINATION_INFO_STORAGE_KEY, ResultPaginationInfo } from "../../types/result-table.interface";
+import { sessionGetObject, sessionSetObject } from "../../../common/util/storage";
describe("ExecuteWorkflowService", () => {
let service: ExecuteWorkflowService;
@@ -82,6 +92,19 @@ describe("ExecuteWorkflowService", () => {
mockWorkflowSnapshotService = TestBed.inject(WorkflowSnapshotService);
});
+ afterEach(() => {
+ // sendExecutionRequest reads/writes sessionStorage; keep tests isolated
+ sessionStorage.clear();
+ });
+
+ // Push an event through the stream the real WorkflowWebsocketService exposes, so the
+ // constructor subscription in ExecuteWorkflowService runs exactly as it would in production.
+ // websocketEvent() publicly returns the underlying Subject; casting to Subject for the test
+ // avoids reaching into the service's private fields.
+ const emitWsEvent = (event: TexeraWebsocketEvent): void => {
+ (TestBed.inject(WorkflowWebsocketService).websocketEvent() as Subject<TexeraWebsocketEvent>).next(event);
+ };
+
it("should be created", inject([ExecuteWorkflowService], (injectedService: ExecuteWorkflowService) => {
expect(injectedService).toBeTruthy();
}));
@@ -182,4 +205,287 @@ describe("ExecuteWorkflowService", () => {
expect(resetStatusSpy).not.toHaveBeenCalled();
expect(sendExecutionRequestSpy).not.toHaveBeenCalled();
});
+
+ // ---- constructor: websocket-event routing --------------------------------------------------
+
+ it("routes region and worker-assignment websocket events to their dedicated streams", () => {
+ const regionUpdates: RegionUpdateEvent[] = [];
+ const regionStates: RegionStateEvent[] = [];
+ service.getRegionUpdateStream().subscribe(event => regionUpdates.push(event));
+ service.getRegionStateStream().subscribe(event => regionStates.push(event));
+
+ emitWsEvent({ type: "RegionUpdateEvent", regions: [] });
+ emitWsEvent({ type: "RegionStateEvent", id: 7, state: "COMPLETED" });
+ emitWsEvent({ type: "WorkerAssignmentUpdateEvent", operatorId: "opX", workerIds: ["w1", "w2"] });
+
+ expect(regionUpdates).toEqual([{ type: "RegionUpdateEvent", regions: [] }]);
+ expect(regionStates).toEqual([{ type: "RegionStateEvent", id: 7, state: "COMPLETED" }]);
+ expect(service.getWorkerIds("opX")).toEqual(["w1", "w2"]);
+ });
+
+ it("routes execution-status websocket events into the execution-state stream", () => {
+ const states: ExecutionState[] = [];
+ service.getExecutionStateStream().subscribe(event => states.push(event.current.state));
+
+ emitWsEvent({ type: "WorkflowStateEvent", state: ExecutionState.Running });
+
+ expect(service.getExecutionState().state).toBe(ExecutionState.Running);
+ expect(states).toEqual([ExecutionState.Running]);
+ });
+
+ it("does not re-emit when a websocket event maps to the unchanged execution state", () => {
+ const states: ExecutionState[] = [];
+ service.getExecutionStateStream().subscribe(event => states.push(event.current.state));
+
+ emitWsEvent({ type: "WorkflowStateEvent", state: ExecutionState.Running });
+ emitWsEvent({ type: "WorkflowStateEvent", state: ExecutionState.Running });
+
+ // second identical event is deduped by updateExecutionState's isEqual guard
+ expect(states).toEqual([ExecutionState.Running]);
+ });
+
+ // ---- handleReconfigurationEvent ------------------------------------------------------------
+
+ it("handleReconfigurationEvent shows an error notification for an invalid ModifyLogicResponse", () => {
+ const errorSpy = vi.spyOn(service["notificationService"], "error").mockImplementation(() => undefined as any);
+ service.handleReconfigurationEvent({
+ type: "ModifyLogicResponse",
+ opId: "1",
+ isValid: false,
+ errorMessage: "boom",
+ });
+ expect(errorSpy).toHaveBeenCalledWith("boom");
+ });
+
+ it("handleReconfigurationEvent confirms a valid ModifyLogicResponse", () => {
+ const infoSpy = vi.spyOn(service["notificationService"], "info").mockImplementation(() => undefined as any);
+ service.handleReconfigurationEvent({ type: "ModifyLogicResponse", opId: "1", isValid: true, errorMessage: "" });
+ expect(infoSpy).toHaveBeenCalledWith("reconfiguration registered");
+ });
+
+ it("handleReconfigurationEvent announces a ModifyLogicCompletedEvent with the operator ids", () => {
+ const infoSpy = vi.spyOn(service["notificationService"], "info").mockImplementation(() => undefined as any);
+ service.handleReconfigurationEvent({ type: "ModifyLogicCompletedEvent", opIds: ["a", "b"] });
+ expect(infoSpy).toHaveBeenCalledWith("reconfiguration on operator(s) a,b complete");
+ });
+
+ // ---- handleExecutionEvent ------------------------------------------------------------------
+
+ it("handleExecutionEvent enters Paused with empty tuples from a non-paused state", () => {
+ (service as any).currentState = { state: ExecutionState.Running };
+ const result = service.handleExecutionEvent({ type: "WorkflowStateEvent", state: ExecutionState.Paused });
+ expect(result).toEqual({ state: ExecutionState.Paused, currentTuples: {} });
+ });
+
+ it("handleExecutionEvent keeps the current state when already Paused", () => {
+ const paused = { state: ExecutionState.Paused, currentTuples: { a: { operatorID: "a", tuples: [] } } };
+ (service as any).currentState = paused;
+ const result = service.handleExecutionEvent({ type: "WorkflowStateEvent", state: ExecutionState.Paused });
+ expect(result).toBe(paused);
+ });
+
+ it("handleExecutionEvent defers the Failed state to the follow-up error event", () => {
+ const result = service.handleExecutionEvent({ type: "WorkflowStateEvent", state: ExecutionState.Failed });
+ expect(result).toBeUndefined();
+ });
+
+ it("handleExecutionEvent passes through other workflow states", () => {
+ const result = service.handleExecutionEvent({ type: "WorkflowStateEvent", state: ExecutionState.Running });
+ expect(result).toEqual({ state: ExecutionState.Running });
+ });
+
+ it("handleExecutionEvent maps RecoveryStartedEvent to the Recovering state", () => {
+ const result = service.handleExecutionEvent({ type: "RecoveryStartedEvent" });
+ expect(result).toEqual({ state: ExecutionState.Recovering });
+ });
+
+ it("handleExecutionEvent merges current tuples on top of the existing paused tuples", () => {
+ const existing = { opA: { operatorID: "opA", tuples: [] } };
+ (service as any).currentState = { state: ExecutionState.Paused, currentTuples: existing };
+ const event: TexeraWebsocketEvent = { type: "OperatorCurrentTuplesUpdateEvent", operatorID: "opB", tuples: [] };
+ const result = service.handleExecutionEvent(event);
+ expect(result).toEqual({
+ state: ExecutionState.Paused,
+ currentTuples: { opB: event, opA: existing.opA },
+ });
+ });
+
+ it("handleExecutionEvent starts a fresh paused-tuples map when not already paused", () => {
+ (service as any).currentState = { state: ExecutionState.Running };
+ const event: TexeraWebsocketEvent = { type: "OperatorCurrentTuplesUpdateEvent", operatorID: "opB", tuples: [] };
+ const result = service.handleExecutionEvent(event);
+ expect(result).toEqual({ state: ExecutionState.Paused, currentTuples: { opB: event } });
+ });
+
+ it("handleExecutionEvent maps WorkflowErrorEvent to Failed and rewrites newline escapes", () => {
+ const fatalError = {
+ message: "line1\\nline2",
+ details: "stack",
+ operatorId: "op1",
+ workerId: "w1",
+ type: { name: "RuntimeError" },
+ timestamp: { nanos: 1, seconds: 2 },
+ };
+ const result = service.handleExecutionEvent({ type: "WorkflowErrorEvent", fatalErrors: [fatalError] });
+ expect(result).toEqual({
+ state: ExecutionState.Failed,
+ errorMessages: [{ ...fatalError, message: "line1<br>line2" }],
+ });
+ });
+
+ it("handleExecutionEvent ignores unrelated events", () => {
+ expect(service.handleExecutionEvent({ type: "HeartBeatResponse" })).toBeUndefined();
+ });
+
+ // ---- getErrorMessages ----------------------------------------------------------------------
+
+ it("getErrorMessages returns the fatal errors while the execution is Failed", () => {
+ const fatalError = {
+ message: "boom",
+ details: "stack",
+ operatorId: "op1",
+ workerId: "w1",
+ type: { name: "RuntimeError" },
+ timestamp: { nanos: 1, seconds: 2 },
+ };
+ (service as any).currentState = { state: ExecutionState.Failed, errorMessages: [fatalError] };
+ expect(service.getErrorMessages()).toEqual([fatalError]);
+ });
+
+ // ---- execution entry points ----------------------------------------------------------------
+
+ it("executeWorkflow delegates to executeWorkflowWithEmailNotification with email disabled", () => {
+ const delegateSpy = vi.spyOn(service, "executeWorkflowWithEmailNotification").mockImplementation(() => {});
+ service.executeWorkflow("run-1", "op-9");
+ expect(delegateSpy).toHaveBeenCalledWith("run-1", false, "op-9");
+ });
+
+ it("executeWorkflowWithReplay builds a replay request and forwards the replay info", () => {
+ const logicalPlanSpy = vi.spyOn(ExecuteWorkflowService, "getLogicalPlanRequest").mockReturnValue({} as LogicalPlan);
+ const resetExecutionStateSpy = vi.spyOn(service, "resetExecutionState");
+ const resetStatusSpy = vi.spyOn(service["workflowStatusService"], "resetStatus");
+ const sendExecutionRequestSpy = vi.spyOn(service, "sendExecutionRequest").mockImplementation(() => {});
+ const replayInfo: ReplayExecutionInfo = { eid: 42, interaction: "step-3" };
+
+ service.executeWorkflowWithReplay(replayInfo);
+
+ expect(logicalPlanSpy).toHaveBeenCalledWith(service["workflowActionService"].getTexeraGraph());
+ expect(resetExecutionStateSpy).toHaveBeenCalled();
+ expect(resetStatusSpy).toHaveBeenCalled();
+ expect(sendExecutionRequestSpy).toHaveBeenCalledWith(
+ "Replay run of 42 to step-3",
+ expect.any(Object),
+ expect.any(Object),
+ false,
+ replayInfo
+ );
+ });
+
+ // ---- sendExecutionRequest branches ---------------------------------------------------------
+
+ it("sendExecutionRequest includes the selected computing unit id in the request", fakeAsync(() => {
+ vi.spyOn(service["computingUnitStatusService"], "getSelectedComputingUnitValue").mockReturnValue({
+ computingUnit: { cuid: 99 },
+ } as any);
+ const wsSendSpy = vi.spyOn(service["workflowWebsocketService"], "send");
+ const settings = service["workflowActionService"].getWorkflowSettings();
+
+ service.sendExecutionRequest("exec", {} as LogicalPlan, settings, true, undefined);
+ tick(FORM_DEBOUNCE_TIME_MS + 1);
+ flush();
+
+ expect(wsSendSpy).toHaveBeenCalledWith(
+ "WorkflowExecuteRequest",
+ expect.objectContaining({ computingUnitId: 99, emailNotificationEnabled: true, executionName: "exec" })
+ );
+ }));
+
+ it("sendExecutionRequest flags stored pagination info as belonging to a new execution", fakeAsync(() => {
+ sessionSetObject(PAGINATION_INFO_STORAGE_KEY, { newWorkflowExecuted: false });
+ const settings = service["workflowActionService"].getWorkflowSettings();
+
+ service.sendExecutionRequest("exec", {} as LogicalPlan, settings, false, undefined);
+ tick(FORM_DEBOUNCE_TIME_MS + 1);
+ flush();
+
+ const stored = sessionGetObject<ResultPaginationInfo>(PAGINATION_INFO_STORAGE_KEY);
+ expect(stored?.newWorkflowExecuted).toBe(true);
+ }));
+
+ // ---- control commands ----------------------------------------------------------------------
+
+ it("pauseWorkflow sends a pause request while running", () => {
+ (service as any).currentState = { state: ExecutionState.Running };
+ const wsSendSpy = vi.spyOn(service["workflowWebsocketService"], "send");
+ service.pauseWorkflow();
+ expect(wsSendSpy).toHaveBeenCalledWith("WorkflowPauseRequest", {});
+ });
+
+ it("killWorkflow throws in a terminal state and sends a kill request otherwise", () => {
+ (service as any).currentState = { state: ExecutionState.Uninitialized };
+ expect(() => service.killWorkflow()).toThrowError(
+ "cannot kill workflow, the current execution state is Uninitialized"
+ );
+
+ (service as any).currentState = { state: ExecutionState.Running };
+ const wsSendSpy = vi.spyOn(service["workflowWebsocketService"], "send");
+ service.killWorkflow();
+ expect(wsSendSpy).toHaveBeenCalledWith("WorkflowKillRequest", {});
+ });
+
+ it("takeGlobalCheckpoint throws when completed and sends a checkpoint request otherwise", () => {
+ (service as any).currentState = { state: ExecutionState.Completed };
+ expect(() => service.takeGlobalCheckpoint()).toThrowError(
+ "cannot take checkpoint, the current execution state is Completed"
+ );
+
+ (service as any).currentState = { state: ExecutionState.Running };
+ const wsSendSpy = vi.spyOn(service["workflowWebsocketService"], "send");
+ service.takeGlobalCheckpoint();
+ expect(wsSendSpy).toHaveBeenCalledWith("WorkflowCheckpointRequest", {});
+ });
+
+ it("resumeWorkflow sends a resume request while paused", () => {
+ (service as any).currentState = { state: ExecutionState.Paused };
+ const wsSendSpy = vi.spyOn(service["workflowWebsocketService"], "send");
+ service.resumeWorkflow();
+ expect(wsSendSpy).toHaveBeenCalledWith("WorkflowResumeRequest", {});
+ });
+
+ it("skipTuples requires a paused state and forwards the workers", () => {
+ (service as any).currentState = { state: ExecutionState.Running };
+ expect(() => service.skipTuples(["w1"])).toThrowError("cannot skip tuples, the current execution state is Running");
+
+ (service as any).currentState = { state: ExecutionState.Paused };
+ const wsSendSpy = vi.spyOn(service["workflowWebsocketService"], "send");
+ service.skipTuples(["w1", "w2"]);
+ expect(wsSendSpy).toHaveBeenCalledWith("SkipTupleRequest", { workers: ["w1", "w2"] });
+ });
+
+ it("retryExecution requires a paused state and forwards the workers", () => {
+ (service as any).currentState = { state: ExecutionState.Running };
+ expect(() => service.retryExecution(["w1"])).toThrowError(
+ "cannot retry the current tuple, the current execution state is Running"
+ );
+
+ (service as any).currentState = { state: ExecutionState.Paused };
+ const wsSendSpy = vi.spyOn(service["workflowWebsocketService"], "send");
+ service.retryExecution(["w9"]);
+ expect(wsSendSpy).toHaveBeenCalledWith("RetryRequest", { workers: ["w9"] });
+ });
+
+ it("modifyOperatorLogic requires a paused state and sends the operator logic", () => {
+ (service as any).currentState = { state: ExecutionState.Running };
+ expect(() => service.modifyOperatorLogic("1")).toThrowError(
+ "cannot modify logic, the current execution state is Running"
+ );
+
+ (service as any).currentState = { state: ExecutionState.Paused };
+ vi.spyOn(service["workflowActionService"].getTexeraGraph(), "getOperator").mockReturnValue(mockScanPredicate);
+ const wsSendSpy = vi.spyOn(service["workflowWebsocketService"], "send");
+ service.modifyOperatorLogic("1");
+ expect(wsSendSpy).toHaveBeenCalledWith("ModifyLogicRequest", {
+ operator: { operatorID: "1", operatorType: "ScanSource" },
+ });
+ });
});