diff options
Diffstat (limited to 'Lib/test/test_asyncio/test_base_events.py')
| -rw-r--r-- | Lib/test/test_asyncio/test_base_events.py | 30 |
1 files changed, 19 insertions, 11 deletions
diff --git a/Lib/test/test_asyncio/test_base_events.py b/Lib/test/test_asyncio/test_base_events.py index d812bc9..4c0d09e 100644 --- a/Lib/test/test_asyncio/test_base_events.py +++ b/Lib/test/test_asyncio/test_base_events.py @@ -647,7 +647,7 @@ class BaseEventLoopTests(test_utils.TestCase): 'Exception in callback.*zero'), exc_info=(ZeroDivisionError, MOCK_ANY, MOCK_ANY)) - assert not mock_handler.called + self.assertFalse(mock_handler.called) def test_set_exc_handler_broken(self): def run_loop(): @@ -1007,22 +1007,26 @@ class MyProto(asyncio.Protocol): if create_future: self.done = asyncio.get_running_loop().create_future() + def _assert_state(self, *expected): + if self.state not in expected: + raise AssertionError(f'state: {self.state!r}, expected: {expected!r}') + def connection_made(self, transport): self.transport = transport - assert self.state == 'INITIAL', self.state + self._assert_state('INITIAL') self.state = 'CONNECTED' transport.write(b'GET / HTTP/1.0\r\nHost: example.com\r\n\r\n') def data_received(self, data): - assert self.state == 'CONNECTED', self.state + self._assert_state('CONNECTED') self.nbytes += len(data) def eof_received(self): - assert self.state == 'CONNECTED', self.state + self._assert_state('CONNECTED') self.state = 'EOF' def connection_lost(self, exc): - assert self.state in ('CONNECTED', 'EOF'), self.state + self._assert_state('CONNECTED', 'EOF') self.state = 'CLOSED' if self.done: self.done.set_result(None) @@ -1037,20 +1041,24 @@ class MyDatagramProto(asyncio.DatagramProtocol): if create_future: self.done = loop.create_future() + def _assert_state(self, expected): + if self.state != expected: + raise AssertionError(f'state: {self.state!r}, expected: {expected!r}') + def connection_made(self, transport): self.transport = transport - assert self.state == 'INITIAL', self.state + self._assert_state('INITIAL') self.state = 'INITIALIZED' def datagram_received(self, data, addr): - assert self.state == 'INITIALIZED', self.state + self._assert_state('INITIALIZED') self.nbytes += len(data) def error_received(self, exc): - assert self.state == 'INITIALIZED', self.state + self._assert_state('INITIALIZED') def connection_lost(self, exc): - assert self.state == 'INITIALIZED', self.state + self._assert_state('INITIALIZED') self.state = 'CLOSED' if self.done: self.done.set_result(None) @@ -1399,7 +1407,7 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase): addr = ('00:01:02:03:04:05', 1) def getaddrinfo(host, port, *args, **kw): - assert (host, port) == addr + self.assertEqual((host, port), addr) return [(999, 1, 999, '', (addr, 1))] m_socket.getaddrinfo = getaddrinfo @@ -1698,7 +1706,7 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase): lambda: MyDatagramProto(create_future=True, loop=self.loop), family=socket.AF_UNIX) transport, protocol = self.loop.run_until_complete(fut) - assert transport._sock.family == socket.AF_UNIX + self.assertEqual(transport._sock.family, socket.AF_UNIX) transport.close() self.loop.run_until_complete(protocol.done) self.assertEqual('CLOSED', protocol.state) |