diff options
Diffstat (limited to 'tests/test_windows_utils.py')
| -rw-r--r-- | tests/test_windows_utils.py | 41 |
1 files changed, 21 insertions, 20 deletions
diff --git a/tests/test_windows_utils.py b/tests/test_windows_utils.py index 3e7a211..ac6d736 100644 --- a/tests/test_windows_utils.py +++ b/tests/test_windows_utils.py @@ -2,18 +2,19 @@ import socket import sys -import test.support import unittest -from test.support import IPV6_ENABLED -from unittest import mock if sys.platform != 'win32': - raise unittest.SkipTest('Windows only') + from trollius.test_utils import SkipTest + raise SkipTest('Windows only') -import _winapi - -from asyncio import windows_utils -from asyncio import _overlapped +from trollius import _overlapped +from trollius import py33_winapi as _winapi +from trollius import test_utils +from trollius import windows_utils +from trollius.test_support import IPV6_ENABLED +from trollius.test_utils import mock +import trollius.test_support as support class WinsocketpairTests(unittest.TestCase): @@ -28,14 +29,14 @@ class WinsocketpairTests(unittest.TestCase): ssock, csock = windows_utils.socketpair() self.check_winsocketpair(ssock, csock) - @unittest.skipUnless(IPV6_ENABLED, 'IPv6 not supported or enabled') + @test_utils.skipUnless(IPV6_ENABLED, 'IPv6 not supported or enabled') def test_winsocketpair_ipv6(self): ssock, csock = windows_utils.socketpair(family=socket.AF_INET6) self.check_winsocketpair(ssock, csock) - @unittest.skipIf(hasattr(socket, 'socketpair'), - 'socket.socketpair is available') - @mock.patch('asyncio.windows_utils.socket') + @test_utils.skipIf(hasattr(socket, 'socketpair'), + 'socket.socketpair is available') + @mock.patch('trollius.windows_utils.socket') def test_winsocketpair_exc(self, m_socket): m_socket.AF_INET = socket.AF_INET m_socket.SOCK_STREAM = socket.SOCK_STREAM @@ -53,9 +54,9 @@ class WinsocketpairTests(unittest.TestCase): self.assertRaises(ValueError, windows_utils.socketpair, proto=1) - @unittest.skipIf(hasattr(socket, 'socketpair'), - 'socket.socketpair is available') - @mock.patch('asyncio.windows_utils.socket') + @test_utils.skipIf(hasattr(socket, 'socketpair'), + 'socket.socketpair is available') + @mock.patch('trollius.windows_utils.socket') def test_winsocketpair_close(self, m_socket): m_socket.AF_INET = socket.AF_INET m_socket.SOCK_STREAM = socket.SOCK_STREAM @@ -81,7 +82,7 @@ class PipeTests(unittest.TestCase): ERROR_IO_INCOMPLETE = 996 try: ov1.getresult() - except OSError as e: + except WindowsError as e: self.assertEqual(e.winerror, ERROR_IO_INCOMPLETE) else: raise RuntimeError('expected ERROR_IO_INCOMPLETE') @@ -91,15 +92,15 @@ class PipeTests(unittest.TestCase): self.assertEqual(ov2.error, 0) ov2.WriteFile(h2, b"hello") - self.assertIn(ov2.error, {0, _winapi.ERROR_IO_PENDING}) + self.assertIn(ov2.error, set((0, _winapi.ERROR_IO_PENDING))) - res = _winapi.WaitForMultipleObjects([ov2.event], False, 100) + res = _winapi.WaitForSingleObject(ov2.event, 100) self.assertEqual(res, _winapi.WAIT_OBJECT_0) self.assertFalse(ov1.pending) self.assertEqual(ov1.error, ERROR_IO_INCOMPLETE) self.assertFalse(ov2.pending) - self.assertIn(ov2.error, {0, _winapi.ERROR_IO_PENDING}) + self.assertIn(ov2.error, set((0, _winapi.ERROR_IO_PENDING))) self.assertEqual(ov1.getresult(), b"hello") finally: _winapi.CloseHandle(h1) @@ -114,7 +115,7 @@ class PipeTests(unittest.TestCase): # check garbage collection of p closes handle del p - test.support.gc_collect() + support.gc_collect() try: _winapi.CloseHandle(h) except OSError as e: |