From 3737b706757f9121e125a3bfe36759c2ad94c45c Mon Sep 17 00:00:00 2001 From: mcmahj45 Date: Thu, 11 Jun 2026 15:56:54 +0100 Subject: [PATCH 01/10] Websocket implementation 2.0 --- backend/src/main/kotlin/dev/dres/DRES.kt | 2 +- .../main/kotlin/dev/dres/api/rest/RestApi.kt | 3 +- .../main/kotlin/dev/dres/run/RunExecutor.kt | 241 +++++++++--------- .../src/app/services/websocket.service.ts | 119 +++++++++ .../src/app/viewer/run-viewer.component.ts | 28 +- 5 files changed, 269 insertions(+), 124 deletions(-) create mode 100644 frontend/src/app/services/websocket.service.ts diff --git a/backend/src/main/kotlin/dev/dres/DRES.kt b/backend/src/main/kotlin/dev/dres/DRES.kt index 783364c0..7592914a 100644 --- a/backend/src/main/kotlin/dev/dres/DRES.kt +++ b/backend/src/main/kotlin/dev/dres/DRES.kt @@ -117,7 +117,7 @@ object DRES { RunExecutor.init(store) /* Initialize EventStreamProcessor */ - EventStreamProcessor.register( /* Add handlers here */) + EventStreamProcessor.register(RunExecutor) EventStreamProcessor.init() /* Initialize Rest API. */ diff --git a/backend/src/main/kotlin/dev/dres/api/rest/RestApi.kt b/backend/src/main/kotlin/dev/dres/api/rest/RestApi.kt index 43bbb0b3..ca6c3e73 100644 --- a/backend/src/main/kotlin/dev/dres/api/rest/RestApi.kt +++ b/backend/src/main/kotlin/dev/dres/api/rest/RestApi.kt @@ -30,6 +30,7 @@ import dev.dres.api.rest.types.status.ErrorStatus import dev.dres.api.rest.types.users.ApiRole import dev.dres.data.model.config.Config import dev.dres.mgmt.cache.CacheManager +import dev.dres.run.RunExecutor import dev.dres.utilities.NamedThreadFactory import io.javalin.Javalin import io.javalin.apibuilder.ApiBuilder.* @@ -328,7 +329,7 @@ object RestApi { } } } - //ws("ws/run", runExecutor) + ws("ws/run", RunExecutor::accept) } } diff --git a/backend/src/main/kotlin/dev/dres/run/RunExecutor.kt b/backend/src/main/kotlin/dev/dres/run/RunExecutor.kt index 2dc5b44c..7fa8d79f 100644 --- a/backend/src/main/kotlin/dev/dres/run/RunExecutor.kt +++ b/backend/src/main/kotlin/dev/dres/run/RunExecutor.kt @@ -1,15 +1,22 @@ package dev.dres.run - +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper import dev.dres.api.rest.types.ViewerInfo +import dev.dres.api.rest.types.evaluation.websocket.ClientMessage +import dev.dres.api.rest.types.evaluation.websocket.ClientMessageType +import dev.dres.api.rest.types.evaluation.websocket.ServerMessage +import dev.dres.api.rest.types.evaluation.websocket.ServerMessageType import dev.dres.data.model.run.* import dev.dres.data.model.run.interfaces.EvaluationId +import dev.dres.run.eventstream.* import dev.dres.run.validation.interfaces.JudgementValidator +import io.javalin.websocket.WsConfig import io.javalin.websocket.WsContext import jetbrains.exodus.database.TransientEntityStore import kotlinx.dnq.query.* import org.slf4j.LoggerFactory import java.util.* +import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.Executors import java.util.concurrent.Future import java.util.concurrent.locks.ReentrantReadWriteLock @@ -17,14 +24,15 @@ import kotlin.concurrent.read import kotlin.concurrent.write /** - * The execution environment for [RunManager]s + * The execution environment for [RunManager]s. * * @author Ralph Gasser - * @version 1.3.0 + * @version 1.4.0 */ -object RunExecutor { +object RunExecutor : StreamEventHandler { private val logger = LoggerFactory.getLogger(this.javaClass) + private val mapper = jacksonObjectMapper() /** Thread Pool Executor which is used to execute the [RunManager]s. */ private val executor = Executors.newCachedThreadPool() @@ -35,22 +43,23 @@ object RunExecutor { /** List of [JudgementValidator]s registered with this [RunExecutor]. */ private val judgementValidators = LinkedList() - /** List of [WsContext] that are currently connected. */ - private val connectedClients = HashMap() + /** WebSocket sessions currently connected, keyed by session ID. */ + private val connectedClients = ConcurrentHashMap() - /** List of session IDs that are currently observing an evaluation. */ - private val observingClients = HashMap>() + /** Session IDs currently observing each evaluation. */ + private val observingClients = HashMap>() -// /** Lock for accessing and changing all data structures related to WebSocket clients. */ -// private val clientLock = StampedLock() + /** Lock for WebSocket client data structures. */ + private val clientLock = ReentrantReadWriteLock() /** Lock for accessing and changing all data structures related to [RunManager]s. */ private val runManagerLock = ReentrantReadWriteLock() - /** Internal array of [Future]s for cleaning after [RunManager]s. See [RunExecutor.cleanerThread]*/ + /** Internal array of [Future]s for cleaning after [RunManager]s. */ private val results = HashMap, EvaluationId>() - /** Initializes the [RunExecutor] singleton. + /** + * Initializes the [RunExecutor] singleton. * * @param store The [TransientEntityStore] instance used to access persistent data. */ @@ -58,7 +67,7 @@ object RunExecutor { store.transactional { DbEvaluation.filter { (it.ended eq null) }.asSequence().forEach { evaluation -> try { - this.schedule(evaluation.toRunManager(store)) /* Re-schedule evaluations. */ + this.schedule(evaluation.toRunManager(store)) } catch (e: RuntimeException) { when (e) { is IllegalStateException, @@ -66,7 +75,6 @@ object RunExecutor { logger.error("Could not re-schedule previous run: ${e.message}") evaluation.ended = System.currentTimeMillis() } - else -> { logger.error("Fatal error during re-scheduling of previous run (${evaluation.evaluationId}): ${e.message}") throw e @@ -86,31 +94,23 @@ object RunExecutor { if (this.runManagers.containsKey(manager.id)) { throw IllegalArgumentException("This RunExecutor already runs a RunManager with the given ID ${manager.id}. The same RunManager cannot be executed twice!") } - - /* Setup all the required data structures. */ this.runManagers[manager.id] = manager this.observingClients[manager.id] = HashSet() - this.results[this.executor.submit(manager)] = manager.id /* Register future for cleanup thread. */ + this.results[this.executor.submit(manager)] = manager.id } - /** A thread that cleans after [RunManager] have finished. */ + /** A thread that cleans after [RunManager]s have finished. */ private val cleanerThread = Thread { while (!this@RunExecutor.executor.isShutdown) { -// var stamp = this@RunExecutor.runManagerLock.readLock() this@RunExecutor.runManagerLock.read { - //try { this@RunExecutor.results.entries.removeIf { entry -> val k = entry.key val v = entry.value if (k.isDone || k.isCancelled) { logger.info("RunManager $v (done = ${k.isDone}, cancelled = ${k.isCancelled}) will be removed!") -// stamp = this@RunExecutor.runManagerLock.tryConvertToWriteLock(stamp) -// if (stamp > -1L) { this@RunExecutor.runManagerLock.write { - /* Cleanup. */ this@RunExecutor.runManagers.remove(v) this@RunExecutor.observingClients.remove(v) - } true } else { @@ -118,9 +118,6 @@ object RunExecutor { } } } -// } finally { -// this@RunExecutor.runManagerLock.unlock(stamp) -// } Thread.sleep(500) } } @@ -132,60 +129,107 @@ object RunExecutor { this.cleanerThread.start() } -// /** -// * Callback for when registering this [RunExecutor] as handler for Javalin's WebSocket. -// * -// * @param t The [WsConfig] of the WebSocket endpoint. -// */ -// fun accept(t: WsConfig) { //TODO remove -// t.onConnect { -// /* Add WSContext to set of connected clients. */ -// this@RunExecutor.clientLock.write { -// val connection = WebSocketConnection(it) -// this.connectedClients[connection.httpSessionId] = connection -// } -// } -// t.onClose { -// val session = WebSocketConnection(it) -// this@RunExecutor.clientLock.write { -// val connection = WebSocketConnection(it) -// this.connectedClients.remove(connection.httpSessionId) -// this.runManagerLock.read { -// for (m in this.runManagers) { -// if (this.observingClients[m.key]?.remove(connection) == true) { -// m.value.wsMessageReceived(session, ClientMessage(m.key, ClientMessageType.UNREGISTER)) /* Send implicit unregister message associated with a disconnect. */ -// } -// } -// } -// } -// } -// t.onMessage { -// val message = try { -// it.messageAsClass() -// } catch (e: Exception) { -// logger.warn("Cannot parse WebSocket message: ${e.localizedMessage}") -// return@onMessage -// } -// val session = WebSocketConnection(it) -// logger.debug("Received WebSocket message: $message from ${it.session.policy}") -// this.runManagerLock.read { -// if (this.runManagers.containsKey(message.evaluationId)) { -// when (message.type) { -// ClientMessageType.ACK -> {} -// ClientMessageType.REGISTER -> this@RunExecutor.clientLock.write { this.observingClients[message.evaluationId]?.add(WebSocketConnection(it)) } -// ClientMessageType.UNREGISTER -> this@RunExecutor.clientLock.write { this.observingClients[message.evaluationId]?.remove(WebSocketConnection(it)) } -// ClientMessageType.PING -> it.send(ServerMessage(message.evaluationId, ServerMessageType.PING)) -// } -// this.runManagers[message.evaluationId]!!.wsMessageReceived(session, message) /* Forward message to RunManager. */ -// } -// } -// } -// } + /** + * Callback for registering this [RunExecutor] as Javalin's WebSocket handler. + * + * @param ws The [WsConfig] to configure. + */ + fun accept(ws: WsConfig) { + ws.onConnect { ctx -> + this.clientLock.write { + this.connectedClients[ctx.sessionId()] = ctx + } + logger.debug("WebSocket client connected: ${ctx.sessionId()}") + } + + ws.onClose { ctx -> + this.clientLock.write { + this.connectedClients.remove(ctx.sessionId()) + this.runManagerLock.read { + this.observingClients.values.forEach { it.remove(ctx.sessionId()) } + } + } + logger.debug("WebSocket client disconnected: ${ctx.sessionId()}") + } + + ws.onMessage { ctx -> + val message = try { + ctx.messageAsClass() + } catch (e: Exception) { + logger.warn("Cannot parse WebSocket message: ${e.localizedMessage}") + return@onMessage + } + logger.debug("Received WebSocket message: $message from ${ctx.sessionId()}") + this.runManagerLock.read { + if (this.runManagers.containsKey(message.evaluationId)) { + when (message.type) { + ClientMessageType.ACK -> {} + ClientMessageType.REGISTER -> this.clientLock.write { + this.observingClients[message.evaluationId]?.add(ctx.sessionId()) + } + ClientMessageType.UNREGISTER -> this.clientLock.write { + this.observingClients[message.evaluationId]?.remove(ctx.sessionId()) + } + ClientMessageType.PING -> ctx.send( + mapper.writeValueAsString(ServerMessage(message.evaluationId, ServerMessageType.PING)) + ) + } + } + } + } + + ws.onError { ctx -> + logger.error("WebSocket error for session ${ctx.sessionId()}: ${ctx.error()?.message}") + this.clientLock.write { + this.connectedClients.remove(ctx.sessionId()) + } + } + } /** - * Lists all [RunManager]s currently executed by this [RunExecutor] + * Broadcasts a [ServerMessage] to all clients observing the specified evaluation. * - * @return Immutable list of all [RunManager]s currently executed. + * @param message The [ServerMessage] to broadcast. + */ + fun broadcastWsMessage(message: ServerMessage) { + val json = try { + mapper.writeValueAsString(message) + } catch (e: Exception) { + logger.error("Failed to serialize ServerMessage: ${e.message}") + return + } + val observers = this.clientLock.read { + this.runManagerLock.read { + this.observingClients[message.evaluationId]?.toSet() ?: emptySet() + } + } + observers.forEach { sessionId -> + try { + this.connectedClients[sessionId]?.send(json) + } catch (e: Exception) { + logger.warn("Failed to send WebSocket message to session $sessionId: ${e.message}") + } + } + } + + /** + * Handles [StreamEvent]s from the [EventStreamProcessor] by broadcasting + * the appropriate [ServerMessage] to all registered observers. + */ + override fun handleStreamEvent(event: StreamEvent) { + when (event) { + is RunStartEvent -> broadcastWsMessage(ServerMessage(event.runId, ServerMessageType.COMPETITION_START)) + is RunEndEvent -> broadcastWsMessage(ServerMessage(event.runId, ServerMessageType.COMPETITION_END)) + is TaskStartEvent -> broadcastWsMessage(ServerMessage(event.runId, ServerMessageType.TASK_START, event.taskId)) + is TaskEndEvent -> broadcastWsMessage(ServerMessage(event.runId, ServerMessageType.TASK_END, event.taskId)) + is ScoreUpdateEvent -> broadcastWsMessage(ServerMessage(event.runId, ServerMessageType.COMPETITION_UPDATE)) + is SubmissionEvent -> broadcastWsMessage(ServerMessage(event.runId, ServerMessageType.TASK_UPDATED)) + else -> {} + } + } + + /** + * Lists all [RunManager]s currently executed by this [RunExecutor]. */ fun managers(): List = this.runManagerLock.read { return this.runManagers.values.toList() @@ -193,54 +237,15 @@ object RunExecutor { /** * Returns the [RunManager] for the given ID if such a [RunManager] exists. - * - * @param evaluationId The ID for which to return the [RunManager]. - * @return Optional [RunManager]. */ fun managerForId(evaluationId: EvaluationId): RunManager? = this.runManagerLock.read { return this.runManagers[evaluationId] } -// /** -// * Broadcasts a [ServerMessage] to all clients currently connected and observing a specific [RunManager]. -// * -// * @param message The [ServerMessage] that should be broadcast. -// */ -// fun broadcastWsMessage(message: ServerMessage) = this.clientLock.read { -// this.runManagerLock.read { -// this.connectedClients.values.filter { -// this.observingClients[message.evaluationId]?.contains(it) ?: false -// }.forEach { -// it.send(message) -// } -// } -// } - -// /** -// * Broadcasts a [ServerMessage] to all clients currently connected and observing a specific [RunManager] and are member of the specified team. -// * -// * @param teamId The [TeamId] of the relevant team -// * @param message The [ServerMessage] that should be broadcast. -// */ -// fun broadcastWsMessage(teamId: TeamId, message: ServerMessage) = this.clientLock.read { -// val manager = managerForId(message.evaluationId) -// if (manager != null) { -// val teamMembers = manager.template.teams.filter { it.id eq teamId }.flatMapDistinct { it.users }.asSequence().map { it.userId }.toList() -// this.runManagerLock.read { -// this.connectedClients.values.filter { -// this.observingClients[message.evaluationId]?.contains(it) ?: false && AccessManager.userIdForSession(it.sessionId) in teamMembers -// }.forEach { -// it.send(message) -// } -// } -// } -// } - /** - * Stops all runs + * Stops all runs. */ fun stop() { this.executor.shutdownNow() } - } diff --git a/frontend/src/app/services/websocket.service.ts b/frontend/src/app/services/websocket.service.ts new file mode 100644 index 00000000..0b9798fa --- /dev/null +++ b/frontend/src/app/services/websocket.service.ts @@ -0,0 +1,119 @@ +import { Injectable, OnDestroy } from '@angular/core'; +import { Observable, Subject, Subscription, interval } from 'rxjs'; +import { AppConfig } from '../app.config'; +import { IWsServerMessage } from '../model/ws/ws-server-message.interface'; +import { IWsClientMessage } from '../model/ws/ws-client-message.interface'; +import { ClientMessageType } from '../model/ws/client-message-type.enum'; + +@Injectable({ + providedIn: 'root', +}) +export class WebSocketService implements OnDestroy { + private socket: WebSocket | null = null; + private messageSubject = new Subject(); + private pingSubscription: Subscription | null = null; + private reconnectTimeout: ReturnType | null = null; + private currentEvaluationId: string | null = null; + + /** Observable stream of messages pushed by the server. */ + readonly messages$: Observable = this.messageSubject.asObservable(); + + constructor(private config: AppConfig) {} + + /** + * Connects to the WebSocket endpoint and registers for the given evaluation. + * Safe to call multiple times — no-ops if already connected to the same evaluation. + */ + connect(evaluationId: string): void { + if (this.socket?.readyState === WebSocket.OPEN && this.currentEvaluationId === evaluationId) { + return; + } + this.closeSocket(); + this.currentEvaluationId = evaluationId; + this.openConnection(); + } + + /** Disconnects from the WebSocket endpoint and stops reconnect attempts. */ + disconnect(): void { + this.currentEvaluationId = null; + this.closeSocket(); + } + + private openConnection(): void { + const url = this.config.webSocketUrl; + this.socket = new WebSocket(url); + + this.socket.onopen = () => { + this.send({ + evaluationId: this.currentEvaluationId, + type: ClientMessageType.ClientMessageTypeEnum.REGISTER, + }); + this.startPing(); + }; + + this.socket.onmessage = (event: MessageEvent) => { + try { + const message: IWsServerMessage = JSON.parse(event.data as string); + this.messageSubject.next(message); + } catch { + console.warn('[WebSocketService] Could not parse message:', event.data); + } + }; + + this.socket.onclose = () => { + this.stopPing(); + if (this.currentEvaluationId) { + this.reconnectTimeout = setTimeout(() => this.openConnection(), 5000); + } + }; + + this.socket.onerror = (error: Event) => { + console.error('[WebSocketService] WebSocket error:', error); + }; + } + + private send(message: IWsClientMessage): void { + if (this.socket?.readyState === WebSocket.OPEN) { + this.socket.send(JSON.stringify(message)); + } + } + + private closeSocket(): void { + if (this.reconnectTimeout !== null) { + clearTimeout(this.reconnectTimeout); + this.reconnectTimeout = null; + } + this.stopPing(); + if (this.socket) { + if (this.socket.readyState === WebSocket.OPEN) { + this.send({ + evaluationId: this.currentEvaluationId, + type: ClientMessageType.ClientMessageTypeEnum.UNREGISTER, + }); + this.socket.close(); + } + this.socket = null; + } + } + + private startPing(): void { + this.pingSubscription = interval(30_000).subscribe(() => { + if (this.currentEvaluationId) { + this.send({ + evaluationId: this.currentEvaluationId, + type: ClientMessageType.ClientMessageTypeEnum.PING, + }); + } + }); + } + + private stopPing(): void { + this.pingSubscription?.unsubscribe(); + this.pingSubscription = null; + } + + ngOnDestroy(): void { + this.disconnect(); + this.messageSubject.complete(); + } +} diff --git a/frontend/src/app/viewer/run-viewer.component.ts b/frontend/src/app/viewer/run-viewer.component.ts index 4c7376eb..905ecf60 100644 --- a/frontend/src/app/viewer/run-viewer.component.ts +++ b/frontend/src/app/viewer/run-viewer.component.ts @@ -1,15 +1,19 @@ import {AfterViewInit, Component, Inject, OnDestroy, OnInit, ViewContainerRef, DOCUMENT} from '@angular/core'; import { ActivatedRoute, ActivationEnd, Params, Router } from "@angular/router"; -import {interval, merge, mergeMap, Observable, of, zip} from 'rxjs'; +import {merge, Observable, of, zip} from 'rxjs'; import { catchError, filter, map, pairwise, shareReplay, - switchMap, tap + switchMap, + take, + tap } from "rxjs/operators"; import { AppConfig } from '../app.config'; +import { WebSocketService } from '../services/websocket.service'; +import { ServerMessageType } from '../model/ws/server-message-type.enum'; import { MatSnackBar } from '@angular/material/snack-bar'; import { Position } from './model/run-viewer-position'; import { Widget } from './model/run-viewer-widgets'; @@ -92,6 +96,7 @@ export class RunViewerComponent implements OnInit, AfterViewInit, OnDestroy { private snackBar: MatSnackBar, private titleService: Title, private overlay: Overlay, + private wsService: WebSocketService, @Inject(DOCUMENT) private document: Document, private _viewContainerRef: ViewContainerRef ) { @@ -158,7 +163,21 @@ export class RunViewerComponent implements OnInit, AfterViewInit, OnDestroy { shareReplay({ bufferSize: 1, refCount: true }) ); - this.state = interval(1000).pipe(mergeMap(() => this.evaluationId)).pipe( + /* Trigger a state fetch on route change or on any relevant WebSocket event. */ + const wsRefresh$ = this.wsService.messages$.pipe( + filter((msg) => [ + ServerMessageType.ServerMessageTypeEnum.TASK_START, + ServerMessageType.ServerMessageTypeEnum.TASK_END, + ServerMessageType.ServerMessageTypeEnum.TASK_PREPARE, + ServerMessageType.ServerMessageTypeEnum.TASK_UPDATED, + ServerMessageType.ServerMessageTypeEnum.COMPETITION_START, + ServerMessageType.ServerMessageTypeEnum.COMPETITION_UPDATE, + ServerMessageType.ServerMessageTypeEnum.COMPETITION_END, + ].includes(msg.type)), + switchMap(() => this.evaluationId.pipe(take(1))) + ); + + this.state = merge(this.evaluationId, wsRefresh$).pipe( switchMap((id) => this.runService.getApiV2EvaluationByEvaluationIdState(id)), catchError((err, o) => { console.log( @@ -209,7 +228,7 @@ export class RunViewerComponent implements OnInit, AfterViewInit, OnDestroy { * Registers this RunViewerComponent on view initialization and creates the WebSocket subscription. */ ngOnInit(): void { - + this.evaluationId.subscribe((id) => this.wsService.connect(id)); } /** @@ -224,6 +243,7 @@ export class RunViewerComponent implements OnInit, AfterViewInit, OnDestroy { * Unregisters this RunViewerComponent on view destruction and cleans the WebSocket subscription. */ ngOnDestroy(): void { + this.wsService.disconnect(); this.titleService.setTitle('DRES'); } From 3ac4c9617b2254b1878f43325bef6d39584a0522 Mon Sep 17 00:00:00 2001 From: mcmahj45 Date: Thu, 11 Jun 2026 16:36:42 +0100 Subject: [PATCH 02/10] Unit Testing for websockets --- .../app/services/websocket.service.spec.ts | 220 ++++++++++++++++++ frontend/tsconfig.spec.json | 3 +- 2 files changed, 222 insertions(+), 1 deletion(-) create mode 100644 frontend/src/app/services/websocket.service.spec.ts diff --git a/frontend/src/app/services/websocket.service.spec.ts b/frontend/src/app/services/websocket.service.spec.ts new file mode 100644 index 00000000..bf7c7b90 --- /dev/null +++ b/frontend/src/app/services/websocket.service.spec.ts @@ -0,0 +1,220 @@ +import { TestBed, fakeAsync, tick } from '@angular/core/testing'; +import { WebSocketService } from './websocket.service'; +import { AppConfig } from '../app.config'; +import { ClientMessageType } from '../model/ws/client-message-type.enum'; +import { ServerMessageType } from '../model/ws/server-message-type.enum'; +import { IWsServerMessage } from '../model/ws/ws-server-message.interface'; + +// ── Minimal WebSocket mock ──────────────────────────────────────────────────── + +class MockWebSocket { + static instance: MockWebSocket | null = null; + + readyState: number = WebSocket.CONNECTING; + sentMessages: string[] = []; + + onopen: ((e: Event) => void) | null = null; + onmessage: ((e: MessageEvent) => void) | null = null; + onclose: ((e: CloseEvent) => void) | null = null; + onerror: ((e: Event) => void) | null = null; + + constructor(public url: string) { + MockWebSocket.instance = this; + } + + send(data: string) { this.sentMessages.push(data); } + close() { this.readyState = WebSocket.CLOSED; this.onclose?.(new CloseEvent('close')); } + + /** Test helper — simulate a successful connection. */ + simulateOpen() { + this.readyState = WebSocket.OPEN; + this.onopen?.(new Event('open')); + } + + /** Test helper — simulate an incoming server message. */ + simulateMessage(msg: IWsServerMessage) { + this.onmessage?.(new MessageEvent('message', { data: JSON.stringify(msg) })); + } +} + +// ── Helpers ─────────────────────────────────────────────────────────────────── + +const WS_URL = 'ws://localhost:8080/api/ws/run'; + +function makeConfig(): Partial { + return { webSocketUrl: WS_URL } as Partial; +} + +function serverMsg(type: ServerMessageType.ServerMessageTypeEnum, evaluationId = 'eval-1'): IWsServerMessage { + return { evaluationId, type, timestamp: Date.now() }; +} + +// ── Tests ───────────────────────────────────────────────────────────────────── + +describe('WebSocketService', () => { + let service: WebSocketService; + + beforeEach(() => { + MockWebSocket.instance = null; + (window as any).WebSocket = MockWebSocket; + + TestBed.configureTestingModule({ + providers: [ + WebSocketService, + { provide: AppConfig, useValue: makeConfig() }, + ], + }); + + service = TestBed.inject(WebSocketService); + }); + + afterEach(() => { + service.disconnect(); + }); + + // ── connect() ────────────────────────────────────────────────────────────── + + describe('connect', () => { + it('opens a WebSocket to the configured URL', () => { + service.connect('eval-1'); + expect(MockWebSocket.instance).not.toBeNull(); + expect(MockWebSocket.instance!.url).toBe(WS_URL); + }); + + it('sends REGISTER message on open', () => { + service.connect('eval-1'); + MockWebSocket.instance!.simulateOpen(); + + const sent = JSON.parse(MockWebSocket.instance!.sentMessages[0]); + expect(sent.type).toBe(ClientMessageType.ClientMessageTypeEnum.REGISTER); + expect(sent.evaluationId).toBe('eval-1'); + }); + + it('is a no-op when already connected to the same evaluation', () => { + service.connect('eval-1'); + MockWebSocket.instance!.simulateOpen(); + const first = MockWebSocket.instance; + + service.connect('eval-1'); + + expect(MockWebSocket.instance).toBe(first); + }); + + it('reconnects when called with a different evaluationId', () => { + service.connect('eval-1'); + MockWebSocket.instance!.simulateOpen(); + const first = MockWebSocket.instance; + + service.connect('eval-2'); + + expect(MockWebSocket.instance).not.toBe(first); + }); + }); + + // ── messages$ ────────────────────────────────────────────────────────────── + + describe('messages$', () => { + it('emits parsed server messages', () => { + const received: IWsServerMessage[] = []; + service.messages$.subscribe((m) => received.push(m)); + + service.connect('eval-1'); + MockWebSocket.instance!.simulateOpen(); + MockWebSocket.instance!.simulateMessage(serverMsg(ServerMessageType.ServerMessageTypeEnum.TASK_START)); + + expect(received.length).toBe(1); + expect(received[0].type).toBe(ServerMessageType.ServerMessageTypeEnum.TASK_START); + }); + + it('emits multiple consecutive messages', () => { + const received: IWsServerMessage[] = []; + service.messages$.subscribe((m) => received.push(m)); + + service.connect('eval-1'); + MockWebSocket.instance!.simulateOpen(); + MockWebSocket.instance!.simulateMessage(serverMsg(ServerMessageType.ServerMessageTypeEnum.TASK_START)); + MockWebSocket.instance!.simulateMessage(serverMsg(ServerMessageType.ServerMessageTypeEnum.TASK_END)); + + expect(received.length).toBe(2); + }); + + it('silently ignores malformed JSON', () => { + const received: IWsServerMessage[] = []; + service.messages$.subscribe((m) => received.push(m)); + + service.connect('eval-1'); + MockWebSocket.instance!.simulateOpen(); + MockWebSocket.instance!.onmessage?.(new MessageEvent('message', { data: 'not-json' })); + + expect(received.length).toBe(0); + }); + }); + + // ── disconnect() ─────────────────────────────────────────────────────────── + + describe('disconnect', () => { + it('sends UNREGISTER before closing', () => { + service.connect('eval-1'); + MockWebSocket.instance!.simulateOpen(); + service.disconnect(); + + const msgs = MockWebSocket.instance!.sentMessages.map((s) => JSON.parse(s)); + const unregister = msgs.find((m) => m.type === ClientMessageType.ClientMessageTypeEnum.UNREGISTER); + expect(unregister).toBeDefined(); + }); + + it('sets socket to null after disconnect', () => { + service.connect('eval-1'); + MockWebSocket.instance!.simulateOpen(); + service.disconnect(); + + expect((service as any).socket).toBeNull(); + }); + + it('does not attempt reconnect after explicit disconnect', fakeAsync(() => { + service.connect('eval-1'); + MockWebSocket.instance!.simulateOpen(); + service.disconnect(); + + const socketAfterDisconnect = MockWebSocket.instance; + tick(6000); + + expect(MockWebSocket.instance).toBe(socketAfterDisconnect); + })); + }); + + // ── auto-reconnect ───────────────────────────────────────────────────────── + + describe('auto-reconnect', () => { + it('reconnects after 5 seconds when connection drops unexpectedly', fakeAsync(() => { + service.connect('eval-1'); + MockWebSocket.instance!.simulateOpen(); + const first = MockWebSocket.instance; + + // Simulate unexpected server-side close + MockWebSocket.instance!.readyState = WebSocket.CLOSED; + MockWebSocket.instance!.onclose?.(new CloseEvent('close')); + + tick(5000); + + expect(MockWebSocket.instance).not.toBe(first); + expect(MockWebSocket.instance!.url).toBe(WS_URL); + })); + }); + + // ── ping ─────────────────────────────────────────────────────────────────── + + describe('ping', () => { + it('sends a PING message every 30 seconds', fakeAsync(() => { + service.connect('eval-1'); + MockWebSocket.instance!.simulateOpen(); + + tick(30_000); + + const msgs = MockWebSocket.instance!.sentMessages.map((s) => JSON.parse(s)); + const ping = msgs.find((m) => m.type === ClientMessageType.ClientMessageTypeEnum.PING); + expect(ping).toBeDefined(); + expect(ping.evaluationId).toBe('eval-1'); + })); + }); +}); diff --git a/frontend/tsconfig.spec.json b/frontend/tsconfig.spec.json index 430cf757..370d7efb 100644 --- a/frontend/tsconfig.spec.json +++ b/frontend/tsconfig.spec.json @@ -2,7 +2,8 @@ "extends": "./tsconfig.json", "compilerOptions": { "outDir": "./out-tsc/spec", - "types": ["jasmine", "node"] + "types": ["jasmine", "node"], + "skipLibCheck": true }, "files": ["src/test.ts", "src/polyfills.ts"], "include": ["src/**/*.spec.ts", "src/**/*.d.ts"] From e87e1ee3a0a6e7024eed45b6902080b033ff1cb9 Mon Sep 17 00:00:00 2001 From: mcmahj45 Date: Thu, 11 Jun 2026 16:45:33 +0100 Subject: [PATCH 03/10] More testing --- .../app/viewer/run-viewer.component.spec.ts | 138 ++++++++++++++++++ 1 file changed, 138 insertions(+) create mode 100644 frontend/src/app/viewer/run-viewer.component.spec.ts diff --git a/frontend/src/app/viewer/run-viewer.component.spec.ts b/frontend/src/app/viewer/run-viewer.component.spec.ts new file mode 100644 index 00000000..c70fbe9d --- /dev/null +++ b/frontend/src/app/viewer/run-viewer.component.spec.ts @@ -0,0 +1,138 @@ +import { NO_ERRORS_SCHEMA } from '@angular/core'; +import { TestBed, fakeAsync, tick } from '@angular/core/testing'; +import { ActivatedRoute, Router } from '@angular/router'; +import { MatSnackBar } from '@angular/material/snack-bar'; +import { Title } from '@angular/platform-browser'; +import { Overlay } from '@angular/cdk/overlay'; +import { of, Subject } from 'rxjs'; + +import { RunViewerComponent } from './run-viewer.component'; +import { AppConfig } from '../app.config'; +import { WebSocketService } from '../services/websocket.service'; +import { EvaluationService } from '../../../openapi'; +import { IWsServerMessage } from '../model/ws/ws-server-message.interface'; +import { ServerMessageType } from '../model/ws/server-message-type.enum'; + +// ── Helpers ─────────────────────────────────────────────────────────────────── + +function wsMsg(type: ServerMessageType.ServerMessageTypeEnum): IWsServerMessage { + return { evaluationId: 'eval-1', type, timestamp: Date.now() }; +} + +const IRRELEVANT_TYPES: ServerMessageType.ServerMessageTypeEnum[] = [ + ServerMessageType.ServerMessageTypeEnum.PING, +]; + +const REFRESH_TYPES: ServerMessageType.ServerMessageTypeEnum[] = [ + ServerMessageType.ServerMessageTypeEnum.TASK_START, + ServerMessageType.ServerMessageTypeEnum.TASK_END, + ServerMessageType.ServerMessageTypeEnum.TASK_PREPARE, + ServerMessageType.ServerMessageTypeEnum.TASK_UPDATED, + ServerMessageType.ServerMessageTypeEnum.COMPETITION_START, + ServerMessageType.ServerMessageTypeEnum.COMPETITION_UPDATE, + ServerMessageType.ServerMessageTypeEnum.COMPETITION_END, +]; + +// ── Suite ───────────────────────────────────────────────────────────────────── + +describe('RunViewerComponent WebSocket wiring', () => { + let component: RunViewerComponent; + let wsService: jasmine.SpyObj; + let runService: jasmine.SpyObj; + let messages$: Subject; + + beforeEach(() => { + messages$ = new Subject(); + + wsService = jasmine.createSpyObj('WebSocketService', ['connect', 'disconnect'], { + messages$: messages$.asObservable(), + }); + + runService = jasmine.createSpyObj('EvaluationService', [ + 'getApiV2EvaluationByEvaluationIdState', + 'getApiV2EvaluationByEvaluationIdInfo', + ]); + runService.getApiV2EvaluationByEvaluationIdState.and.returnValue( + of({ taskStatus: 'RUNNING', taskTemplateId: 't1' } as any) + ); + runService.getApiV2EvaluationByEvaluationIdInfo.and.returnValue( + of({ name: 'Test Eval' } as any) + ); + + TestBed.configureTestingModule({ + declarations: [RunViewerComponent], + schemas: [NO_ERRORS_SCHEMA], + providers: [ + { provide: WebSocketService, useValue: wsService }, + { provide: EvaluationService, useValue: runService }, + { provide: ActivatedRoute, useValue: { params: of({ runId: 'eval-1' }), paramMap: of({ params: { runId: 'eval-1' }, get: () => 'eval-1' }) } }, + { provide: Router, useValue: jasmine.createSpyObj('Router', ['navigate'], { url: '/viewer/eval-1' }) }, + { provide: AppConfig, useValue: { webSocketUrl: 'ws://localhost:8080/api/ws/run' } }, + { provide: MatSnackBar, useValue: jasmine.createSpyObj('MatSnackBar', ['open']) }, + { provide: Title, useValue: jasmine.createSpyObj('Title', ['setTitle']) }, + { provide: Overlay, useValue: {} }, + { provide: 'DOCUMENT', useValue: document }, + ], + }); + + const fixture = TestBed.createComponent(RunViewerComponent); + component = fixture.componentInstance; + }); + + // ── connect / disconnect ─────────────────────────────────────────────────── + + it('connects the WebSocket with the evaluationId on init', () => { + component.ngOnInit(); + expect(wsService.connect).toHaveBeenCalledWith('eval-1'); + }); + + it('disconnects the WebSocket on destroy', () => { + component.ngOnInit(); + component.ngOnDestroy(); + expect(wsService.disconnect).toHaveBeenCalled(); + }); + + // ── state refresh on WS message ──────────────────────────────────────────── + + REFRESH_TYPES.forEach((type) => { + it(`triggers a state fetch when ${type} message is received`, fakeAsync(() => { + component.ngOnInit(); + runService.getApiV2EvaluationByEvaluationIdState.calls.reset(); + + // Subscribe to state so the observable is active + component.state.subscribe(); + + messages$.next(wsMsg(type)); + tick(); + + expect(runService.getApiV2EvaluationByEvaluationIdState).toHaveBeenCalledWith('eval-1'); + })); + }); + + // ── PING is ignored ──────────────────────────────────────────────────────── + + IRRELEVANT_TYPES.forEach((type) => { + it(`does not trigger a state fetch for ${type} message`, fakeAsync(() => { + component.ngOnInit(); + component.state.subscribe(); + runService.getApiV2EvaluationByEvaluationIdState.calls.reset(); + + messages$.next(wsMsg(type)); + tick(); + + expect(runService.getApiV2EvaluationByEvaluationIdState).not.toHaveBeenCalled(); + })); + }); + + // ── no polling ───────────────────────────────────────────────────────────── + + it('does not fetch state on a timer when no WS message arrives', fakeAsync(() => { + component.ngOnInit(); + component.state.subscribe(); + runService.getApiV2EvaluationByEvaluationIdState.calls.reset(); + + tick(5000); // advance time — no interval should fire + + expect(runService.getApiV2EvaluationByEvaluationIdState).not.toHaveBeenCalled(); + })); +}); From 7adf58a4da4f45dd2188e3d405ebb7bee1493f28 Mon Sep 17 00:00:00 2001 From: mcmahj45 Date: Thu, 11 Jun 2026 16:50:44 +0100 Subject: [PATCH 04/10] Mapping tests --- .../main/kotlin/dev/dres/run/RunExecutor.kt | 24 +- .../dres/run/RunExecutorWebSocketTest.kt | 206 ++++++++++++++++++ 2 files changed, 221 insertions(+), 9 deletions(-) create mode 100644 backend/src/test/kotlin/dres/run/RunExecutorWebSocketTest.kt diff --git a/backend/src/main/kotlin/dev/dres/run/RunExecutor.kt b/backend/src/main/kotlin/dev/dres/run/RunExecutor.kt index 7fa8d79f..27123259 100644 --- a/backend/src/main/kotlin/dev/dres/run/RunExecutor.kt +++ b/backend/src/main/kotlin/dev/dres/run/RunExecutor.kt @@ -212,20 +212,26 @@ object RunExecutor : StreamEventHandler { } } + /** + * Maps a [StreamEvent] to the [ServerMessage] that should be broadcast, or null + * if the event type requires no WebSocket notification. + */ + internal fun eventToMessage(event: StreamEvent): ServerMessage? = when (event) { + is RunStartEvent -> ServerMessage(event.runId, ServerMessageType.COMPETITION_START) + is RunEndEvent -> ServerMessage(event.runId, ServerMessageType.COMPETITION_END) + is TaskStartEvent -> ServerMessage(event.runId, ServerMessageType.TASK_START, event.taskId) + is TaskEndEvent -> ServerMessage(event.runId, ServerMessageType.TASK_END, event.taskId) + is ScoreUpdateEvent -> ServerMessage(event.runId, ServerMessageType.COMPETITION_UPDATE) + is SubmissionEvent -> ServerMessage(event.runId, ServerMessageType.TASK_UPDATED) + else -> null + } + /** * Handles [StreamEvent]s from the [EventStreamProcessor] by broadcasting * the appropriate [ServerMessage] to all registered observers. */ override fun handleStreamEvent(event: StreamEvent) { - when (event) { - is RunStartEvent -> broadcastWsMessage(ServerMessage(event.runId, ServerMessageType.COMPETITION_START)) - is RunEndEvent -> broadcastWsMessage(ServerMessage(event.runId, ServerMessageType.COMPETITION_END)) - is TaskStartEvent -> broadcastWsMessage(ServerMessage(event.runId, ServerMessageType.TASK_START, event.taskId)) - is TaskEndEvent -> broadcastWsMessage(ServerMessage(event.runId, ServerMessageType.TASK_END, event.taskId)) - is ScoreUpdateEvent -> broadcastWsMessage(ServerMessage(event.runId, ServerMessageType.COMPETITION_UPDATE)) - is SubmissionEvent -> broadcastWsMessage(ServerMessage(event.runId, ServerMessageType.TASK_UPDATED)) - else -> {} - } + eventToMessage(event)?.let { broadcastWsMessage(it) } } /** diff --git a/backend/src/test/kotlin/dres/run/RunExecutorWebSocketTest.kt b/backend/src/test/kotlin/dres/run/RunExecutorWebSocketTest.kt new file mode 100644 index 00000000..df13e64a --- /dev/null +++ b/backend/src/test/kotlin/dres/run/RunExecutorWebSocketTest.kt @@ -0,0 +1,206 @@ +package dres.run + +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper +import com.fasterxml.jackson.module.kotlin.readValue +import dev.dres.api.rest.types.evaluation.submission.ApiClientSubmission +import dev.dres.api.rest.types.evaluation.websocket.ServerMessage +import dev.dres.api.rest.types.evaluation.websocket.ServerMessageType +import dev.dres.api.rest.types.template.ApiEvaluationTemplate +import dev.dres.api.rest.types.template.tasks.ApiTaskTemplate +import dev.dres.run.RunExecutor +import dev.dres.run.eventstream.* +import org.junit.jupiter.api.Assertions.* +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.assertDoesNotThrow + +/** + * Unit tests for the WebSocket event handling in [RunExecutor]. + * + * Covers: + * 1. [RunExecutor.eventToMessage] — asserts the correct [ServerMessageType] and + * field values are produced for every handled [StreamEvent] subtype. + * 2. [ServerMessage] JSON serialisation — round-trip and field correctness. + * 3. Smoke tests — [RunExecutor.handleStreamEvent] does not throw for any event. + */ +class RunExecutorWebSocketTest { + + private val mapper = jacksonObjectMapper() + + private val runId = "eval-001" + private val taskId = "task-001" + + // ── eventToMessage mapping ──────────────────────────────────────────────── + + @Test + fun `RunStartEvent maps to COMPETITION_START`() { + val msg = RunExecutor.eventToMessage(RunStartEvent(runId, minimalTemplate())) + assertNotNull(msg) + assertEquals(ServerMessageType.COMPETITION_START, msg!!.type) + assertEquals(runId, msg.evaluationId) + assertNull(msg.taskId) + } + + @Test + fun `RunEndEvent maps to COMPETITION_END`() { + val msg = RunExecutor.eventToMessage(RunEndEvent(runId)) + assertNotNull(msg) + assertEquals(ServerMessageType.COMPETITION_END, msg!!.type) + assertEquals(runId, msg.evaluationId) + assertNull(msg.taskId) + } + + @Test + fun `TaskStartEvent maps to TASK_START with correct taskId`() { + val msg = RunExecutor.eventToMessage(TaskStartEvent(runId, taskId, minimalTaskTemplate())) + assertNotNull(msg) + assertEquals(ServerMessageType.TASK_START, msg!!.type) + assertEquals(runId, msg.evaluationId) + assertEquals(taskId, msg.taskId) + } + + @Test + fun `TaskEndEvent maps to TASK_END with correct taskId`() { + val msg = RunExecutor.eventToMessage(TaskEndEvent(runId, taskId)) + assertNotNull(msg) + assertEquals(ServerMessageType.TASK_END, msg!!.type) + assertEquals(runId, msg.evaluationId) + assertEquals(taskId, msg.taskId) + } + + @Test + fun `ScoreUpdateEvent maps to COMPETITION_UPDATE`() { + val msg = RunExecutor.eventToMessage(ScoreUpdateEvent(runId, "scoreboard", emptyMap())) + assertNotNull(msg) + assertEquals(ServerMessageType.COMPETITION_UPDATE, msg!!.type) + assertEquals(runId, msg.evaluationId) + } + + @Test + fun `SubmissionEvent maps to TASK_UPDATED`() { + val msg = RunExecutor.eventToMessage( + SubmissionEvent("session-1", runId, ApiClientSubmission(answerSets = emptyList())) + ) + assertNotNull(msg) + assertEquals(ServerMessageType.TASK_UPDATED, msg!!.type) + assertEquals(runId, msg.evaluationId) + } + + @Test + fun `unhandled event type maps to null`() { + val msg = RunExecutor.eventToMessage(InvalidRequestEvent("session-1", runId, "bad data")) + assertNull(msg) + } + + @Test + fun `every handled event type carries the correct evaluationId`() { + val events: List = listOf( + RunStartEvent(runId, minimalTemplate()), + RunEndEvent(runId), + TaskStartEvent(runId, taskId, minimalTaskTemplate()), + TaskEndEvent(runId, taskId), + ScoreUpdateEvent(runId, "board", emptyMap()), + SubmissionEvent("s", runId, ApiClientSubmission(answerSets = emptyList())) + ) + events.forEach { event -> + val msg = RunExecutor.eventToMessage(event) + assertNotNull(msg, "Expected non-null message for ${event::class.simpleName}") + assertEquals(runId, msg!!.evaluationId, "Wrong evaluationId for ${event::class.simpleName}") + } + } + + // ── ServerMessage JSON serialisation ────────────────────────────────────── + + @Test + fun `ServerMessage round-trips through JSON`() { + val original = ServerMessage(runId, ServerMessageType.TASK_START, taskId) + val decoded: ServerMessage = mapper.readValue(mapper.writeValueAsString(original)) + + assertEquals(original.evaluationId, decoded.evaluationId) + assertEquals(original.type, decoded.type) + assertEquals(original.taskId, decoded.taskId) + } + + @Test + fun `ServerMessage JSON contains evaluationId, type and timestamp`() { + val tree = mapper.readTree( + mapper.writeValueAsString(ServerMessage(runId, ServerMessageType.COMPETITION_END)) + ) + assertEquals(runId, tree["evaluationId"].asText()) + assertEquals("COMPETITION_END", tree["type"].asText()) + assertNotNull(tree["timestamp"]) + } + + @Test + fun `ServerMessage without taskId serialises taskId as null`() { + val tree = mapper.readTree( + mapper.writeValueAsString(ServerMessage(runId, ServerMessageType.COMPETITION_START)) + ) + assertTrue(tree["taskId"].isNull) + } + + @Test + fun `ServerMessage with taskId serialises taskId correctly`() { + val tree = mapper.readTree( + mapper.writeValueAsString(ServerMessage(runId, ServerMessageType.TASK_START, taskId)) + ) + assertEquals(taskId, tree["taskId"].asText()) + } + + @Test + fun `all ServerMessageTypes serialise without error`() { + ServerMessageType.entries.forEach { type -> + assertDoesNotThrow("Failed for type $type") { + mapper.writeValueAsString(ServerMessage(runId, type)) + } + } + } + + // ── handleStreamEvent smoke tests ───────────────────────────────────────── + + @Test + fun `handleStreamEvent does not throw for any event type`() { + val events: List = listOf( + RunStartEvent(runId, minimalTemplate()), + RunEndEvent(runId), + TaskStartEvent(runId, taskId, minimalTaskTemplate()), + TaskEndEvent(runId, taskId), + ScoreUpdateEvent(runId, "board", emptyMap()), + SubmissionEvent("s", runId, ApiClientSubmission(answerSets = emptyList())), + InvalidRequestEvent("s", runId, "bad data") + ) + events.forEach { event -> + assertDoesNotThrow("handleStreamEvent threw for ${event::class.simpleName}") { + RunExecutor.handleStreamEvent(event) + } + } + } + + // ── Fixture helpers ─────────────────────────────────────────────────────── + + private fun minimalTemplate() = ApiEvaluationTemplate( + id = runId, + name = "Test Evaluation", + description = null, + created = null, + modified = null, + taskTypes = emptyList(), + taskGroups = emptyList(), + tasks = emptyList(), + teams = emptyList(), + teamGroups = emptyList(), + judges = emptyList(), + viewers = emptyList() + ) + + private fun minimalTaskTemplate() = ApiTaskTemplate( + id = taskId, + name = "Task 1", + taskGroup = "group-1", + taskType = "KIS", + duration = 300L, + collectionId = "col-1", + targets = emptyList(), + hints = emptyList(), + comment = null + ) +} From e8a958c2a70d5ae474057cb0d70f5ba7434eac03 Mon Sep 17 00:00:00 2001 From: mcmahj45 Date: Fri, 12 Jun 2026 17:01:21 +0100 Subject: [PATCH 05/10] replacing the two judgment viewer interval with websocket events --- .../judgement/judgement-viewer.component.ts | 37 ++++++++++++++----- 1 file changed, 27 insertions(+), 10 deletions(-) diff --git a/frontend/src/app/judgement/judgement-viewer.component.ts b/frontend/src/app/judgement/judgement-viewer.component.ts index dc5521f1..2c4fd04e 100644 --- a/frontend/src/app/judgement/judgement-viewer.component.ts +++ b/frontend/src/app/judgement/judgement-viewer.component.ts @@ -1,7 +1,7 @@ import {AfterViewInit, Component, HostListener, Input, OnDestroy, ViewChild} from '@angular/core'; -import {BehaviorSubject, interval, Observable, of, Subscription, timer} from 'rxjs'; +import {BehaviorSubject, merge, Observable, of, Subscription, timer} from 'rxjs'; import {ActivatedRoute, Router} from '@angular/router'; -import {catchError, filter, map, switchMap, withLatestFrom} from 'rxjs/operators'; +import {catchError, filter, map, switchMap, take, withLatestFrom} from 'rxjs/operators'; import {JudgementMediaViewerComponent} from './judgement-media-viewer.component'; import {MatSnackBar} from '@angular/material/snack-bar'; import { HttpErrorResponse, HttpResponse } from '@angular/common/http'; @@ -10,6 +10,8 @@ import {MatDialog} from '@angular/material/dialog'; import {JudgementDialogComponent} from './judgement-dialog/judgement-dialog.component'; import {JudgementDialogContent} from './judgement-dialog/judgement-dialog-content.model'; import {ApiJudgement, ApiJudgementRequest, ApiVerdictStatus, JudgementService} from '../../../openapi'; +import {WebSocketService} from '../services/websocket.service'; +import {ServerMessageType} from '../model/ws/server-message-type.enum'; /** * This component subscribes to the websocket for submissions. @@ -63,11 +65,14 @@ export class JudgementViewerComponent implements AfterViewInit, OnDestroy { private activeRoute: ActivatedRoute, private snackBar: MatSnackBar, private router: Router, - private dialog: MatDialog + private dialog: MatDialog, + private wsService: WebSocketService ) { } ngAfterViewInit(): void { + this.activeRoute.params.pipe(map((p) => p.runId), take(1)).subscribe((id) => this.wsService.connect(id)); + const dialogRef = this.dialog.open(JudgementDialogComponent, { width: '400px', data: { @@ -123,11 +128,22 @@ export class JudgementViewerComponent implements AfterViewInit, OnDestroy { init(): void { /* Subscription and current run id */ this.runId = this.activeRoute.params.pipe(map((p) => p.runId)); - /* Poll for score status in a given interval */ - this.statusSub = interval(this.pollingFrequency) + + /* Trigger on relevant WebSocket events, with a 30s fallback poll in case the socket drops. */ + const wsRefresh$ = this.wsService.messages$.pipe( + filter((msg) => [ + ServerMessageType.ServerMessageTypeEnum.TASK_UPDATED, + ServerMessageType.ServerMessageTypeEnum.TASK_START, + ServerMessageType.ServerMessageTypeEnum.TASK_END, + ].includes(msg.type)) + ); + const trigger$ = merge(timer(0, 30_000), wsRefresh$); + + /* Fetch judge status on websocket event or fallback interval. */ + this.statusSub = trigger$ .pipe( withLatestFrom(this.runId), - switchMap(([i, runId]) => { + switchMap(([_, runId]) => { return this.judgementService.getApiV2EvaluationByEvaluationIdJudgeStatus(runId).pipe( catchError((err) => { console.log('Error in JudgeStatus'); @@ -149,12 +165,12 @@ export class JudgementViewerComponent implements AfterViewInit, OnDestroy { this.updateProgress(pending, open); }); - /* Poll for score updates in a given interval. */ - this.requestSub = interval(this.pollingFrequency) + /* Fetch next judgement request on websocket event or fallback interval. */ + this.requestSub = trigger$ .pipe( withLatestFrom(this.runId), - switchMap(([i, runId]) => { - /* Stop polling while judgment is ongooing */ + switchMap(([_, runId]) => { + /* Only fetch when no judgement is currently in progress. */ if (this.runId && !this.isJudgmentAvailable) { return this.judgementService.getApiV2EvaluationByEvaluationIdJudgeNext(runId, 'response').pipe( map((req: HttpResponse) => { @@ -217,6 +233,7 @@ export class JudgementViewerComponent implements AfterViewInit, OnDestroy { * */ ngOnDestroy(): void { + this.wsService.disconnect(); this.stopAll(); } From b40de5c5360077164f017fd3ca807dfcfe1ffdd3 Mon Sep 17 00:00:00 2001 From: mcmahj45 Date: Sun, 14 Jun 2026 14:59:06 +0100 Subject: [PATCH 06/10] websockets in submission lists --- .../submissions-list.component.ts | 19 +++++++++- .../judgement-voting-viewer.component.ts | 26 ++++++++++--- .../src/app/run/run-admin-view.component.ts | 38 +++++++++++++++---- .../run-async-admin-view.component.ts | 28 +++++++++++--- 4 files changed, 90 insertions(+), 21 deletions(-) diff --git a/frontend/src/app/evaluation/admin/submission/submissions-list/submissions-list.component.ts b/frontend/src/app/evaluation/admin/submission/submissions-list/submissions-list.component.ts index 3cfe6360..72244b01 100644 --- a/frontend/src/app/evaluation/admin/submission/submissions-list/submissions-list.component.ts +++ b/frontend/src/app/evaluation/admin/submission/submissions-list/submissions-list.component.ts @@ -12,7 +12,9 @@ import { TemplateService } from "../../../../../../openapi"; import { AppConfig } from "../../../../app.config"; -import { catchError, filter, map, switchMap, withLatestFrom } from "rxjs/operators"; +import { catchError, filter, map, switchMap, take, withLatestFrom } from "rxjs/operators"; +import { WebSocketService } from "../../../../services/websocket.service"; +import { ServerMessageType } from "../../../../model/ws/server-message-type.enum"; @Component({ selector: 'app-submissions-list', @@ -53,15 +55,27 @@ export class SubmissionsListComponent implements AfterViewInit, OnDestroy{ private evaluationService: EvaluationAdministratorService, private templateService: TemplateService, public config: AppConfig, + private wsService: WebSocketService, ) { this.runId = this.activeRoute.paramMap.pipe(map((params) => params.get('runId'))); this.taskId = this.activeRoute.paramMap.pipe(map((params) => params.get('taskId'))); } ngAfterViewInit(): void { + this.runId.pipe(take(1)).subscribe((id) => this.wsService.connect(id)); + + /* Refresh on new/updated submissions pushed via WebSocket, in addition to manual refresh and periodic polling. */ + const wsRefresh$ = this.wsService.messages$.pipe( + filter((msg) => [ + ServerMessageType.ServerMessageTypeEnum.TASK_UPDATED, + ServerMessageType.ServerMessageTypeEnum.TASK_END, + ].includes(msg.type)) + ); + this.subscription = merge( timer(0, this.pollingFrequencyInSeconds * 1000) .pipe(filter((_) => this.polling)), - this.refreshSubject) + this.refreshSubject, + wsRefresh$) .pipe( withLatestFrom(this.runId, this.taskId), switchMap(([_,r,t]) => this.evaluationService.getApiV2EvaluationAdminByEvaluationIdSubmissionListByTemplateId(r,t)), @@ -98,6 +112,7 @@ export class SubmissionsListComponent implements AfterViewInit, OnDestroy{ } ngOnDestroy(): void { + this.wsService.disconnect(); this.subscription?.unsubscribe(); this.subscription = null; this.sub?.unsubscribe(); diff --git a/frontend/src/app/judgement/judgement-voting-viewer.component.ts b/frontend/src/app/judgement/judgement-voting-viewer.component.ts index f62e6c26..462486ba 100644 --- a/frontend/src/app/judgement/judgement-voting-viewer.component.ts +++ b/frontend/src/app/judgement/judgement-voting-viewer.component.ts @@ -1,12 +1,14 @@ import { Component, Input, OnDestroy, OnInit, ViewChild } from '@angular/core'; -import { BehaviorSubject, interval, Observable, of, Subscription } from 'rxjs'; -import { catchError, filter, map, switchMap, withLatestFrom } from 'rxjs/operators'; +import { BehaviorSubject, merge, Observable, of, Subscription, timer } from 'rxjs'; +import { catchError, filter, map, switchMap, take, withLatestFrom } from 'rxjs/operators'; import { ActivatedRoute, Router } from '@angular/router'; import { AppConfig } from '../app.config'; import { HttpErrorResponse, HttpResponse } from '@angular/common/http'; import { MatSnackBar } from '@angular/material/snack-bar'; import { JudgementMediaViewerComponent } from './judgement-media-viewer.component'; import {ApiJudgementRequest, JudgementService} from '../../../openapi'; +import { WebSocketService } from '../services/websocket.service'; +import { ServerMessageType } from '../model/ws/server-message-type.enum'; @Component({ selector: 'app-judgement-voting-viewer', @@ -33,18 +35,29 @@ export class JudgementVotingViewerComponent implements OnInit, OnDestroy { private activeRoute: ActivatedRoute, private config: AppConfig, private snackBar: MatSnackBar, - private router: Router + private router: Router, + private wsService: WebSocketService ) {} ngOnInit(): void { this.runId = this.activeRoute.params.pipe(map((p) => p.runId)); this.voteClientPath = this.runId.pipe(map((id) => this.config.resolveUrl(`vote#${id}`))); - /* Poll for score updates in a given interval. */ - this.requestSub = interval(this.pollingFrequency) + this.runId.pipe(take(1)).subscribe((id) => this.wsService.connect(id)); + + const wsRefresh$ = this.wsService.messages$.pipe( + filter((msg) => [ + ServerMessageType.ServerMessageTypeEnum.TASK_UPDATED, + ServerMessageType.ServerMessageTypeEnum.TASK_START, + ServerMessageType.ServerMessageTypeEnum.TASK_END, + ].includes(msg.type)) + ); + + /* Fetch next vote request on websocket event or 30s fallback poll. */ + this.requestSub = merge(timer(0, 30_000), wsRefresh$) .pipe( withLatestFrom(this.runId), - switchMap(([i, runId]) => { + switchMap(([_, runId]) => { if (this.runId) { return this.judgementService.getApiV2EvaluationByEvaluationIdVoteNext(runId, 'response').pipe( map((req: HttpResponse) => { @@ -97,6 +110,7 @@ export class JudgementVotingViewerComponent implements OnInit, OnDestroy { } ngOnDestroy(): void { + this.wsService.disconnect(); this.requestSub.unsubscribe(); this.requestSub = null; if (this.judgePlayer) { diff --git a/frontend/src/app/run/run-admin-view.component.ts b/frontend/src/app/run/run-admin-view.component.ts index 5da8c826..d41912f2 100644 --- a/frontend/src/app/run/run-admin-view.component.ts +++ b/frontend/src/app/run/run-admin-view.component.ts @@ -1,8 +1,10 @@ -import { Component } from '@angular/core'; +import { Component, OnDestroy, OnInit } from '@angular/core'; import { ActivatedRoute, Router } from '@angular/router'; import { AppConfig } from '../app.config'; import { combineLatest, merge, mergeMap, Observable, of, Subject, timer} from 'rxjs'; -import { catchError, filter, map, shareReplay, switchMap } from "rxjs/operators"; +import { catchError, filter, map, shareReplay, switchMap, take } from "rxjs/operators"; +import { WebSocketService } from '../services/websocket.service'; +import { ServerMessageType } from '../model/ws/server-message-type.enum'; import { MatSnackBar } from '@angular/material/snack-bar'; import { MatDialog } from '@angular/material/dialog'; import { RunInfoOverviewTuple } from './admin-run-list.component'; @@ -26,7 +28,7 @@ export interface CombinedRun { styleUrls: ['./run-admin-view.component.scss'], standalone: false }) -export class RunAdminViewComponent { +export class RunAdminViewComponent implements OnInit, OnDestroy { private static VIEWER_POLLING_FREQUENCY = 3 * 1000; //ms private static STATE_POLLING_FREQUENCY = 1 * 1000; //ms @@ -50,8 +52,20 @@ export class RunAdminViewComponent { private competitionService: TemplateService, private runAdminService: EvaluationAdministratorService, private snackBar: MatSnackBar, - private dialog: MatDialog + private dialog: MatDialog, + private wsService: WebSocketService ) { + const wsRefresh$ = this.wsService.messages$.pipe( + filter((msg) => [ + ServerMessageType.ServerMessageTypeEnum.TASK_START, + ServerMessageType.ServerMessageTypeEnum.TASK_END, + ServerMessageType.ServerMessageTypeEnum.TASK_UPDATED, + ServerMessageType.ServerMessageTypeEnum.TASK_PREPARE, + ServerMessageType.ServerMessageTypeEnum.COMPETITION_UPDATE, + ServerMessageType.ServerMessageTypeEnum.COMPETITION_END, + ].includes(msg.type)) + ); + this.runId = this.activeRoute.params.pipe(map((a) => a.runId)); this.run = this.runId.pipe( switchMap((runId) => @@ -71,7 +85,7 @@ export class RunAdminViewComponent { }), filter((q) => q != null) ), - merge(timer(0, RunAdminViewComponent.STATE_POLLING_FREQUENCY), this.refreshSubject).pipe(switchMap((index) => this.runService.getApiV2EvaluationByEvaluationIdState(runId))), + merge(timer(0, 30_000), this.refreshSubject, wsRefresh$).pipe(switchMap(() => this.runService.getApiV2EvaluationByEvaluationIdState(runId))), ]) ), map(([i, s]) => { @@ -138,8 +152,8 @@ export class RunAdminViewComponent { }), filter((q) => q != null) ), - merge(timer(0, RunAdminViewComponent.OVERVIEW_POLLING_FREQUENCY), this.refreshSubject).pipe( - switchMap((index) => this.runAdminService.getApiV2EvaluationAdminByEvaluationIdOverview(runId)) + merge(timer(0, 30_000), this.refreshSubject, wsRefresh$).pipe( + switchMap(() => this.runAdminService.getApiV2EvaluationAdminByEvaluationIdOverview(runId)) ), ]) ), @@ -150,7 +164,7 @@ export class RunAdminViewComponent { ); this.viewers = this.runId.pipe( - mergeMap((runId) => timer(0, RunAdminViewComponent.VIEWER_POLLING_FREQUENCY).pipe(switchMap((i) => this.runAdminService.getApiV2EvaluationAdminByEvaluationIdViewerList(runId)))) + mergeMap((runId) => merge(timer(0, 30_000), wsRefresh$).pipe(switchMap(() => this.runAdminService.getApiV2EvaluationAdminByEvaluationIdViewerList(runId)))) ); @@ -162,6 +176,14 @@ export class RunAdminViewComponent { shareReplay({ bufferSize: 1, refCount: true }) ); } + ngOnInit(): void { + this.runId.pipe(take(1)).subscribe((id) => this.wsService.connect(id)); + } + + ngOnDestroy(): void { + this.wsService.disconnect(); + } + stateFromCombined(combined: Observable): Observable{ return combined.pipe(map((c) => c.state)) } diff --git a/frontend/src/app/run/run-async-admin-view/run-async-admin-view.component.ts b/frontend/src/app/run/run-async-admin-view/run-async-admin-view.component.ts index 3bf82e68..1e22c50e 100644 --- a/frontend/src/app/run/run-async-admin-view/run-async-admin-view.component.ts +++ b/frontend/src/app/run/run-async-admin-view/run-async-admin-view.component.ts @@ -5,6 +5,8 @@ import { AppConfig } from '../../app.config'; import { MatSnackBar } from '@angular/material/snack-bar'; import { MatDialog } from '@angular/material/dialog'; import { catchError, filter, map, shareReplay, switchMap, take } from 'rxjs/operators'; +import { WebSocketService } from '../../services/websocket.service'; +import { ServerMessageType } from '../../model/ws/server-message-type.enum'; import { RunInfoOverviewTuple } from '../admin-run-list.component'; import { MatAccordion } from '@angular/material/expansion'; import { @@ -47,9 +49,22 @@ export class RunAsyncAdminViewComponent implements AfterViewInit, OnDestroy { private scoreService: EvaluationScoresService, private downloadService: DownloadService, private snackBar: MatSnackBar, - private dialog: MatDialog + private dialog: MatDialog, + private wsService: WebSocketService ) { this.activeRoute.params.pipe(map((a) => a.runId)).subscribe(this.runId); + + const wsRefresh$ = this.wsService.messages$.pipe( + filter((msg) => [ + ServerMessageType.ServerMessageTypeEnum.TASK_START, + ServerMessageType.ServerMessageTypeEnum.TASK_END, + ServerMessageType.ServerMessageTypeEnum.TASK_UPDATED, + ServerMessageType.ServerMessageTypeEnum.TASK_PREPARE, + ServerMessageType.ServerMessageTypeEnum.COMPETITION_UPDATE, + ServerMessageType.ServerMessageTypeEnum.COMPETITION_END, + ].includes(msg.type)) + ); + this.run = this.runId.pipe( switchMap((runId) => combineLatest([ @@ -66,8 +81,8 @@ export class RunAsyncAdminViewComponent implements AfterViewInit, OnDestroy { }), filter((q) => q != null) ), - merge(timer(0, 1000), this.update).pipe( - switchMap((index) => this.runAdminService.getApiV2EvaluationAdminByEvaluationIdOverview(runId)) + merge(timer(0, 30_000), this.update, wsRefresh$).pipe( + switchMap(() => this.runAdminService.getApiV2EvaluationAdminByEvaluationIdOverview(runId)) ), ]) ), @@ -84,7 +99,7 @@ export class RunAsyncAdminViewComponent implements AfterViewInit, OnDestroy { shareReplay({ bufferSize: 1, refCount: true }) /* Cache last successful loading. */ ); - this.taskSubmissionCounts = merge(timer(0, 15000), this.update).pipe( + this.taskSubmissionCounts = merge(timer(0, 30_000), this.update, wsRefresh$).pipe( switchMap(() => this.run.pipe(take(1))), switchMap(run => { const runId = this.runId.getValue(); @@ -121,6 +136,8 @@ export class RunAsyncAdminViewComponent implements AfterViewInit, OnDestroy { } ngAfterViewInit(): void { + this.runId.pipe(take(1)).subscribe((id) => this.wsService.connect(id)); + /* Cache past tasks initially */ this.runId.subscribe((runId) => { this.runAdminService.getApiV2EvaluationAdminByEvaluationIdTaskPastList(runId).subscribe((arr) => (this.pastTasksValue = arr)); @@ -149,6 +166,7 @@ export class RunAsyncAdminViewComponent implements AfterViewInit, OnDestroy { } ngOnDestroy(): void { - this.update?.unsubscribe() + this.wsService.disconnect(); + this.update?.unsubscribe(); } } From f650869d19714b16a425109f0f0339dc2b54d116 Mon Sep 17 00:00:00 2001 From: mcmahj45 Date: Sun, 14 Jun 2026 15:46:02 +0100 Subject: [PATCH 07/10] Finishing touchs and bug fixs to pass tests --- .../submissions-list.component.spec.ts | 127 ++++++++++++++ .../judgement-viewer.component.spec.ts | 5 + .../judgement-voting-viewer.component.spec.ts | 116 +++++++++++++ .../app/run/run-admin-view.component.spec.ts | 153 +++++++++++++++++ .../run-async-admin-view.component.spec.ts | 159 ++++++++++++++++++ 5 files changed, 560 insertions(+) create mode 100644 frontend/src/app/evaluation/admin/submission/submissions-list/submissions-list.component.spec.ts create mode 100644 frontend/src/app/judgement/judgement-voting-viewer.component.spec.ts create mode 100644 frontend/src/app/run/run-admin-view.component.spec.ts create mode 100644 frontend/src/app/run/run-async-admin-view/run-async-admin-view.component.spec.ts diff --git a/frontend/src/app/evaluation/admin/submission/submissions-list/submissions-list.component.spec.ts b/frontend/src/app/evaluation/admin/submission/submissions-list/submissions-list.component.spec.ts new file mode 100644 index 00000000..fab422b0 --- /dev/null +++ b/frontend/src/app/evaluation/admin/submission/submissions-list/submissions-list.component.spec.ts @@ -0,0 +1,127 @@ +import { NO_ERRORS_SCHEMA } from '@angular/core'; +import { ComponentFixture, TestBed, fakeAsync, tick, discardPeriodicTasks } from '@angular/core/testing'; +import { ActivatedRoute } from '@angular/router'; +import { MatSnackBar } from '@angular/material/snack-bar'; +import { MatDialog } from '@angular/material/dialog'; +import { of, Subject } from 'rxjs'; + +import { SubmissionsListComponent } from './submissions-list.component'; +import { AppConfig } from '../../../../app.config'; +import { WebSocketService } from '../../../../services/websocket.service'; +import { EvaluationAdministratorService, EvaluationService, TemplateService } from '../../../../../../openapi'; +import { IWsServerMessage } from '../../../../model/ws/ws-server-message.interface'; +import { ServerMessageType } from '../../../../model/ws/server-message-type.enum'; + +// ── Helpers ─────────────────────────────────────────────────────────────────── + +function wsMsg(type: ServerMessageType.ServerMessageTypeEnum): IWsServerMessage { + return { evaluationId: 'eval-1', type, timestamp: Date.now() }; +} + +const REFRESH_TYPES: ServerMessageType.ServerMessageTypeEnum[] = [ + ServerMessageType.ServerMessageTypeEnum.TASK_UPDATED, + ServerMessageType.ServerMessageTypeEnum.TASK_END, +]; + +const IRRELEVANT_TYPES: ServerMessageType.ServerMessageTypeEnum[] = [ + ServerMessageType.ServerMessageTypeEnum.PING, + ServerMessageType.ServerMessageTypeEnum.COMPETITION_UPDATE, + ServerMessageType.ServerMessageTypeEnum.TASK_START, +]; + +// ── Suite ───────────────────────────────────────────────────────────────────── + +describe('SubmissionsListComponent WebSocket wiring', () => { + let component: SubmissionsListComponent; + let fixture: ComponentFixture; + let wsService: jasmine.SpyObj; + let evalService: jasmine.SpyObj; + let evaluationAdminService: jasmine.SpyObj; + let templateService: jasmine.SpyObj; + let messages$: Subject; + + beforeEach(() => { + messages$ = new Subject(); + + wsService = jasmine.createSpyObj('WebSocketService', ['connect', 'disconnect'], { + messages$: messages$.asObservable(), + }); + + evalService = jasmine.createSpyObj('EvaluationService', ['getApiV2EvaluationByEvaluationIdInfo']); + evalService.getApiV2EvaluationByEvaluationIdInfo.and.returnValue(of({ id: 'eval-1', templateId: 'tpl-1' } as any)); + + evaluationAdminService = jasmine.createSpyObj('EvaluationAdministratorService', [ + 'getApiV2EvaluationAdminByEvaluationIdSubmissionListByTemplateId', + ]); + evaluationAdminService.getApiV2EvaluationAdminByEvaluationIdSubmissionListByTemplateId.and.returnValue(of([] as any)); + + templateService = jasmine.createSpyObj('TemplateService', ['getApiV2TemplateByTemplateIdTaskList']); + templateService.getApiV2TemplateByTemplateIdTaskList.and.returnValue(of([{ id: 'task-1' }] as any)); + + TestBed.configureTestingModule({ + declarations: [SubmissionsListComponent], + schemas: [NO_ERRORS_SCHEMA], + providers: [ + { provide: EvaluationService, useValue: evalService }, + { provide: EvaluationAdministratorService, useValue: evaluationAdminService }, + { provide: TemplateService, useValue: templateService }, + { + provide: ActivatedRoute, + useValue: { paramMap: of({ get: (key: string) => (key === 'runId' ? 'eval-1' : 'task-1') }) }, + }, + { provide: AppConfig, useValue: { resolveApiUrl: (p: string) => `http://localhost${p}` } }, + { provide: MatSnackBar, useValue: jasmine.createSpyObj('MatSnackBar', ['open']) }, + { provide: MatDialog, useValue: {} }, + { provide: WebSocketService, useValue: wsService }, + ], + }); + + fixture = TestBed.createComponent(SubmissionsListComponent); + component = fixture.componentInstance; + }); + + // ── connect / disconnect ─────────────────────────────────────────────────── + + it('connects the WebSocket with the evaluationId on init', () => { + component.ngAfterViewInit(); + expect(wsService.connect).toHaveBeenCalledWith('eval-1'); + }); + + it('disconnects the WebSocket on destroy', () => { + component.ngAfterViewInit(); + fixture.destroy(); + expect(wsService.disconnect).toHaveBeenCalled(); + }); + + // ── submission list refresh on WS message ───────────────────────────────── + + REFRESH_TYPES.forEach((type) => { + it(`refreshes the submission list when ${type} message is received`, fakeAsync(() => { + component.ngAfterViewInit(); + tick(); + evaluationAdminService.getApiV2EvaluationAdminByEvaluationIdSubmissionListByTemplateId.calls.reset(); + + messages$.next(wsMsg(type)); + tick(); + + expect(evaluationAdminService.getApiV2EvaluationAdminByEvaluationIdSubmissionListByTemplateId).toHaveBeenCalledWith('eval-1', 'task-1'); + discardPeriodicTasks(); + })); + }); + + // ── PING / unrelated messages are ignored ────────────────────────────────── + + IRRELEVANT_TYPES.forEach((type) => { + it(`does not refresh the submission list for ${type} message`, fakeAsync(() => { + component.ngAfterViewInit(); + tick(); + evaluationAdminService.getApiV2EvaluationAdminByEvaluationIdSubmissionListByTemplateId.calls.reset(); + + messages$.next(wsMsg(type)); + tick(); + + expect(evaluationAdminService.getApiV2EvaluationAdminByEvaluationIdSubmissionListByTemplateId).not.toHaveBeenCalled(); + discardPeriodicTasks(); + })); + }); +}); diff --git a/frontend/src/app/judgement/judgement-viewer.component.spec.ts b/frontend/src/app/judgement/judgement-viewer.component.spec.ts index 840346b4..e8a30eba 100644 --- a/frontend/src/app/judgement/judgement-viewer.component.spec.ts +++ b/frontend/src/app/judgement/judgement-viewer.component.spec.ts @@ -7,6 +7,7 @@ import { of, Subject } from 'rxjs'; import { JudgementViewerComponent } from './judgement-viewer.component'; import { JudgementService } from '../../../openapi'; import { ApiJudgementRequest } from '../../../openapi'; +import { WebSocketService } from '../services/websocket.service'; function makeRequest(token: string, desc = 'Find a cat'): ApiJudgementRequest { return { token, taskDescription: desc, validator: 'v1', answerSet: { answers: [] } } as any; @@ -38,6 +39,10 @@ describe('JudgementViewerComponent', () => { { provide: MatSnackBar, useValue: jasmine.createSpyObj('MatSnackBar', ['open']) }, { provide: Router, useValue: jasmine.createSpyObj('Router', ['navigate']) }, { provide: MatDialog, useValue: mockDialog }, + { + provide: WebSocketService, + useValue: jasmine.createSpyObj('WebSocketService', ['connect', 'disconnect'], { messages$: new Subject().asObservable() }), + }, ], }); diff --git a/frontend/src/app/judgement/judgement-voting-viewer.component.spec.ts b/frontend/src/app/judgement/judgement-voting-viewer.component.spec.ts new file mode 100644 index 00000000..e9643544 --- /dev/null +++ b/frontend/src/app/judgement/judgement-voting-viewer.component.spec.ts @@ -0,0 +1,116 @@ +import { NO_ERRORS_SCHEMA } from '@angular/core'; +import { ComponentFixture, TestBed, fakeAsync, tick, discardPeriodicTasks } from '@angular/core/testing'; +import { ActivatedRoute, Router } from '@angular/router'; +import { MatSnackBar } from '@angular/material/snack-bar'; +import { of, Subject } from 'rxjs'; + +import { JudgementVotingViewerComponent } from './judgement-voting-viewer.component'; +import { AppConfig } from '../app.config'; +import { WebSocketService } from '../services/websocket.service'; +import { JudgementService } from '../../../openapi'; +import { IWsServerMessage } from '../model/ws/ws-server-message.interface'; +import { ServerMessageType } from '../model/ws/server-message-type.enum'; + +// ── Helpers ─────────────────────────────────────────────────────────────────── + +function wsMsg(type: ServerMessageType.ServerMessageTypeEnum): IWsServerMessage { + return { evaluationId: 'eval-1', type, timestamp: Date.now() }; +} + +const REFRESH_TYPES: ServerMessageType.ServerMessageTypeEnum[] = [ + ServerMessageType.ServerMessageTypeEnum.TASK_UPDATED, + ServerMessageType.ServerMessageTypeEnum.TASK_START, + ServerMessageType.ServerMessageTypeEnum.TASK_END, +]; + +const IRRELEVANT_TYPES: ServerMessageType.ServerMessageTypeEnum[] = [ + ServerMessageType.ServerMessageTypeEnum.PING, + ServerMessageType.ServerMessageTypeEnum.COMPETITION_UPDATE, +]; + +// ── Suite ───────────────────────────────────────────────────────────────────── + +describe('JudgementVotingViewerComponent WebSocket wiring', () => { + let component: JudgementVotingViewerComponent; + let wsService: jasmine.SpyObj; + let judgementService: jasmine.SpyObj; + let messages$: Subject; + let fixture: ComponentFixture; + + beforeEach(() => { + messages$ = new Subject(); + + wsService = jasmine.createSpyObj('WebSocketService', ['connect', 'disconnect'], { + messages$: messages$.asObservable(), + }); + + judgementService = jasmine.createSpyObj('JudgementService', ['getApiV2EvaluationByEvaluationIdVoteNext']); + judgementService.getApiV2EvaluationByEvaluationIdVoteNext.and.returnValue(of({ status: 202, body: null } as any)); + + TestBed.configureTestingModule({ + declarations: [JudgementVotingViewerComponent], + schemas: [NO_ERRORS_SCHEMA], + providers: [ + { provide: JudgementService, useValue: judgementService }, + { provide: ActivatedRoute, useValue: { params: of({ runId: 'eval-1' }) } }, + { provide: AppConfig, useValue: { resolveUrl: (p: string) => `http://localhost/${p}` } }, + { provide: MatSnackBar, useValue: jasmine.createSpyObj('MatSnackBar', ['open']) }, + { provide: Router, useValue: jasmine.createSpyObj('Router', ['navigate']) }, + { provide: WebSocketService, useValue: wsService }, + ], + }); + + fixture = TestBed.createComponent(JudgementVotingViewerComponent); + component = fixture.componentInstance; + }); + + afterEach(() => { + (component as any).requestSub?.unsubscribe(); + }); + + // ── connect / disconnect ─────────────────────────────────────────────────── + + it('connects the WebSocket with the evaluationId on init', () => { + component.ngOnInit(); + expect(wsService.connect).toHaveBeenCalledWith('eval-1'); + }); + + it('disconnects the WebSocket on destroy', () => { + component.ngOnInit(); + fixture.destroy(); + expect(wsService.disconnect).toHaveBeenCalled(); + }); + + // ── vote request refresh on WS message ───────────────────────────────────── + + REFRESH_TYPES.forEach((type) => { + it(`fetches the next vote request when ${type} message is received`, fakeAsync(() => { + component.ngOnInit(); + judgementService.getApiV2EvaluationByEvaluationIdVoteNext.calls.reset(); + + messages$.next(wsMsg(type)); + tick(); + + expect(judgementService.getApiV2EvaluationByEvaluationIdVoteNext.calls.mostRecent().args).toEqual(['eval-1', 'response']); + + discardPeriodicTasks(); + })); + }); + + // ── PING / unrelated messages are ignored ────────────────────────────────── + + IRRELEVANT_TYPES.forEach((type) => { + it(`does not fetch the next vote request for ${type} message`, fakeAsync(() => { + component.ngOnInit(); + tick(); // consume the initial fallback-poll emission + judgementService.getApiV2EvaluationByEvaluationIdVoteNext.calls.reset(); + + messages$.next(wsMsg(type)); + tick(); + + expect(judgementService.getApiV2EvaluationByEvaluationIdVoteNext).not.toHaveBeenCalled(); + + discardPeriodicTasks(); + })); + }); +}); diff --git a/frontend/src/app/run/run-admin-view.component.spec.ts b/frontend/src/app/run/run-admin-view.component.spec.ts new file mode 100644 index 00000000..e888e372 --- /dev/null +++ b/frontend/src/app/run/run-admin-view.component.spec.ts @@ -0,0 +1,153 @@ +import { NO_ERRORS_SCHEMA } from '@angular/core'; +import { ComponentFixture, TestBed, fakeAsync, tick, discardPeriodicTasks } from '@angular/core/testing'; +import { ActivatedRoute, Router } from '@angular/router'; +import { MatSnackBar } from '@angular/material/snack-bar'; +import { MatDialog } from '@angular/material/dialog'; +import { of, Subject } from 'rxjs'; + +import { RunAdminViewComponent } from './run-admin-view.component'; +import { AppConfig } from '../app.config'; +import { WebSocketService } from '../services/websocket.service'; +import { EvaluationAdministratorService, EvaluationService, TemplateService } from '../../../openapi'; +import { IWsServerMessage } from '../model/ws/ws-server-message.interface'; +import { ServerMessageType } from '../model/ws/server-message-type.enum'; + +// ── Helpers ─────────────────────────────────────────────────────────────────── + +function wsMsg(type: ServerMessageType.ServerMessageTypeEnum): IWsServerMessage { + return { evaluationId: 'eval-1', type, timestamp: Date.now() }; +} + +const REFRESH_TYPES: ServerMessageType.ServerMessageTypeEnum[] = [ + ServerMessageType.ServerMessageTypeEnum.TASK_START, + ServerMessageType.ServerMessageTypeEnum.TASK_END, + ServerMessageType.ServerMessageTypeEnum.TASK_UPDATED, + ServerMessageType.ServerMessageTypeEnum.TASK_PREPARE, + ServerMessageType.ServerMessageTypeEnum.COMPETITION_UPDATE, + ServerMessageType.ServerMessageTypeEnum.COMPETITION_END, +]; + +const IRRELEVANT_TYPES: ServerMessageType.ServerMessageTypeEnum[] = [ + ServerMessageType.ServerMessageTypeEnum.PING, + ServerMessageType.ServerMessageTypeEnum.COMPETITION_START, +]; + +// ── Suite ───────────────────────────────────────────────────────────────────── + +describe('RunAdminViewComponent WebSocket wiring', () => { + let component: RunAdminViewComponent; + let fixture: ComponentFixture; + let wsService: jasmine.SpyObj; + let runService: jasmine.SpyObj; + let runAdminService: jasmine.SpyObj; + let templateService: jasmine.SpyObj; + let messages$: Subject; + + beforeEach(() => { + messages$ = new Subject(); + + wsService = jasmine.createSpyObj('WebSocketService', ['connect', 'disconnect'], { + messages$: messages$.asObservable(), + }); + + runService = jasmine.createSpyObj('EvaluationService', [ + 'getApiV2EvaluationByEvaluationIdInfo', + 'getApiV2EvaluationByEvaluationIdState', + ]); + runService.getApiV2EvaluationByEvaluationIdInfo.and.returnValue(of({ id: 'eval-1', templateId: 'tpl-1', name: 'Test' } as any)); + runService.getApiV2EvaluationByEvaluationIdState.and.returnValue(of({ taskTemplateId: 't1', taskStatus: 'RUNNING' } as any)); + + runAdminService = jasmine.createSpyObj('EvaluationAdministratorService', [ + 'getApiV2EvaluationAdminByEvaluationIdOverview', + 'getApiV2EvaluationAdminByEvaluationIdViewerList', + 'getApiV2EvaluationAdminByEvaluationIdTaskPastList', + 'getApiV2EvaluationAdminByEvaluationIdSubmissionListByTemplateId', + ]); + runAdminService.getApiV2EvaluationAdminByEvaluationIdOverview.and.returnValue(of({} as any)); + runAdminService.getApiV2EvaluationAdminByEvaluationIdViewerList.and.returnValue(of([] as any)); + runAdminService.getApiV2EvaluationAdminByEvaluationIdTaskPastList.and.returnValue(of([] as any)); + runAdminService.getApiV2EvaluationAdminByEvaluationIdSubmissionListByTemplateId.and.returnValue(of([] as any)); + + templateService = jasmine.createSpyObj('TemplateService', ['getApiV2TemplateByTemplateIdTeamList']); + templateService.getApiV2TemplateByTemplateIdTeamList.and.returnValue(of([] as any)); + + TestBed.configureTestingModule({ + declarations: [RunAdminViewComponent], + schemas: [NO_ERRORS_SCHEMA], + providers: [ + { provide: EvaluationService, useValue: runService }, + { provide: EvaluationAdministratorService, useValue: runAdminService }, + { provide: TemplateService, useValue: templateService }, + { provide: ActivatedRoute, useValue: { params: of({ runId: 'eval-1' }) } }, + { provide: AppConfig, useValue: { resolveApiUrl: (p: string) => `http://localhost${p}` } }, + { provide: MatSnackBar, useValue: jasmine.createSpyObj('MatSnackBar', ['open']) }, + { provide: Router, useValue: jasmine.createSpyObj('Router', ['navigate', 'navigateByUrl']) }, + { provide: MatDialog, useValue: {} }, + { provide: WebSocketService, useValue: wsService }, + ], + }); + + fixture = TestBed.createComponent(RunAdminViewComponent); + component = fixture.componentInstance; + }); + + // ── connect / disconnect ─────────────────────────────────────────────────── + + it('connects the WebSocket with the evaluationId on init', () => { + component.ngOnInit(); + expect(wsService.connect).toHaveBeenCalledWith('eval-1'); + }); + + it('disconnects the WebSocket on destroy', () => { + component.ngOnInit(); + fixture.destroy(); + expect(wsService.disconnect).toHaveBeenCalled(); + }); + + // ── state refresh on WS message ──────────────────────────────────────────── + + REFRESH_TYPES.forEach((type) => { + it(`refreshes run state when ${type} message is received`, fakeAsync(() => { + component.ngOnInit(); + component.run.subscribe(); + tick(); + runService.getApiV2EvaluationByEvaluationIdState.calls.reset(); + + messages$.next(wsMsg(type)); + tick(); + + expect(runService.getApiV2EvaluationByEvaluationIdState).toHaveBeenCalledWith('eval-1'); + discardPeriodicTasks(); + })); + + it(`refreshes the viewer list when ${type} message is received`, fakeAsync(() => { + component.ngOnInit(); + component.viewers.subscribe(); + tick(); + runAdminService.getApiV2EvaluationAdminByEvaluationIdViewerList.calls.reset(); + + messages$.next(wsMsg(type)); + tick(); + + expect(runAdminService.getApiV2EvaluationAdminByEvaluationIdViewerList).toHaveBeenCalledWith('eval-1'); + discardPeriodicTasks(); + })); + }); + + // ── PING / unrelated messages are ignored ────────────────────────────────── + + IRRELEVANT_TYPES.forEach((type) => { + it(`does not refresh run state for ${type} message`, fakeAsync(() => { + component.ngOnInit(); + component.run.subscribe(); + tick(); + runService.getApiV2EvaluationByEvaluationIdState.calls.reset(); + + messages$.next(wsMsg(type)); + tick(); + + expect(runService.getApiV2EvaluationByEvaluationIdState).not.toHaveBeenCalled(); + discardPeriodicTasks(); + })); + }); +}); diff --git a/frontend/src/app/run/run-async-admin-view/run-async-admin-view.component.spec.ts b/frontend/src/app/run/run-async-admin-view/run-async-admin-view.component.spec.ts new file mode 100644 index 00000000..07b346a8 --- /dev/null +++ b/frontend/src/app/run/run-async-admin-view/run-async-admin-view.component.spec.ts @@ -0,0 +1,159 @@ +import { NO_ERRORS_SCHEMA } from '@angular/core'; +import { ComponentFixture, TestBed, fakeAsync, tick, discardPeriodicTasks } from '@angular/core/testing'; +import { ActivatedRoute, Router } from '@angular/router'; +import { MatSnackBar } from '@angular/material/snack-bar'; +import { MatDialog } from '@angular/material/dialog'; +import { Observable, of, Subject } from 'rxjs'; + +import { RunAsyncAdminViewComponent } from './run-async-admin-view.component'; +import { AppConfig } from '../../app.config'; +import { WebSocketService } from '../../services/websocket.service'; +import { + DownloadService, + EvaluationAdministratorService, + EvaluationClientService, + EvaluationScoresService, + EvaluationService, +} from '../../../../openapi'; +import { IWsServerMessage } from '../../model/ws/ws-server-message.interface'; +import { ServerMessageType } from '../../model/ws/server-message-type.enum'; + +// ── Helpers ─────────────────────────────────────────────────────────────────── + +function wsMsg(type: ServerMessageType.ServerMessageTypeEnum): IWsServerMessage { + return { evaluationId: 'eval-1', type, timestamp: Date.now() }; +} + +const REFRESH_TYPES: ServerMessageType.ServerMessageTypeEnum[] = [ + ServerMessageType.ServerMessageTypeEnum.TASK_START, + ServerMessageType.ServerMessageTypeEnum.TASK_END, + ServerMessageType.ServerMessageTypeEnum.TASK_UPDATED, + ServerMessageType.ServerMessageTypeEnum.TASK_PREPARE, + ServerMessageType.ServerMessageTypeEnum.COMPETITION_UPDATE, + ServerMessageType.ServerMessageTypeEnum.COMPETITION_END, +]; + +const IRRELEVANT_TYPES: ServerMessageType.ServerMessageTypeEnum[] = [ + ServerMessageType.ServerMessageTypeEnum.PING, + ServerMessageType.ServerMessageTypeEnum.COMPETITION_START, +]; + +// ── Suite ───────────────────────────────────────────────────────────────────── + +describe('RunAsyncAdminViewComponent WebSocket wiring', () => { + let component: RunAsyncAdminViewComponent; + let fixture: ComponentFixture; + let wsService: jasmine.SpyObj; + let evaluationService: jasmine.SpyObj; + let runAdminService: jasmine.SpyObj; + let messages$: Subject; + + beforeEach(() => { + messages$ = new Subject(); + + wsService = jasmine.createSpyObj('WebSocketService', ['connect', 'disconnect'], { + messages$: messages$.asObservable(), + }); + + evaluationService = jasmine.createSpyObj('EvaluationService', ['getApiV2EvaluationByEvaluationIdInfo']); + evaluationService.getApiV2EvaluationByEvaluationIdInfo.and.returnValue( + of({ id: 'eval-1', templateId: 'tpl-1', name: 'Test', teams: [], taskTemplates: [{ templateId: 't1' }] } as any) + ); + + runAdminService = jasmine.createSpyObj('EvaluationAdministratorService', [ + 'getApiV2EvaluationAdminByEvaluationIdOverview', + 'getApiV2EvaluationAdminByEvaluationIdTaskPastList', + 'getApiV2EvaluationAdminByEvaluationIdSubmissionListByTemplateId', + ]); + runAdminService.getApiV2EvaluationAdminByEvaluationIdOverview.and.returnValue(of({} as any)); + runAdminService.getApiV2EvaluationAdminByEvaluationIdTaskPastList.and.returnValue(of([] as any)); + runAdminService.getApiV2EvaluationAdminByEvaluationIdSubmissionListByTemplateId.and.returnValue(of([] as any)); + + TestBed.configureTestingModule({ + declarations: [RunAsyncAdminViewComponent], + schemas: [NO_ERRORS_SCHEMA], + providers: [ + { provide: EvaluationService, useValue: evaluationService }, + { provide: EvaluationAdministratorService, useValue: runAdminService }, + { provide: EvaluationClientService, useValue: {} }, + { provide: EvaluationScoresService, useValue: {} }, + { provide: DownloadService, useValue: {} }, + { + provide: ActivatedRoute, + // Use a non-completing Observable: the component subscribes its `runId` + // BehaviorSubject directly to `params`, so a source like `of(...)` that + // completes synchronously would also complete (close) that subject. + useValue: { params: new Observable((subscriber) => subscriber.next({ runId: 'eval-1' })) }, + }, + { provide: AppConfig, useValue: { resolveApiUrl: (p: string) => `http://localhost${p}` } }, + { provide: MatSnackBar, useValue: jasmine.createSpyObj('MatSnackBar', ['open']) }, + { provide: Router, useValue: jasmine.createSpyObj('Router', ['navigate', 'navigateByUrl']) }, + { provide: MatDialog, useValue: {} }, + { provide: WebSocketService, useValue: wsService }, + ], + }); + + fixture = TestBed.createComponent(RunAsyncAdminViewComponent); + component = fixture.componentInstance; + }); + + // ── connect / disconnect ─────────────────────────────────────────────────── + + it('connects the WebSocket with the evaluationId on init', () => { + component.ngAfterViewInit(); + expect(wsService.connect).toHaveBeenCalledWith('eval-1'); + }); + + it('disconnects the WebSocket on destroy', () => { + component.ngAfterViewInit(); + fixture.destroy(); + expect(wsService.disconnect).toHaveBeenCalled(); + }); + + // ── refresh on WS message ────────────────────────────────────────────────── + + REFRESH_TYPES.forEach((type) => { + it(`refreshes the run overview when ${type} message is received`, fakeAsync(() => { + component.ngAfterViewInit(); + component.run.subscribe(); + tick(); + runAdminService.getApiV2EvaluationAdminByEvaluationIdOverview.calls.reset(); + + messages$.next(wsMsg(type)); + tick(); + + expect(runAdminService.getApiV2EvaluationAdminByEvaluationIdOverview).toHaveBeenCalledWith('eval-1'); + discardPeriodicTasks(); + })); + + it(`refreshes the task submission counts when ${type} message is received`, fakeAsync(() => { + component.ngAfterViewInit(); + component.taskSubmissionCounts.subscribe(); + tick(); + runAdminService.getApiV2EvaluationAdminByEvaluationIdSubmissionListByTemplateId.calls.reset(); + + messages$.next(wsMsg(type)); + tick(); + + expect(runAdminService.getApiV2EvaluationAdminByEvaluationIdSubmissionListByTemplateId).toHaveBeenCalledWith('eval-1', 't1'); + discardPeriodicTasks(); + })); + }); + + // ── PING / unrelated messages are ignored ────────────────────────────────── + + IRRELEVANT_TYPES.forEach((type) => { + it(`does not refresh the run overview for ${type} message`, fakeAsync(() => { + component.ngAfterViewInit(); + component.run.subscribe(); + tick(); + runAdminService.getApiV2EvaluationAdminByEvaluationIdOverview.calls.reset(); + + messages$.next(wsMsg(type)); + tick(); + + expect(runAdminService.getApiV2EvaluationAdminByEvaluationIdOverview).not.toHaveBeenCalled(); + discardPeriodicTasks(); + })); + }); +}); From 9e871a390ff24350a8b6cbf4379c63bff3f52439 Mon Sep 17 00:00:00 2001 From: mcmahj45 Date: Sun, 14 Jun 2026 19:57:33 +0100 Subject: [PATCH 08/10] Tests for judgement viewer aswell --- .../judgement-viewer.component.spec.ts | 123 +++++++++++++++++- 1 file changed, 122 insertions(+), 1 deletion(-) diff --git a/frontend/src/app/judgement/judgement-viewer.component.spec.ts b/frontend/src/app/judgement/judgement-viewer.component.spec.ts index e8a30eba..5fe7cf83 100644 --- a/frontend/src/app/judgement/judgement-viewer.component.spec.ts +++ b/frontend/src/app/judgement/judgement-viewer.component.spec.ts @@ -1,5 +1,5 @@ import { NO_ERRORS_SCHEMA } from '@angular/core'; -import { TestBed } from '@angular/core/testing'; +import { ComponentFixture, TestBed, fakeAsync, tick, discardPeriodicTasks } from '@angular/core/testing'; import { ActivatedRoute, Router } from '@angular/router'; import { MatSnackBar } from '@angular/material/snack-bar'; import { MatDialog, MatDialogRef } from '@angular/material/dialog'; @@ -8,11 +8,28 @@ import { JudgementViewerComponent } from './judgement-viewer.component'; import { JudgementService } from '../../../openapi'; import { ApiJudgementRequest } from '../../../openapi'; import { WebSocketService } from '../services/websocket.service'; +import { IWsServerMessage } from '../model/ws/ws-server-message.interface'; +import { ServerMessageType } from '../model/ws/server-message-type.enum'; function makeRequest(token: string, desc = 'Find a cat'): ApiJudgementRequest { return { token, taskDescription: desc, validator: 'v1', answerSet: { answers: [] } } as any; } +function wsMsg(type: ServerMessageType.ServerMessageTypeEnum): IWsServerMessage { + return { evaluationId: 'run1', type, timestamp: Date.now() }; +} + +const REFRESH_TYPES: ServerMessageType.ServerMessageTypeEnum[] = [ + ServerMessageType.ServerMessageTypeEnum.TASK_UPDATED, + ServerMessageType.ServerMessageTypeEnum.TASK_START, + ServerMessageType.ServerMessageTypeEnum.TASK_END, +]; + +const IRRELEVANT_TYPES: ServerMessageType.ServerMessageTypeEnum[] = [ + ServerMessageType.ServerMessageTypeEnum.PING, + ServerMessageType.ServerMessageTypeEnum.COMPETITION_UPDATE, +]; + describe('JudgementViewerComponent', () => { let component: JudgementViewerComponent; let mockDialog: jasmine.SpyObj; @@ -135,3 +152,107 @@ describe('JudgementViewerComponent', () => { }); }); }); + +// ── WebSocket wiring ──────────────────────────────────────────────────────── + +describe('JudgementViewerComponent WebSocket wiring', () => { + let component: JudgementViewerComponent; + let fixture: ComponentFixture; + let wsService: jasmine.SpyObj; + let judgementService: jasmine.SpyObj; + let messages$: Subject; + let dialogAfterClosed: Subject; + + beforeEach(() => { + messages$ = new Subject(); + dialogAfterClosed = new Subject(); + + wsService = jasmine.createSpyObj('WebSocketService', ['connect', 'disconnect'], { + messages$: messages$.asObservable(), + }); + + judgementService = jasmine.createSpyObj('JudgementService', [ + 'postApiV2EvaluationByEvaluationIdJudge', + 'getApiV2EvaluationByEvaluationIdJudgeNext', + 'getApiV2EvaluationByEvaluationIdJudgeStatus', + ]); + judgementService.getApiV2EvaluationByEvaluationIdJudgeStatus.and.returnValue(of([] as any)); + judgementService.getApiV2EvaluationByEvaluationIdJudgeNext.and.returnValue(of({ status: 202, body: null } as any)); + + const fakeDialogRef = { afterClosed: () => dialogAfterClosed.asObservable() } as MatDialogRef; + const mockDialog = jasmine.createSpyObj('MatDialog', ['open']); + mockDialog.open.and.returnValue(fakeDialogRef); + + TestBed.configureTestingModule({ + declarations: [JudgementViewerComponent], + schemas: [NO_ERRORS_SCHEMA], + providers: [ + { provide: JudgementService, useValue: judgementService }, + { provide: ActivatedRoute, useValue: { params: of({ runId: 'run1' }) } }, + { provide: MatSnackBar, useValue: jasmine.createSpyObj('MatSnackBar', ['open']) }, + { provide: Router, useValue: jasmine.createSpyObj('Router', ['navigate']) }, + { provide: MatDialog, useValue: mockDialog }, + { provide: WebSocketService, useValue: wsService }, + ], + }); + + fixture = TestBed.createComponent(JudgementViewerComponent); + component = fixture.componentInstance; + component.judgePlayer = jasmine.createSpyObj('JudgementMediaViewerComponent', ['stop', 'togglePlaying']); + }); + + // ── connect / disconnect ─────────────────────────────────────────────────── + + it('connects the WebSocket with the evaluationId on init', () => { + component.ngAfterViewInit(); + expect(wsService.connect).toHaveBeenCalledWith('run1'); + }); + + it('disconnects the WebSocket on destroy', () => { + component.ngAfterViewInit(); + fixture.destroy(); + expect(wsService.disconnect).toHaveBeenCalled(); + }); + + // ── judgement request / status refresh on WS message ─────────────────────── + + REFRESH_TYPES.forEach((type) => { + it(`fetches the next judgement request and status when ${type} message is received`, fakeAsync(() => { + component.ngAfterViewInit(); + dialogAfterClosed.next(); + tick(); + + judgementService.getApiV2EvaluationByEvaluationIdJudgeNext.calls.reset(); + judgementService.getApiV2EvaluationByEvaluationIdJudgeStatus.calls.reset(); + + messages$.next(wsMsg(type)); + tick(); + + expect(judgementService.getApiV2EvaluationByEvaluationIdJudgeNext.calls.mostRecent().args).toEqual(['run1', 'response']); + expect(judgementService.getApiV2EvaluationByEvaluationIdJudgeStatus).toHaveBeenCalledWith('run1'); + + discardPeriodicTasks(); + })); + }); + + // ── PING / unrelated messages are ignored ────────────────────────────────── + + IRRELEVANT_TYPES.forEach((type) => { + it(`does not fetch the next judgement request for ${type} message`, fakeAsync(() => { + component.ngAfterViewInit(); + dialogAfterClosed.next(); + tick(); + + judgementService.getApiV2EvaluationByEvaluationIdJudgeNext.calls.reset(); + judgementService.getApiV2EvaluationByEvaluationIdJudgeStatus.calls.reset(); + + messages$.next(wsMsg(type)); + tick(); + + expect(judgementService.getApiV2EvaluationByEvaluationIdJudgeNext).not.toHaveBeenCalled(); + expect(judgementService.getApiV2EvaluationByEvaluationIdJudgeStatus).not.toHaveBeenCalled(); + + discardPeriodicTasks(); + })); + }); +}); From c165545c8ad1a820e6682ba3e0328f9f1b41d943 Mon Sep 17 00:00:00 2001 From: mcmahj45 Date: Sun, 14 Jun 2026 21:16:22 +0100 Subject: [PATCH 09/10] Fixed a self deadlock issue --- .../main/kotlin/dev/dres/run/RunExecutor.kt | 61 +++++++------------ 1 file changed, 21 insertions(+), 40 deletions(-) diff --git a/backend/src/main/kotlin/dev/dres/run/RunExecutor.kt b/backend/src/main/kotlin/dev/dres/run/RunExecutor.kt index 27123259..ec0f9cbc 100644 --- a/backend/src/main/kotlin/dev/dres/run/RunExecutor.kt +++ b/backend/src/main/kotlin/dev/dres/run/RunExecutor.kt @@ -47,10 +47,7 @@ object RunExecutor : StreamEventHandler { private val connectedClients = ConcurrentHashMap() /** Session IDs currently observing each evaluation. */ - private val observingClients = HashMap>() - - /** Lock for WebSocket client data structures. */ - private val clientLock = ReentrantReadWriteLock() + private val observingClients = ConcurrentHashMap>() /** Lock for accessing and changing all data structures related to [RunManager]s. */ private val runManagerLock = ReentrantReadWriteLock() @@ -95,26 +92,26 @@ object RunExecutor : StreamEventHandler { throw IllegalArgumentException("This RunExecutor already runs a RunManager with the given ID ${manager.id}. The same RunManager cannot be executed twice!") } this.runManagers[manager.id] = manager - this.observingClients[manager.id] = HashSet() + this.observingClients[manager.id] = ConcurrentHashMap.newKeySet() this.results[this.executor.submit(manager)] = manager.id } /** A thread that cleans after [RunManager]s have finished. */ private val cleanerThread = Thread { while (!this@RunExecutor.executor.isShutdown) { - this@RunExecutor.runManagerLock.read { - this@RunExecutor.results.entries.removeIf { entry -> - val k = entry.key - val v = entry.value - if (k.isDone || k.isCancelled) { + /* Determine which futures have finished without holding the write lock. */ + val finished = this@RunExecutor.runManagerLock.read { + this@RunExecutor.results.entries.filter { (k, _) -> k.isDone || k.isCancelled } + } + if (finished.isNotEmpty()) { + /* Acquiring the write lock separately (rather than upgrading from the read lock above, + * which ReentrantReadWriteLock does not support and would deadlock). */ + this@RunExecutor.runManagerLock.write { + finished.forEach { (k, v) -> logger.info("RunManager $v (done = ${k.isDone}, cancelled = ${k.isCancelled}) will be removed!") - this@RunExecutor.runManagerLock.write { - this@RunExecutor.runManagers.remove(v) - this@RunExecutor.observingClients.remove(v) - } - true - } else { - false + this@RunExecutor.results.remove(k) + this@RunExecutor.runManagers.remove(v) + this@RunExecutor.observingClients.remove(v) } } } @@ -136,19 +133,13 @@ object RunExecutor : StreamEventHandler { */ fun accept(ws: WsConfig) { ws.onConnect { ctx -> - this.clientLock.write { - this.connectedClients[ctx.sessionId()] = ctx - } + this.connectedClients[ctx.sessionId()] = ctx logger.debug("WebSocket client connected: ${ctx.sessionId()}") } ws.onClose { ctx -> - this.clientLock.write { - this.connectedClients.remove(ctx.sessionId()) - this.runManagerLock.read { - this.observingClients.values.forEach { it.remove(ctx.sessionId()) } - } - } + this.connectedClients.remove(ctx.sessionId()) + this.observingClients.values.forEach { it.remove(ctx.sessionId()) } logger.debug("WebSocket client disconnected: ${ctx.sessionId()}") } @@ -164,12 +155,8 @@ object RunExecutor : StreamEventHandler { if (this.runManagers.containsKey(message.evaluationId)) { when (message.type) { ClientMessageType.ACK -> {} - ClientMessageType.REGISTER -> this.clientLock.write { - this.observingClients[message.evaluationId]?.add(ctx.sessionId()) - } - ClientMessageType.UNREGISTER -> this.clientLock.write { - this.observingClients[message.evaluationId]?.remove(ctx.sessionId()) - } + ClientMessageType.REGISTER -> this.observingClients[message.evaluationId]?.add(ctx.sessionId()) + ClientMessageType.UNREGISTER -> this.observingClients[message.evaluationId]?.remove(ctx.sessionId()) ClientMessageType.PING -> ctx.send( mapper.writeValueAsString(ServerMessage(message.evaluationId, ServerMessageType.PING)) ) @@ -180,9 +167,7 @@ object RunExecutor : StreamEventHandler { ws.onError { ctx -> logger.error("WebSocket error for session ${ctx.sessionId()}: ${ctx.error()?.message}") - this.clientLock.write { - this.connectedClients.remove(ctx.sessionId()) - } + this.connectedClients.remove(ctx.sessionId()) } } @@ -198,11 +183,7 @@ object RunExecutor : StreamEventHandler { logger.error("Failed to serialize ServerMessage: ${e.message}") return } - val observers = this.clientLock.read { - this.runManagerLock.read { - this.observingClients[message.evaluationId]?.toSet() ?: emptySet() - } - } + val observers = this.observingClients[message.evaluationId]?.toSet() ?: emptySet() observers.forEach { sessionId -> try { this.connectedClients[sessionId]?.send(json) From e5a69d6cdf15eae5e3e76b537cfc491bd63d5846 Mon Sep 17 00:00:00 2001 From: mcmahj45 Date: Sun, 14 Jun 2026 21:46:18 +0100 Subject: [PATCH 10/10] Fixing some websocket auth issues --- .../kotlin/dev/dres/api/rest/AccessManager.kt | 17 ++++++++++++++ .../main/kotlin/dev/dres/run/RunExecutor.kt | 13 +++++++++-- .../utilities/extensions/ContextExtensions.kt | 7 ++++++ .../src/app/services/websocket.service.ts | 22 ++++++++++++++++++- 4 files changed, 56 insertions(+), 3 deletions(-) diff --git a/backend/src/main/kotlin/dev/dres/api/rest/AccessManager.kt b/backend/src/main/kotlin/dev/dres/api/rest/AccessManager.kt index 784fac71..8b737474 100644 --- a/backend/src/main/kotlin/dev/dres/api/rest/AccessManager.kt +++ b/backend/src/main/kotlin/dev/dres/api/rest/AccessManager.kt @@ -165,4 +165,21 @@ object AccessManager { fun getRunManagerForUser(userId: UserId): Set = this.locks.read { return this.usersToRunMap[userId] ?: emptySet() } + + /** + * Checks whether the user associated with the given [SessionToken] is allowed to observe + * (e.g., via WebSocket) the given [runManager], mirroring the access rules applied to the REST API. + * + * @param sessionId The [SessionToken] to check. + * @param runManager The [RunManager] the session wants to observe. + * @return True if access is permitted, false otherwise. + */ + fun canViewEvaluation(sessionId: SessionToken?, runManager: RunManager): Boolean { + val roles = rolesOfSession(sessionId) + if (roles.contains(ApiRole.ADMIN)) return true + val userId = userIdForSession(sessionId) ?: return false + return (roles.contains(ApiRole.JUDGE) && runManager.template.judges.contains(userId)) || + (roles.contains(ApiRole.VIEWER) && runManager.template.viewers.contains(userId)) || + (roles.contains(ApiRole.PARTICIPANT) && runManager.template.hasParticipant(userId)) + } } diff --git a/backend/src/main/kotlin/dev/dres/run/RunExecutor.kt b/backend/src/main/kotlin/dev/dres/run/RunExecutor.kt index ec0f9cbc..0222a802 100644 --- a/backend/src/main/kotlin/dev/dres/run/RunExecutor.kt +++ b/backend/src/main/kotlin/dev/dres/run/RunExecutor.kt @@ -1,6 +1,7 @@ package dev.dres.run import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper +import dev.dres.api.rest.AccessManager import dev.dres.api.rest.types.ViewerInfo import dev.dres.api.rest.types.evaluation.websocket.ClientMessage import dev.dres.api.rest.types.evaluation.websocket.ClientMessageType @@ -10,6 +11,7 @@ import dev.dres.data.model.run.* import dev.dres.data.model.run.interfaces.EvaluationId import dev.dres.run.eventstream.* import dev.dres.run.validation.interfaces.JudgementValidator +import dev.dres.utilities.extensions.sessionToken import io.javalin.websocket.WsConfig import io.javalin.websocket.WsContext import jetbrains.exodus.database.TransientEntityStore @@ -152,10 +154,17 @@ object RunExecutor : StreamEventHandler { } logger.debug("Received WebSocket message: $message from ${ctx.sessionId()}") this.runManagerLock.read { - if (this.runManagers.containsKey(message.evaluationId)) { + val manager = this.runManagers[message.evaluationId] + if (manager != null) { when (message.type) { ClientMessageType.ACK -> {} - ClientMessageType.REGISTER -> this.observingClients[message.evaluationId]?.add(ctx.sessionId()) + ClientMessageType.REGISTER -> { + if (AccessManager.canViewEvaluation(ctx.sessionToken(), manager)) { + this.observingClients[message.evaluationId]?.add(ctx.sessionId()) + } else { + logger.warn("WebSocket session ${ctx.sessionId()} is not permitted to observe evaluation ${message.evaluationId}") + } + } ClientMessageType.UNREGISTER -> this.observingClients[message.evaluationId]?.remove(ctx.sessionId()) ClientMessageType.PING -> ctx.send( mapper.writeValueAsString(ServerMessage(message.evaluationId, ServerMessageType.PING)) diff --git a/backend/src/main/kotlin/dev/dres/utilities/extensions/ContextExtensions.kt b/backend/src/main/kotlin/dev/dres/utilities/extensions/ContextExtensions.kt index 67650ea3..9297435b 100644 --- a/backend/src/main/kotlin/dev/dres/utilities/extensions/ContextExtensions.kt +++ b/backend/src/main/kotlin/dev/dres/utilities/extensions/ContextExtensions.kt @@ -11,6 +11,7 @@ import dev.dres.run.RunExecutor import dev.dres.run.RunManager import dev.dres.run.RunManagerStatus import io.javalin.http.Context +import io.javalin.websocket.WsContext import kotlinx.dnq.query.filter import kotlinx.dnq.query.flatMapDistinct import kotlinx.dnq.query.isNotEmpty @@ -72,6 +73,12 @@ fun Context.sendFile(file: File) { fun Context.sessionToken(): String? = this.attribute("session") +/** + * Returns the session token for this [WsContext], as set by the `before` handler on the + * underlying HTTP upgrade request. + */ +fun WsContext.sessionToken(): String? = this.attribute("session") + fun Context.getOrCreateSessionToken(): String { val attributeId = this.attribute("session") if (attributeId != null) { diff --git a/frontend/src/app/services/websocket.service.ts b/frontend/src/app/services/websocket.service.ts index 0b9798fa..3ae22e25 100644 --- a/frontend/src/app/services/websocket.service.ts +++ b/frontend/src/app/services/websocket.service.ts @@ -9,10 +9,16 @@ import { ClientMessageType } from '../model/ws/client-message-type.enum'; providedIn: 'root', }) export class WebSocketService implements OnDestroy { + /** Base delay for the first reconnect attempt. */ + private static readonly RECONNECT_BASE_DELAY_MS = 1_000; + /** Upper bound for the reconnect delay, reached after repeated failures. */ + private static readonly RECONNECT_MAX_DELAY_MS = 30_000; + private socket: WebSocket | null = null; private messageSubject = new Subject(); private pingSubscription: Subscription | null = null; private reconnectTimeout: ReturnType | null = null; + private reconnectAttempts = 0; private currentEvaluationId: string | null = null; /** Observable stream of messages pushed by the server. */ @@ -44,6 +50,7 @@ export class WebSocketService implements OnDestroy { this.socket = new WebSocket(url); this.socket.onopen = () => { + this.reconnectAttempts = 0; this.send({ evaluationId: this.currentEvaluationId, type: ClientMessageType.ClientMessageTypeEnum.REGISTER, @@ -63,7 +70,8 @@ export class WebSocketService implements OnDestroy { this.socket.onclose = () => { this.stopPing(); if (this.currentEvaluationId) { - this.reconnectTimeout = setTimeout(() => this.openConnection(), 5000); + const delay = this.nextReconnectDelay(); + this.reconnectTimeout = setTimeout(() => this.openConnection(), delay); } }; @@ -79,6 +87,7 @@ export class WebSocketService implements OnDestroy { } private closeSocket(): void { + this.reconnectAttempts = 0; if (this.reconnectTimeout !== null) { clearTimeout(this.reconnectTimeout); this.reconnectTimeout = null; @@ -96,6 +105,17 @@ export class WebSocketService implements OnDestroy { } } + /** + * Computes the delay before the next reconnect attempt using exponential backoff with jitter, + * capped at {@link RECONNECT_MAX_DELAY_MS}. + */ + private nextReconnectDelay(): number { + const exponentialDelay = WebSocketService.RECONNECT_BASE_DELAY_MS * 2 ** this.reconnectAttempts; + const delay = Math.min(exponentialDelay, WebSocketService.RECONNECT_MAX_DELAY_MS); + this.reconnectAttempts++; + return delay / 2 + Math.random() * (delay / 2); + } + private startPing(): void { this.pingSubscription = interval(30_000).subscribe(() => { if (this.currentEvaluationId) {