Return to thread2.py CVS log

Up to [ELWIX - Embedded LightWeight unIX -] / embedaddon / libxml2 / python / tests

libxml2 2.9.1

    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()