1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
|
"""Tests for SSL context passthrough in C4Websocket."""
from unittest.mock import AsyncMock, MagicMock, patch
import aiohttp
import socketio
import pytest
from pyControl4.websocket import C4Websocket
@pytest.mark.asyncio
async def test_sio_connect_without_session():
"""Test that sio_connect uses ssl_verify=False without http_session when
no session is provided."""
ws = C4Websocket("192.168.1.1")
with patch.object(
socketio.AsyncClient, "__init__", return_value=None
) as mock_init, patch.object(
socketio.AsyncClient, "register_namespace"
), patch.object(
socketio.AsyncClient, "connect", new_callable=AsyncMock
):
await ws.sio_connect("test-token")
mock_init.assert_called_once_with(ssl_verify=False)
@pytest.mark.asyncio
async def test_sio_connect_with_session():
"""Test that sio_connect creates a new session sharing the caller's
connector and passes it as http_session, so engineio can safely close
it without affecting the caller's session."""
mock_connector = MagicMock()
mock_session = MagicMock()
mock_session.connector = mock_connector
ws = C4Websocket("192.168.1.1", session_no_verify_ssl=mock_session)
with patch.object(
socketio.AsyncClient, "__init__", return_value=None
) as mock_init, patch.object(
socketio.AsyncClient, "register_namespace"
), patch.object(
socketio.AsyncClient, "connect", new_callable=AsyncMock
), patch.object(
aiohttp, "ClientSession"
) as mock_session_cls:
mock_http_session = MagicMock()
mock_session_cls.return_value = mock_http_session
await ws.sio_connect("test-token")
mock_session_cls.assert_called_once_with(
connector=mock_connector, connector_owner=False
)
mock_init.assert_called_once_with(
ssl_verify=False, http_session=mock_http_session
)
|
