Package ldaptor :: Package test :: Module test_proxy
[hide private]
[frames] | no frames]

Source Code for Module ldaptor.test.test_proxy

 1  """ 
 2  Test cases for ldaptor.protocols.ldap.proxy module. 
 3  """ 
 4   
 5  from twisted.trial import unittest 
 6  from twisted.internet import reactor, error 
 7  from ldaptor.protocols.ldap import proxy, ldaperrors 
 8  from ldaptor.protocols import pureldap 
 9  from ldaptor import testutil 
10   
11 -class Proxy(unittest.TestCase):
12 - def createServer(self, *responses):
13 return testutil.createServer(proxy.Proxy, *responses)
14
15 - def test_bind(self):
16 server = self.createServer([ pureldap.LDAPBindResponse(resultCode=0), 17 ]) 18 server.dataReceived(str(pureldap.LDAPMessage(pureldap.LDAPBindRequest(), id=4))) 19 reactor.iterate() #TODO 20 self.assertEquals(server.transport.value(), 21 str(pureldap.LDAPMessage(pureldap.LDAPBindResponse(resultCode=0), id=4)))
22
23 - def test_search(self):
24 server = self.createServer([ pureldap.LDAPBindResponse(resultCode=0), 25 ], 26 [ pureldap.LDAPSearchResultEntry('cn=foo,dc=example,dc=com', [('a', ['b'])]), 27 pureldap.LDAPSearchResultEntry('cn=bar,dc=example,dc=com', [('b', ['c'])]), 28 pureldap.LDAPSearchResultDone(ldaperrors.Success.resultCode), 29 ], 30 ) 31 server.dataReceived(str(pureldap.LDAPMessage(pureldap.LDAPBindRequest(), id=2))) 32 server.dataReceived(str(pureldap.LDAPMessage(pureldap.LDAPSearchRequest(), id=3))) 33 reactor.iterate() #TODO 34 self.assertEquals(server.transport.value(), 35 str(pureldap.LDAPMessage(pureldap.LDAPBindResponse(resultCode=0), id=2)) 36 +str(pureldap.LDAPMessage(pureldap.LDAPSearchResultEntry('cn=foo,dc=example,dc=com', [('a', ['b'])]), id=3)) 37 +str(pureldap.LDAPMessage(pureldap.LDAPSearchResultEntry('cn=bar,dc=example,dc=com', [('b', ['c'])]), id=3)) 38 +str(pureldap.LDAPMessage(pureldap.LDAPSearchResultDone(ldaperrors.Success.resultCode), id=3)))
39
41 server = self.createServer([ pureldap.LDAPBindResponse(resultCode=0), 42 ], 43 [], 44 ) 45 server.dataReceived(str(pureldap.LDAPMessage(pureldap.LDAPBindRequest(), id=2))) 46 reactor.iterate() #TODO 47 client = server.client 48 client.assertSent(pureldap.LDAPBindRequest()) 49 self.assertEquals(server.transport.value(), 50 str(pureldap.LDAPMessage(pureldap.LDAPBindResponse(resultCode=0), id=2))) 51 server.dataReceived(str(pureldap.LDAPMessage(pureldap.LDAPUnbindRequest(), id=3))) 52 server.connectionLost(error.ConnectionDone) 53 reactor.iterate() #TODO 54 client.assertSent(pureldap.LDAPBindRequest(), 55 pureldap.LDAPUnbindRequest()) 56 self.assertEquals(server.transport.value(), 57 str(pureldap.LDAPMessage(pureldap.LDAPBindResponse(resultCode=0), id=2)))
58
59 - def test_unbind_clientEOF(self):
60 server = self.createServer([ pureldap.LDAPBindResponse(resultCode=0), 61 ], 62 [], 63 ) 64 server.dataReceived(str(pureldap.LDAPMessage(pureldap.LDAPBindRequest(), id=2))) 65 reactor.iterate() #TODO 66 client = server.client 67 client.assertSent(pureldap.LDAPBindRequest()) 68 self.assertEquals(server.transport.value(), 69 str(pureldap.LDAPMessage(pureldap.LDAPBindResponse(resultCode=0), id=2))) 70 server.connectionLost(error.ConnectionDone) 71 reactor.iterate() #TODO 72 client.assertSent(pureldap.LDAPBindRequest(), 73 'fake-unbind-by-LDAPClientTestDriver') 74 self.assertEquals(server.transport.value(), 75 str(pureldap.LDAPMessage(pureldap.LDAPBindResponse(resultCode=0), id=2)))
76