test(terminal): prove the WS ownership refusal through the real route (#14960)

autobot-backend/api/terminal_websocket_route_test.py

@@ -40,6 +40,7 @@
 """
 
 import json
+import logging
 import os
 import subprocess
 import sys
@@ -50,6 +51,7 @@
 import pytest
 from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request, WebSocket
 from starlette.testclient import TestClient
+from starlette.websockets import WebSocketDisconnect
 
 from api.terminal import check_admin_permission
 from api.terminal import router as terminal_router
@@ -350,8 +352,6 @@ def test_unauthenticated_caller_is_refused_not_500(self, terminal_client, owned_
         for EVERY caller, admin or not. After the fix, an unauthenticated
         caller reaches the explicit per-route guard and is refused cleanly.
         """
-        from starlette.websockets import WebSocketDisconnect
-
         with _no_ws_credentials(), _accept_spy() as calls:
             with pytest.raises(WebSocketDisconnect) as exc_info:
                 with terminal_client.websocket_connect(f"/ws/{owned_session_id}"):
@@ -360,6 +360,23 @@ def test_unauthenticated_caller_is_refused_not_500(self, terminal_client, owned_
         assert exc_info.value.code == 1008
         assert len(calls) == 0
 
+    def test_authenticated_non_owner_is_refused(self, terminal_client, owned_session_id, caplog):
+        """#14960 AC2 at route level: a non-owner is refused 1008 before accept().
+        caplog pins the ownership branch; #14961's unknown-session branch also closes 1008."""
+        with (
+            patch("auth_middleware.authenticate_websocket", new=AsyncMock(return_value={"username": "mallory"})),
+            caplog.at_level(logging.WARNING, logger="api.terminal"),
+            _accept_spy() as calls,
+        ):
+            with pytest.raises(WebSocketDisconnect) as exc_info:
+                with terminal_client.websocket_connect(f"/ws/{owned_session_id}"):
+                    pass
+
+        assert exc_info.value.code == 1008
+        assert len(calls) == 0
+        assert any("is not the owner" in r.message for r in caplog.records)
+        assert not any("unknown session_id" in r.message for r in caplog.records)
+
 
 class TestSshTerminalWebsocketRouteHandshake:
     """Real handshake on /ws/ssh/{host_id} via TestClient -- not a handler call."""
@@ -383,8 +400,6 @@ def test_authenticated_admin_connects(self, terminal_client):
         assert len(calls) == 1
 
     def test_unauthenticated_caller_is_refused_not_500(self, terminal_client):
-        from starlette.websockets import WebSocketDisconnect
-
         with _no_ws_credentials(), _accept_spy() as calls:
             with pytest.raises(WebSocketDisconnect) as exc_info:
                 with terminal_client.websocket_connect("/ws/ssh/prod-host-1"):