1  """Tests for window_utils"""
       2  
       3  import sys
       4  import unittest
       5  import warnings
       6  
       7  if sys.platform != 'win32':
       8      raise unittest.SkipTest('Windows only')
       9  
      10  import _overlapped
      11  import _winapi
      12  
      13  import asyncio
      14  from asyncio import windows_utils
      15  from test import support
      16  
      17  
      18  def tearDownModule():
      19      asyncio.set_event_loop_policy(None)
      20  
      21  
      22  class ESC[4;38;5;81mPipeTests(ESC[4;38;5;149munittestESC[4;38;5;149m.ESC[4;38;5;149mTestCase):
      23  
      24      def test_pipe_overlapped(self):
      25          h1, h2 = windows_utils.pipe(overlapped=(True, True))
      26          try:
      27              ov1 = _overlapped.Overlapped()
      28              self.assertFalse(ov1.pending)
      29              self.assertEqual(ov1.error, 0)
      30  
      31              ov1.ReadFile(h1, 100)
      32              self.assertTrue(ov1.pending)
      33              self.assertEqual(ov1.error, _winapi.ERROR_IO_PENDING)
      34              ERROR_IO_INCOMPLETE = 996
      35              try:
      36                  ov1.getresult()
      37              except OSError as e:
      38                  self.assertEqual(e.winerror, ERROR_IO_INCOMPLETE)
      39              else:
      40                  raise RuntimeError('expected ERROR_IO_INCOMPLETE')
      41  
      42              ov2 = _overlapped.Overlapped()
      43              self.assertFalse(ov2.pending)
      44              self.assertEqual(ov2.error, 0)
      45  
      46              ov2.WriteFile(h2, b"hello")
      47              self.assertIn(ov2.error, {0, _winapi.ERROR_IO_PENDING})
      48  
      49              res = _winapi.WaitForMultipleObjects([ov2.event], False, 100)
      50              self.assertEqual(res, _winapi.WAIT_OBJECT_0)
      51  
      52              self.assertFalse(ov1.pending)
      53              self.assertEqual(ov1.error, ERROR_IO_INCOMPLETE)
      54              self.assertFalse(ov2.pending)
      55              self.assertIn(ov2.error, {0, _winapi.ERROR_IO_PENDING})
      56              self.assertEqual(ov1.getresult(), b"hello")
      57          finally:
      58              _winapi.CloseHandle(h1)
      59              _winapi.CloseHandle(h2)
      60  
      61      def test_pipe_handle(self):
      62          h, _ = windows_utils.pipe(overlapped=(True, True))
      63          _winapi.CloseHandle(_)
      64          p = windows_utils.PipeHandle(h)
      65          self.assertEqual(p.fileno(), h)
      66          self.assertEqual(p.handle, h)
      67  
      68          # check garbage collection of p closes handle
      69          with warnings.catch_warnings():
      70              warnings.filterwarnings("ignore", "",  ResourceWarning)
      71              del p
      72              support.gc_collect()
      73          try:
      74              _winapi.CloseHandle(h)
      75          except OSError as e:
      76              self.assertEqual(e.winerror, 6)     # ERROR_INVALID_HANDLE
      77          else:
      78              raise RuntimeError('expected ERROR_INVALID_HANDLE')
      79  
      80  
      81  class ESC[4;38;5;81mPopenTests(ESC[4;38;5;149munittestESC[4;38;5;149m.ESC[4;38;5;149mTestCase):
      82  
      83      def test_popen(self):
      84          command = r"""if 1:
      85              import sys
      86              s = sys.stdin.readline()
      87              sys.stdout.write(s.upper())
      88              sys.stderr.write('stderr')
      89              """
      90          msg = b"blah\n"
      91  
      92          p = windows_utils.Popen([sys.executable, '-c', command],
      93                                  stdin=windows_utils.PIPE,
      94                                  stdout=windows_utils.PIPE,
      95                                  stderr=windows_utils.PIPE)
      96  
      97          for f in [p.stdin, p.stdout, p.stderr]:
      98              self.assertIsInstance(f, windows_utils.PipeHandle)
      99  
     100          ovin = _overlapped.Overlapped()
     101          ovout = _overlapped.Overlapped()
     102          overr = _overlapped.Overlapped()
     103  
     104          ovin.WriteFile(p.stdin.handle, msg)
     105          ovout.ReadFile(p.stdout.handle, 100)
     106          overr.ReadFile(p.stderr.handle, 100)
     107  
     108          events = [ovin.event, ovout.event, overr.event]
     109          # Super-long timeout for slow buildbots.
     110          res = _winapi.WaitForMultipleObjects(events, True,
     111                                               int(support.SHORT_TIMEOUT * 1000))
     112          self.assertEqual(res, _winapi.WAIT_OBJECT_0)
     113          self.assertFalse(ovout.pending)
     114          self.assertFalse(overr.pending)
     115          self.assertFalse(ovin.pending)
     116  
     117          self.assertEqual(ovin.getresult(), len(msg))
     118          out = ovout.getresult().rstrip()
     119          err = overr.getresult().rstrip()
     120  
     121          self.assertGreater(len(out), 0)
     122          self.assertGreater(len(err), 0)
     123          # allow for partial reads...
     124          self.assertTrue(msg.upper().rstrip().startswith(out))
     125          self.assertTrue(b"stderr".startswith(err))
     126  
     127          # The context manager calls wait() and closes resources
     128          with p:
     129              pass
     130  
     131  
     132  if __name__ == '__main__':
     133      unittest.main()