python/tests/thread2.py

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