1 # Copyright (c) Twisted Matrix Laboratories.
2 # See LICENSE for details.
6 Test cases for twisted.protocols.ident module.
11 from twisted.protocols import ident
12 from twisted.python import failure
13 from twisted.internet import error
14 from twisted.internet import defer
16 from twisted.trial import unittest
17 from twisted.test.proto_helpers import StringTransport
21 class ClassParserTestCase(unittest.TestCase):
23 Test parsing of ident responses.
28 Create a ident client used in tests.
30 self.client = ident.IdentClient()
33 def test_indentError(self):
35 'UNKNOWN-ERROR' error should map to the L{ident.IdentError} exception.
38 self.client.queries.append((d, 123, 456))
39 self.client.lineReceived('123, 456 : ERROR : UNKNOWN-ERROR')
40 return self.assertFailure(d, ident.IdentError)
43 def test_noUSerError(self):
45 'NO-USER' error should map to the L{ident.NoUser} exception.
48 self.client.queries.append((d, 234, 456))
49 self.client.lineReceived('234, 456 : ERROR : NO-USER')
50 return self.assertFailure(d, ident.NoUser)
53 def test_invalidPortError(self):
55 'INVALID-PORT' error should map to the L{ident.InvalidPort} exception.
58 self.client.queries.append((d, 345, 567))
59 self.client.lineReceived('345, 567 : ERROR : INVALID-PORT')
60 return self.assertFailure(d, ident.InvalidPort)
63 def test_hiddenUserError(self):
65 'HIDDEN-USER' error should map to the L{ident.HiddenUser} exception.
68 self.client.queries.append((d, 567, 789))
69 self.client.lineReceived('567, 789 : ERROR : HIDDEN-USER')
70 return self.assertFailure(d, ident.HiddenUser)
73 def test_lostConnection(self):
75 A pending query which failed because of a ConnectionLost should
76 receive an L{ident.IdentError}.
79 self.client.queries.append((d, 765, 432))
80 self.client.connectionLost(failure.Failure(error.ConnectionLost()))
81 return self.assertFailure(d, ident.IdentError)
85 class TestIdentServer(ident.IdentServer):
86 def lookup(self, serverAddress, clientAddress):
87 return self.resultValue
90 class TestErrorIdentServer(ident.IdentServer):
91 def lookup(self, serverAddress, clientAddress):
92 raise self.exceptionType()
95 class NewException(RuntimeError):
99 class ServerParserTestCase(unittest.TestCase):
100 def testErrors(self):
101 p = TestErrorIdentServer()
102 p.makeConnection(StringTransport())
104 p.sendLine = L.append
106 p.exceptionType = ident.IdentError
107 p.lineReceived('123, 345')
108 self.assertEqual(L[0], '123, 345 : ERROR : UNKNOWN-ERROR')
110 p.exceptionType = ident.NoUser
111 p.lineReceived('432, 210')
112 self.assertEqual(L[1], '432, 210 : ERROR : NO-USER')
114 p.exceptionType = ident.InvalidPort
115 p.lineReceived('987, 654')
116 self.assertEqual(L[2], '987, 654 : ERROR : INVALID-PORT')
118 p.exceptionType = ident.HiddenUser
119 p.lineReceived('756, 827')
120 self.assertEqual(L[3], '756, 827 : ERROR : HIDDEN-USER')
122 p.exceptionType = NewException
123 p.lineReceived('987, 789')
124 self.assertEqual(L[4], '987, 789 : ERROR : UNKNOWN-ERROR')
125 errs = self.flushLoggedErrors(NewException)
126 self.assertEqual(len(errs), 1)
128 for port in -1, 0, 65536, 65537:
130 p.lineReceived('%d, 5' % (port,))
131 p.lineReceived('5, %d' % (port,))
133 L, ['%d, 5 : ERROR : INVALID-PORT' % (port,),
134 '5, %d : ERROR : INVALID-PORT' % (port,)])
136 def testSuccess(self):
137 p = TestIdentServer()
138 p.makeConnection(StringTransport())
140 p.sendLine = L.append
142 p.resultValue = ('SYS', 'USER')
143 p.lineReceived('123, 456')
144 self.assertEqual(L[0], '123, 456 : USERID : SYS : USER')
147 if struct.pack('=L', 1)[0] == '\x01':
155 class ProcMixinTestCase(unittest.TestCase):
156 line = ('4: %s:0019 %s:02FA 0A 00000000:00000000 '
157 '00:00000000 00000000 0 0 10927 1 f72a5b80 '
158 '3000 0 0 2 -1') % (_addr1, _addr2)
160 def testDottedQuadFromHexString(self):
161 p = ident.ProcServerMixin()
162 self.assertEqual(p.dottedQuadFromHexString(_addr1), '127.0.0.1')
164 def testUnpackAddress(self):
165 p = ident.ProcServerMixin()
166 self.assertEqual(p.unpackAddress(_addr1 + ':0277'),
169 def testLineParser(self):
170 p = ident.ProcServerMixin()
172 p.parseLine(self.line),
173 (('127.0.0.1', 25), ('1.2.3.4', 762), 0))
175 def testExistingAddress(self):
177 p = ident.ProcServerMixin()
178 p.entries = lambda: iter([self.line])
179 p.getUsername = lambda uid: (username.append(uid), 'root')[1]
181 p.lookup(('127.0.0.1', 25), ('1.2.3.4', 762)),
182 (p.SYSTEM_NAME, 'root'))
183 self.assertEqual(username, [0])
185 def testNonExistingAddress(self):
186 p = ident.ProcServerMixin()
187 p.entries = lambda: iter([self.line])
188 self.assertRaises(ident.NoUser, p.lookup, ('127.0.0.1', 26),
190 self.assertRaises(ident.NoUser, p.lookup, ('127.0.0.1', 25),
192 self.assertRaises(ident.NoUser, p.lookup, ('127.0.0.1', 25),