# #region Test.AppModule.WsEvents [C:3] [TYPE Module] [SEMANTICS test,app,ws,events,maintenance,translate,dataset] # @BRIEF Tests for app.py — task_events_websocket, maintenance_events_websocket, dataset_websocket_endpoint, translate_run_websocket. # @RELATION BINDS_TO -> [AppModule] import asyncio from pathlib import Path import sys sys.path.insert(0, str(Path(__file__).parent.parent / "src")) import pytest from unittest.mock import AsyncMock, MagicMock, patch from starlette.websockets import WebSocketDisconnect class TestTaskEventsWebSocket: """task_events_websocket — global task events stream.""" # #region test_auth_rejected [C:2] [TYPE Function] @pytest.mark.asyncio async def test_auth_rejected(self): from src.app import task_events_websocket ws = MagicMock(); ws.query_params = {}; ws.close = AsyncMock() with patch("src.app._authenticate_websocket", return_value=False): await task_events_websocket(ws) ws.close.assert_called_once_with(code=4001, reason="Authentication required") ws.accept.assert_not_called() # #endregion test_auth_rejected # #region test_accepts_and_streams [C:2] [TYPE Function] @pytest.mark.asyncio async def test_accepts_and_streams(self): from src.app import task_events_websocket ws = MagicMock(); ws.query_params = {"token": "valid"} ws.send_json = AsyncMock(side_effect=[None, WebSocketDisconnect()]) ws.accept = AsyncMock() q = asyncio.Queue() await q.put({"type": "task_status", "task_id": "t1", "task": {"status": "RUNNING"}}) await q.put({"type": "task_status", "task_id": "t1", "task": {"status": "COMPLETED"}}) async def sub(): return q with ( patch("src.app._authenticate_websocket", return_value=True), patch("src.app.get_task_manager") as mg, ): tm = MagicMock(); tm.subscribe_task_events = sub; tm.unsubscribe_task_events = MagicMock() mg.return_value = tm await task_events_websocket(ws) ws.accept.assert_called_once() ws.send_json.assert_any_call({"type": "task_status", "task_id": "t1", "task": {"status": "RUNNING"}}) tm.unsubscribe_task_events.assert_called_once() # #endregion test_accepts_and_streams class TestMaintenanceEventsWebSocket: """maintenance_events_websocket — maintenance event stream.""" # #region test_auth_rejected [C:2] [TYPE Function] @pytest.mark.asyncio async def test_auth_rejected(self): from src.app import maintenance_events_websocket ws = MagicMock(); ws.query_params = {}; ws.close = AsyncMock() with patch("src.app._authenticate_websocket", return_value=False): await maintenance_events_websocket(ws) ws.close.assert_called_once_with(code=4001, reason="Authentication required") # #endregion test_auth_rejected # #region test_accepts [C:2] [TYPE Function] @pytest.mark.asyncio async def test_accepts(self): from src.app import maintenance_events_websocket ws = MagicMock(); ws.query_params = {"token": "valid"} ws.send_json = AsyncMock(side_effect=[None, WebSocketDisconnect()]) ws.accept = AsyncMock() q = asyncio.Queue() await q.put({"type": "maintenance.event_created", "maintenance_id": "m-1"}) await q.put({"type": "maintenance.event_ended", "maintenance_id": "m-1"}) async def sub(): return q with ( patch("src.app._authenticate_websocket", return_value=True), patch("src.app.get_task_manager") as mg, ): tm = MagicMock(); tm.subscribe_maintenance_events = sub; tm.unsubscribe_maintenance_events = MagicMock() mg.return_value = tm await maintenance_events_websocket(ws) ws.accept.assert_called_once() evt = ws.send_json.call_args_list[0][0][0] assert evt.get("type") == "maintenance.event_created" tm.unsubscribe_maintenance_events.assert_called_once() # #endregion test_accepts class TestDatasetWebSocket: """dataset_websocket_endpoint — dataset.updated event stream.""" # #region test_auth_rejected [C:2] [TYPE Function] @pytest.mark.asyncio async def test_auth_rejected(self): from src.app import dataset_websocket_endpoint ws = MagicMock(); ws.query_params = {}; ws.close = AsyncMock() with patch("src.app._authenticate_websocket", return_value=False): await dataset_websocket_endpoint(ws, "env-1") ws.close.assert_called_once_with(code=4001, reason="Authentication required") # #endregion test_auth_rejected # #region test_accepts [C:2] [TYPE Function] @pytest.mark.asyncio async def test_accepts(self): from src.app import dataset_websocket_endpoint ws = MagicMock(); ws.query_params = {"token": "valid"} ws.send_json = AsyncMock(side_effect=[None, WebSocketDisconnect()]) ws.accept = AsyncMock() q = asyncio.Queue() await q.put({"type": "dataset.updated", "dataset_id": "ds-1"}) await q.put({"type": "dataset.updated", "dataset_id": "ds-2"}) async def sub(e): return q with ( patch("src.app._authenticate_websocket", return_value=True), patch("src.app.get_task_manager") as mg, ): tm = MagicMock(); tm.subscribe_dataset_events = sub; tm.unsubscribe_dataset_events = MagicMock() mg.return_value = tm await dataset_websocket_endpoint(ws, "env-1") ws.accept.assert_called_once() tm.unsubscribe_dataset_events.assert_called_once() # #endregion test_accepts class TestTranslateRunWebSocket: """translate_run_websocket — translation run progress stream.""" # #region test_auth_rejected [C:2] [TYPE Function] @pytest.mark.asyncio async def test_auth_rejected(self): from src.app import translate_run_websocket ws = MagicMock(); ws.query_params = {}; ws.close = AsyncMock() with patch("src.app._authenticate_websocket", return_value=False): await translate_run_websocket(ws, "run-1") ws.close.assert_called_once_with(code=4001, reason="Authentication required") # #endregion test_auth_rejected # #region test_accepts_and_streams [C:2] [TYPE Function] @pytest.mark.asyncio async def test_accepts_and_streams(self): from src.app import translate_run_websocket ws = MagicMock(); ws.query_params = {"token": "valid"} ws.send_json = AsyncMock(); ws.accept = AsyncMock() with ( patch("src.app._authenticate_websocket", return_value=True), patch("src.core.database.SessionLocal") as msl, patch("src.plugins.translate.orchestrator_aggregator.TranslationResultAggregator") as ma, patch("src.plugins.translate.events.TranslationEventLog"), ): msl.return_value = MagicMock() agg = MagicMock() agg.get_run_status.return_value = { "status": "COMPLETED", "run_id": "run-1", "total_records": 100, "successful_records": 100, "failed_records": 0, "skipped_records": 0, } ma.return_value = agg await translate_run_websocket(ws, "run-1") ws.accept.assert_called_once() sent = ws.send_json.call_args[0][0] assert sent.get("status") == "COMPLETED" assert sent.get("progressPct") == 100 # #endregion test_accepts_and_streams # #region test_error_tick [C:2] [TYPE Function] @pytest.mark.asyncio async def test_error_tick(self): from src.app import translate_run_websocket ws = MagicMock(); ws.query_params = {"token": "valid"} ws.send_json = AsyncMock(); ws.accept = AsyncMock() with ( patch("src.app._authenticate_websocket", return_value=True), patch("src.core.database.SessionLocal", side_effect=Exception("DB failed")), ): await translate_run_websocket(ws, "run-1") ws.accept.assert_called_once() sent = ws.send_json.call_args[0][0] assert "error" in sent # #endregion test_error_tick # #endregion Test.AppModule.WsEvents