def __call__(self, result = None):
# For the COM suite's sake, always ensure we don't leak
# gateways/interfaces
from pythoncom import _GetInterfaceCount, _GetGatewayCount
gc.collect()
ni = _GetInterfaceCount()
ng = _GetGatewayCount()
self.real_test(result)
# Failed - no point checking anything else
if result.shouldStop or not result.wasSuccessful():
return
self._do_leak_tests(result)
gc.collect()
lost_i = _GetInterfaceCount() - ni
lost_g = _GetGatewayCount() - ng
if lost_i or lost_g:
msg = "%d interface objects and %d gateway objects leaked" \
% (lost_i, lost_g)
exc = AssertionError(msg)
result.addFailure(self.real_test, (exc.__class__, exc, None))
python类_GetGatewayCount()的实例源码
def test(fn):
print "The main thread is %d" % (win32api.GetCurrentThreadId())
GIT = CreateGIT()
interp = win32com.client.Dispatch("Python.Interpreter")
cookie = GIT.RegisterInterfaceInGlobal(interp._oleobj_, pythoncom.IID_IDispatch)
events = fn(4, cookie)
numFinished = 0
while 1:
try:
rc = win32event.MsgWaitForMultipleObjects(events, 0, 2000, win32event.QS_ALLINPUT)
if rc >= win32event.WAIT_OBJECT_0 and rc < win32event.WAIT_OBJECT_0+len(events):
numFinished = numFinished + 1
if numFinished >= len(events):
break
elif rc==win32event.WAIT_OBJECT_0 + len(events): # a message
# This is critical - whole apartment model demo will hang.
pythoncom.PumpWaitingMessages()
else: # Timeout
print "Waiting for thread to stop with interfaces=%d, gateways=%d" % (pythoncom._GetInterfaceCount(), pythoncom._GetGatewayCount())
except KeyboardInterrupt:
break
GIT.RevokeInterfaceFromGlobal(cookie)
del interp
del GIT
def test(fn):
print "The main thread is %d" % (win32api.GetCurrentThreadId())
GIT = CreateGIT()
interp = win32com.client.Dispatch("Python.Interpreter")
cookie = GIT.RegisterInterfaceInGlobal(interp._oleobj_, pythoncom.IID_IDispatch)
events = fn(4, cookie)
numFinished = 0
while 1:
try:
rc = win32event.MsgWaitForMultipleObjects(events, 0, 2000, win32event.QS_ALLINPUT)
if rc >= win32event.WAIT_OBJECT_0 and rc < win32event.WAIT_OBJECT_0+len(events):
numFinished = numFinished + 1
if numFinished >= len(events):
break
elif rc==win32event.WAIT_OBJECT_0 + len(events): # a message
# This is critical - whole apartment model demo will hang.
pythoncom.PumpWaitingMessages()
else: # Timeout
print "Waiting for thread to stop with interfaces=%d, gateways=%d" % (pythoncom._GetInterfaceCount(), pythoncom._GetGatewayCount())
except KeyboardInterrupt:
break
GIT.RevokeInterfaceFromGlobal(cookie)
del interp
del GIT
def __call__(self, result = None):
# For the COM suite's sake, always ensure we don't leak
# gateways/interfaces
from pythoncom import _GetInterfaceCount, _GetGatewayCount
gc.collect()
ni = _GetInterfaceCount()
ng = _GetGatewayCount()
self.real_test(result)
# Failed - no point checking anything else
if result.shouldStop or not result.wasSuccessful():
return
self._do_leak_tests(result)
gc.collect()
lost_i = _GetInterfaceCount() - ni
lost_g = _GetGatewayCount() - ng
if lost_i or lost_g:
msg = "%d interface objects and %d gateway objects leaked" \
% (lost_i, lost_g)
exc = AssertionError(msg)
result.addFailure(self.real_test, (exc.__class__, exc, None))
def test(fn):
print "The main thread is %d" % (win32api.GetCurrentThreadId())
GIT = CreateGIT()
interp = win32com.client.Dispatch("Python.Interpreter")
cookie = GIT.RegisterInterfaceInGlobal(interp._oleobj_, pythoncom.IID_IDispatch)
events = fn(4, cookie)
numFinished = 0
while 1:
try:
rc = win32event.MsgWaitForMultipleObjects(events, 0, 2000, win32event.QS_ALLINPUT)
if rc >= win32event.WAIT_OBJECT_0 and rc < win32event.WAIT_OBJECT_0+len(events):
numFinished = numFinished + 1
if numFinished >= len(events):
break
elif rc==win32event.WAIT_OBJECT_0 + len(events): # a message
# This is critical - whole apartment model demo will hang.
pythoncom.PumpWaitingMessages()
else: # Timeout
print "Waiting for thread to stop with interfaces=%d, gateways=%d" % (pythoncom._GetInterfaceCount(), pythoncom._GetGatewayCount())
except KeyboardInterrupt:
break
GIT.RevokeInterfaceFromGlobal(cookie)
del interp
del GIT
def __call__(self, result = None):
# For the COM suite's sake, always ensure we don't leak
# gateways/interfaces
from pythoncom import _GetInterfaceCount, _GetGatewayCount
gc.collect()
ni = _GetInterfaceCount()
ng = _GetGatewayCount()
self.real_test(result)
# Failed - no point checking anything else
if result.shouldStop or not result.wasSuccessful():
return
self._do_leak_tests(result)
gc.collect()
lost_i = _GetInterfaceCount() - ni
lost_g = _GetGatewayCount() - ng
if lost_i or lost_g:
msg = "%d interface objects and %d gateway objects leaked" \
% (lost_i, lost_g)
exc = AssertionError(msg)
result.addFailure(self.real_test, (exc.__class__, exc, None))
def test(fn):
print("The main thread is %d" % (win32api.GetCurrentThreadId()))
GIT = CreateGIT()
interp = win32com.client.Dispatch("Python.Interpreter")
cookie = GIT.RegisterInterfaceInGlobal(interp._oleobj_, pythoncom.IID_IDispatch)
events = fn(4, cookie)
numFinished = 0
while 1:
try:
rc = win32event.MsgWaitForMultipleObjects(events, 0, 2000, win32event.QS_ALLINPUT)
if rc >= win32event.WAIT_OBJECT_0 and rc < win32event.WAIT_OBJECT_0+len(events):
numFinished = numFinished + 1
if numFinished >= len(events):
break
elif rc==win32event.WAIT_OBJECT_0 + len(events): # a message
# This is critical - whole apartment model demo will hang.
pythoncom.PumpWaitingMessages()
else: # Timeout
print("Waiting for thread to stop with interfaces=%d, gateways=%d" % (pythoncom._GetInterfaceCount(), pythoncom._GetGatewayCount()))
except KeyboardInterrupt:
break
GIT.RevokeInterfaceFromGlobal(cookie)
del interp
del GIT
def _DoTestMarshal(self, fn, bCoWait = 0):
#print "The main thread is %d" % (win32api.GetCurrentThreadId())
threads, events = fn(2)
numFinished = 0
while 1:
try:
if bCoWait:
rc = pythoncom.CoWaitForMultipleHandles(0, 2000, events)
else:
# Specifying "bWaitAll" here will wait for messages *and* all events
# (which is pretty useless)
rc = win32event.MsgWaitForMultipleObjects(events, 0, 2000, win32event.QS_ALLINPUT)
if rc >= win32event.WAIT_OBJECT_0 and rc < win32event.WAIT_OBJECT_0+len(events):
numFinished = numFinished + 1
if numFinished >= len(events):
break
elif rc==win32event.WAIT_OBJECT_0 + len(events): # a message
# This is critical - whole apartment model demo will hang.
pythoncom.PumpWaitingMessages()
else: # Timeout
print "Waiting for thread to stop with interfaces=%d, gateways=%d" % (pythoncom._GetInterfaceCount(), pythoncom._GetGatewayCount())
except KeyboardInterrupt:
break
for t in threads:
t.join(2)
self.failIf(t.isAlive(), "thread failed to stop!?")
threads = None # threads hold references to args
# Seems to be a leak here I can't locate :(
#self.failUnlessEqual(pythoncom._GetInterfaceCount(), 0)
#self.failUnlessEqual(pythoncom._GetGatewayCount(), 0)
def CheckClean():
# Ensure no lingering exceptions - Python should have zero outstanding
# COM objects
try:
sys.exc_clear()
except AttributeError:
pass # py3k
c = _GetInterfaceCount()
if c:
print "Warning - %d com interface objects still alive" % c
c = _GetGatewayCount()
if c:
print "Warning - %d com gateway objects still alive" % c
def _DoTestMarshal(self, fn, bCoWait = 0):
#print "The main thread is %d" % (win32api.GetCurrentThreadId())
threads, events = fn(2)
numFinished = 0
while 1:
try:
if bCoWait:
rc = pythoncom.CoWaitForMultipleHandles(0, 2000, events)
else:
# Specifying "bWaitAll" here will wait for messages *and* all events
# (which is pretty useless)
rc = win32event.MsgWaitForMultipleObjects(events, 0, 2000, win32event.QS_ALLINPUT)
if rc >= win32event.WAIT_OBJECT_0 and rc < win32event.WAIT_OBJECT_0+len(events):
numFinished = numFinished + 1
if numFinished >= len(events):
break
elif rc==win32event.WAIT_OBJECT_0 + len(events): # a message
# This is critical - whole apartment model demo will hang.
pythoncom.PumpWaitingMessages()
else: # Timeout
print "Waiting for thread to stop with interfaces=%d, gateways=%d" % (pythoncom._GetInterfaceCount(), pythoncom._GetGatewayCount())
except KeyboardInterrupt:
break
for t in threads:
t.join(2)
self.failIf(t.isAlive(), "thread failed to stop!?")
threads = None # threads hold references to args
# Seems to be a leak here I can't locate :(
#self.failUnlessEqual(pythoncom._GetInterfaceCount(), 0)
#self.failUnlessEqual(pythoncom._GetGatewayCount(), 0)
def CheckClean():
# Ensure no lingering exceptions - Python should have zero outstanding
# COM objects
try:
sys.exc_clear()
except AttributeError:
pass # py3k
c = _GetInterfaceCount()
if c:
print "Warning - %d com interface objects still alive" % c
c = _GetGatewayCount()
if c:
print "Warning - %d com gateway objects still alive" % c
def testall():
dotestall()
pythoncom.CoUninitialize()
print "AXScript Host worked correctly - %d/%d COM objects left alive." % (pythoncom._GetInterfaceCount(), pythoncom._GetGatewayCount())
def _DoTestMarshal(self, fn, bCoWait = 0):
#print "The main thread is %d" % (win32api.GetCurrentThreadId())
threads, events = fn(2)
numFinished = 0
while 1:
try:
if bCoWait:
rc = pythoncom.CoWaitForMultipleHandles(0, 2000, events)
else:
# Specifying "bWaitAll" here will wait for messages *and* all events
# (which is pretty useless)
rc = win32event.MsgWaitForMultipleObjects(events, 0, 2000, win32event.QS_ALLINPUT)
if rc >= win32event.WAIT_OBJECT_0 and rc < win32event.WAIT_OBJECT_0+len(events):
numFinished = numFinished + 1
if numFinished >= len(events):
break
elif rc==win32event.WAIT_OBJECT_0 + len(events): # a message
# This is critical - whole apartment model demo will hang.
pythoncom.PumpWaitingMessages()
else: # Timeout
print "Waiting for thread to stop with interfaces=%d, gateways=%d" % (pythoncom._GetInterfaceCount(), pythoncom._GetGatewayCount())
except KeyboardInterrupt:
break
for t in threads:
t.join(2)
self.failIf(t.isAlive(), "thread failed to stop!?")
threads = None # threads hold references to args
# Seems to be a leak here I can't locate :(
#self.failUnlessEqual(pythoncom._GetInterfaceCount(), 0)
#self.failUnlessEqual(pythoncom._GetGatewayCount(), 0)
def CheckClean():
# Ensure no lingering exceptions - Python should have zero outstanding
# COM objects
try:
sys.exc_clear()
except AttributeError:
pass # py3k
c = _GetInterfaceCount()
if c:
print "Warning - %d com interface objects still alive" % c
c = _GetGatewayCount()
if c:
print "Warning - %d com gateway objects still alive" % c
def testall():
dotestall()
pythoncom.CoUninitialize()
print("AXScript Host worked correctly - %d/%d COM objects left alive." % (pythoncom._GetInterfaceCount(), pythoncom._GetGatewayCount()))
def _DoTestMarshal(self, fn, bCoWait = 0):
#print "The main thread is %d" % (win32api.GetCurrentThreadId())
threads, events = fn(2)
numFinished = 0
while 1:
try:
if bCoWait:
rc = pythoncom.CoWaitForMultipleHandles(0, 2000, events)
else:
# Specifying "bWaitAll" here will wait for messages *and* all events
# (which is pretty useless)
rc = win32event.MsgWaitForMultipleObjects(events, 0, 2000, win32event.QS_ALLINPUT)
if rc >= win32event.WAIT_OBJECT_0 and rc < win32event.WAIT_OBJECT_0+len(events):
numFinished = numFinished + 1
if numFinished >= len(events):
break
elif rc==win32event.WAIT_OBJECT_0 + len(events): # a message
# This is critical - whole apartment model demo will hang.
pythoncom.PumpWaitingMessages()
else: # Timeout
print("Waiting for thread to stop with interfaces=%d, gateways=%d" % (pythoncom._GetInterfaceCount(), pythoncom._GetGatewayCount()))
except KeyboardInterrupt:
break
for t in threads:
t.join(2)
self.failIf(t.isAlive(), "thread failed to stop!?")
threads = None # threads hold references to args
# Seems to be a leak here I can't locate :(
#self.failUnlessEqual(pythoncom._GetInterfaceCount(), 0)
#self.failUnlessEqual(pythoncom._GetGatewayCount(), 0)
def CheckClean():
# Ensure no lingering exceptions - Python should have zero outstanding
# COM objects
try:
sys.exc_clear()
except AttributeError:
pass # py3k
c = _GetInterfaceCount()
if c:
print("Warning - %d com interface objects still alive" % c)
c = _GetGatewayCount()
if c:
print("Warning - %d com gateway objects still alive" % c)
