Reduce annotations, stay bytes longer, make raw_events non-private

python · hauntsaninja · Aug 24, 2024 · Jun 28, 2024 · Jun 28, 2024 · Jun 29, 2024
commit 736d5bc0139ed19d9c4c3592eedf722378c01164
diff --git a/Lib/test/support/strace_helper.py b/Lib/test/support/strace_helper.py
@@ -5,13 +5,12 @@
 from dataclasses import dataclass
 from test import support
 from test.support.script_helper import run_python_until_end
-from typing import Dict, List
 
 _strace_binary = "/usr/bin/strace"
 _syscall_regex = re.compile(
     r"(?P<syscall>[^(]*)\((?P<args>[^)]*)\)\s*[=]\s*(?P<returncode>.+)")
 _returncode_regex = re.compile(
-    r"\+\+\+ exited with (?P<returncode>\d+) \+\+\+")
+    br"\+\+\+ exited with (?P<returncode>\d+) \+\+\+")
 
 # Cached value of whether or not there is a compatible strace binary
 _strace_working: bool | None = None
@@ -20,36 +19,43 @@
 @dataclass
 class StraceEvent:
     syscall: str
-    args: List[str]
+    args: list[str]
     returncode: str
 
 
 @dataclass
 class StraceResult:
     strace_returncode: int
     python_returncode: int
-    _raw_events: str
-    stdout: str
-    stderr: str
 
-    def events(self) -> List[StraceEvent]:
-        """Extract the call list information from _raw_events"""
+    """The event messages generated by strace. This is very similar to the
+    stderr strace produces with returncode marker section removed."""
+    raw_events: bytes
+    stdout: bytes
+    stderr: bytes
+
+    def events(self):
+        """Parse raw_events data into system calls for easier processing.
+
+        This assumes the program under inspection doesn't print any non-utf8
+        strings which would mix into the strace output."""
+        decoded_events = self.raw_events.decode('utf-8')
         matches = [
             _syscall_regex.match(event)
-            for event in self._raw_events.splitlines()
+            for event in decoded_events.splitlines()
         ]
         return [
             StraceEvent(match["syscall"],
                         [arg.strip() for arg in (match["args"].split(","))],
                         match["returncode"]) for match in matches if match
         ]
 
-    def sections(self) -> Dict[str:List[StraceEvent]]:
+    def sections(self):
         """Find all "MARK <X>" writes and use them to make groups of events.
 
-        This is useful to avoid variable / overhead strace events, like that
-        at interpreter startup, so a test can just check does the small case
-        under study work."""
+        This is useful to avoid variable / overhead events, like those at
+        interpreter startup or when opening a file so a test can verify just
+        the small case under study."""
         current_section = "__startup"
         sections = {current_section: []}
         for event in self.events():
@@ -68,20 +74,17 @@ def sections(self) -> Dict[str:List[StraceEvent]]:
 
 
 @support.requires_subprocess()
-def strace_python(code: str,
-                  strace_flags: List[str],
-                  check: bool = True) -> StraceResult:
+def strace_python(code, strace_flags, check = True):
     """Run strace and return the trace.
 
-    Sets strace_returncode and python_returncode to `-1` on error
-    """
+    Sets strace_returncode and python_returncode to `-1` on error."""
     res = None
 
     def _make_error(reason, details):
         return StraceResult(
             strace_returncode=-1,
             python_returncode=-1,
-            _raw_events=f"error({reason},details={details}) = -1",
+            raw_events=f"error({reason},details={details}) = -1".encode('utf-8'),
             stdout=res.out if res else "",
             stderr=res.err if res else "")
 
@@ -98,12 +101,11 @@ def _make_error(reason, details):
         res.fail(cmd_line)
 
     # Get out program returncode
-    decoded = res.err.decode().strip()
-
-    output = decoded.rsplit("\n", 1)
+    stripped = res.err.strip()
+    output = stripped.rsplit(b"\n", 1)
     if len(output) != 2:
         return _make_error("Expected strace events and exit code line",
-                           decoded[-50:])
+                           stripped[-50:])
 
     returncode_match = _returncode_regex.match(output[1])
     if not returncode_match:
@@ -116,13 +118,12 @@ def _make_error(reason, details):
 
     return StraceResult(strace_returncode=res.rc,
                         python_returncode=python_returncode,
-                        _raw_events=output[0],
+                        raw_events=output[0],
                         stdout=res.out,
                         stderr=res.err)
 
 
-def _get_events(code: str, strace_flags: List[str], prelude: str,
-                cleanup: str) -> List[StraceEvent]:
+def _get_events(code, strace_flags, prelude, cleanup):
     # NOTE: The flush is currently required to prevent the prints from getting
     # buffered and done all at once at exit
     prelude = textwrap.dedent(prelude)
@@ -142,10 +143,7 @@ def _get_events(code: str, strace_flags: List[str], prelude: str,
     return all_sections['code']
 
 
-def get_syscalls(code: str,
-                 strace_flags: List[str],
-                 prelude: str = "",
-                 cleanup: str = "") -> List[str]:
+def get_syscalls(code, strace_flags, prelude = "", cleanup = ""):
     """Get the syscalls which a given chunk of python code generates"""
     events = _get_events(code, strace_flags, prelude=prelude, cleanup=cleanup)
     return [ev.syscall for ev in events]

@@ -3461,7 +3461,7 @@ def test_vfork_used_when_expected(self):
                 strace_args
             )
             # Match both vfork() and clone(..., flags=...|CLONE_VFORK|...)
-            self.assertRegex(vfork_result._raw_events, r"(?i)vfork")
+            self.assertRegex(vfork_result.raw_events, rb"(?i)vfork")
             # Do NOT check that fork() or other clones did not happen.
             # If the OS denys the vfork it'll fallback to plain fork().
 
@@ -3487,7 +3487,7 @@ def test_vfork_used_when_expected(self):
                     strace_args
                 )
                 # Ensure neither vfork() or clone(..., flags=...|CLONE_VFORK|...).
-                self.assertNotRegex(non_vfork_result._raw_events, r"(?i)vfork")
+                self.assertNotRegex(non_vfork_result.raw_events, rb"(?i)vfork")
 
 
 @unittest.skipUnless(mswindows, "Windows specific tests")