text: """This script contains the actual auditing tests.
  
  It should not be imported directly, but should be run by the test_audit
  module with arguments identifying each test.
  
  """
  
  import contextlib
  import sys
  
  
  class TestHook:
      """Used in standard hook tests to collect any logged events.
  
      Should be used in a with block to ensure that it has no impact
      after the test completes.
      """
  
      def __init__(self, raise_on_events=None, exc_type=RuntimeError):
          self.raise_on_events = raise_on_events or ()
          self.exc_type = exc_type
          self.seen = []
          self.closed = False
  
      def __enter__(self, *a):
          sys.addaudithook(self)
          return self
  
      def __exit__(self, *a):
          self.close()
  
      def close(self):
          self.closed = True
  
      @property
      def seen_events(self):
          return [i[0] for i in self.seen]
  
      def __call__(self, event, args):
          if self.closed:
              return
          self.seen.append((event, args))
          if event in self.raise_on_events:
              raise self.exc_type("saw event " + event)
  
  
  # Simple helpers, since we are not in unittest here
  def assertEqual(x, y):
      if x != y:
          raise AssertionError(f"{x!r} should equal {y!r}")
  
  
  def assertIn(el, series):
      if el not in series:
          raise AssertionError(f"{el!r} should be in {series!r}")
  
  
  def assertNotIn(el, series):
      if el in series:
          raise AssertionError(f"{el!r} should not be in {series!r}")
  
  
  def assertSequenceEqual(x, y):
      if len(x) != len(y):
          raise AssertionError(f"{x!r} should equal {y!r}")
      if any(ix != iy for ix, iy in zip(x, y)):
          raise AssertionError(f"{x!r} should equal {y!r}")
  
  
  @contextlib.contextmanager
  def assertRaises(ex_type):
      try:
          yield
          assert False, f"expected {ex_type}"
      except BaseException as ex:
          if isinstance(ex, AssertionError):
              raise
          assert type(ex) is ex_type, f"{ex} should be {ex_type}"
  
  
  def test_basic():
      with TestHook() as hook:
          sys.audit("test_event", 1, 2, 3)
          assertEqual(hook.seen[0][0], "test_event")
          assertEqual(hook.seen[0][1], (1, 2, 3))
  
  
  def test_block_add_hook():
      # Raising an exception should prevent a new hook from being added,
      # but will not propagate out.
      with TestHook(raise_on_events="sys.addaudithook") as hook1:
          with TestHook() as hook2:
              sys.audit("test_event")
              assertIn("test_event", hook1.seen_events)
              assertNotIn("test_event", hook2.seen_events)
  
  
  def test_block_add_hook_baseexception():
      # Raising BaseException will propagate out when adding a hook
      with assertRaises(BaseException):
          with TestHook(
              raise_on_events="sys.addaudithook", exc_type=BaseException
          ) as hook1:
              # Adding this next hook should raise BaseException
              with TestHook() as hook2:
                  pass
  
  
  def test_pickle():
      import pickle
  
      class PicklePrint:
          def __reduce_ex__(self, p):
              return str, ("Pwned!",)
  
      payload_1 = pickle.dumps(PicklePrint())
      payload_2 = pickle.dumps(("a", "b", "c", 1, 2, 3))
  
      # Before we add the hook, ensure our malicious pickle loads
      assertEqual("Pwned!", pickle.loads(payload_1))
  
      with TestHook(raise_on_events="pickle.find_class") as hook:
          with assertRaises(RuntimeError):
              # With the hook enabled, loading globals is not allowed
              pickle.loads(payload_1)
          # pickles with no globals are okay
          pickle.loads(payload_2)
  
  
  def test_monkeypatch():
      class A:
          pass
  
      class B:
          pass
  
      class C(A):
          pass
  
      a = A()
  
      with TestHook() as hook:
          # Catch name changes
          C.__name__ = "X"
          # Catch type changes
          C.__bases__ = (B,)
          # Ensure bypassing __setattr__ is still caught
          type.__dict__["__bases__"].__set__(C, (B,))
          # Catch attribute replacement
          C.__init__ = B.__init__
          # Catch attribute addition
          C.new_attr = 123
          # Catch class changes
          a.__class__ = B
  
      actual = [(a[0], a[1]) for e, a in hook.seen if e == "object.__setattr__"]
      assertSequenceEqual(
          [(C, "__name__"), (C, "__bases__"), (C, "__bases__"), (a, "__class__")], actual
      )
  
  
  def test_open():
      # SSLContext.load_dh_params uses _Py_fopen_obj rather than normal open()
      try:
          import ssl
  
          load_dh_params = ssl.create_default_context().load_dh_params
      except ImportError:
          load_dh_params = None
  
      # Try a range of "open" functions.
      # All of them should fail
      with TestHook(raise_on_events={"open"}) as hook:
          for fn, *args in [
              (open, sys.argv[2], "r"),
              (open, sys.executable, "rb"),
              (open, 3, "wb"),
              (open, sys.argv[2], "w", -1, None, None, None, False, lambda *a: 1),
              (load_dh_params, sys.argv[2]),
          ]:
              if not fn:
                  continue
              with assertRaises(RuntimeError):
                  fn(*args)
  
      actual_mode = [(a[0], a[1]) for e, a in hook.seen if e == "open" and a[1]]
      actual_flag = [(a[0], a[2]) for e, a in hook.seen if e == "open" and not a[1]]
      assertSequenceEqual(
          [
              i
              for i in [
                  (sys.argv[2], "r"),
                  (sys.executable, "r"),
                  (3, "w"),
                  (sys.argv[2], "w"),
                  (sys.argv[2], "rb") if load_dh_params else None,
              ]
              if i is not None
          ],
          actual_mode,
      )
      assertSequenceEqual([], actual_flag)
  
  
  def test_cantrace():
      traced = []
  
      def trace(frame, event, *args):
          if frame.f_code == TestHook.__call__.__code__:
              traced.append(event)
  
      old = sys.settrace(trace)
      try:
          with TestHook() as hook:
              # No traced call
              eval("1")
  
              # No traced call
              hook.__cantrace__ = False
              eval("2")
  
              # One traced call
              hook.__cantrace__ = True
              eval("3")
  
              # Two traced calls (writing to private member, eval)
              hook.__cantrace__ = 1
              eval("4")
  
              # One traced call (writing to private member)
              hook.__cantrace__ = 0
      finally:
          sys.settrace(old)
  
      assertSequenceEqual(["call"] * 4, traced)
  
  
  def test_mmap():
      import mmap
  
      with TestHook() as hook:
          mmap.mmap(-1, 8)
          assertEqual(hook.seen[0][1][:2], (-1, 8))
  
  
  def test_excepthook():
      def excepthook(exc_type, exc_value, exc_tb):
          if exc_type is not RuntimeError:
              sys.__excepthook__(exc_type, exc_value, exc_tb)
  
      def hook(event, args):
          if event == "sys.excepthook":
              if not isinstance(args[2], args[1]):
                  raise TypeError(f"Expected isinstance({args[2]!r}, " f"{args[1]!r})")
              if args[0] != excepthook:
                  raise ValueError(f"Expected {args[0]} == {excepthook}")
              print(event, repr(args[2]))
  
      sys.addaudithook(hook)
      sys.excepthook = excepthook
      raise RuntimeError("fatal-error")
  
  
  def test_unraisablehook():
      from _testcapi import write_unraisable_exc
  
      def unraisablehook(hookargs):
          pass
  
      def hook(event, args):
          if event == "sys.unraisablehook":
              if args[0] != unraisablehook:
                  raise ValueError(f"Expected {args[0]} == {unraisablehook}")
              print(event, repr(args[1].exc_value), args[1].err_msg)
  
      sys.addaudithook(hook)
      sys.unraisablehook = unraisablehook
      write_unraisable_exc(RuntimeError("nonfatal-error"), "for audit hook test", None)
  
  
  def test_winreg():
      from winreg import OpenKey, EnumKey, CloseKey, HKEY_LOCAL_MACHINE
  
      def hook(event, args):
          if not event.startswith("winreg."):
              return
          print(event, *args)
  
      sys.addaudithook(hook)
  
      k = OpenKey(HKEY_LOCAL_MACHINE, "Software")
      EnumKey(k, 0)
      try:
          EnumKey(k, 10000)
      except OSError:
          pass
      else:
          raise RuntimeError("Expected EnumKey(HKLM, 10000) to fail")
  
      kv = k.Detach()
      CloseKey(kv)
  
  
  def test_socket():
      import socket
  
      def hook(event, args):
          if event.startswith("socket."):
              print(event, *args)
  
      sys.addaudithook(hook)
  
      socket.gethostname()
  
      # Don't care if this fails, we just want the audit message
      sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
      try:
          # Don't care if this fails, we just want the audit message
          sock.bind(('127.0.0.1', 8080))
      except Exception:
          pass
      finally:
          sock.close()
  
  
  if __name__ == "__main__":
      from test.support import suppress_msvcrt_asserts
  
      suppress_msvcrt_asserts()
  
      test = sys.argv[1]
      globals()[test]()
  