77 lines
2.4 KiB
Python
77 lines
2.4 KiB
Python
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
from PySide6.QtCore import QByteArray
|
|
from PySide6.QtNetwork import QNetworkReply
|
|
|
|
from aare.gui.threads.sse_client import SSEClient
|
|
|
|
|
|
@pytest.fixture
|
|
def sse_client(qtbot):
|
|
client = SSEClient()
|
|
return client
|
|
|
|
|
|
def test_sse_client_init(sse_client):
|
|
assert not sse_client.is_connected()
|
|
assert sse_client._reconnect_delay == 1000
|
|
|
|
|
|
def test_sse_client_parse_message(sse_client, qtbot):
|
|
# Test simple message
|
|
with qtbot.wait_signal(sse_client.message_received) as blocker:
|
|
sse_client._parse_sse_line("data: hello")
|
|
sse_client._parse_sse_line("")
|
|
assert blocker.args == ["hello"]
|
|
|
|
|
|
def test_sse_client_parse_event(sse_client, qtbot):
|
|
# Test event with data
|
|
with qtbot.wait_signal(sse_client.event_received) as blocker:
|
|
sse_client._parse_sse_line("event: update")
|
|
sse_client._parse_sse_line("data: some data")
|
|
sse_client._parse_sse_line("")
|
|
assert blocker.args == ["update", "some data"]
|
|
|
|
|
|
def test_sse_client_multiline_data(sse_client, qtbot):
|
|
with qtbot.wait_signal(sse_client.message_received) as blocker:
|
|
sse_client._parse_sse_line("data: line1")
|
|
sse_client._parse_sse_line("data: line2")
|
|
sse_client._parse_sse_line("")
|
|
assert blocker.args == ["line1\nline2"]
|
|
|
|
|
|
def test_sse_client_buffer_processing(sse_client, qtbot):
|
|
sse_client._buffer = QByteArray(b"data: chunk1\n\n")
|
|
with qtbot.wait_signal(sse_client.message_received, timeout=1000) as blocker:
|
|
sse_client._process_buffer()
|
|
assert blocker.args == ["chunk1"]
|
|
|
|
sse_client._buffer = QByteArray(b"data: chunk2\n\n")
|
|
with qtbot.wait_signal(sse_client.message_received, timeout=1000) as blocker:
|
|
sse_client._process_buffer()
|
|
assert blocker.args == ["chunk2"]
|
|
|
|
|
|
def test_sse_client_retry_parsing(sse_client):
|
|
sse_client._parse_sse_line("retry: 5000")
|
|
assert sse_client._reconnect_delay == 5000
|
|
|
|
sse_client._parse_sse_line("retry: invalid")
|
|
assert sse_client._reconnect_delay == 5000
|
|
|
|
|
|
def test_sse_client_disconnect(sse_client, qtbot):
|
|
# Mock a reply
|
|
mock_reply = MagicMock(spec=QNetworkReply)
|
|
sse_client._reply = mock_reply
|
|
sse_client._connected = True
|
|
|
|
with qtbot.wait_signal(sse_client.disconnected):
|
|
sse_client.disconnect_from_sse()
|
|
|
|
assert not sse_client._connected
|
|
mock_reply.abort.assert_called_once()
|