bpo-37363: Add audit events on startup for the run commands (GH-14524)

1 files changed, 36 insertions, 4 deletions

diff --git a/Lib/test/test_embed.py b/Lib/test/test_embed.py
index b2cd550..37f542b 100644
--- a/Lib/test/test_embed.py
+++ b/Lib/test/test_embed.py
@@ -57,7 +57,8 @@ class EmbeddingTestsMixin:
     def tearDown(self):
         os.chdir(self.oldcwd)
 
-    def run_embedded_interpreter(self, *args, env=None):
+    def run_embedded_interpreter(self, *args, env=None,
+                                 timeout=None, returncode=0, input=None):
         """Runs a test in the embedded interpreter"""
         cmd = [self.test_exe]
         cmd.extend(args)
@@ -73,18 +74,18 @@ class EmbeddingTestsMixin:
                              universal_newlines=True,
                              env=env)
         try:
-            (out, err) = p.communicate()
+            (out, err) = p.communicate(input=input, timeout=timeout)
         except:
             p.terminate()
             p.wait()
             raise
-        if p.returncode != 0 and support.verbose:
+        if p.returncode != returncode and support.verbose:
             print(f"--- {cmd} failed ---")
             print(f"stdout:\n{out}")
             print(f"stderr:\n{err}")
             print(f"------")
 
-        self.assertEqual(p.returncode, 0,
+        self.assertEqual(p.returncode, returncode,
                          "bad returncode %d, stderr is %r" %
                          (p.returncode, err))
         return out, err
@@ -955,6 +956,37 @@ class AuditingTests(EmbeddingTestsMixin, unittest.TestCase):
     def test_audit_subinterpreter(self):
         self.run_embedded_interpreter("test_audit_subinterpreter")
 
+    def test_audit_run_command(self):
+        self.run_embedded_interpreter("test_audit_run_command", timeout=3, returncode=1)
+
+    def test_audit_run_file(self):
+        self.run_embedded_interpreter("test_audit_run_file", timeout=3, returncode=1)
+
+    def test_audit_run_interactivehook(self):
+        startup = os.path.join(self.oldcwd, support.TESTFN) + ".py"
+        with open(startup, "w", encoding="utf-8") as f:
+            print("import sys", file=f)
+            print("sys.__interactivehook__ = lambda: None", file=f)
+        try:
+            env = {**remove_python_envvars(), "PYTHONSTARTUP": startup}
+            self.run_embedded_interpreter("test_audit_run_interactivehook", timeout=5,
+                                          returncode=10, env=env)
+        finally:
+            os.unlink(startup)
+
+    def test_audit_run_startup(self):
+        startup = os.path.join(self.oldcwd, support.TESTFN) + ".py"
+        with open(startup, "w", encoding="utf-8") as f:
+            print("pass", file=f)
+        try:
+            env = {**remove_python_envvars(), "PYTHONSTARTUP": startup}
+            self.run_embedded_interpreter("test_audit_run_startup", timeout=5,
+                                          returncode=10, env=env)
+        finally:
+            os.unlink(startup)
+
+    def test_audit_run_stdin(self):
+        self.run_embedded_interpreter("test_audit_run_stdin", timeout=3, returncode=1)
 
 if __name__ == "__main__":
     unittest.main()