Add a partial-line-marker on Apple system logger and optimize the iOS testbed runner

Author: timrid

Lib/_apple_support.py

@@ -37,6 +37,13 @@ def write(self, s):
 
 
 class LogStream(io.RawIOBase):
+    # Marker appended to any log message that does not end with a newline,
+    # so a cooperating log reader can re-join it with the following message
+    # instead of rendering it as a spurious standalone line. Placing the
+    # marker after the data also prevents the system log from stripping any
+    # trailing whitespace. Uses ASCII Unit Separator (0x1f).
+    PARTIAL_LINE_MARKER = b"\x1f"
+
     def __init__(self, log_write, level):
         self.log_write = log_write
         self.level = level
@@ -59,8 +66,14 @@ def write(self, b):
         # Writing an empty string to the stream should have no effect.
         if b:
             # Encode null bytes using "modified UTF-8" to avoid truncating the
-            # message. This should not affect the return value, as the caller
-            # may be expecting it to match the length of the input.
-            self.log_write(self.level, b.replace(b"\x00", b"\xc0\x80"))
+            # message.
+            data = b.replace(b"\x00", b"\xc0\x80")
+
+            # Append a marker to partial lines (see PARTIAL_LINE_MARKER).
+            if not b.endswith(b"\n"):
+                data += self.PARTIAL_LINE_MARKER
+            self.log_write(self.level, data)
 
+        # Modifications of the changed data should not affect the return value, as
+        # the caller may be expecting it to match the length of the input.
         return len(b)

Lib/test/test_apple.py

@@ -1,11 +1,12 @@
 import unittest
-from _apple_support import SystemLog
+from _apple_support import LogStream, SystemLog
 from test.support import is_apple
 from unittest.mock import Mock, call
 
 if not is_apple:
     raise unittest.SkipTest("Apple-specific")
 
+MARKER = LogStream.PARTIAL_LINE_MARKER
 
 # Test redirection of stdout and stderr to the Apple system log.
 class TestAppleSystemLogOutput(unittest.TestCase):
@@ -52,6 +53,21 @@ def test_simple_str(self):
 
         self.assert_writes([b"hello world\n"])
 
+    def test_partial_line_marker(self):
+        self.log.write("complete\n")
+        self.assert_writes([b"complete\n"])
+
+        self.log.write("partial")
+        self.log.flush()
+        self.assert_writes([b"partial" + MARKER])
+
+        self.log.write("trailing whitespace ")
+        self.log.flush()
+        self.assert_writes([b"trailing whitespace " + MARKER])
+
+        self.log.write("done\n")
+        self.assert_writes([b"done\n"])
+
     def test_buffered_str(self):
         self.log.write("h")
         self.log.write("ello")
@@ -60,7 +76,7 @@ def test_buffered_str(self):
         self.log.write("goodbye.")
         self.log.flush()
 
-        self.assert_writes([b"hello world\n", b"goodbye."])
+        self.assert_writes([b"hello world\n", b"goodbye." + MARKER])
 
     def test_manual_flush(self):
         self.log.write("Hello")
@@ -74,7 +90,7 @@ def test_manual_flush(self):
         self.assert_writes([b"Goodbye world\n"])
 
         self.log.flush()
-        self.assert_writes([b"Hello again"])
+        self.assert_writes([b"Hello again" + MARKER])
 
     def test_non_ascii(self):
         # Spanish
@@ -142,7 +158,7 @@ def test_byteslike_in_buffer(self):
         self.log.buffer.write(b"goodbye")
         self.log.flush()
 
-        self.assert_writes([b"hello", b"goodbye"])
+        self.assert_writes([b"hello" + MARKER, b"goodbye" + MARKER])
 
     def test_non_byteslike_in_buffer(self):
         for obj in ["hello", None, 42]:

Platforms/Apple/testbed/__main__.py

@@ -24,6 +24,23 @@
     r"\s+iOSTestbed\[\d+:\w+\] "  # Process/thread ID
 )
 
+# The system log escapes non-printable bytes using caret notation (e.g. ESC
+# becomes "\^["). This regex matches those sequences so they can be restored.
+LOG_CTRL_CHAR_REGEX = re.compile(r"\\\^(.)")
+
+# Matches the partial-line marker (ASCII Unit Separator, 0x1f) appended by
+# Lib/_apple_support.py to log messages that did not end with a newline,
+# followed by the log-appended newline. Removing both causes the next write
+# to continue on the same line rather than starting a new one.
+LOG_PARTIAL_LINE_REGEX = re.compile(r"\x1f\n$")
+
+
+def _decode_log_ctrl_char(match):
+    char = match.group(1)
+    # Caret notation: "^?" is DEL (0x7f); otherwise XOR the character with
+    # 0x40 (e.g. "^[" -> ESC 0x1b).
+    return "\x7f" if char == "?" else chr(ord(char) ^ 0x40)
+
 
 # Select a simulator device to use.
 def select_simulator_device(platform):
@@ -99,6 +116,16 @@ def xcode_test(location: Path, platform: str, simulator: str, verbose: bool):
     while line := (process.stdout.readline()).decode(*DECODE_ARGS):
         # Strip the timestamp/process prefix from each log line
         line = LOG_PREFIX_REGEX.sub("", line)
+
+        # Restore control characters (e.g. ANSI color escapes) escaped by the
+        # system log.
+        line = LOG_CTRL_CHAR_REGEX.sub(_decode_log_ctrl_char, line)
+
+        # A marker immediately before the message's trailing newline means the
+        # writer did not emit a newline: it is a partial line that should be
+        # joined with the following message rather than shown on its own line.
+        line = LOG_PARTIAL_LINE_REGEX.sub("", line)
+
         sys.stdout.write(line)
         sys.stdout.flush()