python/tests/thread2.py

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