1'''Tests for WindowsConsoleIO
2'''
3
4import io
5import os
6import sys
7import tempfile
8import unittest
9from test import support
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        fd, _ = tempfile.mkstemp()
29        try:
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        finally:
35            os.close(fd)
36
37        try:
38            f = ConIO(0)
39        except ValueError:
40            # cannot open console because it's not a real console
41            pass
42        else:
43            self.assertTrue(f.readable())
44            self.assertFalse(f.writable())
45            self.assertEqual(0, f.fileno())
46            f.close()   # multiple close should not crash
47            f.close()
48
49        try:
50            f = ConIO(1, 'w')
51        except ValueError:
52            # cannot open console because it's not a real console
53            pass
54        else:
55            self.assertFalse(f.readable())
56            self.assertTrue(f.writable())
57            self.assertEqual(1, f.fileno())
58            f.close()
59            f.close()
60
61        try:
62            f = ConIO(2, 'w')
63        except ValueError:
64            # cannot open console because it's not a real console
65            pass
66        else:
67            self.assertFalse(f.readable())
68            self.assertTrue(f.writable())
69            self.assertEqual(2, f.fileno())
70            f.close()
71            f.close()
72
73    def test_open_name(self):
74        self.assertRaises(ValueError, ConIO, sys.executable)
75
76        f = ConIO("CON")
77        self.assertTrue(f.readable())
78        self.assertFalse(f.writable())
79        self.assertIsNotNone(f.fileno())
80        f.close()   # multiple close should not crash
81        f.close()
82
83        f = ConIO('CONIN$')
84        self.assertTrue(f.readable())
85        self.assertFalse(f.writable())
86        self.assertIsNotNone(f.fileno())
87        f.close()
88        f.close()
89
90        f = ConIO('CONOUT$', 'w')
91        self.assertFalse(f.readable())
92        self.assertTrue(f.writable())
93        self.assertIsNotNone(f.fileno())
94        f.close()
95        f.close()
96
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(support.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            if sys.getwindowsversion()[:2] > (6, 1):
120                self.assertIsInstance(f, ConIO)
121            else:
122                self.assertNotIsInstance(f, ConIO)
123
124    def test_write_empty_data(self):
125        with ConIO('CONOUT$', 'w') as f:
126            self.assertEqual(f.write(b''), 0)
127
128    def assertStdinRoundTrip(self, text):
129        stdin = open('CONIN$', 'r')
130        old_stdin = sys.stdin
131        try:
132            sys.stdin = stdin
133            write_input(
134                stdin.buffer.raw,
135                (text + '\r\n').encode('utf-16-le', 'surrogatepass')
136            )
137            actual = input()
138        finally:
139            sys.stdin = old_stdin
140        self.assertEqual(actual, text)
141
142    def test_input(self):
143        # ASCII
144        self.assertStdinRoundTrip('abc123')
145        # Non-ASCII
146        self.assertStdinRoundTrip('ϼўТλФЙ')
147        # Combining characters
148        self.assertStdinRoundTrip('A͏B ﬖ̳AA̝')
149        # Non-BMP
150        self.assertStdinRoundTrip('\U00100000\U0010ffff\U0010fffd')
151
152    def test_partial_reads(self):
153        # Test that reading less than 1 full character works when stdin
154        # contains multibyte UTF-8 sequences
155        source = 'ϼўТλФЙ\r\n'.encode('utf-16-le')
156        expected = 'ϼўТλФЙ\r\n'.encode('utf-8')
157        for read_count in range(1, 16):
158            with open('CONIN$', 'rb', buffering=0) as stdin:
159                write_input(stdin, source)
160
161                actual = b''
162                while not actual.endswith(b'\n'):
163                    b = stdin.read(read_count)
164                    actual += b
165
166                self.assertEqual(actual, expected, 'stdin.read({})'.format(read_count))
167
168    def test_partial_surrogate_reads(self):
169        # Test that reading less than 1 full character works when stdin
170        # contains surrogate pairs that cannot be decoded to UTF-8 without
171        # reading an extra character.
172        source = '\U00101FFF\U00101001\r\n'.encode('utf-16-le')
173        expected = '\U00101FFF\U00101001\r\n'.encode('utf-8')
174        for read_count in range(1, 16):
175            with open('CONIN$', 'rb', buffering=0) as stdin:
176                write_input(stdin, source)
177
178                actual = b''
179                while not actual.endswith(b'\n'):
180                    b = stdin.read(read_count)
181                    actual += b
182
183                self.assertEqual(actual, expected, 'stdin.read({})'.format(read_count))
184
185    def test_ctrl_z(self):
186        with open('CONIN$', 'rb', buffering=0) as stdin:
187            source = '\xC4\x1A\r\n'.encode('utf-16-le')
188            expected = '\xC4'.encode('utf-8')
189            write_input(stdin, source)
190            a, b = stdin.read(1), stdin.readall()
191            self.assertEqual(expected[0:1], a)
192            self.assertEqual(expected[1:], b)
193
194if __name__ == "__main__":
195    unittest.main()
196