← Back to branch summary

~justin-fathomdb/nova/justinsb-openstack-api-volumes

« back to all changes in this revision

Viewing changes to vendor/Twisted-10.0.0/twisted/test/test_htb.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_htb.py
 
1
# -*- Python -*-
 
2
 
 
3
__version__ = '$Revision: 1.3 $'[11:-2]
 
4
 
 
5
from twisted.trial import unittest
 
6
from twisted.protocols import htb
 
7
 
 
8
class DummyClock:
 
9
    time = 0
 
10
    def set(self, when):
 
11
        self.time = when
 
12
 
 
13
    def __call__(self):
 
14
        return self.time
 
15
 
 
16
class SomeBucket(htb.Bucket):
 
17
    maxburst = 100
 
18
    rate = 2
 
19
 
 
20
class TestBucketBase(unittest.TestCase):
 
21
    def setUp(self):
 
22
        self._realTimeFunc = htb.time
 
23
        self.clock = DummyClock()
 
24
        htb.time = self.clock
 
25
 
 
26
    def tearDown(self):
 
27
        htb.time = self._realTimeFunc
 
28
 
 
29
class TestBucket(TestBucketBase):
 
30
    def testBucketSize(self):
 
31
        """Testing the size of the bucket."""
 
32
        b = SomeBucket()
 
33
        fit = b.add(1000)
 
34
        self.failUnlessEqual(100, fit)
 
35
 
 
36
    def testBucketDrian(self):
 
37
        """Testing the bucket's drain rate."""
 
38
        b = SomeBucket()
 
39
        fit = b.add(1000)
 
40
        self.clock.set(10)
 
41
        fit = b.add(1000)
 
42
        self.failUnlessEqual(20, fit)
 
43
 
 
44
class TestBucketNesting(TestBucketBase):
 
45
    def setUp(self):
 
46
        TestBucketBase.setUp(self)
 
47
        self.parent = SomeBucket()
 
48
        self.child1 = SomeBucket(self.parent)
 
49
        self.child2 = SomeBucket(self.parent)
 
50
 
 
51
    def testBucketParentSize(self):
 
52
        # Use up most of the parent bucket.
 
53
        self.child1.add(90)
 
54
        fit = self.child2.add(90)
 
55
        self.failUnlessEqual(10, fit)
 
56
 
 
57
    def testBucketParentRate(self):
 
58
        # Make the parent bucket drain slower.
 
59
        self.parent.rate = 1
 
60
        # Fill both child1 and parent.
 
61
        self.child1.add(100)
 
62
        self.clock.set(10)
 
63
        fit = self.child1.add(100)
 
64
        # How much room was there?  The child bucket would have had 20,
 
65
        # but the parent bucket only ten (so no, it wouldn't make too much
 
66
        # sense to have a child bucket draining faster than its parent in a real
 
67
        # application.)
 
68
        self.failUnlessEqual(10, fit)
 
69
 
 
70
 
 
71
# TODO: Test the Transport stuff?
 
72
 
 
73
from test_pcp import DummyConsumer
 
74
 
 
75
class ConsumerShaperTest(TestBucketBase):
 
76
    def setUp(self):
 
77
        TestBucketBase.setUp(self)
 
78
        self.underlying = DummyConsumer()
 
79
        self.bucket = SomeBucket()
 
80
        self.shaped = htb.ShapedConsumer(self.underlying, self.bucket)
 
81
 
 
82
    def testRate(self):
 
83
        # Start off with a full bucket, so the burst-size dosen't factor in
 
84
        # to the calculations.
 
85
        delta_t = 10
 
86
        self.bucket.add(100)
 
87
        self.shaped.write("x" * 100)
 
88
        self.clock.set(delta_t)
 
89
        self.shaped.resumeProducing()
 
90
        self.failUnlessEqual(len(self.underlying.getvalue()),
 
91
                             delta_t * self.bucket.rate)
 
92
 
 
93
    def testBucketRefs(self):
 
94
        self.failUnlessEqual(self.bucket._refcount, 1)
 
95
        self.shaped.stopProducing()
 
96
        self.failUnlessEqual(self.bucket._refcount, 0)