Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 10 additions & 2 deletions at_client/connections/atconnection.py
Original file line number Diff line number Diff line change
Expand Up @@ -98,8 +98,16 @@ def disconnect(self):
"""
Close the socket connection.
"""
self._secure_root_socket.close()
self._connected = False
# Always clear _connected, even if close() fails on an already-broken socket
# (e.g. "Bad file descriptor"). Otherwise _connected stays True and the monitor
# restart path (guarded by `if not self._connected`) never rebuilds the socket,
# so the client silently stops receiving notifications until it is recreated.
try:
self._secure_root_socket.close()
except Exception:
pass
finally:
self._connected = False

@abstractmethod
def parse_raw_response(self, raw_response:str) -> Response:
Expand Down
35 changes: 35 additions & 0 deletions test/disconnect_test.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
import unittest
from unittest.mock import MagicMock

from at_client.connections.atconnection import AtConnection


class _ConcreteConnection(AtConnection):
"""Minimal concrete subclass so we can exercise the base disconnect()."""

def parse_raw_response(self, raw_response):
return None


class DisconnectTest(unittest.TestCase):
"""disconnect() must always clear _connected, even if the socket close fails."""

def test_disconnect_clears_connected_on_clean_close(self):
conn = _ConcreteConnection.__new__(_ConcreteConnection) # bypass network __init__
conn._connected = True
conn._secure_root_socket = MagicMock()
conn.disconnect()
conn._secure_root_socket.close.assert_called_once()
self.assertFalse(conn._connected)

def test_disconnect_clears_connected_even_when_close_raises(self):
conn = _ConcreteConnection.__new__(_ConcreteConnection)
conn._connected = True
conn._secure_root_socket = MagicMock()
conn._secure_root_socket.close.side_effect = OSError(9, "Bad file descriptor")
conn.disconnect() # must not propagate
self.assertFalse(conn._connected) # so the restart path can rebuild the socket


if __name__ == "__main__":
unittest.main()