← Back to branch summary

~ntt-pf-lab/nova/monkey_patch_notification

« back to all changes in this revision

Viewing changes to vendor/Twisted-10.0.0/twisted/test/test_timeoutqueue.py

  • Committer: Jesse Andrews
  • Date: 2010-05-28 06:05:26 UTC
  • Revision ID: git-v1:bf6e6e718cdc7488e2da87b21e258ccc065fe499
initial commit

Show diffs side-by-side

added added

removed removed

Lines of Context:
vendor/Twisted-10.0.0/twisted/test/test_timeoutqueue.py
 
1
 
 
2
# Copyright (c) 2001-2007 Twisted Matrix Laboratories.
 
3
# See LICENSE for details.
 
4
 
 
5
"""
 
6
Test cases for timeoutqueue module.
 
7
"""
 
8
 
 
9
import time
 
10
 
 
11
from twisted.python import timeoutqueue
 
12
from twisted.trial import unittest, util
 
13
from twisted.internet import reactor, interfaces
 
14
 
 
15
timeoutqueueSuppression = util.suppress(
 
16
    message="timeoutqueue is deprecated since Twisted 8.0",
 
17
    category=DeprecationWarning)
 
18
 
 
19
 
 
20
class TimeoutQueueTest(unittest.TestCase):
 
21
    """
 
22
    Test L{timeoutqueue.TimeoutQueue} class.
 
23
    """
 
24
 
 
25
    def tearDown(self):
 
26
        del self.q
 
27
 
 
28
    def put(self):
 
29
        time.sleep(1)
 
30
        self.q.put(1)
 
31
 
 
32
    def test_timeout(self):
 
33
        q = self.q = timeoutqueue.TimeoutQueue()
 
34
 
 
35
        try:
 
36
            q.wait(1)
 
37
        except timeoutqueue.TimedOut:
 
38
            pass
 
39
        else:
 
40
            self.fail("Didn't time out")
 
41
    test_timeout.suppress = [timeoutqueueSuppression]
 
42
 
 
43
    def test_get(self):
 
44
        q = self.q = timeoutqueue.TimeoutQueue()
 
45
 
 
46
        start = time.time()
 
47
        threading.Thread(target=self.put).start()
 
48
        q.wait(1.5)
 
49
        assert time.time() - start < 2
 
50
 
 
51
        result = q.get(0)
 
52
        if result != 1:
 
53
            self.fail("Didn't get item we put in")
 
54
    test_get.suppress = [timeoutqueueSuppression]
 
55
 
 
56
    def test_deprecation(self):
 
57
        """
 
58
        Test that L{timeoutqueue.TimeoutQueue} prints a warning message.
 
59
        """
 
60
        def createQueue():
 
61
            return timeoutqueue.TimeoutQueue()
 
62
        self.q = self.assertWarns(
 
63
            DeprecationWarning,
 
64
            "timeoutqueue is deprecated since Twisted 8.0",
 
65
            __file__,
 
66
            createQueue)
 
67
 
 
68
    if interfaces.IReactorThreads(reactor, None) is None:
 
69
        test_get.skip = "No thread support, no way to test putting during a blocked get"
 
70
    else:
 
71
        global threading
 
72
        import threading
 
73