← Back to branch summary

~certify-web-dev/twisted/certify-trunk

« back to all changes in this revision

Viewing changes to twisted/conch/test/test_mixin.py

  • Committer: Bazaar Package Importer
  • Author(s): Matthias Klose
  • Date: 2006-01-16 19:56:10 UTC
  • mfrom: (1.1.3 upstream)
  • Revision ID: james.westby@ubuntu.com-20060116195610-ykmxbia4mnnod9o2
Tags: 2.1.0-0ubuntu2
debian/copyright: Include copyright for python 2.3; some 2.3 files
are included in the upstream tarball, but not in the binary packages.

Show diffs side-by-side

added added

removed removed

Lines of Context:
twisted/conch/test/test_mixin.py
1
 
# -*- twisted.conch.test.test_mixin -*-
2
 
# Copyright (c) 2001-2004 Twisted Matrix Laboratories.
3
 
# See LICENSE for details.
4
 
 
5
 
import time
6
 
 
7
 
from twisted.internet import reactor, protocol
8
 
 
9
 
from twisted.trial import unittest
10
 
from twisted.test.proto_helpers import StringTransport
11
 
 
12
 
from twisted.conch import mixin
13
 
 
14
 
class BufferingTest(unittest.TestCase):
15
 
    def testBuffering(self):
16
 
        p = mixin.BufferingMixin()
17
 
        t = p.transport = StringTransport()
18
 
 
19
 
        L = ['foo', 'bar', 'baz', 'quux']
20
 
 
21
 
        for s in L:
22
 
            p.write(s)
23
 
            self.assertEquals(t.value(), '')
24
 
 
25
 
        for i in range(10):
26
 
            reactor.iterate(0.01)
27
 
            if t.value():
28
 
                break
29
 
 
30
 
        self.assertEquals(t.value(), ''.join(L))
31
 
 
32
 
class BufferingProtocol(protocol.Protocol, mixin.BufferingMixin):
33
 
    pass
34
 
 
35
 
class UnbufferingProtocol(protocol.Protocol):
36
 
    def connectionMade(self):
37
 
        self.write = self.transport.write
38
 
        self.flush = lambda: None
39
 
 
40
 
class BufferingTiming(unittest.TestCase):
41
 
    def setUp(self):
42
 
        f = protocol.ServerFactory()
43
 
        f.protocol = protocol.Protocol
44
 
        self.server = reactor.listenTCP(0, f)
45
 
 
46
 
        f2 = protocol.ClientCreator(reactor, BufferingProtocol)
47
 
        self.buffered = f2.connectTCP('127.0.0.1', self.server.getHost().port)
48
 
 
49
 
        f3 = protocol.ClientCreator(reactor, UnbufferingProtocol)
50
 
        self.unbuffered = f3.connectTCP('127.0.0.1', self.server.getHost().port)
51
 
 
52
 
    def benchmarkBuffering(self, clock=time.clock, sleep=time.sleep):
53
 
        bufp = unittest.deferredResult(self.buffered)
54
 
        unbufp = unittest.deferredResult(self.unbuffered)
55
 
 
56
 
        one = 'x'
57
 
        ten = one * 10
58
 
        hundred = ten * 10
59
 
        thousand = hundred * 10
60
 
 
61
 
        for p in bufp, unbufp:
62
 
            write = p.write
63
 
            iteration = xrange(100)
64
 
            start = clock()
65
 
 
66
 
            write(one)
67
 
            for i in iteration:
68
 
                write(ten)
69
 
 
70
 
            end = clock()
71
 
            print 'Took', end - start
72