• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1'''Tests for WindowsConsoleIO
2'''
3
4import io
5import os
6import sys
7import tempfile
8import unittest
9from test.support import os_helper
10
11if sys.platform != 'win32':
12    raise unittest.SkipTest("test only relevant on win32")
13
14from _testconsole import write_input
15
16ConIO = io._WindowsConsoleIO
17
18class WindowsConsoleIOTests(unittest.TestCase):
19    def test_abc(self):
20        self.assertTrue(issubclass(ConIO, io.RawIOBase))
21        self.assertFalse(issubclass(ConIO, io.BufferedIOBase))
22        self.assertFalse(issubclass(ConIO, io.TextIOBase))
23
24    def test_open_fd(self):
25        self.assertRaisesRegex(ValueError,
26            "negative file descriptor", ConIO, -1)
27
28        with tempfile.TemporaryFile() as tmpfile:
29            fd = tmpfile.fileno()
30            # Windows 10: "Cannot open non-console file"
31            # Earlier: "Cannot open console output buffer for reading"
32            self.assertRaisesRegex(ValueError,
33                "Cannot open (console|non-console file)", ConIO, fd)
34
35        try:
36            f = ConIO(0)
37        except ValueError:
38            # cannot open console because it's not a real console
39            pass
40        else:
41            self.assertTrue(f.readable())
42            self.assertFalse(f.writable())
43            self.assertEqual(0, f.fileno())
44            f.close()   # multiple close should not crash
45            f.close()
46
47        try:
48            f = ConIO(1, 'w')
49        except ValueError:
50            # cannot open console because it's not a real console
51            pass
52        else:
53            self.assertFalse(f.readable())
54            self.assertTrue(f.writable())
55            self.assertEqual(1, f.fileno())
56            f.close()
57            f.close()
58
59        try:
60            f = ConIO(2, 'w')
61        except ValueError:
62            # cannot open console because it's not a real console
63            pass
64        else:
65            self.assertFalse(f.readable())
66            self.assertTrue(f.writable())
67            self.assertEqual(2, f.fileno())
68            f.close()
69            f.close()
70
71    def test_open_name(self):
72        self.assertRaises(ValueError, ConIO, sys.executable)
73
74        f = ConIO("CON")
75        self.assertTrue(f.readable())
76        self.assertFalse(f.writable())
77        self.assertIsNotNone(f.fileno())
78        f.close()   # multiple close should not crash
79        f.close()
80
81        f = ConIO('CONIN$')
82        self.assertTrue(f.readable())
83        self.assertFalse(f.writable())
84        self.assertIsNotNone(f.fileno())
85        f.close()
86        f.close()
87
88        f = ConIO('CONOUT$', 'w')
89        self.assertFalse(f.readable())
90        self.assertTrue(f.writable())
91        self.assertIsNotNone(f.fileno())
92        f.close()
93        f.close()
94
95        # bpo-45354: Windows 11 changed MS-DOS device name handling
96        if sys.getwindowsversion()[:3] < (10, 0, 22000):
97            f = open('C:/con', 'rb', buffering=0)
98            self.assertIsInstance(f, ConIO)
99            f.close()
100
101    @unittest.skipIf(sys.getwindowsversion()[:2] <= (6, 1),
102        "test does not work on Windows 7 and earlier")
103    def test_conin_conout_names(self):
104        f = open(r'\\.\conin$', 'rb', buffering=0)
105        self.assertIsInstance(f, ConIO)
106        f.close()
107
108        f = open('//?/conout$', 'wb', buffering=0)
109        self.assertIsInstance(f, ConIO)
110        f.close()
111
112    def test_conout_path(self):
113        temp_path = tempfile.mkdtemp()
114        self.addCleanup(os_helper.rmtree, temp_path)
115
116        conout_path = os.path.join(temp_path, 'CONOUT$')
117
118        with open(conout_path, 'wb', buffering=0) as f:
119            # bpo-45354: Windows 11 changed MS-DOS device name handling
120            if (6, 1) < sys.getwindowsversion()[:3] < (10, 0, 22000):
121                self.assertIsInstance(f, ConIO)
122            else:
123                self.assertNotIsInstance(f, ConIO)
124
125    def test_write_empty_data(self):
126        with ConIO('CONOUT$', 'w') as f:
127            self.assertEqual(f.write(b''), 0)
128
129    def assertStdinRoundTrip(self, text):
130        stdin = open('CONIN$', 'r')
131        old_stdin = sys.stdin
132        try:
133            sys.stdin = stdin
134            write_input(
135                stdin.buffer.raw,
136                (text + '\r\n').encode('utf-16-le', 'surrogatepass')
137            )
138            actual = input()
139        finally:
140            sys.stdin = old_stdin
141        self.assertEqual(actual, text)
142
143    def test_input(self):
144        # ASCII
145        self.assertStdinRoundTrip('abc123')
146        # Non-ASCII
147        self.assertStdinRoundTrip('ϼўТλФЙ')
148        # Combining characters
149        self.assertStdinRoundTrip('A͏B ﬖ̳AA̝')
150
151    # bpo-38325
152    @unittest.skipIf(True, "Handling Non-BMP characters is broken")
153    def test_input_nonbmp(self):
154        # Non-BMP
155        self.assertStdinRoundTrip('\U00100000\U0010ffff\U0010fffd')
156
157    def test_partial_reads(self):
158        # Test that reading less than 1 full character works when stdin
159        # contains multibyte UTF-8 sequences
160        source = 'ϼўТλФЙ\r\n'.encode('utf-16-le')
161        expected = 'ϼўТλФЙ\r\n'.encode('utf-8')
162        for read_count in range(1, 16):
163            with open('CONIN$', 'rb', buffering=0) as stdin:
164                write_input(stdin, source)
165
166                actual = b''
167                while not actual.endswith(b'\n'):
168                    b = stdin.read(read_count)
169                    actual += b
170
171                self.assertEqual(actual, expected, 'stdin.read({})'.format(read_count))
172
173    # bpo-38325
174    @unittest.skipIf(True, "Handling Non-BMP characters is broken")
175    def test_partial_surrogate_reads(self):
176        # Test that reading less than 1 full character works when stdin
177        # contains surrogate pairs that cannot be decoded to UTF-8 without
178        # reading an extra character.
179        source = '\U00101FFF\U00101001\r\n'.encode('utf-16-le')
180        expected = '\U00101FFF\U00101001\r\n'.encode('utf-8')
181        for read_count in range(1, 16):
182            with open('CONIN$', 'rb', buffering=0) as stdin:
183                write_input(stdin, source)
184
185                actual = b''
186                while not actual.endswith(b'\n'):
187                    b = stdin.read(read_count)
188                    actual += b
189
190                self.assertEqual(actual, expected, 'stdin.read({})'.format(read_count))
191
192    def test_ctrl_z(self):
193        with open('CONIN$', 'rb', buffering=0) as stdin:
194            source = '\xC4\x1A\r\n'.encode('utf-16-le')
195            expected = '\xC4'.encode('utf-8')
196            write_input(stdin, source)
197            a, b = stdin.read(1), stdin.readall()
198            self.assertEqual(expected[0:1], a)
199            self.assertEqual(expected[1:], b)
200
201if __name__ == "__main__":
202    unittest.main()
203