1
# Copyright (c) 2001-2004 Twisted Matrix Laboratories.
2
# See LICENSE for details.
6
Test cases for twisted.internet.app.
9
from twisted.trial import unittest, util
10
from twisted.internet import app, protocol, error
11
from twisted.internet.defer import succeed, fail, SUCCESS, FAILURE
12
from twisted.python import log
15
class AppTestCase(unittest.TestCase):
16
suppress = [util.suppress(message='twisted.internet.app is deprecated',
17
category=DeprecationWarning)]
19
def testListenUnlistenTCP(self):
20
a = app.Application("foo")
21
f = protocol.ServerFactory()
24
self.assertEquals(len(a.tcpPorts), 2)
26
self.assertEquals(len(a.tcpPorts), 1)
27
a.listenTCP(9999, f, interface='127.0.0.1')
28
self.assertEquals(len(a.tcpPorts), 2)
29
a.unlistenTCP(9999, '127.0.0.1')
30
self.assertEquals(len(a.tcpPorts), 1)
32
self.assertEquals(len(a.tcpPorts), 0)
34
def testListenUnlistenUDP(self):
35
a = app.Application("foo")
36
f = protocol.DatagramProtocol()
39
self.assertEquals(len(a.udpPorts), 2)
41
self.assertEquals(len(a.udpPorts), 1)
42
a.listenUDP(9999, f, interface='127.0.0.1')
43
self.assertEquals(len(a.udpPorts), 2)
44
a.unlistenUDP(9999, '127.0.0.1')
45
self.assertEquals(len(a.udpPorts), 1)
47
self.assertEquals(len(a.udpPorts), 0)
49
def testListenUnlistenUNIX(self):
50
a = app.Application("foo")
51
f = protocol.ServerFactory()
52
a.listenUNIX("xxx", f)
53
self.assertEquals(len(a.unixPorts), 1)
55
self.assertEquals(len(a.unixPorts), 0)
57
def testIllegalUnlistens(self):
58
a = app.Application("foo")
60
self.assertRaises(error.NotListeningError, a.unlistenTCP, 1010)
61
self.assertRaises(error.NotListeningError, a.unlistenUNIX, '1010')
62
self.assertRaises(error.NotListeningError, a.unlistenSSL, 1010)
63
self.assertRaises(error.NotListeningError, a.unlistenUDP, 1010)
65
class ServiceTestCase(unittest.TestCase):
67
def testRegisterService(self):
68
a = app.Application("foo")
69
svc = app.ApplicationService("service", a)
70
self.assertEquals(a.getServiceNamed("service"), svc)
71
self.assertEquals(a, svc.serviceParent)
72
testRegisterService.suppress = [util.suppress(message='twisted.internet.app is deprecated',
73
category=DeprecationWarning)]
75
class StopError(Exception): pass
77
class StoppingService(app.ApplicationService):
79
def __init__(self, name, succeed):
80
app.ApplicationService.__init__(self, name)
81
self.succeed = succeed
83
def stopService(self):
85
return succeed("yay!")
87
return fail(StopError('boo'))
89
class StoppingServiceII(app.ApplicationService):
90
def stopService(self):
91
# The default stopService returns None.
92
return None # return app.ApplicationService.stopService(self)
94
class MultiServiceTestCase(unittest.TestCase):
98
def testDeferredStopService(self):
99
ms = app.MultiService("MultiService")
100
self.s1 = StoppingService("testService", 0)
101
self.s2 = StoppingService("testService2", 1)
102
ms.addService(self.s1)
103
ms.addService(self.s2)
104
ms.stopService().addCallback(self.woohoo)
105
log.flushErrors (StopError)
107
def woohoo(self, res):
109
self.assertEqual(res[self.s1][0], 0)
110
self.assertEqual(res[self.s2][0], 1)
112
def testStopServiceNone(self):
113
"""MultiService.stopService returns Deferred when service returns None.
115
ms = app.MultiService("MultiService")
116
self.s1 = StoppingServiceII("testService")
117
ms.addService(self.s1)
119
d.addCallback(self.cb_nonetest)
120
log.flushErrors (StopError)
122
def cb_nonetest(self, res):
124
self.assertEqual((SUCCESS, None), res[self.s1])
126
def testEmptyStopService(self):
127
"""MutliService.stopService returns Deferred when empty."""
128
ms = app.MultiService("MultiService")
130
d.addCallback(self.cb_emptytest)
132
def cb_emptytest(self, res):
134
self.assertEqual(len(res), 0)
137
log.flushErrors (StopError)
138
self.failUnless(self.callbackRan, "Callback was never run.")
added
removed
twisted/test/test_app.py
