Copilot commented on code in PR #6765:
URL: https://github.com/apache/texera/pull/6765#discussion_r3627696791
##########
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:
In this async test, the subscriptions created with `subscribe(...)` are
never unsubscribed. `closeWebsocket()` does not complete
`webSocketResponseSubject`/`connectionStatusSubject`, so these observers can
leak past the test and retain references unnecessarily. Capture the returned
subscriptions and unsubscribe them in `finally`.
##########
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:
This test starts the private polling interval but never stops it. If the
polling Subscription survives `vi.useRealTimers()`, it can keep a live
interval/subscription around after the test and cause flaky cross-test
interference. Clean up by stopping polling in `finally`, and prefer referencing
the service’s `REFRESH_INTERVAL_MS` instead of hard-coding `2000` to keep the
test resilient to future interval changes.
--
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]
