tests: fix flaky TestTwistedConnection.test_connection_initialization

Patch reactor.running to False in setUp() so that maybe_start() always
enters the branch that spawns the reactor thread. Without this, leaked
global reactor state from prior tests can leave reactor.running as True,
causing maybe_start() to skip thread creation and the reactor.run mock
to never be called — making the assertion in test_connection_initialization
fail intermittently.

Observed in CI on PyPy 3.11 + macOS x86 (Rosetta 2), where timing
differences make the reactor state leak more likely.

Author: mykaul

tests/unit/io/test_twistedreactor.py

@@ -30,6 +30,7 @@
 
 from tests.unit.io.utils import TimerTestMixin
 
+
 class TestTwistedTimer(TimerTestMixin, unittest.TestCase):
     """
     Simple test class that is used to validate that the TimerManager, and timer
@@ -54,7 +55,6 @@ def setUp(self):
 
 
 class TestTwistedProtocol(unittest.TestCase):
-
     def setUp(self):
         if twistedreactor is None:
             raise unittest.SkipTest("Twisted libraries not available")
@@ -84,7 +84,7 @@ def test_receiving_data(self):
         the connection object's buffer and calls handle_read().
         """
         self.obj_ut.makeConnection(self.tr)
-        self.obj_ut.dataReceived('foobar')
+        self.obj_ut.dataReceived("foobar")
         assert self.mock_connection.handle_read.called
         self.mock_connection._iobuf.write.assert_called_with("foobar")
 
@@ -96,17 +96,27 @@ def setUp(self):
         if twistedreactor.TwistedConnection._loop:
             twistedreactor.TwistedConnection._loop._cleanup()
         twistedreactor.TwistedConnection.initialize_reactor()
-        self.reactor_cft_patcher = patch(
-            'twisted.internet.reactor.callFromThread')
-        self.reactor_run_patcher = patch('twisted.internet.reactor.run')
+        self.reactor_cft_patcher = patch("twisted.internet.reactor.callFromThread")
+        self.reactor_run_patcher = patch("twisted.internet.reactor.run")
+        # Patch reactor.running to False so maybe_start() always enters
+        # the branch that spawns the reactor thread. Without this, leaked
+        # reactor state from prior tests can cause reactor.running to be
+        # True, making maybe_start() a no-op and the reactor.run mock
+        # never called — leading to a flaky test_connection_initialization.
+        self.reactor_running_patcher = patch(
+            "twisted.internet.reactor.running", new=False
+        )
         self.mock_reactor_cft = self.reactor_cft_patcher.start()
         self.mock_reactor_run = self.reactor_run_patcher.start()
-        self.obj_ut = twistedreactor.TwistedConnection(DefaultEndPoint('1.2.3.4'),
-                                                       cql_version='3.0.1')
+        self.reactor_running_patcher.start()
+        self.obj_ut = twistedreactor.TwistedConnection(
+            DefaultEndPoint("1.2.3.4"), cql_version="3.0.1"
+        )
 
     def tearDown(self):
         self.reactor_cft_patcher.stop()
         self.reactor_run_patcher.stop()
+        self.reactor_running_patcher.stop()
 
     def test_connection_initialization(self):
         """
@@ -124,7 +134,7 @@ def test_client_connection_made(self):
         self.obj_ut.client_connection_made(Mock())
         self.obj_ut._send_options_message.assert_called_with()
 
-    @patch('twisted.internet.reactor.connectTCP')
+    @patch("twisted.internet.reactor.connectTCP")
     def test_close(self, mock_connectTCP):
         """
         Verify that close() disconnects the connector and errors callbacks.
@@ -144,16 +154,22 @@ def test_handle_read__incomplete(self):
         Verify that handle_read() processes incomplete messages properly.
         """
         self.obj_ut.process_msg = Mock()
-        assert self.obj_ut._iobuf.getvalue() == b''  # buf starts empty
+        assert self.obj_ut._iobuf.getvalue() == b""  # buf starts empty
         # incomplete header
-        self.obj_ut._iobuf.write(b'\x84\x00\x00\x00\x00')
+        self.obj_ut._iobuf.write(b"\x84\x00\x00\x00\x00")
         self.obj_ut.handle_read()
-        assert self.obj_ut._io_buffer.cql_frame_buffer.getvalue() == b'\x84\x00\x00\x00\x00'
+        assert (
+            self.obj_ut._io_buffer.cql_frame_buffer.getvalue()
+            == b"\x84\x00\x00\x00\x00"
+        )
 
         # full header, but incomplete body
-        self.obj_ut._iobuf.write(b'\x00\x00\x00\x15')
+        self.obj_ut._iobuf.write(b"\x00\x00\x00\x15")
         self.obj_ut.handle_read()
-        assert self.obj_ut._io_buffer.cql_frame_buffer.getvalue() == b'\x84\x00\x00\x00\x00\x00\x00\x00\x15'
+        assert (
+            self.obj_ut._io_buffer.cql_frame_buffer.getvalue()
+            == b"\x84\x00\x00\x00\x00\x00\x00\x00\x15"
+        )
         assert self.obj_ut._current_frame.end_pos == 30
 
         # verify we never attempted to process the incomplete message
@@ -164,27 +180,34 @@ def test_handle_read__fullmessage(self):
         Verify that handle_read() processes complete messages properly.
         """
         self.obj_ut.process_msg = Mock()
-        assert self.obj_ut._iobuf.getvalue() == b''  # buf starts empty
+        assert self.obj_ut._iobuf.getvalue() == b""  # buf starts empty
 
         # write a complete message, plus 'NEXT' (to simulate next message)
         # assumes protocol v3+ as default Connection.protocol_version
-        body = b'this is the drum roll'
-        extra = b'NEXT'
-        self.obj_ut._iobuf.write(
-            b'\x84\x01\x00\x02\x03\x00\x00\x00\x15' + body + extra)
+        body = b"this is the drum roll"
+        extra = b"NEXT"
+        self.obj_ut._iobuf.write(b"\x84\x01\x00\x02\x03\x00\x00\x00\x15" + body + extra)
         self.obj_ut.handle_read()
         assert self.obj_ut._io_buffer.cql_frame_buffer.getvalue() == extra
         self.obj_ut.process_msg.assert_called_with(
-            _Frame(version=4, flags=1, stream=2, opcode=3, body_offset=9, end_pos=9 + len(body)), body)
-
-    @patch('twisted.internet.reactor.connectTCP')
+            _Frame(
+                version=4,
+                flags=1,
+                stream=2,
+                opcode=3,
+                body_offset=9,
+                end_pos=9 + len(body),
+            ),
+            body,
+        )
+
+    @patch("twisted.internet.reactor.connectTCP")
     def test_push(self, mock_connectTCP):
         """
         Verifiy that push() calls transport.write(data).
         """
         self.obj_ut.add_connection()
         transport_mock = Mock()
         self.obj_ut.transport = transport_mock
-        self.obj_ut.push('123 pickup')
-        self.mock_reactor_cft.assert_called_with(
-            transport_mock.write, '123 pickup')
+        self.obj_ut.push("123 pickup")
+        self.mock_reactor_cft.assert_called_with(transport_mock.write, "123 pickup")