Skip to content
Closed
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Updated test_base_events.py
  • Loading branch information
terryluan12 committed Dec 29, 2025
commit ff252da9287859dee842684fc1d2ea65f82efb46
17 changes: 17 additions & 0 deletions Lib/test/test_asyncio/test_base_events.py
Original file line number Diff line number Diff line change
Expand Up @@ -948,6 +948,10 @@ def test_run_forever_pre_stopped(self):
self.loop.run_forever()
self.loop._selector.select.assert_called_once_with(0)


# TODO: RUSTPYTHON
# 'BaseEventLoop' object has no attribute '_run_forever_setup
@unittest.expectedFailure
def test_custom_run_forever_integration(self):
# Test that the run_forever_setup() and run_forever_cleanup() primitives
# can be used to implement a custom run_forever loop.
Expand Down Expand Up @@ -1019,6 +1023,10 @@ async def iter_one():
asyncio.create_task(iter_one())
return status

# TODO: RUSTPYTHON
# self.assertTrue(status['finalized'])
# AssertionError: False is not true
@unittest.expectedFailure
def test_asyncgen_finalization_by_gc(self):
# Async generators should be finalized when garbage collected.
self.loop._process_events = mock.Mock()
Expand All @@ -1034,6 +1042,10 @@ def test_asyncgen_finalization_by_gc(self):
test_utils.run_briefly(self.loop)
self.assertTrue(status['finalized'])

# TODO: RUSTPYTHON
# self.assertTrue(status['finalized'])
# AssertionError: False is not true
@unittest.expectedFailure
def test_asyncgen_finalization_by_gc_in_other_thread(self):
# Python issue 34769: If garbage collector runs in another
# thread, async generators will not finalize in debug
Expand Down Expand Up @@ -1194,7 +1206,9 @@ def getaddrinfo(*args, **kw):
self.loop.run_until_complete(coro)
self.assertTrue(sock.close.called)


@patch_socket
@unittest.skip('TODO: RUSTPYTHON')
def test_create_connection_happy_eyeballs_empty_exceptions(self, m_socket):
# See gh-135836: Fix IndexError when Happy Eyeballs algorithm
# results in empty exceptions list
Expand Down Expand Up @@ -1255,6 +1269,7 @@ def test_create_server_ssl_timeout_for_plain_socket(self):

@unittest.skipUnless(hasattr(socket, 'SOCK_NONBLOCK'),
'no socket.SOCK_NONBLOCK (linux only)')
@unittest.skip('TODO: RUSTPYTHON')
def test_create_server_stream_bittype(self):
sock = socket.socket(
socket.AF_INET, socket.SOCK_STREAM | socket.SOCK_NONBLOCK)
Expand Down Expand Up @@ -1331,6 +1346,7 @@ def getaddrinfo_task(*args, **kwds):
self.assertIsInstance(cm.exception.exceptions[0], OSError)

@patch_socket
@unittest.skip('TODO: RUSTPYTHON')
def test_create_connection_connect_non_os_err_close_err(self, m_socket):
# Test the case when sock_connect() raises non-OSError exception
# and sock.close() raises OSError.
Expand Down Expand Up @@ -1499,6 +1515,7 @@ def test_create_connection_no_inet_pton(self, m_socket):
support.is_android and platform.android_ver().api_level < 23,
"Issue gh-71123: this fails on Android before API level 23"
)
@unittest.skip('TODO: RUSTPYTHON')
def test_create_connection_service_name(self, m_socket):
m_socket.getaddrinfo = socket.getaddrinfo
sock = m_socket.socket.return_value
Expand Down