Wait on _ever_connected in connect(), not _connected

pylutron/__init__.py

@@ -110,16 +110,26 @@ def __init__(self, host: str, user: str, password: str, recv_callback: Callable[
     self.daemon = True
 
   def connect(self) -> None:
-    """Connects to the lutron controller."""
+    """Connects to the lutron controller.
+
+    Returns once the initial connection has succeeded, or raises if it could
+    not be established. Returning means we got connected at least once, not
+    that the connection is up right now -- the session may already have dropped
+    and be in the reconnect loop. That is the only guarantee available here,
+    since the controller can close the session at any moment.
+    """
     if self._connected or self.is_alive():
       raise ConnectionExistsError("Already connected")
     # After starting the thread we wait for it to post us
     # an event signifying that connection is established. This
     # ensures that the caller only resumes when we are fully connected.
     self.start()
     with self._lock:
-      self._connect_cond.wait_for(lambda: self._connected or self._done)
-      if not self._connected and self._done:
+      # _ever_connected, not _connected: a session that comes up and drops
+      # before we reacquire the lock would otherwise leave the predicate false
+      # in both terms, and we would wait forever. _ever_connected latches.
+      self._connect_cond.wait_for(lambda: self._ever_connected or self._done)
+      if not self._ever_connected and self._done:
         if self._exception:
           raise self._exception
         raise LutronConnectionError("Failed to connect to Lutron controller")

tests/test_connection.py

@@ -122,7 +122,10 @@ async def readuntil_side_effect(prompt: bytes) -> bytes:
         self.mock_reader.readline.side_effect = [b"~OUTPUT,1,1,100.0\r\n", b""]
 
         self.conn.connect()
-        self.assertTrue(self.conn._connected)
+        # connect() guarantees we got connected at least once, not that we are
+        # connected right now: this fixture's readline drops the session after
+        # one line, so _connected may already be back to False.
+        self.assertTrue(self.conn._ever_connected)
         time.sleep(0.1)
         self.assertIn('~OUTPUT,1,1,100.0', self.received_lines)
 
@@ -160,7 +163,10 @@ async def mock_do_login_success() -> None:
 
         with patch.object(LutronConnection, '_do_login', side_effect=mock_do_login_success):
             self.conn.connect()
-            self.assertTrue(self.conn._connected)
+            # As in test_thread_start_and_connect: this fixture's readline drops
+            # the session immediately, so _connected may already be False by the
+            # time connect() returns. _ever_connected is what connect() promises.
+            self.assertTrue(self.conn._ever_connected)
             self.conn._done = True
             self.conn.join(timeout=1)
 
@@ -310,6 +316,63 @@ async def do_login() -> None:
 
 
 
+class TestLutronConnectionConnect(unittest.TestCase):
+    """connect() must not wait on a flag the reader may already have cleared."""
+
+    def setUp(self) -> None:
+        self.mock_reader = AsyncMock()
+        self.mock_writer = MagicMock()
+        self.mock_writer.drain = AsyncMock()
+
+        async def factory(host: str, port: int, connect_timeout: Optional[float] = None,
+                          encoding: Optional[str] = None) -> Tuple[AsyncMock, MagicMock]:
+            return self.mock_reader, self.mock_writer
+
+        self.conn = LutronConnection('127.0.0.1', 'user', 'pass', lambda line: None,
+                                     connection_factory=factory)
+
+    def tearDown(self) -> None:
+        self.conn.disconnect(timeout=2.0)
+
+    def test_connect_returns_when_the_session_drops_immediately(self) -> None:
+        """A session accepted and closed at once must not hang connect().
+
+        _main_loop sets _connected then notifies, but if the first read returns
+        nothing the loop falls straight through to _disconnect_locked() and
+        clears the flag again -- possibly before this thread reacquires the lock
+        and re-checks. Waiting on _connected then leaves the predicate false in
+        both terms (_done is False, the loop is retrying) and connect() never
+        returns. Waiting on _ever_connected, which latches, cannot do that.
+
+        connect() runs on a helper thread so a regression fails the test instead
+        of hanging the whole suite.
+        """
+        async def do_login() -> None:
+            self.conn._reader = self.mock_reader
+            self.conn._writer = self.mock_writer
+            self.mock_reader.readline = AsyncMock(return_value=b"")
+
+        returned = threading.Event()
+        errors: List[BaseException] = []
+
+        def call_connect() -> None:
+            try:
+                self.conn.connect()
+            except BaseException as exc:   # noqa: BLE001 - re-checked below
+                errors.append(exc)
+            finally:
+                returned.set()
+
+        with patch.object(LutronConnection, '_do_login', side_effect=do_login):
+            threading.Thread(target=call_connect, daemon=True).start()
+            self.assertTrue(returned.wait(10.0),
+                            "connect() hung after the session dropped immediately")
+
+        self.assertEqual(errors, [], f"connect() raised: {errors}")
+        self.assertTrue(self.conn._ever_connected)
+
+
+
 class TestLutronConnectionDisconnect(unittest.TestCase):
     """disconnect() must stop the reader thread and free the event loop.