| """Tests for sys.audit and sys.addaudithook | |
| """ | |
| import subprocess | |
| import sys | |
| import unittest | |
| from test import support | |
| from test.support import import_helper | |
| from test.support import os_helper | |
| if not hasattr(sys, "addaudithook") or not hasattr(sys, "audit"): | |
| raise unittest.SkipTest("test only relevant when sys.audit is available") | |
| AUDIT_TESTS_PY = support.findfile("audit-tests.py") | |
| class AuditTest(unittest.TestCase): | |
| maxDiff = None | |
| @support.requires_subprocess() | |
| def run_test_in_subprocess(self, *args): | |
| with subprocess.Popen( | |
| [sys.executable, "-X utf8", AUDIT_TESTS_PY, *args], | |
| encoding="utf-8", | |
| errors="backslashreplace", | |
| stdout=subprocess.PIPE, | |
| stderr=subprocess.PIPE, | |
| ) as p: | |
| p.wait() | |
| return p, p.stdout.read(), p.stderr.read() | |
| def do_test(self, *args): | |
| proc, stdout, stderr = self.run_test_in_subprocess(*args) | |
| sys.stdout.write(stdout) | |
| sys.stderr.write(stderr) | |
| if proc.returncode: | |
| self.fail(stderr) | |
| def run_python(self, *args, expect_stderr=False): | |
| events = [] | |
| proc, stdout, stderr = self.run_test_in_subprocess(*args) | |
| if not expect_stderr or support.verbose: | |
| sys.stderr.write(stderr) | |
| return ( | |
| proc.returncode, | |
| [line.strip().partition(" ") for line in stdout.splitlines()], | |
| stderr, | |
| ) | |
| def test_basic(self): | |
| self.do_test("test_basic") | |
| def test_block_add_hook(self): | |
| self.do_test("test_block_add_hook") | |
| def test_block_add_hook_baseexception(self): | |
| self.do_test("test_block_add_hook_baseexception") | |
| def test_block_add_hook_valueerror(self): | |
| self.do_test("test_block_add_hook_valueerror") | |
| def test_marshal(self): | |
| import_helper.import_module("marshal") | |
| self.do_test("test_marshal") | |
| def test_pickle(self): | |
| import_helper.import_module("pickle") | |
| self.do_test("test_pickle") | |
| def test_monkeypatch(self): | |
| self.do_test("test_monkeypatch") | |
| def test_open(self): | |
| self.do_test("test_open", os_helper.TESTFN) | |
| def test_cantrace(self): | |
| self.do_test("test_cantrace") | |
| def test_mmap(self): | |
| self.do_test("test_mmap") | |
| def test_ctypes_call_function(self): | |
| import_helper.import_module("ctypes") | |
| self.do_test("test_ctypes_call_function") | |
| def test_posixsubprocess(self): | |
| import_helper.import_module("_posixsubprocess") | |
| self.do_test("test_posixsubprocess") | |
| def test_excepthook(self): | |
| returncode, events, stderr = self.run_python("test_excepthook") | |
| if not returncode: | |
| self.fail(f"Expected fatal exception\n{stderr}") | |
| self.assertSequenceEqual( | |
| [("sys.excepthook", " ", "RuntimeError('fatal-error')")], events | |
| ) | |
| def test_unraisablehook(self): | |
| import_helper.import_module("_testcapi") | |
| returncode, events, stderr = self.run_python("test_unraisablehook") | |
| if returncode: | |
| self.fail(stderr) | |
| self.assertEqual(events[0][0], "sys.unraisablehook") | |
| self.assertEqual( | |
| events[0][2], | |
| "RuntimeError('nonfatal-error') Exception ignored for audit hook test", | |
| ) | |
| def test_winreg(self): | |
| import_helper.import_module("winreg") | |
| returncode, events, stderr = self.run_python("test_winreg") | |
| if returncode: | |
| self.fail(stderr) | |
| self.assertEqual(events[0][0], "winreg.OpenKey") | |
| self.assertEqual(events[1][0], "winreg.OpenKey/result") | |
| expected = events[1][2] | |
| self.assertTrue(expected) | |
| self.assertSequenceEqual(["winreg.EnumKey", " ", f"{expected} 0"], events[2]) | |
| self.assertSequenceEqual(["winreg.EnumKey", " ", f"{expected} 10000"], events[3]) | |
| self.assertSequenceEqual(["winreg.PyHKEY.Detach", " ", expected], events[4]) | |
| def test_socket(self): | |
| import_helper.import_module("socket") | |
| returncode, events, stderr = self.run_python("test_socket") | |
| if returncode: | |
| self.fail(stderr) | |
| if support.verbose: | |
| print(*events, sep='\n') | |
| self.assertEqual(events[0][0], "socket.gethostname") | |
| self.assertEqual(events[1][0], "socket.__new__") | |
| self.assertEqual(events[2][0], "socket.bind") | |
| self.assertEndsWith(events[2][2], "('127.0.0.1', 8080)") | |
| def test_gc(self): | |
| returncode, events, stderr = self.run_python("test_gc") | |
| if returncode: | |
| self.fail(stderr) | |
| if support.verbose: | |
| print(*events, sep='\n') | |
| self.assertEqual( | |
| [event[0] for event in events], | |
| ["gc.get_objects", "gc.get_referrers", "gc.get_referents"] | |
| ) | |
| @support.requires_resource('network') | |
| def test_http(self): | |
| import_helper.import_module("http.client") | |
| returncode, events, stderr = self.run_python("test_http_client") | |
| if returncode: | |
| self.fail(stderr) | |
| if support.verbose: | |
| print(*events, sep='\n') | |
| self.assertEqual(events[0][0], "http.client.connect") | |
| self.assertEqual(events[0][2], "www.python.org 80") | |
| self.assertEqual(events[1][0], "http.client.send") | |
| if events[1][2] != '[cannot send]': | |
| self.assertIn('HTTP', events[1][2]) | |
| def test_sqlite3(self): | |
| sqlite3 = import_helper.import_module("sqlite3") | |
| returncode, events, stderr = self.run_python("test_sqlite3") | |
| if returncode: | |
| self.fail(stderr) | |
| if support.verbose: | |
| print(*events, sep='\n') | |
| actual = [ev[0] for ev in events] | |
| expected = ["sqlite3.connect", "sqlite3.connect/handle"] * 2 | |
| if hasattr(sqlite3.Connection, "enable_load_extension"): | |
| expected += [ | |
| "sqlite3.enable_load_extension", | |
| "sqlite3.load_extension", | |
| ] | |
| self.assertEqual(actual, expected) | |
| def test_sys_getframe(self): | |
| returncode, events, stderr = self.run_python("test_sys_getframe") | |
| if returncode: | |
| self.fail(stderr) | |
| if support.verbose: | |
| print(*events, sep='\n') | |
| actual = [(ev[0], ev[2]) for ev in events] | |
| expected = [("sys._getframe", "test_sys_getframe")] | |
| self.assertEqual(actual, expected) | |
| def test_sys_getframemodulename(self): | |
| returncode, events, stderr = self.run_python("test_sys_getframemodulename") | |
| if returncode: | |
| self.fail(stderr) | |
| if support.verbose: | |
| print(*events, sep='\n') | |
| actual = [(ev[0], ev[2]) for ev in events] | |
| expected = [("sys._getframemodulename", "0")] | |
| self.assertEqual(actual, expected) | |
| def test_threading(self): | |
| returncode, events, stderr = self.run_python("test_threading") | |
| if returncode: | |
| self.fail(stderr) | |
| if support.verbose: | |
| print(*events, sep='\n') | |
| actual = [(ev[0], ev[2]) for ev in events] | |
| expected = [ | |
| ("_thread.start_new_thread", "(<test_func>, (), None)"), | |
| ("test.test_func", "()"), | |
| ("_thread.start_joinable_thread", "(<test_func>, 1, None)"), | |
| ("test.test_func", "()"), | |
| ] | |
| self.assertEqual(actual, expected) | |
| def test_wmi_exec_query(self): | |
| import_helper.import_module("_wmi") | |
| returncode, events, stderr = self.run_python("test_wmi_exec_query") | |
| if returncode: | |
| self.fail(stderr) | |
| if support.verbose: | |
| print(*events, sep='\n') | |
| actual = [(ev[0], ev[2]) for ev in events] | |
| expected = [("_wmi.exec_query", "SELECT * FROM Win32_OperatingSystem")] | |
| self.assertEqual(actual, expected) | |
| def test_syslog(self): | |
| syslog = import_helper.import_module("syslog") | |
| returncode, events, stderr = self.run_python("test_syslog") | |
| if returncode: | |
| self.fail(stderr) | |
| if support.verbose: | |
| print('Events:', *events, sep='\n ') | |
| self.assertSequenceEqual( | |
| events, | |
| [('syslog.openlog', ' ', f'python 0 {syslog.LOG_USER}'), | |
| ('syslog.syslog', ' ', f'{syslog.LOG_INFO} test'), | |
| ('syslog.setlogmask', ' ', f'{syslog.LOG_DEBUG}'), | |
| ('syslog.closelog', '', ''), | |
| ('syslog.syslog', ' ', f'{syslog.LOG_INFO} test2'), | |
| ('syslog.openlog', ' ', f'audit-tests.py 0 {syslog.LOG_USER}'), | |
| ('syslog.openlog', ' ', f'audit-tests.py {syslog.LOG_NDELAY} {syslog.LOG_LOCAL0}'), | |
| ('syslog.openlog', ' ', f'None 0 {syslog.LOG_USER}'), | |
| ('syslog.closelog', '', '')] | |
| ) | |
| def test_not_in_gc(self): | |
| returncode, _, stderr = self.run_python("test_not_in_gc") | |
| if returncode: | |
| self.fail(stderr) | |
| def test_time(self): | |
| returncode, events, stderr = self.run_python("test_time", "print") | |
| if returncode: | |
| self.fail(stderr) | |
| if support.verbose: | |
| print(*events, sep='\n') | |
| actual = [(ev[0], ev[2]) for ev in events] | |
| expected = [("time.sleep", "0"), | |
| ("time.sleep", "0.0625"), | |
| ("time.sleep", "-1")] | |
| self.assertEqual(actual, expected) | |
| def test_time_fail(self): | |
| returncode, events, stderr = self.run_python("test_time", "fail", | |
| expect_stderr=True) | |
| self.assertNotEqual(returncode, 0) | |
| self.assertIn('hook failed', stderr.splitlines()[-1]) | |
| def test_sys_monitoring_register_callback(self): | |
| returncode, events, stderr = self.run_python("test_sys_monitoring_register_callback") | |
| if returncode: | |
| self.fail(stderr) | |
| if support.verbose: | |
| print(*events, sep='\n') | |
| actual = [(ev[0], ev[2]) for ev in events] | |
| expected = [("sys.monitoring.register_callback", "(None,)")] | |
| self.assertEqual(actual, expected) | |
| def test_winapi_createnamedpipe(self): | |
| winapi = import_helper.import_module("_winapi") | |
| pipe_name = r"\\.\pipe\LOCAL\test_winapi_createnamed_pipe" | |
| returncode, events, stderr = self.run_python("test_winapi_createnamedpipe", pipe_name) | |
| if returncode: | |
| self.fail(stderr) | |
| if support.verbose: | |
| print(*events, sep='\n') | |
| actual = [(ev[0], ev[2]) for ev in events] | |
| expected = [("_winapi.CreateNamedPipe", f"({pipe_name!r}, 3, 8)")] | |
| self.assertEqual(actual, expected) | |
| def test_assert_unicode(self): | |
| # See gh-126018 | |
| returncode, _, stderr = self.run_python("test_assert_unicode") | |
| if returncode: | |
| self.fail(stderr) | |
| @support.support_remote_exec_only | |
| @support.cpython_only | |
| def test_sys_remote_exec(self): | |
| returncode, events, stderr = self.run_python("test_sys_remote_exec") | |
| self.assertTrue(any(["sys.remote_exec" in event for event in events])) | |
| self.assertTrue(any(["cpython.remote_debugger_script" in event for event in events])) | |
| if returncode: | |
| self.fail(stderr) | |
| def test_import_module(self): | |
| self.do_test("test_import_module") | |
| def test_builtin__import__(self): | |
| self.do_test("test_builtin__import__") | |
| def test_import_statement(self): | |
| self.do_test("test_import_statement") | |
| if __name__ == "__main__": | |
| unittest.main() |