lda
Verified
038e152bbc
Checks / tests (ubuntu-latest, 3.11) (push) Canceled after 0s
Checks / tests (ubuntu-latest, 3.14) (push) Canceled after 0s
Checks / tests (windows-latest, 3.11) (push) Canceled after 0s
Checks / tests (windows-latest, 3.14) (push) Canceled after 0s
Checks / package (push) Canceled after 0s
89 lines
4.0 KiB
Python
89 lines
4.0 KiB
Python
"""Verify direct connections, write completion, and disconnect handling."""
|
|
|
|
import socket
|
|
import types
|
|
import unittest
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from controller.transport import RfcommStream, open_stream
|
|
|
|
|
|
class TransportTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.mock_socket = MagicMock()
|
|
# Replace this adapter's socket module reference rather than globally
|
|
# mocking socket.socket, which also breaks asyncio's internal sockets
|
|
# (particularly Windows' Proactor event loop).
|
|
isolated_socket = types.SimpleNamespace(
|
|
socket=self.mock_socket,
|
|
AF_BLUETOOTH=32,
|
|
BTPROTO_RFCOMM=3,
|
|
SOCK_STREAM=socket.SOCK_STREAM,
|
|
timeout=TimeoutError,
|
|
)
|
|
replacement = patch("controller.transport.socket", isolated_socket)
|
|
replacement.start()
|
|
self.addCleanup(replacement.stop)
|
|
self.peer = self.mock_socket.return_value
|
|
|
|
def test_direct_address_bypasses_serial_and_closes(self):
|
|
with patch.dict("sys.modules", {"serial": None}):
|
|
with open_stream("12:34:56:78:9A:BC", 1) as stream:
|
|
self.assertEqual(stream.write(b"packet"), 6)
|
|
self.peer.connect.assert_called_once_with(("12:34:56:78:9A:BC", 1))
|
|
self.peer.sendall.assert_called_once_with(b"packet")
|
|
self.peer.close.assert_called_once()
|
|
|
|
def test_idle_is_not_disconnect(self):
|
|
with RfcommStream("12:34:56:78:9A:BC") as stream:
|
|
self.peer.recv.side_effect = TimeoutError()
|
|
self.assertEqual(stream.read(1024), b"")
|
|
self.peer.recv.side_effect = None
|
|
self.peer.recv.return_value = b""
|
|
with self.assertRaises(ConnectionError):
|
|
stream.read(1024)
|
|
|
|
def test_connect_failure_closes_socket(self):
|
|
self.peer.connect.side_effect = OSError("connection refused")
|
|
with self.assertRaises(OSError):
|
|
with RfcommStream("12:34:56:78:9A:BC"):
|
|
self.fail("Failed connection entered context")
|
|
self.peer.close.assert_called_once()
|
|
|
|
def test_connect_timeout_identifies_phase_and_closes(self):
|
|
self.peer.connect.side_effect = TimeoutError("timed out")
|
|
with self.assertRaisesRegex(TimeoutError, "before the Samsung handshake"):
|
|
with RfcommStream("12:34:56:78:9A:BC"):
|
|
self.fail("Timed out connection entered context")
|
|
self.peer.close.assert_called_once()
|
|
|
|
def test_write_timeout_identifies_phase(self):
|
|
with RfcommStream("12:34:56:78:9A:BC") as stream:
|
|
self.peer.sendall.side_effect = TimeoutError("timed out")
|
|
with self.assertRaisesRegex(TimeoutError, "sending a Samsung packet"):
|
|
stream.write(b"packet")
|
|
|
|
def test_prepared_connection_retains_windows_handle(self):
|
|
device = MagicMock()
|
|
prepare = AsyncMock(return_value=device)
|
|
report = MagicMock()
|
|
with patch("controller.windows_bluetooth.prepare_device", prepare):
|
|
with open_stream("12:34:56:78:9A:BC", 1, prepare=True, report=report):
|
|
device.close.assert_not_called()
|
|
prepare.assert_awaited_once_with("12:34:56:78:9A:BC", report)
|
|
self.peer.connect.assert_called_once_with(("12:34:56:78:9A:BC", 1))
|
|
device.close.assert_called_once()
|
|
|
|
def test_prepared_connect_failure_releases_both_handles(self):
|
|
device = MagicMock()
|
|
# The base adapter's socket cleanup is covered separately. Mock its
|
|
# entry here so asyncio's own internal sockets do not affect counts.
|
|
with (
|
|
patch("controller.windows_bluetooth.prepare_device", AsyncMock(return_value=device)),
|
|
patch.object(RfcommStream, "__enter__", side_effect=OSError("host down")),
|
|
):
|
|
with self.assertRaises(OSError):
|
|
with open_stream("12:34:56:78:9A:BC", 1, prepare=True):
|
|
self.fail("Failed connection entered context")
|
|
device.close.assert_called_once()
|