gh-136336: add tests for PySys_Audit() and PySys_AuditTuple()

Author: alex-semenyuk

Lib/test/test_capi/test_sys.py

@@ -12,11 +12,6 @@
 NULL = None
 
 class CAPITest(unittest.TestCase):
-    # TODO: Test the following functions:
-    #
-    #   PySys_Audit()
-    #   PySys_AuditTuple()
-
     maxDiff = None
 
     @unittest.skipIf(_testlimitedcapi is None, 'need _testlimitedcapi module')
@@ -211,6 +206,142 @@ def test_sys_writestderr(self):
         # Test PySys_WriteStderr()
         self._test_sys_writestream('PySys_WriteStderr', 'stderr')
 
+    @unittest.skipIf(_testlimitedcapi is None, 'need _testlimitedcapi module')
+    def test_sys_audit(self):
+        sys_audit = _testlimitedcapi.sys_audit
+
+        audit_events = []
+        def audit_hook(event, args):
+            audit_events.append((event, args))
+            return None
+
+        import sys
+        sys.addaudithook(audit_hook)
+
+        try:
+            result = sys_audit("cpython.run_command", "")
+            self.assertEqual(result, 0)
+            self.assertEqual(len(audit_events), 1)
+            self.assertEqual(audit_events[-1][0], "cpython.run_command")
+            self.assertEqual(audit_events[-1][1], ())
+
+            result = sys_audit("open", "OOO", "test.txt", "r", 0)
+            self.assertEqual(result, 0)
+            self.assertEqual(len(audit_events), 2)
+            self.assertEqual(audit_events[-1][0], "open")
+            self.assertEqual(len(audit_events[-1][1]), 3)
+            self.assertEqual(audit_events[-1][1][0], "test.txt")
+            self.assertEqual(audit_events[-1][1][1], "r")
+            self.assertEqual(audit_events[-1][1][2], 0)
+
+            test_dict = {"key": "value"}
+            test_list = [1, 2, 3]
+            result = sys_audit("test.objects", "OO", test_dict, test_list)
+            self.assertEqual(result, 0)
+            self.assertEqual(len(audit_events), 3)
+            self.assertEqual(audit_events[-1][0], "test.objects")
+            self.assertEqual(audit_events[-1][1][0], test_dict)
+            self.assertEqual(audit_events[-1][1][1], test_list)
+
+            result = sys_audit("test.mixed_types", "OOO", "string", 42, 123456789)
+            self.assertEqual(result, 0)
+            self.assertEqual(len(audit_events), 4)
+            self.assertEqual(audit_events[-1][0], "test.mixed_types")
+            self.assertEqual(audit_events[-1][1][0], "string")
+            self.assertEqual(audit_events[-1][1][1], 42)
+            self.assertEqual(audit_events[-1][1][2], 123456789)
+
+        finally:
+            sys.audit_hooks = []
+
+        result = sys_audit("cpython.run_file", "")
+        self.assertEqual(result, 0)
+
+        result = sys_audit("os.chdir", "(O)", "/tmp")
+        self.assertEqual(result, 0)
+
+        result = sys_audit("ctypes.dlopen", "O", "libc.so.6")
+        self.assertEqual(result, 0)
+
+        self.assertRaises(TypeError, sys_audit, 123, "O", "arg")
+        self.assertRaises(TypeError, sys_audit, None, "O", "arg")
+        self.assertRaises(TypeError, sys_audit, ["not", "a", "string"], "O", "arg")
+
+        self.assertRaises(TypeError, sys_audit, "test.event", 456, "arg")
+        self.assertRaises(TypeError, sys_audit, "test.event", None, "arg")
+        self.assertRaises(TypeError, sys_audit, "test.event", {"format": "string"}, "arg")
+
+    @unittest.skipIf(_testlimitedcapi is None, 'need _testlimitedcapi module')
+    def test_sys_audittuple(self):
+        sys_audittuple = _testlimitedcapi.sys_audittuple
+
+        # Test with audit hook to verify internal behavior
+        audit_events = []
+        def audit_hook(event, args):
+            audit_events.append((event, args))
+            return None
+
+        import sys
+        sys.addaudithook(audit_hook)
+
+        try:
+            result = sys_audittuple("cpython.run_command", ())
+            self.assertEqual(result, 0)
+            self.assertEqual(len(audit_events), 1)
+            self.assertEqual(audit_events[-1][0], "cpython.run_command")
+            self.assertEqual(audit_events[-1][1], ())
+
+            result = sys_audittuple("os.chdir", ("/tmp",))
+            self.assertEqual(result, 0)
+            self.assertEqual(len(audit_events), 2)
+            self.assertEqual(audit_events[-1][0], "os.chdir")
+            self.assertEqual(audit_events[-1][1], ("/tmp",))
+
+            result = sys_audittuple("open", ("test.txt", "r", 0))
+            self.assertEqual(result, 0)
+            self.assertEqual(len(audit_events), 3)
+            self.assertEqual(audit_events[-1][0], "open")
+            self.assertEqual(audit_events[-1][1], ("test.txt", "r", 0))
+
+            test_dict = {"key": "value"}
+            test_list = [1, 2, 3]
+            result = sys_audittuple("test.objects", (test_dict, test_list))
+            self.assertEqual(result, 0)
+            self.assertEqual(len(audit_events), 4)
+            self.assertEqual(audit_events[-1][0], "test.objects")
+            self.assertEqual(audit_events[-1][1][0], test_dict)
+            self.assertEqual(audit_events[-1][1][1], test_list)
+
+            result = sys_audittuple("test.complex", ("text", 3.14, True, None))
+            self.assertEqual(result, 0)
+            self.assertEqual(len(audit_events), 5)
+            self.assertEqual(audit_events[-1][0], "test.complex")
+            self.assertEqual(audit_events[-1][1][0], "text")
+            self.assertEqual(audit_events[-1][1][1], 3.14)
+            self.assertEqual(audit_events[-1][1][2], True)
+            self.assertEqual(audit_events[-1][1][3], None)
+
+        finally:
+            sys.audit_hooks = []
+
+        result = sys_audittuple("cpython.run_file", ())
+        self.assertEqual(result, 0)
+
+        result = sys_audittuple("ctypes.dlopen", ("libc.so.6",))
+        self.assertEqual(result, 0)
+
+        result = sys_audittuple("sqlite3.connect", ("test.db",))
+        self.assertEqual(result, 0)
+
+        self.assertRaises(TypeError, sys_audittuple, 123, ("arg",))
+        self.assertRaises(TypeError, sys_audittuple, None, ("arg",))
+        self.assertRaises(TypeError, sys_audittuple, ["not", "a", "string"], ("arg",))
+
+        self.assertRaises(TypeError, sys_audittuple, "test.event", "not_a_tuple")
+        self.assertRaises(TypeError, sys_audittuple, "test.event", ["list", "not", "tuple"])
+        self.assertRaises(TypeError, sys_audittuple, "test.event", {"dict": "not_tuple"})
+        self.assertRaises(TypeError, sys_audittuple, "test.event", None)
+
 
 if __name__ == "__main__":
     unittest.main()

