test: Add test for exceeding max line length in torcontrol

diff --git a/test/functional/feature_torcontrol.py b/test/functional/feature_torcontrol.py
index 084f3dfe..2148fc10 100755
--- a/test/functional/feature_torcontrol.py
+++ b/test/functional/feature_torcontrol.py
@@ -107,18 +107,24 @@ class TorControlTest(BitcoinTestFramework):
     def set_test_params(self):
         self.num_nodes = 1
 
-    def test_basic(self):
-        self.log.info("Test Tor control basic functionality")
-        tor_control_port = p2p_port(self.num_nodes + 1)
-        mock_tor = MockTorControlServer(tor_control_port)
-        mock_tor.start()
+    def next_port(self):
+        self._port_counter = getattr(self, '_port_counter', 0) + 1
+        return p2p_port(self.num_nodes + self._port_counter)
 
+    def restart_with_mock(self, mock_tor):
+        mock_tor.start()
         self.restart_node(0, extra_args=[
-            f"-torcontrol=127.0.0.1:{tor_control_port}",
+            f"-torcontrol=127.0.0.1:{mock_tor.port}",
             "-listenonion=1",
-            "-debug=tor"
+            "-debug=tor",
         ])
 
+    def test_basic(self):
+        self.log.info("Test Tor control basic functionality")
+
+        mock_tor = MockTorControlServer(self.next_port())
+        self.restart_with_mock(mock_tor)
+
         # Waiting for Tor control commands
         self.wait_until(lambda: len(mock_tor.received_commands) >= 4, timeout=10)
 
@@ -135,15 +141,8 @@ class TorControlTest(BitcoinTestFramework):
     def test_partial_data(self):
         self.log.info("Test that partial Tor control responses are buffered until complete")
 
-        tor_port = p2p_port(self.num_nodes + 2)
-        mock_tor = MockTorControlServer(tor_port, manual_mode=True)
-        mock_tor.start()
-
-        self.restart_node(0, extra_args=[
-            f"-torcontrol=127.0.0.1:{tor_port}",
-            "-listenonion=1",
-            "-debug=tor"
-        ])
+        mock_tor = MockTorControlServer(self.next_port(), manual_mode=True)
+        self.restart_with_mock(mock_tor)
 
         # Wait for connection and PROTOCOLINFO command
         mock_tor.conn_ready.wait(timeout=10)
@@ -173,8 +172,6 @@ class TorControlTest(BitcoinTestFramework):
     def test_pow_fallback(self):
         self.log.info("Test that ADD_ONION retries without PoW on 512 error")
 
-        tor_port = p2p_port(self.num_nodes + 3)
-
         class NoPowServer(MockTorControlServer):
             def _get_response(self, command):
                 if command.startswith("ADD_ONION"):
@@ -187,14 +184,8 @@ class TorControlTest(BitcoinTestFramework):
                         )
                 return super()._get_response(command)
 
-        mock_tor = NoPowServer(tor_port)
-        mock_tor.start()
-
-        self.restart_node(0, extra_args=[
-            f"-torcontrol=127.0.0.1:{tor_port}",
-            "-listenonion=1",
-            "-debug=tor",
-        ])
+        mock_tor = NoPowServer(self.next_port())
+        self.restart_with_mock(mock_tor)
 
         # Expect: PROTOCOLINFO, AUTHENTICATE, GETINFO, ADD_ONION (with PoW), ADD_ONION (without PoW)
         self.wait_until(lambda: len(mock_tor.received_commands) >= 5, timeout=10)
@@ -210,10 +201,33 @@ class TorControlTest(BitcoinTestFramework):
         # Clean up
         mock_tor.stop()
 
+    def test_oversized_line(self):
+        self.log.info("Test that Tor control disconnects on oversized response lines")
+
+        mock_tor = MockTorControlServer(self.next_port(), manual_mode=True)
+        self.restart_with_mock(mock_tor)
+
+        # Wait for connection and PROTOCOLINFO command.
+        mock_tor.conn_ready.wait(timeout=10)
+        self.wait_until(lambda: len(mock_tor.received_commands) >= 1, timeout=10)
+        assert_equal(mock_tor.received_commands[0], "PROTOCOLINFO 1")
+
+        # Send a single line longer than MAX_LINE_LENGTH. The node should disconnect.
+        MAX_LINE_LENGTH = 100000
+        mock_tor.send_raw("250-" + ("A" * (MAX_LINE_LENGTH + 1)) + "\r\n")
+        ensure_for(duration=2, f=lambda: self.nodes[0].process.poll() is None)
+
+        # Connection should be dropped and retried, causing another PROTOCOLINFO.
+        self.wait_until(lambda: len(mock_tor.received_commands) >= 2, timeout=10)
+        assert_equal(mock_tor.received_commands[1], "PROTOCOLINFO 1")
+
+        mock_tor.stop()
+
     def run_test(self):
         self.test_basic()
         self.test_partial_data()
         self.test_pow_fallback()
+        self.test_oversized_line()
 
 if __name__ == '__main__':
     TorControlTest(__file__).main()