Server : Apache/2.4.41 (Ubuntu) System : Linux absol.cf 5.4.0-198-generic #218-Ubuntu SMP Fri Sep 27 20:18:53 UTC 2024 x86_64 User : www-data ( 33) PHP Version : 7.4.33 Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare, Directory :  /usr/share/doc/python-libxml2/examples/

Upload File : current_dir [ Writeable ] document_root [ Writeable ]

#!/usr/bin/python -u
import string, sys, time
try:
    from _thread import get_ident
except:
    from thread import get_ident
from threading import Thread, Lock

import libxml2

THREADS_COUNT = 15

failed = 0

class ErrorHandler:

    def __init__(self):
        self.errors = []
        self.lock = Lock()

    def handler(self,ctx,str):
        self.lock.acquire()
        self.errors.append(str)
        self.lock.release()

def getLineNumbersDefault():
    old = libxml2.lineNumbersDefault(0)
    libxml2.lineNumbersDefault(old)
    return old

def test(expectedLineNumbersDefault):
    time.sleep(1)
    global failed
    # check a per thread-global
    if expectedLineNumbersDefault != getLineNumbersDefault():
        failed = 1
        print("FAILED to obtain correct value for " \
              "lineNumbersDefault in thread %d" % get_ident())
    # check ther global error handler 
    # (which is NOT per-thread in the python bindings)
    try:
        doc = libxml2.parseFile("bad.xml")
    except:
        pass
    else:
        assert "failed"

# global error handler
eh = ErrorHandler()
libxml2.registerErrorHandler(eh.handler,"")

# set on the main thread only
libxml2.lineNumbersDefault(1) 
test(1)
ec = len(eh.errors)
if ec == 0:
    print("FAILED: should have obtained errors")
    sys.exit(1)

ts = []
for i in range(THREADS_COUNT):
    # expect 0 for lineNumbersDefault because
    # the new value has been set on the main thread only
    ts.append(Thread(target=test,args=(0,)))
for t in ts:
    t.start()
for t in ts:
    t.join()

if len(eh.errors) != ec+THREADS_COUNT*ec:
    print("FAILED: did not obtain the correct number of errors")
    sys.exit(1)

# set lineNumbersDefault for future new threads
libxml2.thrDefLineNumbersDefaultValue(1)
ts = []
for i in range(THREADS_COUNT):
    # expect 1 for lineNumbersDefault
    ts.append(Thread(target=test,args=(1,)))
for t in ts:
    t.start()
for t in ts:
    t.join()

if len(eh.errors) != ec+THREADS_COUNT*ec*2:
    print("FAILED: did not obtain the correct number of errors")
    sys.exit(1)

if failed:
    print("FAILED")
    sys.exit(1)

# Memory debug specific
libxml2.cleanupParser()
if libxml2.debugMemory(1) == 0:
    print("OK")
else:
    print("Memory leak %d bytes" % (libxml2.debugMemory(1)))
    libxml2.dumpMemory()