Merge bitcoin/bitcoin#35583: test: close the listeners before terminating the event loop

e4d80e7 test: close the loop after the network thread has completed (Vasil Dimov)
29fba5d test: close the listeners before terminating the event loop (Vasil Dimov)

Pull request description:

  Whenever a test creates a new `P2PInterface` object a new listener is
  created inside `NetworkThread.create_listen_server()` by calling
  `cls.network_event_loop.create_server()`.

  These listeners are never closed which might result in:

  ```
  2026-06-10T22:13:35.3934880Z Task was destroyed but it is pending!
  2026-06-10T22:13:35.3936020Z task: <Task pending name='Task-54' coro=<BaseSelectorEventLoop._accept_connection2() done, defined at /opt/homebrew/Cellar/python@3.14/3.14.5/Frameworks/Python.framework/Versions/3.14/lib/python3.14/asyncio/selector_events.py:217> wait_for=<Future finished result=None>>
  ```

  when the event loop is closed.

  Fix that by closing the listeners.

  Fixes: bitcoin/bitcoin#35508

ACKs for top commit:
  andrewtoth:
    ACK e4d80e7
  sedited:
    ACK e4d80e7

Tree-SHA512: b93d06526b4eb31ac445a1a0e379e5ec947661f8ea29f2e07ac88b9e4760b0cc5348638b320ab8d60735f34fc163fcdd18eba42b20e0a7726a98a5136433bd64

Author: fanquake

test/functional/test_framework/p2p.py

@@ -748,6 +748,7 @@ def __init__(self):
 
         NetworkThread.listeners = {}
         NetworkThread.protos = {}
+        NetworkThread.protos_accept_done = []
 
     def run(self):
         """Start the network thread."""
@@ -756,10 +757,25 @@ def run(self):
 
     def close(self, *, timeout):
         """Close the connections and network event loop."""
+        for p in NetworkThread.protos_accept_done:
+            p.peer_disconnect()
+        NetworkThread.protos_accept_done.clear()
+
+        listeners = list(NetworkThread.listeners.values())
+        NetworkThread.listeners.clear()
+
+        async def close_listeners():
+            for listener in listeners:
+                listener.close()
+            for listener in listeners:
+                await listener.wait_closed()
+        future = asyncio.run_coroutine_threadsafe(close_listeners(), self.network_event_loop)
+        future.result(timeout=timeout)
+
         self.network_event_loop.call_soon_threadsafe(self.network_event_loop.stop)
         wait_until_helper_internal(lambda: not self.network_event_loop.is_running(), timeout=timeout)
-        self.network_event_loop.close()
         self.join(timeout)
+        self.network_event_loop.close()
         # Safe to remove event loop.
         NetworkThread.network_event_loop = None
 
@@ -795,6 +811,7 @@ def peer_protocol():
             response = cls.protos.get((addr, port))
             # remove protocol function from dict only when reconnection doesn't need to happen/already happened
             if not proto.reconnect:
+                cls.protos_accept_done.append(response)
                 cls.protos[(addr, port)] = None
             return response