Modules/_testlimitedcapi/sys.c

@@ -5,6 +5,7 @@
 #endif
 #include "parts.h"
 #include "util.h"
+#include "audit.h"
 
 
 static PyObject *
@@ -106,6 +107,48 @@ sys_getxoptions(PyObject *Py_UNUSED(module), PyObject *Py_UNUSED(ignored))
     return Py_XNewRef(result);
 }
 
+static PyObject *
+sys_audit(PyObject *Py_UNUSED(module), PyObject *args)
+{
+    const char *event;
+    const char *argFormat;
+    PyObject *arg1 = NULL, *arg2 = NULL, *arg3 = NULL;
+
+    if (!PyArg_ParseTuple(args, "ss|OOO", &event, &argFormat, &arg1, &arg2, &arg3)) {
+        return NULL;
+    }
+
+    int result;
+    if (arg1 == NULL) {
+        result = PySys_Audit(event, argFormat);
+    } else if (arg2 == NULL) {
+        result = PySys_Audit(event, argFormat, arg1);
+    } else if (arg3 == NULL) {
+        result = PySys_Audit(event, argFormat, arg1, arg2);
+    } else {
+        result = PySys_Audit(event, argFormat, arg1, arg2, arg3);
+    }
+
+    RETURN_INT(result);
+}
+
+static PyObject *
+sys_audittuple(PyObject *Py_UNUSED(module), PyObject *args)
+{
+    const char *event;
+    PyObject *tuple_args;
+
+    if (!PyArg_ParseTuple(args, "sO", &event, &tuple_args)) {
+        return NULL;
+    }
+
+    if (!PyTuple_Check(tuple_args)) {
+        PyErr_SetString(PyExc_TypeError, "second argument must be a tuple");
+        return NULL;
+    }
+
+    RETURN_INT(PySys_AuditTuple(event, tuple_args));
+}
 
 static PyMethodDef test_methods[] = {
     {"sys_getattr", sys_getattr, METH_O},
@@ -115,6 +158,8 @@ static PyMethodDef test_methods[] = {
     {"sys_getobject", sys_getobject, METH_O},
     {"sys_setobject", sys_setobject, METH_VARARGS},
     {"sys_getxoptions", sys_getxoptions, METH_NOARGS},
+    {"sys_audit", sys_audit, METH_VARARGS},
+    {"sys_audittuple", sys_audittuple, METH_VARARGS},
     {NULL},
 };