Home | History | Annotate | Download | only in tests 
      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()
    100