tests for windowutils import sys import unittest import warnings if sys platform win32 raise unittest skiptest windows only import overlapped import winapi import asyncio from asyncio import windowsutils from test import support def teardownmodule asyncio seteventlooppolicynone class pipetestsunittest testcase def testpipeoverlappedself h1 h2 windowsutils pipeoverlappedtrue true try ov1 overlapped overlapped self assertfalseov1 pending self assertequalov1 error 0 ov1 readfileh1 100 self asserttrueov1 pending self assertequalov1 error winapi erroriopending errorioincomplete 996 try ov1 getresult except oserror as e self assertequale winerror errorioincomplete else raise runtimeerror expected errorioincomplete ov2 overlapped overlapped self assertfalseov2 pending self assertequalov2 error 0 ov2 writefileh2 bhello self assertinov2 error 0 winapi erroriopending res winapi waitformultipleobjectsov2 event false 100 self assertequalres winapi waitobject0 self assertfalseov1 pending self assertequalov1 error errorioincomplete self assertfalseov2 pending self assertinov2 error 0 winapi erroriopending self assertequalov1 getresult bhello finally winapi closehandleh1 winapi closehandleh2 def testpipehandleself h windowsutils pipeoverlappedtrue true winapi closehandle p windowsutils pipehandleh self assertequalp fileno h self assertequalp handle h check garbage collection of p closes handle with warnings catchwarnings warnings filterwarningsignore resourcewarning del p support gccollect try winapi closehandleh except oserror as e self assertequale winerror 6 errorinvalidhandle else raise runtimeerror expected errorinvalidhandle class popentestsunittest testcase def testpopenself command rif 1 import sys s sys stdin readline sys stdout writes upper sys stderr write stderr superlong timeout for slow buildbots allow for partial reads the context manager calls wait and closes resources check garbage collection of p closes handle error_invalid_handle if 1 import sys s sys stdin readline sys stdout write s upper sys stderr write stderr super long timeout for slow buildbots allow for partial reads the context manager calls wait and closes resources
import sys
import unittest
import warnings

if sys.platform != 'win32':
    raise unittest.SkipTest('Windows only')

import _overlapped
import _winapi

import asyncio
from asyncio import windows_utils
from test import support


def tearDownModule():
    asyncio.set_event_loop_policy(None)


class PipeTests(unittest.TestCase):

    def test_pipe_overlapped(self):
        h1, h2 = windows_utils.pipe(overlapped=(True, True))
        try:
            ov1 = _overlapped.Overlapped()
            self.assertFalse(ov1.pending)
            self.assertEqual(ov1.error, 0)

            ov1.ReadFile(h1, 100)
            self.assertTrue(ov1.pending)
            self.assertEqual(ov1.error, _winapi.ERROR_IO_PENDING)
            ERROR_IO_INCOMPLETE = 996
            try:
                ov1.getresult()
            except OSError as e:
                self.assertEqual(e.winerror, ERROR_IO_INCOMPLETE)
            else:
                raise RuntimeError('expected ERROR_IO_INCOMPLETE')

            ov2 = _overlapped.Overlapped()
            self.assertFalse(ov2.pending)
            self.assertEqual(ov2.error, 0)

            ov2.WriteFile(h2, b"hello")
            self.assertIn(ov2.error, {0, _winapi.ERROR_IO_PENDING})

            res = _winapi.WaitForMultipleObjects([ov2.event], False, 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.assertEqual(ov1.getresult(), b"hello")
        finally:
            _winapi.CloseHandle(h1)
            _winapi.CloseHandle(h2)

    def test_pipe_handle(self):
        h, _ = windows_utils.pipe(overlapped=(True, True))
        _winapi.CloseHandle(_)
        p = windows_utils.PipeHandle(h)
        self.assertEqual(p.fileno(), h)
        self.assertEqual(p.handle, h)

        
        with warnings.catch_warnings():
            warnings.filterwarnings("ignore", "",  ResourceWarning)
            del p
            support.gc_collect()
        try:
            _winapi.CloseHandle(h)
        except OSError as e:
            self.assertEqual(e.winerror, 6)     
        else:
            raise RuntimeError('expected ERROR_INVALID_HANDLE')


class PopenTests(unittest.TestCase):

    def test_popen(self):
        command = r
        msg = b"blah\n"

        p = windows_utils.Popen([sys.executable, '-c', command],
                                stdin=windows_utils.PIPE,
                                stdout=windows_utils.PIPE,
                                stderr=windows_utils.PIPE)

        for f in [p.stdin, p.stdout, p.stderr]:
            self.assertIsInstance(f, windows_utils.PipeHandle)

        ovin = _overlapped.Overlapped()
        ovout = _overlapped.Overlapped()
        overr = _overlapped.Overlapped()

        ovin.WriteFile(p.stdin.handle, msg)
        ovout.ReadFile(p.stdout.handle, 100)
        overr.ReadFile(p.stderr.handle, 100)

        events = [ovin.event, ovout.event, overr.event]
        
        res = _winapi.WaitForMultipleObjects(events, True,
                                             int(support.SHORT_TIMEOUT * 1000))
        self.assertEqual(res, _winapi.WAIT_OBJECT_0)
        self.assertFalse(ovout.pending)
        self.assertFalse(overr.pending)
        self.assertFalse(ovin.pending)

        self.assertEqual(ovin.getresult(), len(msg))
        out = ovout.getresult().rstrip()
        err = overr.getresult().rstrip()

        self.assertGreater(len(out), 0)
        self.assertGreater(len(err), 0)
        
        self.assertTrue(msg.upper().rstrip().startswith(out))
        self.assertTrue(b"stderr".startswith(err))

        
        with p:
            pass


if __name__ == '__main__':
    unittest.main()