Viewing file:  test_timeoutqueue.py (1.81 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
# Copyright (c) 2001-2007 Twisted Matrix Laboratories.
# See LICENSE for details.
"""
Test cases for timeoutqueue module.
"""
import time
from twisted.python import timeoutqueue
from twisted.trial import unittest, util
from twisted.internet import reactor, interfaces
timeoutqueueSuppression = util.suppress(
message="timeoutqueue is deprecated since Twisted 8.0",
category=DeprecationWarning)
class TimeoutQueueTest(unittest.TestCase):
"""
Test L{timeoutqueue.TimeoutQueue} class.
"""
def tearDown(self):
del self.q
def put(self):
time.sleep(1)
self.q.put(1)
def test_timeout(self):
q = self.q = timeoutqueue.TimeoutQueue()
try:
q.wait(1)
except timeoutqueue.TimedOut:
pass
else:
self.fail("Didn't time out")
test_timeout.suppress = [timeoutqueueSuppression]
def test_get(self):
q = self.q = timeoutqueue.TimeoutQueue()
start = time.time()
threading.Thread(target=self.put).start()
q.wait(1.5)
assert time.time() - start < 2
result = q.get(0)
if result != 1:
self.fail("Didn't get item we put in")
test_get.suppress = [timeoutqueueSuppression]
def test_deprecation(self):
"""
Test that L{timeoutqueue.TimeoutQueue} prints a warning message.
"""
def createQueue():
return timeoutqueue.TimeoutQueue()
self.q = self.assertWarns(
DeprecationWarning,
"timeoutqueue is deprecated since Twisted 8.0",
__file__,
createQueue)
if interfaces.IReactorThreads(reactor, None) is None:
test_get.skip = "No thread support, no way to test putting during a blocked get"
else:
global threading
import threading
|