← Back to branch summary

~vcs-imports/shtoom/main

« back to all changes in this revision

Viewing changes to shtoom/livetest/test_upnp.py

  • Committer: anthony
  • Date: 2005-03-31 14:37:57 UTC
  • Revision ID: Arch-1:shtoom@bazaar.ubuntu.com%shtoom--trunk--0--patch-981
* ALSA handles stereo-only devices now
* Microphone samples are asynchronously pushed up by the lower layers
rather than being polled by the upper layers. On Mac, the high-precision
realtime thread is used to push microphone samples. This fixes a bug
with short reads and a bug with inaccurate polling. On ALSA, there is a
LoopingCall in the ALSA driver that polls at an appropriate interval.
* The encoder base class has a buffer to store up the appropriate
bytes of microphone data to make up a media frame. This buffer gets
flushed when the audio device closes or reopens.
* The audio device gets closed and reopened during important state
transitions, namely call start and call end. This fixes the "That jerk!"
bug, in which you could say "I have to call that jerk!" immediately
before a call connected and then the jerk in question would hear you say
it when he answered.
* Remove some extremely detailed diags that measured the number of 
packets sent per second and the number of packets received per second.
Those diags have served well and are now retired.
* The Mac audio loopback test is rewritten, and a bug involving
closing the loopback test versus closing a phone call is fixed.
* The discovery/selection of the appropriate audio device is done
before the construction of the Phone object. This makes the other
platforms' initialization process parallel to the Mac initialization
process, and also I prefer this approach. (The other approach is that
you construct the Phone object and then it discovers/selects the audio
device itself.)
* Plays ringing sounds for inbound and outbound ringing sounds
* New audio_device option for selecting a different ALSA or OSS device
* New playout algorithm

Show diffs side-by-side

added added

removed removed

Lines of Context:
shtoom/livetest/test_upnp.py
 
1
# Copyright (C) 2004 Anthony Baxter
 
2
"""Tests for shtoom.upnp
 
3
"""
 
4
 
 
5
# This is more a functional test than a unit test. oh well.
 
6
 
 
7
from twisted.trial import unittest
 
8
from twisted.trial import util
 
9
from twisted.internet import reactor, defer
 
10
from twisted.internet.protocol import Protocol, Factory, DatagramProtocol
 
11
 
 
12
import random
 
13
 
 
14
def checkUPnP():
 
15
    from shtoom.upnp import getUPnP
 
16
    d = getUPnP()
 
17
    s = Saver()
 
18
    d.addCallback(s.save)
 
19
    util.wait(d, timeout=8)
 
20
    if s.val is None:
 
21
        raise unittest.SkipTest('no UPnP available')
 
22
 
 
23
class TestUPnP:
 
24
 
 
25
    def discovered(self, prot):
 
26
        #print "upnp discovery done,", prot
 
27
        self.prot = prot
 
28
        return self.prot.getExternalIPAddress()
 
29
 
 
30
    def gotexternal(self, res):
 
31
        #print "upnp external address,",res
 
32
        return self.prot.getPortMappings()
 
33
 
 
34
    def gotmappings(self, res):
 
35
        #print "upnp mappings:", res
 
36
        return self.prot.addPortMapping(12367,12367,'test mapping')
 
37
 
 
38
    def added(self, res):
 
39
        #print "upnp port mapping added"
 
40
        return self.prot.getPortMappings()
 
41
 
 
42
    def gotmappings2(self, res):
 
43
        #print "upnp mappings:", res
 
44
        return self.prot.deletePortMapping(12367)
 
45
 
 
46
    def deleted(self, res):
 
47
        #print "upnp port mapping removed"
 
48
        return self.prot.getPortMappings()
 
49
 
 
50
    def gotmappings3(self, res):
 
51
        #print "upnp mappings:", res
 
52
        return self.prot.getExternalIPAddress()
 
53
 
 
54
    def gotexternal2(self, res):
 
55
        #print "upnp external address,",res
 
56
        pass
 
57
 
 
58
    def go(self):
 
59
        from shtoom.upnp import getUPnP
 
60
        d = getUPnP()
 
61
        d.addCallback(self.discovered)
 
62
        d.addCallback(self.gotexternal)
 
63
        d.addCallback(self.gotmappings)
 
64
        d.addCallback(self.added)
 
65
        d.addCallback(self.gotmappings2)
 
66
        d.addCallback(self.deleted)
 
67
        d.addCallback(self.gotmappings3)
 
68
        d.addCallback(self.gotexternal2)
 
69
        return d
 
70
 
 
71
class Saver:
 
72
    def __init__(self):
 
73
        self.val = None
 
74
    def save(self, val):
 
75
        self.val = val
 
76
 
 
77
class TestMapper:
 
78
    def __init__(self, mapper, port):
 
79
        self.port = port
 
80
        self.mapper = mapper
 
81
    def go(self):
 
82
        #print "going"
 
83
        cd = defer.Deferred()
 
84
        d = self.mapper.map(self.port)
 
85
        d.addCallback(lambda x: self.cb_mapped(x, cd))
 
86
        return cd
 
87
    def cb_mapped(self, ext, cd):
 
88
        #print "mapped"
 
89
        self.map_res = ext
 
90
        self.info_res = self.mapper.info(self.port)
 
91
        d = self.mapper.unmap(self.port)
 
92
        d.addCallback(lambda x: self.cb_unmapped(x, cd))
 
93
    def cb_unmapped(self, res, cd):
 
94
        #print "unmapped"
 
95
        self.unmap_res = res
 
96
        cd.callback(None)
 
97
 
 
98
class UPnPTest(unittest.TestCase):
 
99
 
 
100
 
 
101
    def test_upnp(self):
 
102
        checkUPnP()
 
103
        test = TestUPnP()
 
104
        d = test.go()
 
105
        util.wait(d, timeout=32)
 
106
 
 
107
    def test_upnp_mapper(self):
 
108
        from shtoom.upnp import UPnPMapper
 
109
        ae = self.assertEquals
 
110
        ar = self.assertRaises
 
111
        checkUPnP()
 
112
        mapper = UPnPMapper()
 
113
        uprot = DatagramProtocol()
 
114
        uport = reactor.listenUDP(random.randint(10000,20000), uprot)
 
115
        class tfactory(Factory):
 
116
            protocol = Protocol
 
117
        tport = reactor.listenTCP(0, tfactory())
 
118
        for port in uport, tport:
 
119
            ar(ValueError, mapper.unmap, port)
 
120
            ar(ValueError, mapper.info, port)
 
121
            t = TestMapper(mapper, port)
 
122
            d = t.go()
 
123
            util.wait(d, timeout=16)
 
124
            ae(len(t.map_res), 2)
 
125
            ae(t.map_res, t.info_res)
 
126
            ae(t.unmap_res, None)
 
127
            # Can't unmap a port that's not mapped
 
128
            ar(ValueError, mapper.unmap, port)
 
129
            d = port.stopListening()
 
130
            util.wait(d)
 
131
            # Can't map a closed port
 
132
            ar(ValueError, mapper.map, port)
 
133
            # Can't get info on a closed port
 
134
            ar(ValueError, mapper.info, port)