| import os | |
| import subprocess | |
| import sys | |
| import unittest | |
| from textwrap import dedent | |
| from test import support | |
| from test.support import os_helper, requires_resource | |
| from test.support.os_helper import TESTFN, TESTFN_ASCII | |
| if sys.platform != "win32": | |
| raise unittest.SkipTest("windows related tests") | |
| import _winapi | |
| import msvcrt | |
| class TestFileOperations(unittest.TestCase): | |
| def test_locking(self): | |
| with open(TESTFN, "w") as f: | |
| self.addCleanup(os_helper.unlink, TESTFN) | |
| msvcrt.locking(f.fileno(), msvcrt.LK_LOCK, 1) | |
| self.assertRaises(OSError, msvcrt.locking, f.fileno(), msvcrt.LK_NBLCK, 1) | |
| def test_unlockfile(self): | |
| with open(TESTFN, "w") as f: | |
| self.addCleanup(os_helper.unlink, TESTFN) | |
| msvcrt.locking(f.fileno(), msvcrt.LK_LOCK, 1) | |
| msvcrt.locking(f.fileno(), msvcrt.LK_UNLCK, 1) | |
| msvcrt.locking(f.fileno(), msvcrt.LK_LOCK, 1) | |
| def test_setmode(self): | |
| with open(TESTFN, "w") as f: | |
| self.addCleanup(os_helper.unlink, TESTFN) | |
| msvcrt.setmode(f.fileno(), os.O_BINARY) | |
| msvcrt.setmode(f.fileno(), os.O_TEXT) | |
| def test_open_osfhandle(self): | |
| h = _winapi.CreateFile(TESTFN_ASCII, _winapi.GENERIC_WRITE, 0, 0, 1, 128, 0) | |
| self.addCleanup(os_helper.unlink, TESTFN_ASCII) | |
| try: | |
| fd = msvcrt.open_osfhandle(h, os.O_RDONLY) | |
| h = None | |
| os.close(fd) | |
| finally: | |
| if h: | |
| _winapi.CloseHandle(h) | |
| def test_get_osfhandle(self): | |
| with open(TESTFN, "w") as f: | |
| self.addCleanup(os_helper.unlink, TESTFN) | |
| msvcrt.get_osfhandle(f.fileno()) | |
| c = '\u5b57' # unicode CJK char (meaning 'character') for 'wide-char' tests | |
| c_encoded = b'\x57\x5b' # utf-16-le (which windows internally used) encoded char for this CJK char | |
| def has_console(): | |
| # A process created without a console (for example by pythonw.exe, or with | |
| # the DETACHED_PROCESS creation flag) cannot write to the console. | |
| try: | |
| with open('CONOUT$', 'w'): | |
| return True | |
| except OSError: | |
| return False | |
| requires_console = unittest.skipUnless(has_console(), 'requires a console') | |
| class TestConsoleIO(unittest.TestCase): | |
| # CREATE_NEW_CONSOLE creates a "popup" window. | |
| @requires_resource('gui') | |
| def run_in_separated_process(self, code): | |
| # Run test in a separated process to avoid stdin conflicts. | |
| # See: gh-110147 | |
| cmd = [sys.executable, '-c', code] | |
| try: | |
| subprocess.run(cmd, check=True, capture_output=True, | |
| creationflags=subprocess.CREATE_NEW_CONSOLE) | |
| except subprocess.CalledProcessError as exc: | |
| support.skip_on_low_desktop_heap_memory_subprocess(exc.returncode) | |
| raise | |
| def test_kbhit(self): | |
| code = dedent(''' | |
| import msvcrt | |
| assert msvcrt.kbhit() == 0 | |
| ''') | |
| self.run_in_separated_process(code) | |
| def test_getch(self): | |
| msvcrt.ungetch(b'c') | |
| self.assertEqual(msvcrt.getch(), b'c') | |
| def check_getwch(self, funcname): | |
| code = dedent(f''' | |
| import msvcrt | |
| from _testconsole import write_input | |
| with open("CONIN$", "rb", buffering=0) as stdin: | |
| write_input(stdin, {ascii(c_encoded)}) | |
| assert msvcrt.{funcname}() == "{c}" | |
| ''') | |
| self.run_in_separated_process(code) | |
| def test_getwch(self): | |
| self.check_getwch('getwch') | |
| def test_getche(self): | |
| msvcrt.ungetch(b'c') | |
| self.assertEqual(msvcrt.getche(), b'c') | |
| def test_getwche(self): | |
| self.check_getwch('getwche') | |
| @requires_console | |
| def test_putch(self): | |
| msvcrt.putch(b'c') | |
| @requires_console | |
| def test_putwch(self): | |
| msvcrt.putwch(c) | |
| def test_putch_without_console(self): | |
| # gh-69573: putch() and putwch() must report the error instead of | |
| # silently ignoring it when the process has no console attached. | |
| code = dedent(''' | |
| import msvcrt | |
| import sys | |
| for name, arg in (('putch', b'c'), ('putwch', 'c')): | |
| func = getattr(msvcrt, name) | |
| try: | |
| func(arg) | |
| except OSError: | |
| pass | |
| else: | |
| sys.exit(f'msvcrt.{name}() did not raise OSError') | |
| ''') | |
| # DETACHED_PROCESS: the child process is created without a console. | |
| proc = subprocess.run([sys.executable, '-c', code], | |
| creationflags=subprocess.DETACHED_PROCESS, | |
| capture_output=True, text=True) | |
| self.assertEqual(proc.returncode, 0, proc.stderr) | |
| class TestOther(unittest.TestCase): | |
| def test_heap_min(self): | |
| try: | |
| msvcrt.heapmin() | |
| except OSError: | |
| pass | |
| if __name__ == "__main__": | |
| unittest.main() |