Implement fileno method on stdout and stderr

Author: mhsmith

Lib/_android_support.py

@@ -31,15 +31,17 @@ def init_streams(android_log_write, stdout_prio, stderr_prio):
     logcat = Logcat(android_log_write)
 
     sys.stdout = TextLogStream(
-        stdout_prio, "python.stdout", errors=sys.stdout.errors)
+        stdout_prio, "python.stdout", sys.stdout.fileno(),
+        errors=sys.stdout.errors)
     sys.stderr = TextLogStream(
-        stderr_prio, "python.stderr", errors=sys.stderr.errors)
+        stderr_prio, "python.stderr", sys.stderr.fileno(),
+        errors=sys.stderr.errors)
 
 
 class TextLogStream(io.TextIOWrapper):
-    def __init__(self, prio, tag, **kwargs):
+    def __init__(self, prio, tag, fileno=None, **kwargs):
         kwargs.setdefault("encoding", "UTF-8")
-        super().__init__(BinaryLogStream(prio, tag), **kwargs)
+        super().__init__(BinaryLogStream(prio, tag, fileno), **kwargs)
         self._lock = RLock()
         self._pending_bytes = []
         self._pending_bytes_count = 0
@@ -98,9 +100,10 @@ def line_buffering(self):
 
 
 class BinaryLogStream(io.RawIOBase):
-    def __init__(self, prio, tag):
+    def __init__(self, prio, tag, fileno=None):
         self.prio = prio
         self.tag = tag
+        self._fileno = fileno
 
     def __repr__(self):
         return f"<BinaryLogStream {self.tag!r}>"
@@ -122,6 +125,12 @@ def write(self, b):
             logcat.write(self.prio, self.tag, b)
         return len(b)
 
+    # This is needed by the test suite --timeout option, which uses faulthandler.
+    def fileno(self):
+        if self._fileno is None:
+            raise io.UnsupportedOperation("fileno")
+        return self._fileno
+
 
 # When a large volume of data is written to logcat at once, e.g. when a test
 # module fails in --verbose3 mode, there's a risk of overflowing logcat's own

Lib/test/test_android.py

@@ -19,6 +19,9 @@
 
 api_level = platform.android_ver().api_level
 
+# (name, level, fileno)
+STREAM_INFO = [("stdout", "I", 1), ("stderr", "W", 2)]
+
 
 # Test redirection of stdout and stderr to the Android log.
 @unittest.skipIf(
@@ -94,26 +97,29 @@ def stream_context(self, stream_name, level):
         stack = ExitStack()
         stack.enter_context(self.subTest(stream_name))
         stream = getattr(sys, stream_name)
+        native_stream = getattr(sys, f"__{stream_name}__")
         if isinstance(stream, io.StringIO):
             stack.enter_context(
                 patch(
                     f"sys.{stream_name}",
                     TextLogStream(
-                        prio, f"python.{stream_name}", errors="backslashreplace"
+                        prio, f"python.{stream_name}", native_stream.fileno(),
+                        errors="backslashreplace"
                     ),
                 )
             )
         return stack
 
     def test_str(self):
-        for stream_name, level in [("stdout", "I"), ("stderr", "W")]:
+        for stream_name, level, fileno in STREAM_INFO:
             with self.stream_context(stream_name, level):
                 stream = getattr(sys, stream_name)
                 tag = f"python.{stream_name}"
                 self.assertEqual(f"<TextLogStream '{tag}'>", repr(stream))
 
                 self.assertIs(stream.writable(), True)
                 self.assertIs(stream.readable(), False)
+                self.assertEqual(stream.fileno(), fileno)
                 self.assertEqual("UTF-8", stream.encoding)
                 self.assertIs(stream.line_buffering, True)
                 self.assertIs(stream.write_through, False)
@@ -257,13 +263,14 @@ def __str__(self):
                 write("\n", [s * 51])  # 0 bytes in, 510 bytes out
 
     def test_bytes(self):
-        for stream_name, level in [("stdout", "I"), ("stderr", "W")]:
+        for stream_name, level, fileno in STREAM_INFO:
             with self.stream_context(stream_name, level):
                 stream = getattr(sys, stream_name).buffer
                 tag = f"python.{stream_name}"
                 self.assertEqual(f"<BinaryLogStream '{tag}'>", repr(stream))
                 self.assertIs(stream.writable(), True)
                 self.assertIs(stream.readable(), False)
+                self.assertEqual(stream.fileno(), fileno)
 
                 def write(b, lines=None, *, write_len=None):
                     if write_len is None: