mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-09-01 22:47:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

bpo-23819: Fix asyncio tests on python optimized mode (GH-30195)

Browse source
This commit is contained in:
Kumar Aditya 2021-12-26 16:43:14 +05:30 • committed by GitHub
parent f9a4352056
commit a23ab7b6d8
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 9 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

4
Lib/asyncio/base_events.py
View file

@ -1285,8 +1285,8 @@ class BaseEventLoop(events.AbstractEventLoop):
addr_infos = {} # Using order preserving dict addr_infos = {} # Using order preserving dict
for idx, addr in ((0, local_addr), (1, remote_addr)): for idx, addr in ((0, local_addr), (1, remote_addr)):
if addr is not None: if addr is not None:
assert isinstance(addr, tuple) and len(addr) == 2, ( if not (isinstance(addr, tuple) and len(addr) == 2):
'2-tuple is expected') raise TypeError('2-tuple is expected')
infos = await self._ensure_resolved( infos = await self._ensure_resolved(
addr, family=family, type=socket.SOCK_DGRAM, addr, family=family, type=socket.SOCK_DGRAM,

4
Lib/test/test_asyncio/test_base_events.py
View file

@ -1603,11 +1603,11 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase):
coro = self.loop.create_datagram_endpoint( coro = self.loop.create_datagram_endpoint(
MyDatagramProto, local_addr='localhost') MyDatagramProto, local_addr='localhost')
self.assertRaises( self.assertRaises(
AssertionError, self.loop.run_until_complete, coro) TypeError, self.loop.run_until_complete, coro)
coro = self.loop.create_datagram_endpoint( coro = self.loop.create_datagram_endpoint(
MyDatagramProto, local_addr=('localhost', 1, 2, 3)) MyDatagramProto, local_addr=('localhost', 1, 2, 3))
self.assertRaises( self.assertRaises(
AssertionError, self.loop.run_until_complete, coro) TypeError, self.loop.run_until_complete, coro)
def test_create_datagram_endpoint_connect_err(self): def test_create_datagram_endpoint_connect_err(self):
self.loop.sock_connect = mock.Mock() self.loop.sock_connect = mock.Mock()

2
Lib/test/test_asyncio/test_proactor_events.py
View file

@ -77,6 +77,7 @@ class ProactorSocketTransportTests(test_utils.TestCase):
self.loop._proactor.recv_into.assert_called_with(self.sock, called_buf) self.loop._proactor.recv_into.assert_called_with(self.sock, called_buf)
self.protocol.data_received.assert_called_with(bytearray(buf)) self.protocol.data_received.assert_called_with(bytearray(buf))
@unittest.skipIf(sys.flags.optimize, "Assertions are disabled in optimized mode")
def test_loop_reading_no_data(self): def test_loop_reading_no_data(self):
res = self.loop.create_future() res = self.loop.create_future()
res.set_result(0) res.set_result(0)
@ -869,6 +870,7 @@ class BaseProactorEventLoopTests(test_utils.TestCase):
self.protocol.datagram_received.assert_called_with(b'data', ('127.0.0.1', 12068)) self.protocol.datagram_received.assert_called_with(b'data', ('127.0.0.1', 12068))
close_transport(tr) close_transport(tr)
@unittest.skipIf(sys.flags.optimize, "Assertions are disabled in optimized mode")
def test_datagram_loop_reading_no_data(self): def test_datagram_loop_reading_no_data(self):
res = self.loop.create_future() res = self.loop.create_future()
res.set_result((b'', ('127.0.0.1', 12068))) res.set_result((b'', ('127.0.0.1', 12068)))

2
Lib/test/test_asyncio/test_selector_events.py
View file

@ -1,5 +1,6 @@
"""Tests for selector_events.py""" """Tests for selector_events.py"""
import sys
import selectors import selectors
import socket import socket
import unittest import unittest
@ -804,6 +805,7 @@ class SelectorSocketTransportTests(test_utils.TestCase):
self.sock.close.assert_called_with() self.sock.close.assert_called_with()
self.protocol.connection_lost.assert_called_with(None) self.protocol.connection_lost.assert_called_with(None)
@unittest.skipIf(sys.flags.optimize, "Assertions are disabled in optimized mode")
def test_write_ready_no_data(self): def test_write_ready_no_data(self):
transport = self.socket_transport() transport = self.socket_transport()
# This is an internal error. # This is an internal error.

1
Misc/NEWS.d/next/Tests/2021-12-19-08-44-32.bpo-23819.9ueiII.rst Normal file
View file

@ -0,0 +1 @@
Fixed :mod:`asyncio` tests in python optimized mode. Patch by Kumar Aditya.