PG1204 commented on code in PR #8301:
URL: https://github.com/apache/texera/pull/8301#discussion_r3919117431
##########
frontend/src/app/workspace/service/workflow-status/workflow-status.service.spec.ts:
##########
@@ -58,32 +63,77 @@ describe("WorkflowStatusService", () => {
service = TestBed.inject(WorkflowStatusService);
});
- it("forwards an OperatorStatisticsUpdateEvent to the status stream", () => {
- const received: Record<string, OperatorStatistics>[] = [];
- service.getStatusUpdateStream().subscribe(s => received.push(s));
+ it("splits an OperatorStatisticsUpdateEvent into the state and statistics
streams", () => {
+ const stateEmissions: Record<string, OperatorState>[] = [];
+ const statisticsEmissions: Record<string, OperatorStatistics>[] = [];
+ service.getStateUpdateStream().subscribe(s => stateEmissions.push(s));
+ service.getStatisticsUpdateStream().subscribe(s =>
statisticsEmissions.push(s));
- websocketEventSubject.next(statsEvent({ op1: sampleStats }));
+ websocketEventSubject.next(statsEvent({ op1: sampleRuntimeStatus }));
- expect(received).toHaveLength(1);
- expect(received[0]).toEqual({ op1: sampleStats });
- expect(service.getCurrentStatus()).toEqual({ op1: sampleStats });
+ expect(stateEmissions).toHaveLength(1);
+ expect(stateEmissions[0]).toEqual({ op1: OperatorState.Running });
+ expect(statisticsEmissions).toHaveLength(1);
+ expect(statisticsEmissions[0]).toEqual({ op1: sampleStatistics });
+ expect(service.getCurrentState()).toEqual({ op1: OperatorState.Running });
+ expect(service.getCurrentStatistics()).toEqual({ op1: sampleStatistics });
+ });
+
+ it("does not leak the operator state into the statistics concept", () => {
+ websocketEventSubject.next(statsEvent({ op1: sampleRuntimeStatus }));
+
expect(service.getCurrentStatistics()["op1"]).not.toHaveProperty("operatorState");
+ });
+
+ it("exposes state and statistics as independent streams", () => {
+ // A consumer subscribed to only one of the two sub-concepts sees exactly
+ // one emission per update, unaffected by the other stream.
+ let stateCount = 0;
+ let statisticsCount = 0;
+ service.getStateUpdateStream().subscribe(() => stateCount++);
+ service.getStatisticsUpdateStream().subscribe(() => statisticsCount++);
+
+ websocketEventSubject.next(statsEvent({ op1: sampleRuntimeStatus }));
+ websocketEventSubject.next(statsEvent({ op1: { ...sampleRuntimeStatus,
operatorState: OperatorState.Paused } }));
+
+ expect(stateCount).toBe(2);
+ expect(statisticsCount).toBe(2);
+ expect(service.getCurrentState()).toEqual({ op1: OperatorState.Paused });
+ });
+
+ it("emits state before statistics, so a statistics subscriber sees the
matching state snapshot", () => {
+ const order: string[] = [];
+ service.getStateUpdateStream().subscribe(() => order.push("state"));
+ service.getStatisticsUpdateStream().subscribe(() => {
+ order.push("statistics");
+ // The licensed read: by the time statistics arrive, the state snapshot
+ // already reflects the same wire event. The converse does not hold.
+ expect(service.getCurrentState()).toEqual({ op1: OperatorState.Running
});
+ });
+
+ websocketEventSubject.next(statsEvent({ op1: sampleRuntimeStatus }));
+
+ expect(order).toEqual(["state", "statistics"]);
Review Comment:
reworked to capture the snapshot in the subscriber and assert after next()
returns, with a comment noting why the assertion can't live in the callback.
Both the order and the snapshot guarantee now fail the test synchronously if
broken.
--
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]
