mengw15 commented on code in PR #6765:
URL: https://github.com/apache/texera/pull/6765#discussion_r3627720587
##########
frontend/src/app/common/service/computing-unit/computing-unit-status/computing-unit-status.service.spec.ts:
##########
@@ -223,4 +223,72 @@ describe("ComputingUnitStatusService", () => {
expect(listSpy).toHaveBeenCalled();
expect(latest).toEqual(newUnits);
});
+
+ it("updateUnitInList replaces the matching unit and leaves the others
untouched", () => {
+ const unitA = mockUnit(1);
+ const unitB = mockUnit(2);
+ (service as any).allComputingUnitsSubject.next([unitA, unitB]);
+
+ const updatedA = { computingUnit: { cuid: 1 }, status: "Running" } as
unknown as DashboardWorkflowComputingUnit;
+ (service as any).updateUnitInList(updatedA);
+
+ expect((service as any).allComputingUnitsSubject.value).toEqual([updatedA,
unitB]);
+ });
+
+ it("setComputingUnitsState refreshes the selected unit when it is still
present in the new list", () => {
+ (service as any).selectedUnitSubject.next(mockUnit(7));
+
+ const updated = { computingUnit: { cuid: 7 }, status: "Running" } as
unknown as DashboardWorkflowComputingUnit;
+ (service as any).setComputingUnitsState([updated]);
+
+ expect(service.getSelectedComputingUnitValue()).toBe(updated);
+ });
+
+ it("setComputingUnitsState clears the selection and stops polling when the
selected unit disappears", () => {
+ (service as any).selectedUnitSubject.next(mockUnit(7));
+ const stopSpy = vi.spyOn(service as any, "stopPollingSelectedUnit");
+
+ (service as any).setComputingUnitsState([mockUnit(8)]);
+
+ expect(service.getSelectedComputingUnitValue()).toBeNull();
+ expect(stopSpy).toHaveBeenCalled();
+ });
+
+ it("startPollingSelectedUnit polls the unit on each interval tick and merges
the result", () => {
+ vi.useFakeTimers();
+ try {
+ const managing = TestBed.inject(WorkflowComputingUnitManagingService);
+ const polled = { computingUnit: { cuid: 3 }, status: "Running" } as
unknown as DashboardWorkflowComputingUnit;
+ const getSpy = vi.spyOn(managing,
"getComputingUnit").mockReturnValue(of(polled));
+ (service as any).allComputingUnitsSubject.next([mockUnit(3)]);
+
+ (service as any).startPollingSelectedUnit(3);
+ // interval() fires only after the first period elapses
+ expect(getSpy).not.toHaveBeenCalled();
+
+ vi.advanceTimersByTime(2000);
+
+ expect(getSpy).toHaveBeenCalledWith(3);
+ expect((service as
any).allComputingUnitsSubject.value).toEqual([polled]);
+ } finally {
+ vi.useRealTimers();
+ }
+ });
Review Comment:
Done — the poll is now stopped in `finally` and the tick advances by
`REFRESH_INTERVAL_MS` instead of a hard-coded `2000`.
##########
frontend/src/app/workspace/service/workflow-websocket/workflow-websocket.service.spec.ts:
##########
@@ -102,4 +102,61 @@ describe("WorkflowWebsocketService", () => {
service.closeWebsocket();
expect(service.numWorkers).toBe(-1);
});
+
+ it("websocketEvent surfaces events pushed onto the response stream", () => {
+ const received: unknown[] = [];
+ const sub = service.websocketEvent().subscribe(event =>
received.push(event));
+
+ const event = { type: "WorkflowStateEvent", state: "RUNNING" };
+ (service as any).webSocketResponseSubject.next(event);
+ sub.unsubscribe();
+
+ expect(received).toEqual([event]);
+ });
+
+ it("getConnectionStatusStream reflects updateConnectionStatus transitions
and guards duplicates", () => {
+ const emissions: boolean[] = [];
+ const sub = service.getConnectionStatusStream().subscribe(value =>
emissions.push(value));
+
+ // BehaviorSubject seeds `false`; a repeated value is guarded and does not
re-emit.
+ (service as any).updateConnectionStatus(true);
+ (service as any).updateConnectionStatus(true);
+ (service as any).updateConnectionStatus(false);
+ sub.unsubscribe();
+
+ expect(emissions).toEqual([false, true, false]);
+ expect(service.isConnected).toBe(false);
+ });
+
+ it("openWebsocket routes an incoming socket message to websocketEvent and
marks the connection up", async () => {
+ const originalWebSocket = window.WebSocket;
+ const sockets: FakeWebSocket[] = [];
+ class CapturingWebSocket extends FakeWebSocket {
+ constructor(url: string) {
+ super(url);
+ sockets.push(this);
+ }
+ }
+ window.WebSocket = CapturingWebSocket as unknown as typeof WebSocket;
+
+ try {
+ const events: unknown[] = [];
+ service.websocketEvent().subscribe(event => events.push(event));
+ let connected: boolean | undefined;
+ service.getConnectionStatusStream().subscribe(value => (connected =
value));
+
+ service.openWebsocket(1, 1, 1);
+ await Promise.resolve(); // let the fake socket transition to OPEN
+
+ const socket = sockets[sockets.length - 1];
+ const event = { type: "WorkflowStateEvent", state: "RUNNING" };
+ socket.onmessage?.(new MessageEvent("message", { data:
JSON.stringify(event) }));
+
+ expect(events).toContainEqual(event);
+ expect(connected).toBe(true);
+ } finally {
+ service.closeWebsocket();
+ window.WebSocket = originalWebSocket;
+ }
Review Comment:
Done — both stream observers are captured and unsubscribed in `finally`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
