1  #!/usr/bin/env python3
       2  import string, sys, time
       3  try:
       4      from _thread import get_ident
       5  except:
       6      from thread import get_ident
       7  from threading import Thread, Lock
       8  
       9  import setup_test
      10  import libxml2
      11  
      12  # Memory debug specific
      13  libxml2.debugMemory(1)
      14  
      15  THREADS_COUNT = 15
      16  
      17  failed = 0
      18  
      19  class ESC[4;38;5;81mErrorHandler:
      20  
      21      def __init__(self):
      22          self.errors = []
      23          self.lock = Lock()
      24  
      25      def handler(self,ctx,str):
      26          self.lock.acquire()
      27          self.errors.append(str)
      28          self.lock.release()
      29  
      30  def getLineNumbersDefault():
      31      old = libxml2.lineNumbersDefault(0)
      32      libxml2.lineNumbersDefault(old)
      33      return old
      34  
      35  def test(expectedLineNumbersDefault):
      36      time.sleep(1)
      37      global failed
      38      # check a per thread-global
      39      if expectedLineNumbersDefault != getLineNumbersDefault():
      40          failed = 1
      41          print("FAILED to obtain correct value for " \
      42                "lineNumbersDefault in thread %d" % get_ident())
      43      # check ther global error handler 
      44      # (which is NOT per-thread in the python bindings)
      45      try:
      46          doc = libxml2.parseFile("bad.xml")
      47      except:
      48          pass
      49      else:
      50          assert "failed"
      51  
      52  # global error handler
      53  eh = ErrorHandler()
      54  libxml2.registerErrorHandler(eh.handler,"")
      55  
      56  # set on the main thread only
      57  libxml2.lineNumbersDefault(1) 
      58  test(1)
      59  ec = len(eh.errors)
      60  if ec == 0:
      61      print("FAILED: should have obtained errors")
      62      sys.exit(1)
      63  
      64  ts = []
      65  for i in range(THREADS_COUNT):
      66      # expect 0 for lineNumbersDefault because
      67      # the new value has been set on the main thread only
      68      ts.append(Thread(target=test,args=(0,)))
      69  for t in ts:
      70      t.start()
      71  for t in ts:
      72      t.join()
      73  
      74  if len(eh.errors) != ec+THREADS_COUNT*ec:
      75      print("FAILED: did not obtain the correct number of errors")
      76      sys.exit(1)
      77  
      78  # set lineNumbersDefault for future new threads
      79  libxml2.thrDefLineNumbersDefaultValue(1)
      80  ts = []
      81  for i in range(THREADS_COUNT):
      82      # expect 1 for lineNumbersDefault
      83      ts.append(Thread(target=test,args=(1,)))
      84  for t in ts:
      85      t.start()
      86  for t in ts:
      87      t.join()
      88  
      89  if len(eh.errors) != ec+THREADS_COUNT*ec*2:
      90      print("FAILED: did not obtain the correct number of errors")
      91      sys.exit(1)
      92  
      93  if failed:
      94      print("FAILED")
      95      sys.exit(1)
      96  
      97  # Memory debug specific
      98  libxml2.cleanupParser()
      99  # Note that this can leak memory on Windows if the global state
     100  # destructors weren't run yet. They should be called eventually,
     101  # so this leak should be harmless.
     102  if libxml2.debugMemory(1) == 0:
     103      print("OK")
     104  else:
     105      print("Memory leak %d bytes" % (libxml2.debugMemory(1)))