"""Poller socket lifecycle — proxied fd-leak regressions (#79889). On macOS (256 soft fd limit) a gateway routing weixin/email pollers through a local HTTP proxy leaked one TCP socket per failed poll/connect cycle until ``[Errno 24] Too many open files`` crashed the gateway. Two code-side gaps: 1. email: ``imaplib.IMAP4.logout()`` only swallows ``OSError``; on a broken connection ``LOGOUT`` raises ``IMAP4.abort`` *before* the internal ``shutdown()``, so the socket stayed open. And ``connect()`` had no try/finally at all — a login/select failure abandoned the connected socket entirely. 2. weixin: repeated poll failures through a proxy strand sockets in the aiohttp connector; the poll session was never recycled, so they accumulated for the life of the process. """ import asyncio import imaplib import os import unittest from unittest.mock import MagicMock, patch def _make_email_adapter(address="hermes@test.com"): from gateway.config import PlatformConfig with patch.dict(os.environ, { "EMAIL_ADDRESS": address, "EMAIL_PASSWORD": "secret", "EMAIL_IMAP_HOST": "imap.test.com", "EMAIL_SMTP_HOST": "smtp.test.com", }): from plugins.platforms.email.adapter import EmailAdapter return EmailAdapter(PlatformConfig(enabled=True)) class TestCloseImap(unittest.TestCase): """_close_imap must guarantee socket teardown.""" def test_logout_success_no_shutdown_needed(self): from plugins.platforms.email.adapter import _close_imap imap = MagicMock() _close_imap(imap) imap.logout.assert_called_once() imap.shutdown.assert_not_called() def test_logout_abort_falls_back_to_shutdown(self): from plugins.platforms.email.adapter import _close_imap imap = MagicMock() imap.logout.side_effect = imaplib.IMAP4.abort("socket error: EOF") _close_imap(imap) imap.shutdown.assert_called_once() def test_shutdown_failure_is_swallowed(self): from plugins.platforms.email.adapter import _close_imap imap = MagicMock() imap.logout.side_effect = imaplib.IMAP4.abort("broken") imap.shutdown.side_effect = OSError("already closed") _close_imap(imap) # must not raise class TestEmailConnectClosesSocket(unittest.TestCase): """connect() must close the IMAP socket on every path, incl. failures.""" def test_login_failure_still_closes_socket(self): adapter = _make_email_adapter() mock_imap = MagicMock() mock_imap.login.side_effect = imaplib.IMAP4.error("AUTHENTICATIONFAILED") with patch("imaplib.IMAP4_SSL", return_value=mock_imap): result = asyncio.run(adapter.connect()) self.assertFalse(result) # The failed handle must have been torn down (logout attempted; # abort fallback covered by TestCloseImap). mock_imap.logout.assert_called_once() def test_select_failure_still_closes_socket(self): adapter = _make_email_adapter() mock_imap = MagicMock() mock_imap.select.side_effect = imaplib.IMAP4.abort("connection lost") with patch("imaplib.IMAP4_SSL", return_value=mock_imap): result = asyncio.run(adapter.connect()) self.assertFalse(result) mock_imap.logout.assert_called_once() class TestFetchClosesSocketOnBrokenLogout(unittest.TestCase): def test_fetch_logout_abort_falls_back_to_shutdown(self): adapter = _make_email_adapter() mock_imap = MagicMock() mock_imap.uid.return_value = ("OK", [b""]) mock_imap.logout.side_effect = imaplib.IMAP4.abort("EOF") with patch("imaplib.IMAP4_SSL", return_value=mock_imap): results = adapter._fetch_new_messages() self.assertEqual(results, []) mock_imap.shutdown.assert_called_once() # A teardown failure is not a fetch failure. self.assertFalse(adapter._last_fetch_failed) class TestWeixinPollSessionRecycle(unittest.TestCase): """The weixin poll loop must recycle its session after a failure streak.""" def _make_adapter(self): from gateway.config import PlatformConfig from gateway.platforms.weixin import WeixinAdapter return WeixinAdapter( PlatformConfig( enabled=True, token="test-token", extra={"account_id": "test-account"}, ) ) def test_recycle_closes_old_and_installs_fresh_session(self): from gateway.platforms import weixin as weixin_mod adapter = self._make_adapter() adapter._running = True old_session = MagicMock() old_session.closed = False async def _close(): old_session.close_called = True old_session.close = MagicMock(side_effect=lambda: _close()) # close() must return an awaitable closed = {"v": False} async def _aclose(): closed["v"] = True old_session.close = _aclose adapter._poll_session = old_session new_session = MagicMock() with patch.object( weixin_mod.aiohttp, "ClientSession", return_value=new_session ) as mk: asyncio.run(adapter._recycle_poll_session()) mk.assert_called_once() self.assertIs(adapter._poll_session, new_session) self.assertTrue(closed["v"]) def test_recycle_noop_when_not_running(self): adapter = self._make_adapter() adapter._running = False sentinel = MagicMock() adapter._poll_session = sentinel asyncio.run(adapter._recycle_poll_session()) self.assertIs(adapter._poll_session, sentinel) def test_poll_loop_recycles_after_max_consecutive_failures(self): from gateway.platforms import weixin as weixin_mod adapter = self._make_adapter() adapter._running = True adapter._poll_session = MagicMock() calls = {"n": 0, "recycled": 0} async def _failing_get_updates(*args, **kwargs): calls["n"] += 1 if calls["n"] > weixin_mod.MAX_CONSECUTIVE_FAILURES: adapter._running = False raise asyncio.CancelledError raise ConnectionError("Cannot connect to host via proxy") async def _fake_recycle(): calls["recycled"] += 1 async def _no_sleep(_secs): return None with patch.object(weixin_mod, "_get_updates", _failing_get_updates), \ patch.object(weixin_mod, "_load_sync_buf", return_value=""), \ patch.object(weixin_mod.asyncio, "sleep", _no_sleep), \ patch.object(adapter, "_recycle_poll_session", _fake_recycle): asyncio.run(adapter._poll_loop()) self.assertEqual(calls["recycled"], 1) if __name__ == "__main__": unittest.main()